Add support for detection of misordered packets per flow

Only flows from ranges are supported so far. This means that
flows generated through randoms are not supported yet.

Misordered packets can now be detected per flow.
flow definition is hard-coded as 5 tuple (proto, ip/udp src/dst).
The generator writes flow_id and packet_id within the packets (4 + 5 bytes).
This means that, if all latency, packet_id, signature, accuracy, flow_id and packet_id_in_flow
parameters are set, the minimum packet size is 72 bytes.

This feature is enabled through the following parameters:
On the gen side:
- packet id in flow pos: position (5 bytes) for packet id
- flow id pos: flow id pos (4 bytes)

On the lat side
- flow_count: the number of flows received by the latency core. If not enough
  flows are configured, then an error message is displayed (once) and packet_reordering
  count is only measured for those flows
- packet id in flow pos: position (5 bytes) for packet id
- flow id pos: flow id pos (4 bytes)

Misordered packets (and extent) are reported per core as before.
No change related to duplicate packets (not measured per flow as a duplicate packet is not per flow...).

Signed-off-by: Xavier Simonart <simonartxavier@gmail.com>
Change-Id: I07517df87dfebec81408baf1decb647d9a0edd94

1 files changed, 53 insertions, 5 deletions

diff --git a/VNFs/DPPD-PROX/handle_lat.c b/VNFs/DPPD-PROX/handle_lat.c
index 550f3f55..d6943070 100644
--- a/VNFs/DPPD-PROX/handle_lat.c
+++ b/VNFs/DPPD-PROX/handle_lat.c
@@ -91,6 +91,10 @@ struct loss_buffer {
 	uint32_t n;
 };
 
+struct flows {
+	uint32_t packet_id;
+};
+
 struct task_lat {
 	struct task_base base;
 	uint64_t limit;
@@ -125,6 +129,10 @@ struct task_lat {
 	uint32_t loss_buffer_size;
 	struct loss_buffer *loss_buffer;
 	uint32_t loss_id;
+	uint32_t packet_id_in_flow_pos;
+	int32_t flow_id_pos;
+	uint32_t flow_count;
+	struct flows *flows;
 };
 /* This function calculate the difference between rx and tx_time
  * Both values are uint32_t (see handle_lat_bulk)
@@ -490,6 +498,15 @@ static void lat_test_histogram_add(struct lat_test *lat_test, uint64_t lat_tsc)
 	lat_test->buckets[bucket_id]++;
 }
 
+static void lat_test_check_flow_ordering(struct task_lat *task, struct lat_test *lat_test, int32_t flow_id, uint32_t packet_id)
+{
+	if (packet_id < task->flows[flow_id].packet_id) {
+		lat_test->mis_ordered++;
+		lat_test->extent += task->flows[flow_id].packet_id - packet_id;
+	}
+	task->flows[flow_id].packet_id = packet_id;
+}
+
 static void lat_test_check_ordering(struct task_lat *task, struct lat_test *lat_test, uint32_t packet_id, uint8_t generator_id)
 {
 	if (packet_id < task->previous_packet[generator_id]) {
@@ -551,6 +568,7 @@ static void task_lat_store_lat(struct task_lat *task, uint64_t rx_packet_index,
 static int handle_lat_bulk(struct task_base *tbase, struct rte_mbuf **mbufs, uint16_t n_pkts)
 {
 	struct task_lat *task = (struct task_lat *)tbase;
+	static int max_flows_printed = 0;
 	int rc;
 
 	if (n_pkts == 0) {
@@ -639,6 +657,24 @@ static int handle_lat_bulk(struct task_base *tbase, struct rte_mbuf **mbufs, uin
 
 		uint8_t generator_id;
 		uint32_t packet_id;
+		int32_t flow_id = -1;
+		if (task->flow_id_pos) {
+			flow_id = *(int32_t *)(hdr + task->flow_id_pos);
+			if (unlikely(flow_id >= (int32_t)(task->flow_count))) {
+				flow_id = -1;
+				if (!max_flows_printed) {
+					plog_info("Too many flows - increase flow count (only printed once)\n");
+					max_flows_printed = 1;
+				}
+			}
+
+		}
+		if (task->packet_id_in_flow_pos && (flow_id != -1)) {
+			uint32_t packet_id_in_flow;
+			struct unique_id *unique_id = (struct unique_id *)(hdr + task->packet_id_in_flow_pos);
+			unique_id_get(unique_id, &generator_id, &packet_id_in_flow);
+			lat_test_check_flow_ordering(task, task->lat_test, flow_id + generator_id * task->generator_count, packet_id_in_flow);
+		}
 		if (task->unique_id_pos) {
 			struct unique_id *unique_id = (struct unique_id *)(hdr + task->unique_id_pos);
 			unique_id_get(unique_id, &generator_id, &packet_id);
@@ -650,7 +686,9 @@ static int handle_lat_bulk(struct task_base *tbase, struct rte_mbuf **mbufs, uin
 				// Skip unexpected packet
 				continue;
 			}
-			lat_test_check_ordering(task, task->lat_test, packet_id, generator_id);
+			if (flow_id == -1) {
+				lat_test_check_ordering(task, task->lat_test, packet_id, generator_id);
+			}
 			lat_test_check_duplicate(task, task->lat_test, packet_id, generator_id);
 			uint32_t loss =  task_lat_early_loss_detect(task, packet_id, generator_id);
 			if (loss) {
@@ -779,6 +817,8 @@ static void init_task_lat(struct task_base *tbase, struct task_args *targ)
 	task->accur_pos = targ->accur_pos;
 	task->sig_pos = targ->sig_pos;
 	task->sig = targ->sig;
+	task->packet_id_in_flow_pos = targ->packet_id_in_flow_pos;
+	task->flow_id_pos = targ->flow_id_pos;
 
 	task->unique_id_pos = targ->packet_id_pos;
 	task->latency_buffer_size = targ->latency_buffer_size;
@@ -860,10 +900,11 @@ static void init_task_lat(struct task_base *tbase, struct task_args *targ)
 		}
 	}
 	task->loss_buffer_size = targ->loss_buffer_size;
-	task->loss_buffer = prox_zmalloc(task->loss_buffer_size * sizeof(struct loss_buffer), rte_lcore_to_socket_id(targ->lconf->id));
-	PROX_PANIC(task->loss_buffer == NULL,
-		"Failed to allocate %lu bytes (in huge pages) for loss_buffer\n", task->loss_buffer_size * sizeof(struct loss_buffer));
-
+	if (task->loss_buffer_size) {
+		task->loss_buffer = prox_zmalloc(task->loss_buffer_size * sizeof(struct loss_buffer), rte_lcore_to_socket_id(targ->lconf->id));
+		PROX_PANIC(task->loss_buffer == NULL,
+			"Failed to allocate %lu bytes (in huge pages) for loss_buffer\n", task->loss_buffer_size * sizeof(struct loss_buffer));
+	}
 	task->bytes_to_tsc = prox_zmalloc(max_frame_size * sizeof(task->bytes_to_tsc[0]) * MAX_PKT_BURST, rte_lcore_to_socket_id(targ->lconf->id));
 	PROX_PANIC(task->bytes_to_tsc == NULL,
 		"Failed to allocate %lu bytes (in huge pages) for bytes_to_tsc\n", max_frame_size * sizeof(task->bytes_to_tsc[0]) * MAX_PKT_BURST);
@@ -877,6 +918,13 @@ static void init_task_lat(struct task_base *tbase, struct task_args *targ)
 		else
 			task->bytes_to_tsc[i] = (rte_get_tsc_hz() * i * 0.99) / bytes_per_hz;
 	}
+	task->flow_count = targ->flow_count;
+	PROX_PANIC(task->flow_id_pos && (task->flow_count == 0), "flow_count must be configured when flow_id_pos is set\n");
+	if (task->flow_count) {
+		task->flows = prox_zmalloc(task->flow_count * sizeof(struct flows) * task->generator_count, rte_lcore_to_socket_id(targ->lconf->id));
+		PROX_PANIC(task->flows == NULL,
+			"Failed to allocate %lu bytes (in huge pages) for flows\n", task->flow_count * sizeof(struct flows) * task->generator_count);
+	}
 }
 
 static struct task_init task_init_lat = {