| 85 | } |
| 86 | |
| 87 | struct task_result QueueInc::RunTask(Context *ctx, bess::PacketBatch *batch, |
| 88 | void *arg) { |
| 89 | Port *p = port_; |
| 90 | |
| 91 | if (!p->conf().admin_up) { |
| 92 | return {.block = true, .packets = 0, .bits = 0}; |
| 93 | } |
| 94 | |
| 95 | const queue_t qid = (queue_t)(uintptr_t)arg; |
| 96 | |
| 97 | uint64_t received_bytes = 0; |
| 98 | |
| 99 | const int burst = ACCESS_ONCE(burst_); |
| 100 | const int pkt_overhead = 24; |
| 101 | |
| 102 | batch->set_cnt(p->RecvPackets(qid, batch->pkts(), burst)); |
| 103 | uint32_t cnt = batch->cnt(); |
| 104 | p->queue_stats[PACKET_DIR_INC][qid].requested_hist[burst]++; |
| 105 | p->queue_stats[PACKET_DIR_INC][qid].actual_hist[cnt]++; |
| 106 | p->queue_stats[PACKET_DIR_INC][qid].diff_hist[burst - cnt]++; |
| 107 | if (cnt == 0) { |
| 108 | return {.block = true, .packets = 0, .bits = 0}; |
| 109 | } |
| 110 | |
| 111 | // NOTE: we cannot skip this step since it might be used by scheduler. |
| 112 | if (prefetch_) { |
| 113 | for (uint32_t i = 0; i < cnt; i++) { |
| 114 | received_bytes += batch->pkts()[i]->total_len(); |
| 115 | rte_prefetch0(batch->pkts()[i]->head_data()); |
| 116 | } |
| 117 | } else { |
| 118 | for (uint32_t i = 0; i < cnt; i++) { |
| 119 | received_bytes += batch->pkts()[i]->total_len(); |
| 120 | } |
| 121 | } |
| 122 | |
| 123 | if (!(p->GetFlags() & DRIVER_FLAG_SELF_INC_STATS)) { |
| 124 | p->queue_stats[PACKET_DIR_INC][qid].packets += cnt; |
| 125 | p->queue_stats[PACKET_DIR_INC][qid].bytes += received_bytes; |
| 126 | } |
| 127 | |
| 128 | RunNextModule(ctx, batch); |
| 129 | |
| 130 | return {.block = false, |
| 131 | .packets = cnt, |
| 132 | .bits = (received_bytes + cnt * pkt_overhead) * 8}; |
| 133 | } |
| 134 | |
| 135 | CommandResponse QueueInc::CommandSetBurst( |
| 136 | const bess::pb::QueueIncCommandSetBurstArg &arg) { |