Merge pull request #499 from NetSys/modify-taskresult

Modifies task_result to contain block field.

Author: sangjinhan

core/module.h

@@ -15,19 +15,20 @@
 
 using bess::gate_idx_t;
 
+#define INVALID_TASK_ID ((task_id_t)-1)
 #define MAX_NUMA_NODE 16
+#define MAX_TASKS_PER_MODULE 32
 #define UNCONSTRAINED_SOCKET ((0x1ull << MAX_NUMA_NODE) - 1)
 
 struct task_result {
-  uint64_t packets;
+  bool block;
+  uint32_t packets;
   uint64_t bits;
 };
 
 typedef uint16_t task_id_t;
 typedef uint64_t placement_constraint;
 
-#define MAX_TASKS_PER_MODULE 32
-#define INVALID_TASK_ID ((task_id_t)-1)
 
 using module_cmd_func_t =
     pb_func_t<CommandResponse, Module, google::protobuf::Any>;

core/module_bench.cc

@@ -13,7 +13,7 @@ class DummySourceModule : public Module {
 };
 
 [[gnu::noinline]] struct task_result DummySourceModule::RunTask(void *arg) {
-  const size_t batch_size = reinterpret_cast<size_t>(arg);
+  const uint32_t batch_size = reinterpret_cast<size_t>(arg);
   bess::PacketBatch batch;
 
   bess::Packet pkts[bess::PacketBatch::kMaxBurst];
@@ -33,7 +33,7 @@ class DummySourceModule : public Module {
 
   RunNextModule(&batch);
 
-  return {.packets = static_cast<uint64_t>(batch_size), .bits = 0};
+  return { .block = false, .packets = batch_size, .bits = 0 };
 }
 
 class DummyRelayModule : public Module {

core/modules/drr.cc

@@ -116,7 +116,6 @@ void DRR::ProcessBatch(bess::PacketBatch* batch) {
 
 struct task_result DRR::RunTask(void*) {
   bess::PacketBatch batch;
-  struct task_result ret;
   int err = 0;
   batch.clear();
   uint32_t total_bytes = 0;
@@ -130,11 +129,9 @@ struct task_result DRR::RunTask(void*) {
   }
 
   // the number of bits inserted into the packet batch
-  uint64_t cnt = batch.cnt();
+  uint32_t cnt = batch.cnt();
   uint64_t bits_retrieved = (total_bytes + cnt * kPacketOverhead) * 8;
-  ret = (struct task_result){.packets = cnt, .bits = bits_retrieved};
-
-  return ret;
+  return { .block = (cnt == 0), .packets = cnt, .bits = bits_retrieved };
 }
 
 uint32_t DRR::GetNextBatch(bess::PacketBatch* batch, int* err) {

core/modules/flowgen.cc

@@ -479,21 +479,18 @@ void FlowGen::GeneratePackets(bess::PacketBatch *batch) {
 }
 
 struct task_result FlowGen::RunTask(void *) {
-  bess::PacketBatch batch;
-  struct task_result ret;
-
   const int pkt_overhead = 24;
+  bess::PacketBatch batch;
 
   GeneratePackets(&batch);
   RunNextModule(&batch);
 
-  ret = (struct task_result){
-      .packets = static_cast<uint64_t>(batch.cnt()),
-      .bits = static_cast<uint64_t>(
-          ((template_size_ + pkt_overhead) * batch.cnt()) * 8),
+  uint32_t cnt = batch.cnt();
+  return {
+    .block = (cnt == 0),
+    .packets = cnt,
+    .bits = ((template_size_ + pkt_overhead) * cnt) * 8
   };
-
-  return ret;
 }
 
 std::string FlowGen::GetDesc() const {

core/modules/noop.cc

@@ -11,9 +11,7 @@ CommandResponse NoOP::Init(const bess::pb::EmptyArg &) {
 }
 
 struct task_result NoOP::RunTask(void *) {
-  return {
-      .packets = 0, .bits = 0,
-  };
+  return { .block = false, .packets = 0, .bits = 0 };
 }
 
 ADD_MODULE(NoOP, "noop", "creates a task that does nothing")

core/modules/port_inc.cc

@@ -75,46 +75,41 @@ struct task_result PortInc::RunTask(void *arg) {
   const queue_t qid = (queue_t)(uintptr_t)arg;
 
   bess::PacketBatch batch;
-  struct task_result ret;
-
   uint64_t received_bytes = 0;
 
   const int burst = ACCESS_ONCE(burst_);
   const int pkt_overhead = 24;
 
-  uint64_t cnt;
   batch.set_cnt(p->RecvPackets(qid, batch.pkts(), burst));
-  cnt = batch.cnt();
-
+  uint32_t cnt = batch.cnt();
   if (cnt == 0) {
-    ret.packets = 0;
-    ret.bits = 0;
-    return ret;
+    return { .block = true, .packets = 0, .bits = 0 };
   }
 
-  /* NOTE: we cannot skip this step since it might be used by scheduler */
+  // NOTE: we cannot skip this step since it might be used by scheduler.
   if (prefetch_) {
-    for (uint64_t i = 0; i < cnt; i++) {
+    for (uint32_t i = 0; i < cnt; i++) {
       received_bytes += batch.pkts()[i]->total_len();
       rte_prefetch0(batch.pkts()[i]->head_data());
     }
   } else {
-    for (uint64_t i = 0; i < cnt; i++)
+    for (uint32_t i = 0; i < cnt; i++) {
       received_bytes += batch.pkts()[i]->total_len();
+    }
   }
 
-  ret = (struct task_result){
-      .packets = cnt, .bits = (received_bytes + cnt * pkt_overhead) * 8,
-  };
-
   if (!(p->GetFlags() & DRIVER_FLAG_SELF_INC_STATS)) {
     p->queue_stats[PACKET_DIR_INC][qid].packets += cnt;
     p->queue_stats[PACKET_DIR_INC][qid].bytes += received_bytes;
   }
 
   RunNextModule(&batch);
 
-  return ret;
+  return {
+    .block = false,
+    .packets = cnt,
+    .bits = (received_bytes + cnt * pkt_overhead) * 8
+  };
 }
 
 CommandResponse PortInc::CommandSetBurst(

core/modules/queue.cc

@@ -110,35 +110,37 @@ void Queue::ProcessBatch(bess::PacketBatch *batch) {
 /* to downstream */
 struct task_result Queue::RunTask(void *) {
   bess::PacketBatch batch;
-  struct task_result ret;
 
   const int burst = ACCESS_ONCE(burst_);
   const int pkt_overhead = 24;
 
   uint64_t total_bytes = 0;
 
-  uint64_t cnt = llring_sc_dequeue_burst(queue_, (void **)batch.pkts(), burst);
+  uint32_t cnt = llring_sc_dequeue_burst(queue_, (void **)batch.pkts(), burst);
 
-  if (cnt > 0) {
-    batch.set_cnt(cnt);
-    RunNextModule(&batch);
+  if (cnt == 0) {
+    return { .block = true, .packets = 0, .bits = 0 };
   }
 
+  batch.set_cnt(cnt);
+  RunNextModule(&batch);
+
   if (prefetch_) {
-    for (uint64_t i = 0; i < cnt; i++) {
+    for (uint32_t i = 0; i < cnt; i++) {
       total_bytes += batch.pkts()[i]->total_len();
       rte_prefetch0(batch.pkts()[i]->head_data());
     }
   } else {
-    for (uint64_t i = 0; i < cnt; i++)
+    for (uint32_t i = 0; i < cnt; i++) {
       total_bytes += batch.pkts()[i]->total_len();
+    }
   }
 
-  ret = (struct task_result){
-      .packets = cnt, .bits = (total_bytes + cnt * pkt_overhead) * 8,
+  return {
+    .block = false,
+    .packets = cnt,
+    .bits = (total_bytes + cnt * pkt_overhead) * 8
   };
-
-  return ret;
 }
 
 CommandResponse Queue::CommandSetBurst(

core/modules/queue_inc.cc

@@ -60,47 +60,43 @@ struct task_result QueueInc::RunTask(void *arg) {
   const queue_t qid = (queue_t)(uintptr_t)arg;
 
   bess::PacketBatch batch;
-  struct task_result ret;
 
   uint64_t received_bytes = 0;
 
   const int burst = ACCESS_ONCE(burst_);
   const int pkt_overhead = 24;
 
-  uint64_t cnt;
   batch.set_cnt(p->RecvPackets(qid, batch.pkts(), burst));
-  cnt = batch.cnt();
+  uint32_t cnt = batch.cnt();
 
   if (cnt == 0) {
-    ret.packets = 0;
-    ret.bits = 0;
-    return ret;
+    return { .block = true, .packets = 0, .bits = 0 };
   }
 
-  /* NOTE: we cannot skip this step since it might be used by scheduler */
+  // NOTE: we cannot skip this step since it might be used by scheduler.
   if (prefetch_) {
-    for (uint64_t i = 0; i < cnt; i++) {
+    for (uint32_t i = 0; i < cnt; i++) {
       received_bytes += batch.pkts()[i]->total_len();
       rte_prefetch0(batch.pkts()[i]->head_data());
     }
   } else {
-    for (uint64_t i = 0; i < cnt; i++) {
+    for (uint32_t i = 0; i < cnt; i++) {
       received_bytes += batch.pkts()[i]->total_len();
     }
   }
 
-  ret = (struct task_result){
-      .packets = cnt, .bits = (received_bytes + cnt * pkt_overhead) * 8,
-  };
-
   if (!(p->GetFlags() & DRIVER_FLAG_SELF_INC_STATS)) {
     p->queue_stats[PACKET_DIR_INC][qid].packets += cnt;
     p->queue_stats[PACKET_DIR_INC][qid].bytes += received_bytes;
   }
 
   RunNextModule(&batch);
 
-  return ret;
+  return {
+    .block = false,
+    .packets = cnt,
+    .bits = (received_bytes + cnt * pkt_overhead) * 8
+  };
 }
 
 CommandResponse QueueInc::CommandSetBurst(

core/modules/source.cc

@@ -58,14 +58,14 @@ struct task_result Source::RunTask(void *) {
   const int pkt_size = ACCESS_ONCE(pkt_size_);
   const int burst = ACCESS_ONCE(burst_);
 
-  int cnt = bess::Packet::Alloc(batch.pkts(), burst, pkt_size);
-
+  uint32_t cnt = bess::Packet::Alloc(batch.pkts(), burst, pkt_size);
   batch.set_cnt(cnt);
   RunNextModule(&batch);  // it's fine to call this function with cnt==0
 
-  return (struct task_result){
-      .packets = static_cast<uint64_t>(cnt),
-      .bits = static_cast<uint64_t>(pkt_size + pkt_overhead) * cnt * 8,
+  return {
+    .block = (cnt == 0),
+    .packets = cnt,
+    .bits = (pkt_size + pkt_overhead) * cnt * 8
   };
 }
 

core/scheduler.h

@@ -320,12 +320,9 @@ class ExperimentalScheduler : public Scheduler<CallableTask> {
       auto ret = leaf->Task()();
       now = rdtsc();
 
-      if (ret.packets == 0 && ret.bits == 0) {
+      if (ret.packets == 0 && ret.block) {
         constexpr uint64_t kMaxWait = 1ull << 32;
-        uint64_t wait = leaf->wait_cycles() << 1;
-        if (wait > kMaxWait) {
-          wait = kMaxWait;
-        }
+        uint64_t wait = std::min(kMaxWait, leaf->wait_cycles() << 1);
         leaf->set_wait_cycles(wait);
 
         leaf->blocked_ = true;