From c3636005831064e71b03a5f8796a21e89b2a714f Mon Sep 17 00:00:00 2001 From: Dimitri Staessens Date: Thu, 19 Feb 2026 22:03:16 +0100 Subject: irmd: Allow direct rbuff between local processes This allows bypassing the IPCP for local processes that share the same packet pool, lowering latency between processes to comparable levels as Unix sockets (RTT in the order of a microsecond). For local processes, no IPCPs are needed: $ irm b prog oping n oping $ oping -l Ouroboros ping server started. New flow 64. Received 64 bytes on fd 64. The direct IPC can be disabled with the DISABLE_DIRECT_IPC build flag. Note that this is needed for rumba 'local' experiments to emulate network topologies. Without this flag all processes will just communicate directly. Signed-off-by: Dimitri Staessens Signed-off-by: Sander Vrijders --- src/tools/oping/oping_client.c | 122 +++++++++++++++++++++++++++++++++++------ 1 file changed, 106 insertions(+), 16 deletions(-) (limited to 'src/tools/oping/oping_client.c') diff --git a/src/tools/oping/oping_client.c b/src/tools/oping/oping_client.c index 18dd3078..23807f65 100644 --- a/src/tools/oping/oping_client.c +++ b/src/tools/oping/oping_client.c @@ -67,6 +67,26 @@ static void update_rtt_stats(double ms) client.rtt_m2 += d * (ms - client.rtt_avg); } +static double rtt_val(double ms) +{ + return ms < 0.1 ? ms * 1000 : ms; +} + +static const char * rtt_unit(double ms) +{ + return ms < 0.1 ? "µs" : "ms"; +} + +static void print_rtt(int len, int seq, + double ms, const char * suf) +{ + printf("%d bytes from %s: seq=%d " + "time=%.3f %s%s\n", + len, client.s_apn, seq, + rtt_val(ms), rtt_unit(ms), + suf != NULL ? suf : ""); +} + void * reader(void * o) { struct timespec timeout = {client.interval / 1000 + 2, 0}; @@ -127,12 +147,9 @@ void * reader(void * o) (size_t) rtc.tv_nsec / 1000); } - printf("%d bytes from %s: seq=%d time=%.3f ms%s\n", - msg_len, - client.s_apn, - ntohl(msg->id), - ms, - id < exp_id ? " [out-of-order]" : ""); + print_rtt(msg_len, ntohl(msg->id), ms, + id < exp_id ? + " [out-of-order]" : NULL); } update_rtt_stats(ms); @@ -223,16 +240,87 @@ static void print_stats(struct timespec * tic, printf("time: %.3f ms\n", ts_diff_us(toc, tic) / 1000.0); if (client.rcvd > 0) { + double a = client.rtt_avg; + double f = a < 0.1 ? 1000 : 1; printf("rtt min/avg/max/mdev = %.3f/%.3f/%.3f/", - client.rtt_min, - client.rtt_avg, - client.rtt_max); + client.rtt_min * f, client.rtt_avg * f, + client.rtt_max * f); if (client.rcvd > 1) - printf("%.3f ms\n", - sqrt(client.rtt_m2 / (client.rcvd - 1))); + printf("%.3f %s\n", + sqrt(client.rtt_m2 / + (client.rcvd - 1)) * f, + rtt_unit(a)); else - printf("NaN ms\n"); + printf("NaN %s\n", rtt_unit(a)); + } +} + +static int flood_busy_ping(int fd) +{ + char buf[OPING_BUF_SIZE]; + struct oping_msg * msg = (struct oping_msg *) buf; + struct timespec sent; + struct timespec rcvd; + double ms; + int n; + + memset(buf, 0, client.size); + + fccntl(fd, FLOWSFLAGS, + FLOWFRDWR | FLOWFRNOPART | FLOWFRNOBLOCK); + + if (!client.quiet) + printf("Pinging %s with %d bytes" + " of data (%u packets," + " busy-poll):\n\n", + client.s_apn, client.size, + client.count); + + while (!stop && client.sent < client.count) { + clock_gettime(CLOCK_MONOTONIC, &sent); + + msg->type = htonl(ECHO_REQUEST); + msg->id = htonl(client.sent); + msg->tv_sec = sent.tv_sec; + msg->tv_nsec = sent.tv_nsec; + + if (flow_write(fd, buf, + client.size) < 0) { + printf("Failed to send " + "packet.\n"); + break; + } + + ++client.sent; + + do { + n = flow_read(fd, buf, + OPING_BUF_SIZE); + } while (n == -EAGAIN && !stop); + + if (n < 0) + break; + + clock_gettime(CLOCK_MONOTONIC, &rcvd); + + if (ntohl(msg->type) != ECHO_REPLY) + continue; + + ++client.rcvd; + + sent.tv_sec = msg->tv_sec; + sent.tv_nsec = msg->tv_nsec; + ms = ts_diff_us(&rcvd, &sent) / 1000.0; + + update_rtt_stats(ms); + + if (!client.quiet) + print_rtt(client.size, + ntohl(msg->id), ms, + NULL); } + + return 0; } static int flood_ping(int fd) @@ -283,9 +371,9 @@ static int flood_ping(int fd) update_rtt_stats(ms); if (!client.quiet) - printf("%d bytes from %s: seq=%d time=%.3f ms\n", - client.size, client.s_apn, - ntohl(msg->id), ms); + print_rtt(client.size, + ntohl(msg->id), ms, + NULL); } return 0; @@ -337,7 +425,9 @@ static int client_main(void) clock_gettime(CLOCK_REALTIME, &tic); - if (client.flood) + if (client.flood_busy) + flood_busy_ping(fd); + else if (client.flood) flood_ping(fd); else threaded_ping(fd); -- cgit v1.2.3