Skip to content

Share the length-prefixed record reader across the stdin demos #346

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
josephnef merged 3 commits into from
Jul 27, 2026
Merged

Share the length-prefixed record reader across the stdin demos #346

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
3641eba
Share the length-prefixed record reader across the stdin demos
josephnef Jul 27, 2026
ed50a48
Cover the control-opcode escape the duplex demo reads
josephnef Jul 27, 2026
bca86ba
Bench coverage for the duplex control escape and the svctx rate ladder
josephnef Jul 27, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
80 changes: 76 additions & 4 deletions examples/common/stream_stdin.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,10 +14,22 @@
// truncates the first PSDU ("short read on stdin (76/269)") before a
// single frame is transmitted.
//
// 2. read_exact() — the length-prefixed record reader, returning a tri-state
// so each caller keeps its own short-read policy (the TX demo aborts on a
// truncated record; the duplex demo just stops its TX thread and lets RX
// run on).
// 2. read_exact() — the byte reader, returning a tri-state so each caller
// keeps its own short-read policy (the TX demo aborts on a truncated
// record; the duplex demo just stops its TX thread and lets RX run on).
//
// 3. read_record() — one whole <u32_le len><body> record, which is what
// every caller actually wanted. The little-endian assembly and the
// zero/oversize check were re-typed in four demos before this existed,
// and they had already drifted. Callers that need the length word before
// the body — the duplex demo escapes to a control TLV on its top bit —
// compose read_length() and read_body() instead.
//
// The result states are deliberately finer than "it failed": a producer that
// closed cleanly between records is an ordinary end of run, one that closed
// with a length word already written has lost a record, and a length out of
// range is a producer bug rather than a stream end. Each demo maps them to its
// own policy; none of them has to re-derive them.
//
// StreamStdinSelftest + tests/stream_stdin_test.cmake exercise this header
// headlessly (no libusb, no hardware), so a regression in the _WIN32 gate
Expand All @@ -27,6 +39,7 @@
#include <cstddef>
#include <cstdint>
#include <cstdio>
#include <vector>

#if defined(_WIN32)
#include <io.h>
Expand Down Expand Up @@ -71,4 +84,63 @@ inline ReadResult read_exact(std::FILE *f, void *buf, std::size_t n) {
return ReadResult::Ok;
}

// Read the 4-byte little-endian length prefix. `len` is untouched unless Ok.
inline ReadResult read_length(std::FILE *f, std::uint32_t &len) {
std::uint8_t b[4];
const ReadResult r = read_exact(f, b, sizeof(b));
if (r == ReadResult::Ok)
len = static_cast<std::uint32_t>(b[0]) |
(static_cast<std::uint32_t>(b[1]) << 8) |
(static_cast<std::uint32_t>(b[2]) << 16) |
(static_cast<std::uint32_t>(b[3]) << 24);
return r;
}

// Read `n` body bytes into `out`, sizing it to match. A zero-length body is Ok
// and leaves `out` empty — callers that forbid one check the length first.
inline ReadResult read_body(std::FILE *f, std::vector<std::uint8_t> &out,
std::size_t n) {
out.resize(n);
return n ? read_exact(f, out.data(), n) : ReadResult::Ok;
}

enum class RecordResult {
Ok, // a complete record is in `out`
Eof, // clean close: nothing at all where a record would have started
Short, // the stream ended part-way through a record
EofMidBody, // the length word arrived, then the stream closed before a byte
// of its body — one record lost, cleanly
BadLength, // zero, or larger than the caller's bound
};

// Read one whole <u32_le len><body> record. `max` bounds the body. When
// `raw_len` is given it receives the length word verbatim even on BadLength,
// so a caller with its own convention in the high bits can inspect it.
inline RecordResult read_record(std::FILE *f, std::vector<std::uint8_t> &out,
std::size_t max,
std::uint32_t *raw_len = nullptr) {
std::uint32_t len = 0;
switch (read_length(f, len)) {
case ReadResult::Eof:
return RecordResult::Eof;
case ReadResult::Short:
return RecordResult::Short;
case ReadResult::Ok:
break;
}
if (raw_len)
*raw_len = len;
if (len == 0 || len > max)
return RecordResult::BadLength;
switch (read_body(f, out, len)) {
case ReadResult::Ok:
return RecordResult::Ok;
case ReadResult::Eof:
return RecordResult::EofMidBody;
case ReadResult::Short:
return RecordResult::Short;
}
return RecordResult::Short; // unreachable; keeps every toolchain quiet
}

} // namespace stream_stdin
152 changes: 128 additions & 24 deletions examples/common/stream_stdin_selftest.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,8 +5,9 @@
// put into binary mode; the gate has to be `_WIN32` (not `_MSC_VER`) so it also
// fires under mingw/GCC. A regression there is invisible to a build-only CI job
// — it compiles fine and only corrupts bytes at runtime — so this binary
// exercises the exact set_stdin_binary() + read_exact() path the demos use, with
// no libusb and no radio.
// exercises the exact set_stdin_binary() + read_record() path the demos use,
// with no libusb and no radio. It also pins what each record state means,
// since the four demos map those states onto four different policies.
//
// StreamStdinSelftest --gen writes the canonical stream to stdout (binary)
// StreamStdinSelftest reads that stream from stdin and round-trips it
Expand Down Expand Up @@ -38,6 +39,10 @@ static const std::vector<std::vector<uint8_t>> &canonical_records() {
return recs;
}

// Bound for read_record. Comfortably above the canonical records, so a
// BadLength here means the stream desynced, never that the cap was too tight.
static const std::size_t kMaxRecord = 4096;

static void put_u32_le(uint8_t *p, uint32_t v) {
p[0] = static_cast<uint8_t>(v & 0xFF);
p[1] = static_cast<uint8_t>((v >> 8) & 0xFF);
Expand All @@ -62,38 +67,34 @@ static int do_check() {
const auto &expected = canonical_records();
size_t got_records = 0, got_bytes = 0;
for (const auto &exp : expected) {
uint8_t len_bytes[4];
auto r = stream_stdin::read_exact(stdin, len_bytes, sizeof(len_bytes));
if (r != stream_stdin::ReadResult::Ok) {
// read_record is the composed path every demo now uses: the length
// assembly and the range check live in the header rather than four times
// over, so this exercises what the demos actually run.
std::vector<uint8_t> body;
uint32_t len = 0;
const auto r = stream_stdin::read_record(stdin, body, kMaxRecord, &len);
if (r == stream_stdin::RecordResult::BadLength) {
std::fprintf(stderr,
"stream_stdin_selftest: FAIL — len-prefix read for record %zu "
"returned %d (binary stdin likely broken: a 0x1A in a prior "
"record was read as EOF)\n",
"stream_stdin_selftest: FAIL — record %zu length %u out of "
"range (stream desynced; likely CRLF/Ctrl-Z translation)\n",
got_records, len);
return 3;
}
if (r != stream_stdin::RecordResult::Ok) {
std::fprintf(stderr,
"stream_stdin_selftest: FAIL — record %zu returned %d "
"(binary stdin likely broken: a 0x1A in a prior record was "
"read as EOF)\n",
got_records, static_cast<int>(r));
return 2;
}
uint32_t len = static_cast<uint32_t>(len_bytes[0])
| (static_cast<uint32_t>(len_bytes[1]) << 8)
| (static_cast<uint32_t>(len_bytes[2]) << 16)
| (static_cast<uint32_t>(len_bytes[3]) << 24);
if (len != exp.size()) {
std::fprintf(stderr,
"stream_stdin_selftest: FAIL — record %zu length %u != "
"expected %zu (stream desynced; likely CRLF/Ctrl-Z "
"translation)\n",
got_records, len, exp.size());
return 3;
}
std::vector<uint8_t> body(len);
if (len) {
r = stream_stdin::read_exact(stdin, body.data(), len);
if (r != stream_stdin::ReadResult::Ok) {
std::fprintf(stderr,
"stream_stdin_selftest: FAIL — body read for record %zu "
"returned %d\n",
got_records, static_cast<int>(r));
return 4;
}
return 4;
}
if (body != exp) {
std::fprintf(stderr,
Expand All @@ -120,7 +121,110 @@ static int do_check() {
return 0;
}

// Every demo maps read_record's states onto its own policy — streamtx warns on
// one and exits 2 on another, txdemo counts a lost shard, duplex stops its TX
// thread and leaves RX running. So the states have to mean exactly what they
// say. Driven through tmpfile() rather than the pipe, since these are the cases
// a well-formed stream never produces.
static int check_state(const char *what, const std::vector<uint8_t> &bytes,
std::size_t max, stream_stdin::RecordResult want,
uint32_t want_len) {
std::FILE *f = std::tmpfile();
if (!f) {
std::fprintf(stderr, "stream_stdin_selftest: tmpfile() failed\n");
return 7;
}
if (!bytes.empty()) std::fwrite(bytes.data(), 1, bytes.size(), f);
std::rewind(f);
std::vector<uint8_t> body;
uint32_t len = 0;
const auto got = stream_stdin::read_record(f, body, max, &len);
std::fclose(f);
if (got != want) {
std::fprintf(stderr,
"stream_stdin_selftest: FAIL — %s returned %d, expected %d\n",
what, static_cast<int>(got), static_cast<int>(want));
return 7;
}
if (want_len && len != want_len) {
std::fprintf(stderr,
"stream_stdin_selftest: FAIL — %s reported length %u, "
"expected %u\n", what, len, want_len);
return 7;
}
return 0;
}

static int do_states() {
using R = stream_stdin::RecordResult;
auto rec = [](uint32_t len, std::size_t body_bytes) {
std::vector<uint8_t> v(4);
put_u32_le(v.data(), len);
v.insert(v.end(), body_bytes, 0x1A); // 0x1A: EOF to a text-mode read
return v;
};
struct Case {
const char *what;
std::vector<uint8_t> bytes;
std::size_t max;
R want;
uint32_t want_len;
};
const std::vector<Case> cases = {
{"a complete record", rec(3, 3), 16, R::Ok, 3},
{"nothing at all", {}, 16, R::Eof, 0},
{"a truncated length prefix", {0x05, 0x00}, 16, R::Short, 0},
{"a length with no body at all", rec(4, 0), 16, R::EofMidBody, 4},
{"a body cut short", rec(8, 3), 16, R::Short, 8},
{"a zero length", rec(0, 0), 16, R::BadLength, 0},
{"a length past the bound", rec(99, 0), 16, R::BadLength, 99},
};
for (const auto &c : cases) {
const int rc = check_state(c.what, c.bytes, c.max, c.want, c.want_len);
if (rc) return rc;
}

/* The duplex demo escapes to a control TLV on the length's top bit, and
* bounds that body at 256 rather than at its PSDU maximum — so it reads the
* length, masks it, and only then reads the body. That composition is the
* reason read_length and read_body are public at all, and nothing else
* covers it: a record whose length has the top bit set must arrive
* verbatim, and must not be mistaken for an oversize PSDU. */
{
std::vector<uint8_t> bytes(4);
const uint32_t escaped = 0x80000000u | 3u;
put_u32_le(bytes.data(), escaped);
const std::vector<uint8_t> want = {0x01, 0x1A, 0x0D};
bytes.insert(bytes.end(), want.begin(), want.end());
std::FILE *f = std::tmpfile();
if (!f) {
std::fprintf(stderr, "stream_stdin_selftest: tmpfile() failed\n");
return 7;
}
std::fwrite(bytes.data(), 1, bytes.size(), f);
std::rewind(f);
uint32_t len = 0;
std::vector<uint8_t> body;
const bool ok =
stream_stdin::read_length(f, len) == stream_stdin::ReadResult::Ok &&
(len & 0x80000000u) != 0 &&
stream_stdin::read_body(f, body, len & 0x7fffffffu) ==
stream_stdin::ReadResult::Ok &&
body == want;
std::fclose(f);
if (!ok) {
std::fprintf(stderr, "stream_stdin_selftest: FAIL — the control-opcode "
"escape (top length bit) did not round-trip\n");
return 7;
}
}
std::fprintf(stdout, "stream_stdin_selftest: %zu record states OK\n",
cases.size());
return 0;
}

int main(int argc, char **argv) {
if (argc > 1 && std::strcmp(argv[1], "--gen") == 0) return do_gen();
if (const int rc = do_states()) return rc;
return do_check();
}
20 changes: 9 additions & 11 deletions examples/duplex/main.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -196,25 +196,23 @@ static void tx_thread(TxArgs args) {
long tx_count = 0;

while (!args.should_stop->load()) {
uint8_t len_bytes[4];
if (stream_stdin::read_exact(stdin, len_bytes, sizeof(len_bytes)) !=
stream_stdin::ReadResult::Ok) {
// This demo reads the length itself rather than using read_record: its top
// bit escapes to a control TLV with its own, much smaller bound, so the
// length has to be inspected before the body may be read.
uint32_t len = 0;
if (stream_stdin::read_length(stdin, len) != stream_stdin::ReadResult::Ok) {
// Clean EOF or short read — TX side done. RX keeps running.
devourer::Ev(*g_ev, "stream.eof").f("tx_count", tx_count);
break;
}
uint32_t len = static_cast<uint32_t>(len_bytes[0])
| (static_cast<uint32_t>(len_bytes[1]) << 8)
| (static_cast<uint32_t>(len_bytes[2]) << 16)
| (static_cast<uint32_t>(len_bytes[3]) << 24);

// Control-opcode escape: top bit set -> the body is a control TLV (the
// adaptive link's live knobs), not a PSDU. <op:u8><payload...>.
if (len & 0x80000000u) {
uint32_t clen = len & 0x7fffffffu;
if (clen == 0 || clen > 256) break;
std::vector<uint8_t> ctl(clen);
if (stream_stdin::read_exact(stdin, ctl.data(), clen) !=
std::vector<uint8_t> ctl;
if (stream_stdin::read_body(stdin, ctl, clen) !=
stream_stdin::ReadResult::Ok)
break;
uint8_t op = ctl[0];
Expand All @@ -241,8 +239,8 @@ static void tx_thread(TxArgs args) {
args.max_psdu);
break;
}
std::vector<uint8_t> psdu(len);
if (stream_stdin::read_exact(stdin, psdu.data(), len) !=
std::vector<uint8_t> psdu;
if (stream_stdin::read_body(stdin, psdu, len) !=
stream_stdin::ReadResult::Ok) {
/* EOF mid-PSDU: `bytes` = the expected PSDU length that was cut short. */
devourer::Ev(*g_ev, "stream.eof").f("tx_count", tx_count).f("bytes", len);
Expand Down
35 changes: 12 additions & 23 deletions examples/streamtx/main.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -363,33 +363,22 @@ int main(int argc, char **argv) {
std::chrono::high_resolution_clock::now().time_since_epoch().count());
std::vector<uint8_t> sync_buf;
while (true) {
uint8_t len_bytes[4];
std::vector<uint8_t> psdu;
uint32_t len = 0;
{
auto r = stream_stdin::read_exact(stdin, len_bytes, sizeof(len_bytes));
if (r == stream_stdin::ReadResult::Eof) break; // clean stdin close
if (r == stream_stdin::ReadResult::Short) {
logger->error("short read on stdin len-prefix; record truncated");
std::exit(2);
}
}
uint32_t len = static_cast<uint32_t>(len_bytes[0])
| (static_cast<uint32_t>(len_bytes[1]) << 8)
| (static_cast<uint32_t>(len_bytes[2]) << 16)
| (static_cast<uint32_t>(len_bytes[3]) << 24);
if (len == 0 || len > max_psdu) {
logger->error("PSDU length {} out of range (max {}); stopping", len,
max_psdu);
break;
}
std::vector<uint8_t> psdu(len);
{
auto r = stream_stdin::read_exact(stdin, psdu.data(), len);
if (r == stream_stdin::ReadResult::Eof) {
const auto r = stream_stdin::read_record(stdin, psdu, max_psdu, &len);
if (r == stream_stdin::RecordResult::Eof) break; // clean stdin close
if (r == stream_stdin::RecordResult::EofMidBody) {
logger->warn("EOF mid-PSDU (expected {} bytes)", len);
break;
}
if (r == stream_stdin::ReadResult::Short) {
logger->error("short read mid-PSDU (expected {} bytes); record "
if (r == stream_stdin::RecordResult::BadLength) {
logger->error("PSDU length {} out of range (max {}); stopping", len,
max_psdu);
break;
}
if (r != stream_stdin::RecordResult::Ok) {
logger->error("short read on stdin (expected {} bytes); record "
"truncated", len);
std::exit(2);
}
Expand Down
Loading
Loading