pping: Eliminate flow creation/deletion concurrency issue

Only push flow events for opening/closing flows if the
creation/deletion of the flow-state was successful (as indicated by
the bpf_map_*_elem() return value). This should avoid outputting
several flow creation/deletion messages in case multiple instances are
trying to create/delete a flow concurrently, as could theoretically
occur previously.

Also set the last_timestamp value before creating a new flow, to avoid
a race condition where the userspace cleanup might incorrectly
determine that a flow is old before the last_timestamp value can be
set. Explicitly skip the rate-limit for the first packet of a new flow
to avoid it failing the rate-limit. This also fixes an issue where the
first packet of a new flow would previously fail the rate-limit if the
rate-limit was higher than current time uptime (CLOCK_MONOTONIC).

Signed-off-by: Simon Sundberg <simon.sundberg@kau.se>

Author: simosund

pping/pping_kern.c

@@ -350,6 +350,7 @@ static void pping_egress(void *ctx, struct parsing_context *pctx)
 	struct flow_event fe;
 	struct flow_state *f_state;
 	struct flow_state new_state = { 0 };
+	bool new_flow = false;
 	__u64 now;
 
 	if (parse_packet_identifier(pctx, &p_id, &fe.event_info) < 0)
@@ -372,20 +373,25 @@ static void pping_egress(void *ctx, struct parsing_context *pctx)
 
 	// No previous state - attempt to create it and push flow-opening event
 	if (!f_state) {
-		bpf_map_update_elem(&flow_state, &p_id.flow, &new_state,
-				    BPF_NOEXIST);
-		f_state = bpf_map_lookup_elem(&flow_state, &p_id.flow);
+		new_state.last_timestamp = now;
+		if (bpf_map_update_elem(&flow_state, &p_id.flow, &new_state,
+					BPF_NOEXIST) == 0) {
+			new_flow = true;
+
+			if (fe.event_info.event != FLOW_EVENT_OPENING) {
+				fe.event_info.event = FLOW_EVENT_OPENING;
+				fe.event_info.reason =
+					EVENT_REASON_FIRST_OBS_PCKT;
+			}
+			fill_flow_event(&fe, now, &p_id.flow,
+					EVENT_SOURCE_EGRESS);
+			bpf_perf_event_output(ctx, &events, BPF_F_CURRENT_CPU,
+					      &fe, sizeof(fe));
+		}
 
+		f_state = bpf_map_lookup_elem(&flow_state, &p_id.flow);
 		if (!f_state) // Creation failed
 			return;
-
-		if (fe.event_info.event != FLOW_EVENT_OPENING) {
-			fe.event_info.event = FLOW_EVENT_OPENING;
-			fe.event_info.reason = EVENT_REASON_FIRST_OBS_PCKT;
-		}
-		fill_flow_event(&fe, now, &p_id.flow, EVENT_SOURCE_EGRESS);
-		bpf_perf_event_output(ctx, &events, BPF_F_CURRENT_CPU, &fe,
-				      sizeof(fe));
 	}
 
 	f_state->sent_pkts++;
@@ -397,8 +403,8 @@ static void pping_egress(void *ctx, struct parsing_context *pctx)
 	f_state->last_id = p_id.identifier;
 
 	// Check rate-limit
-	if (now < f_state->last_timestamp ||
-	    now - f_state->last_timestamp < config.rate_limit)
+	if (!new_flow && (now < f_state->last_timestamp ||
+			  now - f_state->last_timestamp < config.rate_limit))
 		return;
 
 	/*
@@ -463,8 +469,8 @@ static void pping_ingress(void *ctx, struct parsing_context *pctx)
 
 validflow_out:
 	// Wait with deleting flow until having pushed final RTT message
-	if (fe.event_info.event == FLOW_EVENT_CLOSING && f_state) {
-		bpf_map_delete_elem(&flow_state, &p_id.flow);
+	if (fe.event_info.event == FLOW_EVENT_CLOSING &&
+	    bpf_map_delete_elem(&flow_state, &p_id.flow) == 0) {
 		fill_flow_event(&fe, now, &p_id.flow, EVENT_SOURCE_INGRESS);
 		bpf_perf_event_output(ctx, &events, BPF_F_CURRENT_CPU, &fe,
 				      sizeof(fe));