Merge pull request #510 from ddiproietto/improve_split_v2

Optimize RunModuleSplit() and reduce TLS usage

Author: sangjinhan

core/module.cc

@@ -592,32 +592,33 @@ void Module::RunSplit(const gate_idx_t *out_gates,
   gate_idx_t pending[bess::PacketBatch::kMaxBurst];
   bess::PacketBatch batches[bess::PacketBatch::kMaxBurst];
 
-  bess::PacketBatch *splits = ctx.splits();
+  bess::PacketBatch **splits = ctx.splits();
 
-  /* phase 1: collect unique ogates into pending[] */
+  // phase 1: collect unique ogates into pending[] and add packets to local
+  // batches, using splits to remember the association between an ogate and a
+  // local batch
   for (int i = 0; i < cnt; i++) {
     bess::PacketBatch *batch;
     gate_idx_t ogate;
 
     ogate = out_gates[i];
-    batch = &splits[ogate];
+    batch = splits[ogate];
+    if (!batch) {
+      batch = splits[ogate] = &batches[num_pending];
+      batch->clear();
+      pending[num_pending] = ogate;
+      num_pending++;
+    }
 
     batch->add(*(p_pkt++));
-
-    pending[num_pending] = ogate;
-    num_pending += (batch->cnt() == 1);
   }
 
-  /* phase 2: move batches to local stack, since it may be reentrant */
+  // phase 2: clear splits, since it may be reentrant.
   for (int i = 0; i < num_pending; i++) {
-    bess::PacketBatch *batch;
-
-    batch = &splits[pending[i]];
-    batches[i].Copy(batch);
-    batch->clear();
+    splits[pending[i]] = nullptr;
   }
 
-  /* phase 3: fire */
+  // phase 3: fire
   for (int i = 0; i < num_pending; i++)
     RunChooseModule(pending[i], &batches[i]);
 }

core/worker.h

@@ -91,8 +91,7 @@ class Worker {
   gate_idx_t current_igate() const { return current_igate_; }
   void set_current_igate(gate_idx_t idx) { current_igate_ = idx; }
 
-  /* better be the last field. it's huge */
-  bess::PacketBatch *splits() { return splits_; }
+  bess::PacketBatch **splits() { return splits_; }
 
  private:
   volatile worker_status_t status_;
@@ -115,8 +114,11 @@ class Worker {
    * Modules should use get_igate() for access */
   gate_idx_t current_igate_;
 
-  /* better be the last field. it's huge */
-  bess::PacketBatch splits_[MAX_GATES + 1];
+  // For each possible output gate contains a pointer to a batch, or nullptr,
+  // if no batch has been associated with the output gate yet.
+  //
+  // This should be the last field, since it's huge.
+  bess::PacketBatch *splits_[MAX_GATES + 1];
 };
 
 // NOTE: Do not use "thread_local" here. It requires a function call every time