module: Optimize RunModuleSplit().

This commit adds a level of indirection to the splits_ thread local
variable: instead of keeping an entire batch for each possible output
gate, we keep a pointer to a batch.

This makes RunModuleSplit() faster, because we avoid copying the whole
batch (we only use a pointer now).

This also fixes Issue #443, because now we're using ~64K of stack space
instead of ~2M.

I tried running `samples/acl` on my laptop before and after the change
and looking at how many packets are generated by source0:

| source0 burst | before (Mpps) | after (Mpps) |
| -------------:| -------------:| ------------:|
|            32 |          76.0 |         83.2 |
|             1 |          14.3 |         14.9 |

Author: ddiproietto

core/module.cc

@@ -482,32 +482,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