pping: WIP - add RTT-based sampling

To allow the rtt-rate to be passed as a float (so you can do for ex 1
RTT-sample every 0.1 RTT periods), I've used fixed-point arithmetic as
floats are not supported in BPF (or the kernel in general). I've never
tried to use fixed-point arithmetic before, so hopefully I haven't
screwed it up too badly.

Have limited the RTT-based sampling to a maximum of 10k*RTT in order
for overflow in fixed-point multiplication to be unlikely.

The RTT-interval is based on a smoothed RTT (moving average)
calculated the same way as for the TCP stack (srtt = 7/8*prev_rtt +
1/8*rtt).

Signed-off-by: Simon Sundberg <[email protected]>

Author: simosund

pping/pping.c

@@ -93,6 +93,7 @@ static const struct option long_options[] = {
 	{ "help",             no_argument,       NULL, 'h' },
 	{ "interface",        required_argument, NULL, 'i' }, // Name of interface to run on
 	{ "rate-limit",       required_argument, NULL, 'r' }, // Sampling rate-limit in ms
+	{ "rtt-rate",         required_argument, NULL, 'R' }, // Sampling rate in terms of flow-RTT (ex 1 sample per RTT-interval)
 	{ "force",            no_argument,       NULL, 'f' }, // Overwrite any existing XDP program on interface, remove qdisc on cleanup
 	{ "cleanup-interval", required_argument, NULL, 'c' }, // Map cleaning interval in s
 	{ "format",           required_argument, NULL, 'F' }, // Which format to output in (standard/json/ppviz)
@@ -158,14 +159,14 @@ static int parse_bounded_double(double *res, const char *str, double low,
 static int parse_arguments(int argc, char *argv[], struct pping_config *config)
 {
 	int err, opt;
-	double rate_limit_ms, cleanup_interval_s;
+	double rate_limit_ms, cleanup_interval_s, rtt_rate;
 
 	config->ifindex = 0;
 	config->force = false;
 	config->json_format = false;
 	config->ppviz_format = false;
 
-	while ((opt = getopt_long(argc, argv, "hfi:r:c:F:I:", long_options,
+	while ((opt = getopt_long(argc, argv, "hfi:r:R:c:F:I:", long_options,
 				  NULL)) != -1) {
 		switch (opt) {
 		case 'i':
@@ -193,6 +194,14 @@ static int parse_arguments(int argc, char *argv[], struct pping_config *config)
 			config->bpf_config.rate_limit =
 				rate_limit_ms * NS_PER_MS;
 			break;
+		case 'R':
+			err = parse_bounded_double(&rtt_rate, optarg, 0, 10000,
+						   "rate-limit");
+			if (err)
+				return -EINVAL;
+			config->bpf_config.rtt_rate =
+				DOUBLE_TO_FIXPOINT(rtt_rate);
+
 		case 'c':
 			err = parse_bounded_double(&cleanup_interval_s, optarg,
 						   0, 1000000000,
@@ -443,6 +452,8 @@ static bool flow_timeout(void *key_ptr, void *val_ptr, __u64 now)
 	struct flow_event fe;
 	__u64 ts = ((struct flow_state *)val_ptr)->last_timestamp;
 
+	printf(">>> now: %llu, ts: %llu, diff: %llu\n", now, ts, now - ts);
+
 	if (now > ts && now - ts > FLOW_LIFETIME) {
 		if (print_event_func) {
 			fe.event_type = EVENT_TYPE_FLOW;
@@ -917,7 +928,7 @@ int main(int argc, char *argv[])
 	DECLARE_LIBBPF_OPTS(bpf_tc_opts, tc_egress_opts);
 
 	struct pping_config config = {
-		.bpf_config = { .rate_limit = 100 * NS_PER_MS },
+		.bpf_config = { .rate_limit = 100 * NS_PER_MS, .rtt_rate = 0 },
 		.cleanup_interval = 1 * NS_PER_SECOND,
 		.object_path = "pping_kern.o",
 		.ingress_sec = SEC_INGRESS_XDP,

pping/pping.h

@@ -10,6 +10,11 @@
 #define SEC_INGRESS_TC "classifier/ingress"
 #define SEC_EGRESS_TC "classifier/egress"
 
+typedef __u64 fixpoint64;
+#define FIXPOINT_SHIFT 16
+#define DOUBLE_TO_FIXPOINT(X) ((fixpoint64)((X) * (1UL << FIXPOINT_SHIFT)))
+#define FIXPOINT_TO_UINT(X) ((X) >> FIXPOINT_SHIFT)
+
 /* For the event_type members of rtt_event and flow_event */
 #define EVENT_TYPE_FLOW 1
 #define EVENT_TYPE_RTT 2
@@ -38,6 +43,7 @@ enum __attribute__((__packed__)) flow_event_source {
 
 struct bpf_config {
 	__u64 rate_limit;
+	fixpoint64 rtt_rate;
 };
 
 /*
@@ -68,6 +74,7 @@ struct network_tuple {
 
 struct flow_state {
 	__u64 min_rtt;
+	__u64 srtt;
 	__u64 last_timestamp;
 	__u64 sent_pkts;
 	__u64 sent_bytes;

pping/pping_kern.c

@@ -323,6 +323,33 @@ static __u32 remaining_pkt_payload(struct parsing_context *ctx)
 	return parsed_bytes < ctx->pkt_len ? ctx->pkt_len - parsed_bytes : 0;
 }
 
+/*
+ * Calculate a smooted rtt similar to how TCP stack does it in
+ * net/ipv4/tcp_input.c/tcp_rtt_estimator().
+ *
+ * NOTE: Will cause roundoff errors, but if RTTs > 1000ns errors should be small
+ */
+static __u64 calculate_srtt(__u64 prev_srtt, __u64 rtt)
+{
+	if (!prev_srtt)
+		return rtt;
+	// srtt = 7/8*prev_srtt + 1/8*rtt
+	return prev_srtt - (prev_srtt >> 3) + (rtt >> 3);
+}
+
+static bool wait_for_rate_limit(__u64 now, __u64 last_ts, __u64 rtt)
+{
+	if (now < last_ts)
+		return true;
+
+	// RTT-based rate limit
+	if (config.rtt_rate && rtt)
+		return now - last_ts < FIXPOINT_TO_UINT(config.rtt_rate * rtt);
+
+	// Static rate limit
+	return now - last_ts < config.rate_limit;
+}
+
 /*
  * Fills in event_type, timestamp, flow, source and reserved.
  * Does not fill in the flow_info.
@@ -402,8 +429,8 @@ static void pping_egress(void *ctx, struct parsing_context *pctx)
 	f_state->last_id = p_id.identifier;
 
 	// Check rate-limit
-	if (!new_flow && (now < f_state->last_timestamp ||
-			  now - f_state->last_timestamp < config.rate_limit))
+	if (!new_flow &&
+	    wait_for_rate_limit(now, f_state->last_timestamp, f_state->srtt))
 		return;
 
 	/*
@@ -448,12 +475,12 @@ static void pping_ingress(void *ctx, struct parsing_context *pctx)
 		goto validflow_out;
 
 	re.rtt = now - *p_ts;
-
 	// Delete timestamp entry as soon as RTT is calculated
 	bpf_map_delete_elem(&packet_ts, &p_id);
 
 	if (f_state->min_rtt == 0 || re.rtt < f_state->min_rtt)
 		f_state->min_rtt = re.rtt;
+	f_state->srtt = calculate_srtt(f_state->srtt, re.rtt);
 
 	// Fill event and push to perf-buffer
 	re.event_type = EVENT_TYPE_RTT;
@@ -526,6 +553,11 @@ int pping_xdp_ingress(struct xdp_md *ctx)
 		.is_egress = false,
 	};
 
+	bpf_printk("%llu - %llu\n", config.rate_limit, config.rtt_rate);
+	__u64 rtt = 1000000000;
+	bpf_printk("rtt_rate * %llu = %llu\n", rtt,
+		   FIXPOINT_TO_UINT(config.rtt_rate * rtt));
+
 	pping_ingress(ctx, &pctx);
 
 	return XDP_PASS;