ef: add per-command rate limiting + sendmmsg / TX_RING fast TX paths

CMakeLists.txt

@@ -81,7 +81,9 @@ add_library(libef STATIC
     src/ef-payload.c
     src/ef-profinet.c
     src/ef-ptp.c
+    src/ef-rate.c
     src/ef-sv.c
+    src/ef-txring.c
     src/ef-udp.c
     src/ef-vlan.c
     ${version_file}
@@ -123,6 +125,7 @@ add_executable(ef-tests
     test/test-padding.cxx
     test/test-alloc-free.cxx
     test/test-mld-zero-width.cxx
+    test/test-rate-limit.cxx
 )
 target_compile_options(ef-tests PRIVATE ${EF_SANITIZE_FLAGS})
 

src/ef-args.c

@@ -4,6 +4,7 @@
 #include <stdlib.h>
 #include <unistd.h>
 #include <stdio.h>
+#include <getopt.h>
 
 int argc_frame(int argc, const char *argv[], frame_t *f) {
     int i, j, res, offset;
@@ -79,6 +80,11 @@ void cmd_destruct(cmd_t *c) {
     if (c->frame_mask_buf)
         bfree(c->frame_mask_buf);
 
+    if (c->mmsg)
+        free(c->mmsg);
+    if (c->miov)
+        free(c->miov);
+
     memset(c, 0, sizeof(*c));
 }
 
@@ -97,11 +103,29 @@ void print_help() {
     po("  -h                    Top level help message.\n");
     po("  -p                    No pad. Skip padding frames to 60 bytes,\n");
     po("     allowing runt frames to be sent or matched as-is.\n");
-    po("  -t <timeout-in-ms>    When listening on an interface (rx),\n");
-    po("     When listening on an interface (rx), the tool will always\n");
-    po("     listen during the entire timeout period. This is needed,\n");
-    po("     as we must also check that no frames are received during\n");
-    po("     the test.  Default is 100ms.\n");
+    po("  -Q                    Set PACKET_QDISC_BYPASS on all sockets.\n");
+    po("     Skips the Linux qdisc layer entirely, reducing TX CPU cost.\n");
+    po("  -r                    Use PACKET_TX_RING (TPACKET_V2) for TX.\n");
+    po("     Per-cmd mmap ring; one atomic store per frame plus a periodic\n");
+    po("     send() kick. Off by default; the env var EF_TX_RING=1 has the\n");
+    po("     same effect as -r.\n");
+    po("  -m                    Batch TX with sendmmsg in the rate path.\n");
+    po("     Sends up to 'burst' frames per syscall using a pre-built\n");
+    po("     mmsghdr vector. Requires a 'rate ...' on the tx command\n");
+    po("     (use 'rate <high>G' as a near-unlimited rate to opt in).\n");
+    po("     Same effect via env: EF_USE_SENDMMSG=1.\n");
+    po("  --ignore-link-down    On the PACKET_TX_RING path (-r), retry\n");
+    po("     silently when send() returns ENETDOWN instead of treating it\n");
+    po("     as fatal. Useful for tests that toggle the link mid-run.\n");
+    po("  -t <timeout-in-ms>    Wall-clock deadline. Default 100ms.\n");
+    po("     RX: the tool always listens for the full timeout period\n");
+    po("     so we can verify that no unexpected frames arrive.\n");
+    po("     TX:\n");
+    po("       'rep N' (with or without 'rate'): runs to completion,\n");
+    po("         ignoring -t. Explicit rep is the user contract.\n");
+    po("       'rate ...' with no rep: stops at -t.\n");
+    po("       no rep, no rate: a single frame is sent and the loop\n");
+    po("         exits as soon as RX (if any) is satisfied.\n");
     po("\n");
     po("  -c <if>,[<snaplen>],[<sync>],[<file>],[cnt]\n");
     po("     Use tcpdump to capture traffic on an interface while the\n");
@@ -113,7 +137,7 @@ void print_help() {
     po("\n");
     po("Valid commands:\n");
     po("  tx: Transmit a frame on a interface. Syntax:\n");
-    po("  tx <interface> FRAME | help\n");
+    po("  tx <interface> [rep <N>] [rate <pps>] [burst <N>] FRAME | help\n");
     po("\n");
     po("  rx: Specify a frame which is expected to be received. If no \n");
     po("      frame is specified, then the expectation is that no\n");
@@ -166,6 +190,21 @@ void print_help() {
     po("   Note that the repeat flag must follow the tx <interface> key-word\n");
     po("   Results must be viewed through the PC or DUT interface counters, i.e. outside of 'ef'\n");
     po("\n");
+    po("TX rate limiting:\n");
+    po("   'rate <pps>' limits TX to the given packets per second.\n");
+    po("   'rate <N>K|M|G' limits TX to the given wire rate in Kbps/Mbps/Gbps.\n");
+    po("   Wire rate includes preamble, SFD, FCS and IFG (24 bytes overhead).\n");
+    po("   'rate' without 'rep' implies infinite repeat, bounded by -t timeout.\n");
+    po("   'rep', 'rate' and 'burst' can appear in any order.\n");
+    po("   'burst <N>' overrides the token-bucket burst size (default: 10%%\n");
+    po("   of pps, clamped to [1, 1024]). Useful at low rates where the\n");
+    po("   default burst would send an unwanted packet storm.\n");
+    po("Examples:\n");
+    po("   ef -t 5000 tx eth0 rate 1000 eth dmac ::1 smac ::2\n");
+    po("   ef tx eth0 rep 500 rate 100 eth dmac ::1 smac ::2\n");
+    po("   ef -t 5000 tx eth0 rate 1G eth dmac ::1 smac ::2\n");
+    po("   ef -t 5000 tx eth0 rate 100M eth dmac ::1 smac ::2\n");
+    po("\n");
 }
 
 int argc_cmd(int argc, const char *argv[], cmd_t *c) {
@@ -223,11 +262,54 @@ int argc_cmd(int argc, const char *argv[], cmd_t *c) {
     }
 
     if (c->type == CMD_TYPE_TX) {
+        int rep_given = 0, kw;
+
         c->repeat = 1;
-        if (strcmp(argv[i], "rep") == 0 || strcmp(argv[i], "repeat") == 0) {
-            c->repeat = atoi(argv[i+1]);
-            i += 2;
+        c->rate_pps = 0;
+        c->rate_bps = 0;
+        c->rate_burst = 0;
+
+        for (kw = 0; kw < 3 && i < argc; kw++) {
+            if (strcmp(argv[i], "rep") == 0 ||
+                strcmp(argv[i], "repeat") == 0) {
+                if (i + 1 >= argc)
+                    break;
+                c->repeat = atoi(argv[i + 1]);
+                rep_given = 1;
+                i += 2;
+            } else if (strcmp(argv[i], "rate") == 0) {
+                if (i + 1 >= argc)
+                    break;
+
+                const char *val = argv[i + 1];
+                char *end;
+                unsigned long long num = strtoull(val, &end, 10);
+
+                if (end != val && (*end == 'K' || *end == 'k')) {
+                    c->rate_bps = num * 1000ULL;
+                } else if (end != val && (*end == 'M' || *end == 'm')) {
+                    c->rate_bps = num * 1000000ULL;
+                } else if (end != val && (*end == 'G' || *end == 'g')) {
+                    c->rate_bps = num * 1000000000ULL;
+                } else {
+                    c->rate_pps = (uint32_t)num;
+                }
+
+                i += 2;
+            } else if (strcmp(argv[i], "burst") == 0) {
+                if (i + 1 >= argc)
+                    break;
+                c->rate_burst = atoi(argv[i + 1]);
+                i += 2;
+            } else {
+                break;
+            }
         }
+
+        // rate without rep implies infinite repeat
+        if ((c->rate_pps > 0 || c->rate_bps > 0) && !rep_given)
+            c->repeat = UINT32_MAX;
+        c->rep_explicit = rep_given;
     }
 
     //po("%d, i=%d/%d %s\n", __LINE__, i, argc, argv[i]);
@@ -411,13 +493,39 @@ int argc_cmds(int argc, const char *argv[]) {
 
 int NO_PAD = 0;
 int TIME_OUT_MS = 100;
+int QDISC_BYPASS = 0;
+int TX_RING = 0;
+int MMSG_TX = 0;
+int IGNORE_LINK_DOWN = 0;
 parse_err_ctx_t PARSE_ERR_CTX;
 
 int main_(int argc, const char *argv[]) {
+    static const struct option long_opts[] = {
+        { "ignore-link-down", no_argument, NULL, 1 },
+        { NULL,               0,           NULL, 0 },
+    };
     int opt;
 
-    while ((opt = getopt(argc, (char * const*)argv, "pvht:c:")) != -1) {
+    while ((opt = getopt_long(argc, (char * const*)argv, "pQvhrmt:c:",
+                              long_opts, NULL)) != -1) {
         switch (opt) {
+            case 1:  // --ignore-link-down
+                IGNORE_LINK_DOWN = 1;
+                break;
+
+            case 'Q':
+                QDISC_BYPASS = 1;
+                break;
+
+
+            case 'r':
+                TX_RING = 1;
+                break;
+
+            case 'm':
+                MMSG_TX = 1;
+                break;
+
             case 'p':
                 NO_PAD = 1;
                 break;