From b035f14072efa3ca907256edbc28d72d8e39ba2a Mon Sep 17 00:00:00 2001 From: Jonathan Bennett Date: Sat, 17 Oct 2026 04:45:09 +0000 Subject: [PATCH] DMShellTest: end a session when the output is complete, not at a timer -DDMSHELL_TEST_UNTIL_DONE=1 types a printf after the command whose output is a marker. When the marker arrives in order, all of the command's output has arrived. Typing then stops and a second marker is typed. Its output follows everything typed before it, so seeing it means the shell has caught up, and the session closes. DMSHELL_TEST_SESSION_MS becomes a cap, 30 minutes by default in this mode. The stats add complete_ms and caught_up_ms. With the flag unset nothing changes. Co-Authored-By: Jonathan Bennett Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01REkPVFh6kvG4AZJ5A8AtM9 --- src/modules/DMShellRecovery.h | 30 +++++++ src/modules/DMShellTestModule.cpp | 100 ++++++++++++++++++++--- src/modules/DMShellTestModule.h | 12 +++ test/test_dmshell_recovery/test_main.cpp | 50 ++++++++++++ 4 files changed, 181 insertions(+), 11 deletions(-) diff --git a/src/modules/DMShellRecovery.h b/src/modules/DMShellRecovery.h index ab216af2d6..16ba640c11 100644 --- a/src/modules/DMShellRecovery.h +++ b/src/modules/DMShellRecovery.h @@ -6,6 +6,7 @@ #include "mesh/Throttle.h" #include +#include /// What the receive side wants done with one inbound sequenced frame. struct DMShellRxDecision { @@ -453,3 +454,32 @@ inline bool retransmitDue(uint32_t nowMs, uint32_t lastSentMs, uint32_t interval { return (uint32_t)(nowMs - lastSentMs) >= intervalMs; } + +/// Finds a fixed marker in a byte stream fed one byte at a time, so a match may span frames. +class DMShellOutputMarker +{ + public: + explicit DMShellOutputMarker(const char *marker) : text(marker) {} + + void reset() { matched = 0; } + + /// True on the byte that completes the marker; matching then starts afresh, so matches never overlap. + bool feed(uint8_t b) + { + // The longest prefix of the marker that ends at b: the bytes before b are the prefix matched so far + uint32_t k = matched + 1; + while (k > 0 && ((uint8_t)text[k - 1] != b || memcmp(text, text + matched + 1 - k, k - 1) != 0)) { + k--; + } + matched = k; + if (text[matched] == '\0') { + matched = 0; + return true; + } + return false; + } + + private: + const char *text; + uint32_t matched = 0; +}; diff --git a/src/modules/DMShellTestModule.cpp b/src/modules/DMShellTestModule.cpp index b22afb6792..77ccbe2de8 100644 --- a/src/modules/DMShellTestModule.cpp +++ b/src/modules/DMShellTestModule.cpp @@ -30,10 +30,19 @@ #ifndef DMSHELL_TEST_CR_EVERY #define DMSHELL_TEST_CR_EVERY 10 #endif +// 1 = a session types until the command's output has all arrived, then closes once the shell has caught up with the +// typing; DMSHELL_TEST_SESSION_MS is then only a cap +#ifndef DMSHELL_TEST_UNTIL_DONE +#define DMSHELL_TEST_UNTIL_DONE 0 +#endif // How long each session types for, from OPEN_OK; then it drains its input and closes #ifndef DMSHELL_TEST_SESSION_MS +#if DMSHELL_TEST_UNTIL_DONE +#define DMSHELL_TEST_SESSION_MS 1800000 +#else #define DMSHELL_TEST_SESSION_MS 300000 #endif +#endif #ifndef DMSHELL_TEST_SESSIONS #define DMSHELL_TEST_SESSIONS 1 #endif @@ -78,9 +87,14 @@ static constexpr uint32_t OPEN_TIMEOUT_MS = 180000; static constexpr uint32_t HEARTBEAT_IDLE_MS = 5000; static constexpr uint32_t HEARTBEAT_REPEAT_MS = 15000; static constexpr uint32_t DRAIN_TIMEOUT_MS = 10000; +static constexpr uint32_t CATCH_UP_TIMEOUT_MS = 60000; static constexpr uint32_t CLOSE_WAIT_MS = 7000; static constexpr uint32_t ACTIVE_POLL_MS = 20; +// Typed after the command and when typing stops; printf's arguments keep the terminal's echo from matching the markers +static const char DONE_COMMAND[] = "printf '<%s:%s>\\n' dmshell done\n"; +static const char IDLE_COMMAND[] = "printf '<%s:%s>\\n' dmshell idle\n"; + DMShellTestModule *dmShellTestModule; static const char *opName(meshtastic_RemoteShell_OpCode op) @@ -122,6 +136,8 @@ DMShellTestModule::DMShellTestModule() (unsigned)DMSHELL_TEST_PEER, (unsigned)DMSHELL_TEST_SESSIONS, (unsigned)DMSHELL_TEST_SESSION_MS, (unsigned)DMSHELL_TEST_START_DELAY_MS, DMSHELL_TEST_BATCH_PIPE ? "pipe" : "terminal", (unsigned)DMSHELL_TEST_INPUT_WINDOW, (unsigned)DMSHELL_TEST_COMMAND_AGAIN_MS); + if (DMSHELL_TEST_UNTIL_DONE) + LOG_INFO("DMShellTest: a session ends when its output is complete, capped at %u ms", (unsigned)DMSHELL_TEST_SESSION_MS); } int32_t DMShellTestModule::runOnce() @@ -189,6 +205,15 @@ void DMShellTestModule::startSession(uint32_t now) pendingInputLen = 0; commandTyped = false; commandRepeated = false; + commandsTyped = 0; + doneMarker.reset(); + idleMarker.reset(); + doneMarksSeen = 0; + completeAtMs = 0; + caughtUpAtMs = 0; + idleTyped = false; + typingEnded = false; + typingEndedAtMs = 0; keystrokesTyped = 0; batchLen = 0; batchFirstByteMs = batchLastByteMs = 0; @@ -221,24 +246,42 @@ void DMShellTestModule::endSession(uint32_t now, const char *reason) void DMShellTestModule::serviceActive(uint32_t now) { - const bool typing = !Throttle::deadlinePassedAt(now, openedAtMs + DMSHELL_TEST_SESSION_MS); - if (typing) - typeWorkload(now); - else - closeBatch(); // time is up: whatever is still being batched goes out now, as EOF ends the client's read + if (!typingEnded) { + const bool capped = Throttle::deadlinePassedAt(now, openedAtMs + DMSHELL_TEST_SESSION_MS); + if (capped || completeAtMs) { + typingEnded = true; + typingEndedAtMs = now; + if (DMSHELL_TEST_UNTIL_DONE && !completeAtMs) + LOG_WARN("DMShellTest t=%u cap of %u ms reached before the output completed", (unsigned)now, + (unsigned)DMSHELL_TEST_SESSION_MS); + closeBatch(); // whatever is still being batched goes out now, as EOF ends the client's read + if (completeAtMs) { + // Its output follows everything typed before it, so seeing it means the shell has caught up + typeText(IDLE_COMMAND, now); + closeBatch(); + idleTyped = true; + } + } else { + typeWorkload(now); + } + } flushInput(now); serviceRetransmit(now); if (phase != Phase::Active) return; // the peer stopped answering and the session ended serviceHeartbeat(now); - if (typing) + if (!typingEnded) return; const bool drained = pendingInputLen == 0 && batchLen == 0 && txWindow.outstanding() == 0; - if (drained || Throttle::deadlinePassedAt(now, openedAtMs + DMSHELL_TEST_SESSION_MS + DRAIN_TIMEOUT_MS)) { + const bool caughtUp = !idleTyped || caughtUpAtMs; + if ((drained && caughtUp) || + Throttle::deadlinePassedAt(now, typingEndedAtMs + (idleTyped ? CATCH_UP_TIMEOUT_MS : DRAIN_TIMEOUT_MS))) { if (!drained) LOG_WARN("DMShellTest t=%u closing with input unacknowledged: %u queued, %u outstanding", (unsigned)now, (unsigned)pendingInputLen, (unsigned)txWindow.outstanding()); + if (!caughtUp) + LOG_WARN("DMShellTest t=%u closing before the shell caught up", (unsigned)now); sendFrame(meshtastic_RemoteShell_OpCode_CLOSE, now); closeSent = true; phase = Phase::Closing; @@ -284,17 +327,44 @@ void DMShellTestModule::closeBatch() batchFirstByteMs = 0; } -void DMShellTestModule::typeCommand(uint32_t now) +void DMShellTestModule::typeText(const char *text, uint32_t now) { - static const char command[] = DMSHELL_TEST_COMMAND; - for (size_t i = 0; i + 1 < sizeof(command); i++) { - addByte((uint8_t)command[i], now); + for (const char *c = text; *c; c++) { + addByte((uint8_t)*c, now); if (batchDue(now)) closeBatch(); } +} + +void DMShellTestModule::typeCommand(uint32_t now) +{ + static const char command[] = DMSHELL_TEST_COMMAND; + typeText(command, now); + if (DMSHELL_TEST_UNTIL_DONE) + typeText(DONE_COMMAND, now); + commandsTyped++; LOG_INFO("DMShellTest t=%u typed the command (%u bytes)", (unsigned)now, (unsigned)(sizeof(command) - 1)); } +// Only in-order output reaches here, so a marker means everything the shell printed before it has arrived +void DMShellTestModule::noteOutput(const uint8_t *bytes, size_t len, uint32_t now) +{ + const uint32_t commandsWanted = DMSHELL_TEST_COMMAND_AGAIN_MS ? 2 : 1; + for (size_t i = 0; i < len; i++) { + if (doneMarker.feed(bytes[i])) { + doneMarksSeen++; + LOG_INFO("DMShellTest t=%u command output %u of %u complete", (unsigned)now, (unsigned)doneMarksSeen, + (unsigned)commandsWanted); + if (!completeAtMs && doneMarksSeen >= commandsWanted) + completeAtMs = now; + } + if (idleMarker.feed(bytes[i]) && idleTyped && !caughtUpAtMs) { + caughtUpAtMs = now; + LOG_INFO("DMShellTest t=%u shell caught up", (unsigned)now); + } + } +} + // The bench workload: after the settle time the command once, then one keystroke per interval void DMShellTestModule::typeWorkload(uint32_t now) { @@ -510,6 +580,8 @@ void DMShellTestModule::handleInOrder(const meshtastic_RemoteShell &frame, uint3 break; case meshtastic_RemoteShell_OpCode_OUTPUT: stats.rxOutputBytes += frame.payload.size; + if (DMSHELL_TEST_UNTIL_DONE) + noteOutput(frame.payload.bytes, frame.payload.size, now); break; case meshtastic_RemoteShell_OpCode_ERROR: LOG_WARN("DMShellTest t=%u remote error seq=%u: %.*s", (unsigned)now, (unsigned)frame.seq, (int)frame.payload.size, @@ -705,6 +777,12 @@ void DMShellTestModule::logStats(uint32_t now) (unsigned)stats.replayUnavailable, (unsigned)stats.replayEvicted, (unsigned)stats.openRetries); LOG_INFO("DMShellTest stats session=0x%08x recovery input_window_closed=%u input_dropped=%u", id, (unsigned)stats.inputWindowClosed, (unsigned)stats.pendingInputDropped); + if (DMSHELL_TEST_UNTIL_DONE) { + // From OPEN_OK; 0 = never + LOG_INFO("DMShellTest stats session=0x%08x until_done complete_ms=%u caught_up_ms=%u commands=%u", id, + (unsigned)(completeAtMs ? completeAtMs - openedAtMs : 0), + (unsigned)(caughtUpAtMs ? caughtUpAtMs - openedAtMs : 0), (unsigned)commandsTyped); + } } #endif diff --git a/src/modules/DMShellTestModule.h b/src/modules/DMShellTestModule.h index 8ef4ecd7eb..10811b5791 100644 --- a/src/modules/DMShellTestModule.h +++ b/src/modules/DMShellTestModule.h @@ -80,6 +80,16 @@ class DMShellTestModule : public SinglePortModule, private concurrency::OSThread uint32_t pendingInputLen = 0; bool commandTyped = false; bool commandRepeated = false; + uint32_t commandsTyped = 0; + // DMSHELL_TEST_UNTIL_DONE: what the shell prints once the command's output is complete, and once it has caught up + DMShellOutputMarker doneMarker{""}; + DMShellOutputMarker idleMarker{""}; + uint32_t doneMarksSeen = 0; + uint32_t completeAtMs = 0; + uint32_t caughtUpAtMs = 0; + bool idleTyped = false; + bool typingEnded = false; + uint32_t typingEndedAtMs = 0; uint32_t keystrokesTyped = 0; uint32_t nextKeystrokeMs = 0; // The batch the client's stdin read would return, before it becomes one INPUT @@ -96,6 +106,8 @@ class DMShellTestModule : public SinglePortModule, private concurrency::OSThread void serviceActive(uint32_t now); void typeWorkload(uint32_t now); void typeCommand(uint32_t now); + void typeText(const char *text, uint32_t now); + void noteOutput(const uint8_t *bytes, size_t len, uint32_t now); void addByte(uint8_t b, uint32_t now); bool batchDue(uint32_t now) const; void closeBatch(); diff --git a/test/test_dmshell_recovery/test_main.cpp b/test/test_dmshell_recovery/test_main.cpp index 9bb2da9faf..ab794498aa 100644 --- a/test/test_dmshell_recovery/test_main.cpp +++ b/test/test_dmshell_recovery/test_main.cpp @@ -760,6 +760,52 @@ void test_dmshell_retransmit_due_survives_millis_wrap() TEST_ASSERT_TRUE(retransmitDue(sent + 1400, sent, 1400)); } +static uint32_t feedAll(DMShellOutputMarker &marker, const char *bytes) +{ + uint32_t matches = 0; + for (const char *c = bytes; *c; c++) { + if (marker.feed((uint8_t)*c)) { + matches++; + } + } + return matches; +} + +void test_dmshell_output_marker_matches_across_frames() +{ + // OUTPUT frames split wherever the PTY read did, so the marker can straddle two of them + DMShellOutputMarker marker(""); + TEST_ASSERT_EQUAL_UINT32(0, feedAll(marker, "tail of the file\r\n\r\n$ ")); +} + +void test_dmshell_output_marker_ignores_the_echoed_command() +{ + // The terminal echoes what was typed; printf's arguments are chosen so the echo cannot match + DMShellOutputMarker marker(""); + TEST_ASSERT_EQUAL_UINT32(0, feedAll(marker, "$ printf '<%s:%s>\\n' dmshell done\r\n")); + TEST_ASSERT_EQUAL_UINT32(1, feedAll(marker, "\r\n")); +} + +void test_dmshell_output_marker_recovers_from_a_false_start() +{ + // A partial match broken by a byte that itself starts the marker again + DMShellOutputMarker marker(""); + TEST_ASSERT_EQUAL_UINT32(1, feedAll(marker, "")); + // A marker whose prefix repeats inside it: restarting from its first byte alone would miss this one + DMShellOutputMarker repeating("aab"); + TEST_ASSERT_EQUAL_UINT32(1, feedAll(repeating, "aaab")); +} + +void test_dmshell_output_marker_counts_each_occurrence() +{ + DMShellOutputMarker marker(""); + TEST_ASSERT_EQUAL_UINT32(2, feedAll(marker, "\r\n$ ")); + marker.feed('<'); + marker.reset(); + TEST_ASSERT_EQUAL_UINT32(0, feedAll(marker, "dmshell:done>")); +} + void setup() { initializeTestEnvironment(); @@ -811,6 +857,10 @@ void setup() RUN_TEST(test_dmshell_ack_latency_interval_is_twice_the_estimate_within_bounds); RUN_TEST(test_dmshell_retransmit_waits_for_the_frame_not_the_poll); RUN_TEST(test_dmshell_retransmit_due_survives_millis_wrap); + RUN_TEST(test_dmshell_output_marker_matches_across_frames); + RUN_TEST(test_dmshell_output_marker_ignores_the_echoed_command); + RUN_TEST(test_dmshell_output_marker_recovers_from_a_false_start); + RUN_TEST(test_dmshell_output_marker_counts_each_occurrence); exit(UNITY_END()); }