Bench traffic: jittered gaps, a fixed frame count, and an RX log

BENCH_TRAFFIC_JITTER_PCT=<j> draws each gap uniformly from the mean +- j%
instead of exponentially; 0 is a fixed timeline. BENCH_TRAFFIC_COUNT=<n>
stops after n frames and logs a final "done" summary. Every generator frame
heard from another node is logged with its sequence number, so loss is
counted per direction by packet. A frame the router refuses is counted as
rejected, not sent, and keeps its sequence number.

Co-Authored-By: Jonathan Bennett <jbennett@incomsystems.biz>
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01REkPVFh6kvG4AZJ5A8AtM9

src/modules/BenchTrafficModule.cpp

@@ -25,6 +25,14 @@
 #ifndef BENCH_TRAFFIC_SUMMARY_MS
 #define BENCH_TRAFFIC_SUMMARY_MS 60000
 #endif
+// Stop after this many frames have been sent; 0 runs until reboot
+#ifndef BENCH_TRAFFIC_COUNT
+#define BENCH_TRAFFIC_COUNT 0
+#endif
+// Unset: exponential gaps (Poisson arrivals). Set to j: each gap is the mean +- up to j%, drawn uniformly.
+#ifdef BENCH_TRAFFIC_JITTER_PCT
+static_assert(BENCH_TRAFFIC_JITTER_PCT >= 0 && BENCH_TRAFFIC_JITTER_PCT <= 100, "BENCH_TRAFFIC_JITTER_PCT is 0 to 100");
+#endif
 
 static_assert(BENCH_TRAFFIC_LOAD_PCT >= 0 && BENCH_TRAFFIC_LOAD_PCT <= 50, "BENCH_TRAFFIC_LOAD_PCT is 0 to 50");
 static_assert(BENCH_TRAFFIC_MIN_LEN >= 4 && BENCH_TRAFFIC_MIN_LEN <= BENCH_TRAFFIC_MAX_LEN, "length range");
@@ -40,6 +48,12 @@ BenchTrafficModule::BenchTrafficModule()
     LOG_INFO("BenchTraffic: %u%% of airtime, %u-%u byte broadcasts, hop limit 0, from %u ms after boot",
              (unsigned)BENCH_TRAFFIC_LOAD_PCT, (unsigned)BENCH_TRAFFIC_MIN_LEN, (unsigned)BENCH_TRAFFIC_MAX_LEN,
              (unsigned)BENCH_TRAFFIC_START_DELAY_MS);
+#ifdef BENCH_TRAFFIC_JITTER_PCT
+    LOG_INFO("BenchTraffic: gaps uniform, mean +-%u%%, %u frames", (unsigned)BENCH_TRAFFIC_JITTER_PCT,
+             (unsigned)BENCH_TRAFFIC_COUNT);
+#else
+    LOG_INFO("BenchTraffic: gaps exponential, %u frames", (unsigned)BENCH_TRAFFIC_COUNT);
+#endif
     setIntervalFromNow(BENCH_TRAFFIC_START_DELAY_MS);
 }
 
@@ -50,6 +64,40 @@ uint32_t BenchTrafficModule::exponentialMs(uint32_t meanMs)
     return (uint32_t)(-logf(u) * (float)meanMs + 0.5f);
 }
 
+uint32_t BenchTrafficModule::gapMs(uint32_t meanMs)
+{
+#ifdef BENCH_TRAFFIC_JITTER_PCT
+    const int32_t span = (int32_t)((uint64_t)meanMs * BENCH_TRAFFIC_JITTER_PCT / 100);
+    const int32_t gap = (int32_t)meanMs + (span ? (int32_t)random(-span, span + 1) : 0);
+    return gap > 0 ? (uint32_t)gap : 0;
+#else
+    return exponentialMs(meanMs);
+#endif
+}
+
+ProcessMessage BenchTrafficModule::handleReceived(const meshtastic_MeshPacket &mp)
+{
+    if (isFromUs(&mp) || mp.decoded.payload.size < sizeof(uint32_t))
+        return ProcessMessage::CONTINUE;
+    uint32_t seq;
+    memcpy(&seq, mp.decoded.payload.bytes, sizeof(seq));
+    heard++;
+    LOG_INFO("BenchTraffic: rx from=0x%08x n=%u id=0x%08x len=%u snr=%g rssi=%d", (unsigned)mp.from, (unsigned)seq,
+             (unsigned)mp.id, (unsigned)mp.decoded.payload.size, mp.rx_snr, (int)mp.rx_rssi);
+    return ProcessMessage::CONTINUE;
+}
+
+void BenchTrafficModule::logSummary(uint32_t now, const char *what)
+{
+    const uint32_t elapsed = now - firstSendMs;
+    LOG_INFO("BenchTraffic: %s: %u sent, %u rejected, %u skipped for duty cycle, %u heard, %u ms offered in %u ms (%u.%u%%), "
+             "send blocked %u ms/frame, started late %u ms/frame, %u reanchors",
+             what, (unsigned)sent, (unsigned)rejected, (unsigned)skippedDutyCycle, (unsigned)heard, (unsigned)offeredAirMs,
+             (unsigned)elapsed, (unsigned)(elapsed ? offeredAirMs * 100ULL / elapsed : 0),
+             (unsigned)(elapsed ? offeredAirMs * 1000ULL / elapsed % 10 : 0), (unsigned)(sent ? sendBlockedMs / sent : 0),
+             (unsigned)(sent ? lateMs / sent : 0), (unsigned)reanchors);
+}
+
 uint32_t BenchTrafficModule::sendOne()
 {
     RadioInterface *radio = router ? router->getRadioIface() : nullptr;
@@ -64,7 +112,7 @@ uint32_t BenchTrafficModule::sendOne()
     if (!p)
         return 0;
     const uint32_t len = (uint32_t)random(BENCH_TRAFFIC_MIN_LEN, BENCH_TRAFFIC_MAX_LEN + 1);
-    const uint32_t seq = ++sent;
+    const uint32_t seq = sent + rejected + 1; // a rejected frame keeps its number, so a gap in rx seqs is not all loss
     memcpy(p->decoded.payload.bytes, &seq, sizeof(seq));
     // Random fill, never uniform: the bench reads a uniform-fill frame as a failed SPI read
     for (uint32_t i = sizeof(seq); i < len; i++)
@@ -78,7 +126,13 @@ uint32_t BenchTrafficModule::sendOne()
     p->priority = meshtastic_MeshPacket_Priority_DEFAULT;
     const uint32_t airMs = radio->getPacketTime(p);
     const PacketId id = p->id;
-    service->sendToMesh(p);
+    const ErrorCode res = service->sendToMesh(p);
+    if (res != ERRNO_OK && res != ERRNO_SHOULD_RELEASE) {
+        rejected++;
+        LOG_WARN("BenchTraffic: tx n=%u id=0x%08x refused, err=%d", (unsigned)seq, (unsigned)id, (int)res);
+        return airMs; // still paces the schedule, so a full queue does not turn into a burst
+    }
+    sent++;
     offeredAirMs += airMs;
     LOG_INFO("BenchTraffic: tx n=%u id=0x%08x len=%u air=%u ms", (unsigned)seq, (unsigned)id, (unsigned)len, (unsigned)airMs);
     return airMs;
@@ -90,6 +144,13 @@ int32_t BenchTrafficModule::runOnce()
     return IDLE_POLL_MS; // the load-0 arm: the same image and node on the air, sending nothing of its own
 #else
     uint32_t now = millis();
+    if (done)
+        return IDLE_POLL_MS;
+    if (BENCH_TRAFFIC_COUNT && sent + rejected >= BENCH_TRAFFIC_COUNT) {
+        done = true;
+        logSummary(now, "done");
+        return IDLE_POLL_MS;
+    }
     if (!firstSendMs) {
         firstSendMs = now ? now : 1;
         lastSummaryMs = firstSendMs;
@@ -103,17 +164,11 @@ int32_t BenchTrafficModule::runOnce()
     now = after;
     if (!Throttle::isWithinTimespanMs(lastSummaryMs, BENCH_TRAFFIC_SUMMARY_MS)) {
         lastSummaryMs = now;
-        const uint32_t elapsed = now - firstSendMs;
-        LOG_INFO("BenchTraffic: %u sent, %u skipped for duty cycle, %u ms offered in %u ms (%u.%u%%), send blocked %u ms/frame, "
-                 "started late %u ms/frame, %u reanchors",
-                 (unsigned)sent, (unsigned)skippedDutyCycle, (unsigned)offeredAirMs, (unsigned)elapsed,
-                 (unsigned)(elapsed ? offeredAirMs * 100ULL / elapsed : 0),
-                 (unsigned)(elapsed ? offeredAirMs * 1000ULL / elapsed % 10 : 0), (unsigned)(sent ? sendBlockedMs / sent : 0),
-                 (unsigned)(sent ? lateMs / sent : 0), (unsigned)reanchors);
+        logSummary(now, "summary");
     }
     // Mean gap = this frame's airtime / load, so the offered fraction averages BENCH_TRAFFIC_LOAD_PCT
     const uint32_t meanMs = (airMs ? airMs : 50) * 100 / BENCH_TRAFFIC_LOAD_PCT;
-    nextDueMs += exponentialMs(meanMs);
+    nextDueMs += gapMs(meanMs);
     // Behind schedule: send at once, but past a few mean gaps restart from now rather than burst to catch up
     if (Throttle::deadlinePassedAt(now, nextDueMs + 4 * meanMs)) {
         reanchors++;

src/modules/BenchTrafficModule.h

@@ -1,7 +1,7 @@
 #pragma once
 
-// Bench: -DBENCH_TRAFFIC_LOAD_PCT=<n> makes this node a background traffic source. It broadcasts frames of random
-// length at exponentially distributed intervals, sized so they fill n% of airtime on average, with no relaying.
+// Bench: -DBENCH_TRAFFIC_LOAD_PCT=<n> makes this node a traffic source. It broadcasts frames of random length, sized so
+// they fill n% of airtime on average, with no relaying, and logs every such frame it hears from another node.
 #ifdef BENCH_TRAFFIC_LOAD_PCT
 
 #include "SinglePortModule.h"
@@ -14,6 +14,7 @@ class BenchTrafficModule : public SinglePortModule, private concurrency::OSThrea
 
   protected:
     int32_t runOnce() override;
+    ProcessMessage handleReceived(const meshtastic_MeshPacket &mp) override;
 
   private:
     uint32_t sent = 0;
@@ -25,11 +26,17 @@ class BenchTrafficModule : public SinglePortModule, private concurrency::OSThrea
     uint32_t sendBlockedMs = 0; // total time sendOne() held the loop
     uint32_t lateMs = 0;        // total time runOnce() started after its due time, i.e. other work held the loop
     uint32_t reanchors = 0;     // times the schedule fell too far behind and restarted from now
+    uint32_t rejected = 0;      // frames the router refused (a full TX queue), counted apart from sent
+    uint32_t heard = 0;         // another node's frames received
+    bool done = false;          // BENCH_TRAFFIC_COUNT frames sent and the final summary logged
 
     /// Queue one frame and return its expected airtime in ms, or 0 if nothing was sent
     uint32_t sendOne();
     /// An exponential draw with this mean, so arrivals are Poisson
     static uint32_t exponentialMs(uint32_t meanMs);
+    /// The gap to the next arrival: exponential, or the mean +- BENCH_TRAFFIC_JITTER_PCT uniformly when that is set
+    static uint32_t gapMs(uint32_t meanMs);
+    void logSummary(uint32_t now, const char *what);
 };
 
 extern BenchTrafficModule *benchTrafficModule;