From 43f4ee2493be4a04c4f1acf10aa2bcc89fa4e2f0 Mon Sep 17 00:00:00 2001 From: Jan Pietek Date: Fri, 9 Oct 2026 14:05:16 +0200 Subject: [PATCH 1/4] Expose native packet EOF for orderly live stream drain --- src/avplumber_pybind.cpp | 2 ++ tests/python/smoke_packet_eof.py | 32 ++++++++++++++++++++++++++++++++ 2 files changed, 34 insertions(+) create mode 100644 tests/python/smoke_packet_eof.py diff --git a/src/avplumber_pybind.cpp b/src/avplumber_pybind.cpp index f3a4dc37..c4509099 100644 --- a/src/avplumber_pybind.cpp +++ b/src/avplumber_pybind.cpp @@ -867,6 +867,8 @@ PYBIND11_MODULE(_avplumber, m) { py::class_>(m, "Packet") .def(py::init<>()) + .def_static("eof", []() { return createEofPacket(); }, + "Create the native packet EOF marker (an empty Packet is not EOF).") .def_property_readonly("pts", [](const av::Packet &p) { return p.pts(); }) .def_property_readonly("dts", [](const av::Packet &p) { return p.dts(); }) .def_property_readonly("duration", [](const av::Packet &p) { return p.duration(); }) diff --git a/tests/python/smoke_packet_eof.py b/tests/python/smoke_packet_eof.py new file mode 100644 index 00000000..ab239be8 --- /dev/null +++ b/tests/python/smoke_packet_eof.py @@ -0,0 +1,32 @@ +"""Run under an external timeout; two input EOFs must finish one mux output.""" + +import time + +from _avplumber import Packet +from pyplumber import AVPlumber +from pyplumber.node import Mux + +marker = Packet.eof() +assert marker.size == 1 +assert marker.data == b"\xff" +assert marker.pts.timestamp == Packet().pts.timestamp +assert Packet().size == 0 + +avp = AVPlumber() +avp.addNode(Mux({"name": "mux", "group": "test", "src": ["video", "audio"], + "dst": "merged", "allow_no_encoder": True})) +avp.group("test").startNodes() +avp.getEdge("video", "Packet").enqueue(marker) +time.sleep(0.05) +assert avp.node("mux").isWorking, "Video EOF must not discard unfinished audio" +assert avp.getEdge("merged", "Packet").occupied == 0 +avp.getEdge("audio", "Packet").enqueue(Packet.eof()) +deadline = time.monotonic() + 5 +while avp.node("mux").isWorking and time.monotonic() < deadline: + time.sleep(0.01) +assert not avp.node("mux").isWorking, "Mux did not finish after both EOFs" +merged = avp.getEdge("merged", "Packet") +assert merged.occupied == 1 +assert merged.tryGet(0).data == b"\xff" +avp.shutdown() +print("Packet EOF and mux drain passed", flush=True) From ee7ae4804fa60697317439de4447de57c0bafaad Mon Sep 17 00:00:00 2001 From: Jan Pietek Date: Fri, 9 Oct 2026 14:21:46 +0200 Subject: [PATCH 2/4] Drain live SRT publishers before closing FFmpeg sockets --- ...019-avformat-libsrt-live-close-drain.patch | 107 ++++++++++++++++++ deps/ffmpeg/9/bases.env | 4 +- deps/ffmpeg/README.md | 7 ++ tests/test_ffmpeg_patch_series.py | 9 +- tests/test_ffmpeg_srt_drain.py | 93 +++++++++++++++ 5 files changed, 215 insertions(+), 5 deletions(-) create mode 100644 deps/ffmpeg/9/0019-avformat-libsrt-live-close-drain.patch create mode 100644 tests/test_ffmpeg_srt_drain.py diff --git a/deps/ffmpeg/9/0019-avformat-libsrt-live-close-drain.patch b/deps/ffmpeg/9/0019-avformat-libsrt-live-close-drain.patch new file mode 100644 index 00000000..0c90b654 --- /dev/null +++ b/deps/ffmpeg/9/0019-avformat-libsrt-live-close-drain.patch @@ -0,0 +1,107 @@ +From eecb04adc1ed5ff7c2a48844e21626da5864e8f8 Mon Sep 17 00:00:00 2001 +From: avplumber patches +Date: Fri, 9 Oct 2026 14:19:16 +0200 +Subject: [PATCH] avformat/libsrt: drain live publishers before explicit linger + close + +--- + doc/protocols.texi | 8 +++++++ + libavformat/libsrt.c | 54 +++++++++++++++++++++++++++++++++++++++++++- + 2 files changed, 61 insertions(+), 1 deletion(-) + +diff --git a/doc/protocols.texi b/doc/protocols.texi +index dbb1f166c7..26139fe3bd 100644 +--- a/doc/protocols.texi ++++ b/doc/protocols.texi +@@ -2098,6 +2098,14 @@ Default is -1. -1 means auto (off with 0 seconds in live mode, on with 180 + seconds in file mode). The range for this option is integers in the + 0 - @code{INT_MAX}. + ++For live outputs, a positive value also bounds orderly completion: wait for ++the sender buffer to empty, then keep the connection open for the negotiated ++peer delivery delay plus two measured round-trip times. This lets the peer ++deliver its final buffered packets before shutdown. The entire close drain ++must fit within @var{seconds}; expiration is reported as a timeout. Zero or ++an unset value preserves immediate live close. This changes only stream ++shutdown, not packet delivery latency during transmission. ++ + @item tsbpd=@var{1|0} + When true, use Timestamp-based Packet Delivery mode. The default behavior + depends on the transmission type: enabled in live mode, disabled in file +diff --git a/libavformat/libsrt.c b/libavformat/libsrt.c +index c1d93b2550..5e72d520b2 100644 +--- a/libavformat/libsrt.c ++++ b/libavformat/libsrt.c +@@ -608,16 +608,68 @@ static int libsrt_write(URLContext *h, const uint8_t *buf, int size) + return ret; + } + ++/* Positive linger opts a live publisher into a bounded, application-visible ++ * drain. ACKs clear the sender buffer before the peer's TSBPD delivery time; ++ * keep the connection open for its negotiated delay plus two RTTs afterwards. ++ * srt_close() alone is insufficient: SNDSYN is disabled and final cleanup may ++ * destroy asynchronous linger state. See Haivision/srt issue #2760. */ ++static int libsrt_drain_live_output(URLContext *h) ++{ ++ SRTContext *s = h->priv_data; ++ const int64_t start = av_gettime_relative(); ++ const int64_t deadline = start + (int64_t)s->linger * 1000000; ++ int64_t quiet_until = -1; ++ SRT_TRACEBSTATS stats = { 0 }; ++ size_t blocks = 0, bytes = 0; ++ ++ if (!(h->flags & AVIO_FLAG_WRITE) || s->linger <= 0 || ++ s->transtype == SRTT_FILE) ++ return 0; ++ ++ while (av_gettime_relative() < deadline) { ++ if (ff_check_interrupt(&h->interrupt_callback)) ++ return AVERROR_EXIT; ++ if (srt_getsndbuffer(s->fd, &blocks, &bytes) < 0) ++ return libsrt_neterrno(h); ++ if (!blocks) { ++ if (quiet_until < 0) { ++ double quiet_ms; ++ if (srt_bstats(s->fd, &stats, 0) < 0) ++ return libsrt_neterrno(h); ++ if (!stats.byteSentTotal) ++ return 0; ++ quiet_ms = FFMAX(stats.msSndTsbPdDelay, 0) + ++ 2.0 * FFMAX(stats.msRTT, 0.0); ++ /* Round up so sub-millisecond RTTs are never truncated. */ ++ quiet_until = av_gettime_relative() + (int64_t)(quiet_ms * 1000) + 1000; ++ } ++ if (av_gettime_relative() >= quiet_until) { ++ av_log(h, AV_LOG_VERBOSE, "SRT output drained in %"PRId64" us " ++ "(peer delay %d ms, RTT %.3f ms)\n", ++ av_gettime_relative() - start, stats.msSndTsbPdDelay, stats.msRTT); ++ return 0; ++ } ++ } else { ++ quiet_until = -1; ++ } ++ av_usleep(1000); ++ } ++ av_log(h, AV_LOG_ERROR, "SRT output drain exceeded linger=%d seconds " ++ "(%zu blocks, %zu bytes pending)\n", s->linger, blocks, bytes); ++ return AVERROR(ETIMEDOUT); ++} ++ + static int libsrt_close(URLContext *h) + { + SRTContext *s = h->priv_data; ++ int ret = libsrt_drain_live_output(h); + + srt_epoll_release(s->eid); + srt_close(s->fd); + + srt_cleanup(); + +- return 0; ++ return ret; + } + + static const AVClass libsrt_class = { +-- +2.55.0 + diff --git a/deps/ffmpeg/9/bases.env b/deps/ffmpeg/9/bases.env index 5c203691..593f80b3 100644 --- a/deps/ffmpeg/9/bases.env +++ b/deps/ffmpeg/9/bases.env @@ -1,6 +1,6 @@ upstream_ref=master-98e92563 upstream_commit=98e92563a3b60dbf6d370fd3491d7f896398e4c1 -upstream_tree=ab815f9f1ab4c1644e14192dd713b4f632a30ad9 -patch_count=18 +upstream_tree=51bce4909d067f70630fe970fa05e920a6302222 +patch_count=19 nv_codec_headers_tag=n13.1.15.0 nv_codec_headers_commit=0a6fba9a2820628b8103464f4c8753ee05838baa diff --git a/deps/ffmpeg/README.md b/deps/ffmpeg/README.md index 0a25a05f..3af39843 100644 --- a/deps/ffmpeg/README.md +++ b/deps/ffmpeg/README.md @@ -22,6 +22,13 @@ to fetch both pinned revisions; the default remains `n8.1`. ## Development-series differences +Live SRT outputs with explicit positive `linger` drain the sender buffer and +keep the connection open for the negotiated peer latency plus two measured +RTTs, within that total linger deadline. This protects the receiver's final +TSBPD-buffered packets; asynchronous socket linger alone does not. Timeout is +reported as an error. Unset/zero linger, SRT inputs and file-mode sockets retain +their previous close behavior. See the upstream [live shutdown discussion](https://github.com/Haivision/srt/issues/2760#issuecomment-1625033028). + All seven custom CUDA filters remain: `convert_cuda`, `crop_cuda`, `overlay_many_cuda`, `pad_cuda`, `transition_cuda` (including 10-bit and dip), `tonemap_cuda` and `band_blur_cuda`. The RFC 4175, V4L2, optional NDI registration diff --git a/tests/test_ffmpeg_patch_series.py b/tests/test_ffmpeg_patch_series.py index a76548a4..c7a599fd 100644 --- a/tests/test_ffmpeg_patch_series.py +++ b/tests/test_ffmpeg_patch_series.py @@ -1,15 +1,18 @@ """Guard the shared FFmpeg series and its GPU band-blur registration.""" from pathlib import Path +import pytest + ROOT = Path(__file__).resolve().parents[1] -def test_shared_ffmpeg_patch_manifest(): - series = ROOT / "deps/ffmpeg/8" +@pytest.mark.parametrize("version", ("8", "9")) +def test_shared_ffmpeg_patch_manifest(version): + series = ROOT / "deps/ffmpeg" / version manifest = dict(line.split("=", 1) for line in (series / "bases.env").read_text().splitlines() if "=" in line) assert len(list(series.glob("*.patch"))) == int(manifest["patch_count"]) - for base in ("n80", "n81"): + for base in (("n80", "n81") if version == "8" else ("upstream",)): for suffix in ("commit", "tree"): value = manifest[f"{base}_{suffix}"] assert len(value) == 40 diff --git a/tests/test_ffmpeg_srt_drain.py b/tests/test_ffmpeg_srt_drain.py new file mode 100644 index 00000000..b33de2f7 --- /dev/null +++ b/tests/test_ffmpeg_srt_drain.py @@ -0,0 +1,93 @@ +"""Execute the close-drain helper with a deterministic socket and clock.""" + +import shutil +import subprocess +from pathlib import Path + +import pytest + + +def test_live_srt_close_drain_contract(tmp_path): + cc = shutil.which("cc") + if not cc: + pytest.skip("C compiler unavailable") + patch = (Path(__file__).resolve().parents[1] / + "deps/ffmpeg/9/0019-avformat-libsrt-live-close-drain.patch").read_text() + additions = "\n".join(line[1:] for line in patch.splitlines() + if line.startswith("+") and not line.startswith("+++")) + helper = additions[additions.index("static int libsrt_drain_live_output"):] + helper = helper[:helper.index("\n}\n") + 3] + source = r''' +#include +#include +#include +#include +#include +#define AVIO_FLAG_WRITE 2 +#define SRTT_FILE 1 +#define AVERROR(e) (-(e)) +#define AVERROR_EXIT (-123) +#define AV_LOG_VERBOSE 0 +#define AV_LOG_ERROR 1 +#define FFMAX(a,b) ((a) > (b) ? (a) : (b)) +#define av_log(...) ((void)0) +typedef struct { int linger, fd, transtype; } SRTContext; +typedef struct { void *priv_data; int flags, interrupt_callback; } URLContext; +typedef struct { uint64_t byteSentTotal; int msSndTsbPdDelay; double msRTT; } SRT_TRACEBSTATS; +static int64_t now, pending_until; +static int interrupted, socket_error, calls, delay_ms = 200; +static int64_t av_gettime_relative(void) { return now; } +static int av_usleep(unsigned us) { now += us; return 0; } +static int ff_check_interrupt(void *cb) { return interrupted; } +static int libsrt_neterrno(URLContext *h) { return -EIO; } +static int srt_getsndbuffer(int fd, size_t *blocks, size_t *bytes) { + ++calls; + if (socket_error) return -1; + *blocks = now < pending_until ? 1 : 0; + *bytes = *blocks * 1316; + return 0; +} +static int srt_bstats(int fd, SRT_TRACEBSTATS *s, int clear) { + s->byteSentTotal = 1316; + s->msSndTsbPdDelay = delay_ms; + s->msRTT = 0.5; + return 0; +} +''' + source += helper + source += r''' +int main(void) { + SRTContext s = {.linger=2, .fd=5, .transtype=0}; + URLContext h = {.priv_data=&s, .flags=AVIO_FLAG_WRITE}; + pending_until = 5000; + assert(libsrt_drain_live_output(&h) == 0); + /* Last ACK at 5ms is not enough: retain the connection for the peer's + 200ms TSBPD delay + two 0.5ms RTTs + rounding guard afterwards. */ + assert(now >= 206000 && now < 209000); + + now = calls = 0; s.linger = 0; + assert(libsrt_drain_live_output(&h) == 0 && calls == 0 && now == 0); + s.linger = -1; + assert(libsrt_drain_live_output(&h) == 0 && calls == 0); + s.linger = 2; h.flags = 0; + assert(libsrt_drain_live_output(&h) == 0 && calls == 0); + h.flags = AVIO_FLAG_WRITE; s.transtype = SRTT_FILE; + assert(libsrt_drain_live_output(&h) == 0 && calls == 0); + + s.transtype = 0; pending_until = 3000000; + assert(libsrt_drain_live_output(&h) == -ETIMEDOUT && now == 2000000); + now = 0; pending_until = 0; delay_ms = 3000; + assert(libsrt_drain_live_output(&h) == -ETIMEDOUT && now == 2000000); + now = 0; delay_ms = 200; interrupted = 1; + assert(libsrt_drain_live_output(&h) == AVERROR_EXIT && now == 0); + interrupted = 0; socket_error = 1; + assert(libsrt_drain_live_output(&h) == -EIO && now == 0); + return 0; +} +''' + path = tmp_path / "drain.c" + path.write_text(source) + binary = tmp_path / "drain" + subprocess.run([cc, "-std=c11", str(path), "-o", str(binary)], check=True, + capture_output=True, text=True) + subprocess.run([str(binary)], check=True, timeout=5) From b50c17b0518bd9e854cdd32cf891a1252f095200 Mon Sep 17 00:00:00 2001 From: Jan Pietek Date: Fri, 9 Oct 2026 14:40:39 +0200 Subject: [PATCH 3/4] Drain live input PES and parser packets on explicit EOF --- doc/NODES.md | 9 +++ src/nodes/input_rec.cpp | 48 +++++++++-- tests/python/smoke_input_eof_drain.py | 111 ++++++++++++++++++++++++++ 3 files changed, 162 insertions(+), 6 deletions(-) create mode 100644 tests/python/smoke_input_eof_drain.py diff --git a/doc/NODES.md b/doc/NODES.md index ddc0fcf7..644d57cd 100644 --- a/doc/NODES.md +++ b/doc/NODES.md @@ -33,6 +33,15 @@ It has the same parameters as the `input` node and some additional - `preseek` (float, seconds, default 0) - how many seconds to preseek back to increase chance of finding a keyframe, when seek to timestamp (`seek` command) is requested - `team` (string, name of instance-shared object) - if specified, seeks on the `input` nodes may be synchronized inside specified team +Runtime control: `node.object.set request-eof true` interrupts further +input reads while keeping the producer alive to drain buffered demux/parser +packets. It then sends one common packet EOF and finishes (respecting +`stop_delay`), even if looping or ordinary EOF notification was disabled. +This is irreversible and intended for a running forward input whose live +publisher has finished but whose relay keeps the connection open. It preserves +normal downstream backpressure; the caller must keep consumers running and +enforce a shutdown deadline. `node.stop` remains an immediate abort. + ### `realtime` Rate limit output packets/frames to wallclock. This way, DTS (in diff --git a/src/nodes/input_rec.cpp b/src/nodes/input_rec.cpp index 42e7a18d..4d24323f 100644 --- a/src/nodes/input_rec.cpp +++ b/src/nodes/input_rec.cpp @@ -36,6 +36,7 @@ class RecordingInput: public NodeSingleOutput, public IStreamsInput, protected: av::FormatContext ictx_; std::atomic_bool should_end_ {false}; + std::atomic_bool eof_requested_ {false}; bool input_url_set_ = false; AVTS wait_start_; AVTS wait_max_ = AV_NOPTS_VALUE; @@ -92,6 +93,31 @@ class RecordingInput: public NodeSingleOutput, public IStreamsInput, av::Timestamp shift_ = NOTS; Parameters streams_object_, programs_object_; + av::Packet readPacket() { + std::error_code error; + av::Packet packet = ictx_.readPacket(error); + if (!error) { + return packet; + } + if (!eof_requested_ || error != std::error_code(AVERROR_EXIT, av::ffmpeg_category())) { + throw std::system_error(error); + } + if (packet.isNull()) { + // av_read_frame has exhausted its PES and codec-parser buffers. + return packet; + } + // avcpp reports pb->error even when av_read_frame successfully returns + // a buffered packet, before setting its time base/completeness. During + // explicit EOF drain, retain that packet and finish avcpp's bookkeeping. + // This context does not enable avcpp's subtract-start-time option. + if (ictx_.raw()->pb && ictx_.raw()->pb->error == AVERROR_EXIT && + packet.streamIndex() >= 0 && size_t(packet.streamIndex()) < ictx_.streamsCount()) { + packet.setTimeBase(ictx_.stream(packet.streamIndex()).timeBase()); + packet.setComplete(true); + return packet; + } + throw std::system_error(error); + } void closeInput(bool warn = true) { try { ictx_.close(); @@ -400,6 +426,11 @@ class RecordingInput: public NodeSingleOutput, public IStreamsInput, public: RecordingInput(std::unique_ptr> &&sink): NodeSingleOutput(std::move(sink)) { ictx_.setInterruptCallback([this]() -> int { + if (eof_requested_) { + // Interrupt further reads, not the processing loop: demuxers + // can still return their final PES/parser packets after this. + return 1; + } if ( (wait_max_!=AV_NOPTS_VALUE) && ((wallclock.pts() - wait_start_) > wait_max_) ) { logstream << "Timeout " << wait_max_ << " exceeded"; this->finished_ = true; @@ -669,7 +700,7 @@ class RecordingInput: public NodeSingleOutput, public IStreamsInput, virtual void process() { if (notify_eof_) { this->sink_->put(createEofPacket(video_stream_)); - if (stop_on_eof_) { + if (stop_on_eof_ || eof_requested_) { doStop(); eof_sent_ = false; } @@ -768,7 +799,7 @@ class RecordingInput: public NodeSingleOutput, public IStreamsInput, } } wait_start_ = wallclock.pts(); - av::Packet pkt = ictx_.readPacket(); + av::Packet pkt = readPacket(); if (play_direction_ == IPlaybackControl::EPlaybackDirection::pd_Backward) { // read only video frames, discard all other frames @@ -777,11 +808,11 @@ class RecordingInput: public NodeSingleOutput, public IStreamsInput, while (!pkt.isNull() && rep--) { if (pkt.streamIndex() == video_stream_) break; - pkt = ictx_.readPacket(); + pkt = readPacket(); } } - if (pkt.isNull() && loop_ && (play_direction_ == EPlaybackDirection::pd_Forward)) { + if (pkt.isNull() && loop_ && !eof_requested_ && (play_direction_ == EPlaybackDirection::pd_Forward)) { loopForward(); return; } @@ -789,7 +820,7 @@ class RecordingInput: public NodeSingleOutput, public IStreamsInput, if (!eof_sent_) { // we are at the end os recording logstream << "end of video reached"; - notify_eof_ = send_eof_; + notify_eof_ = send_eof_ || eof_requested_; eof_sent_ = true; } std::this_thread::sleep_for(5ms); @@ -1234,7 +1265,12 @@ class RecordingInput: public NodeSingleOutput, public IStreamsInput, } } virtual void setObject(const std::string name, const Parameters& p) { - if (name == "stream-limits") { + if (name == "request-eof") { + if (!p.is_boolean() || !p.get()) { + throw Error("request-eof expects true; EOF cannot be cancelled"); + } + eof_requested_ = true; + } else if (name == "stream-limits") { if (p.count("start") > 0) { if (p["start"].is_null()) { start_ts_ = StreamTarget(); diff --git a/tests/python/smoke_input_eof_drain.py b/tests/python/smoke_input_eof_drain.py new file mode 100644 index 00000000..5c2fb98c --- /dev/null +++ b/tests/python/smoke_input_eof_drain.py @@ -0,0 +1,111 @@ +"""Compare MPEG-TS file EOF with explicit EOF on a still-open TCP publisher. + +Run remotely with the native module under an external timeout. Supply a short +H.264/AAC TS fixture: this test does no encoding and needs no GPU. It compares +every elementary packet, including those held in the final PES/parser buffers. +""" + +import argparse +import hashlib +import socket +import threading +import time +from collections import defaultdict +from pathlib import Path + +from pyplumber import AVPlumber +from pyplumber.node import InputRec + + +def capture(url, sent=None): + avp = AVPlumber() + errors = [] + avp.on_exception = lambda *args: errors.append(args) + packets = defaultdict(list) + avp.executeCommandsFromString("queue.plan_capacity packets 2\n") + avp.addNode(InputRec({"name": "input", "group": "test", "dst": "packets", + "url": url, "format": "mpegts", "timeout": 10, + "options": {"analyzeduration": "250000", "probesize": "32768"}})) + eof_count = 0 + requested = False + before_drain = None + try: + avp.group("test").startNodes() + edge = avp.getEdge("packets", "Packet") + # Exercise normal bounded-queue backpressure before releasing consumers. + time.sleep(0.05) + deadline = time.monotonic() + 15 + last_packet = time.monotonic() + while avp.node("input").isWorking or edge.occupied: + assert time.monotonic() < deadline, "Input did not finish bounded drain" + assert not errors, errors + packet = edge.tryGet(20) + if packet is not None: + last_packet = time.monotonic() + if packet.size == 1 and packet.data == b"\xff": + eof_count += 1 + else: + assert eof_count == 0, "Media appeared after EOF" + packets[packet.stream_index].append(( + packet.pts.timestamp, packet.dts.timestamp, + packet.size, hashlib.sha256(packet.data).hexdigest())) + if (sent is not None and sent.is_set() and not requested + and time.monotonic() - last_packet > 0.2): + # The publisher is still connected. The input has consumed all + # available bytes, but cannot flush its final PES/parser yet. + before_drain = sum(map(len, packets.values())) + avp.executeCommandsFromString("node.object.set input request-eof true\n") + # The command is idempotent while drain remains in progress. + avp.executeCommandsFromString("node.object.set input request-eof true\n") + requested = True + assert not errors, errors + assert eof_count == 1, f"Expected exactly one EOF, got {eof_count}" + if sent is not None: + assert requested + assert before_drain < sum(map(len, packets.values())), "Fixture did not exercise parser tail" + return dict(packets) + finally: + avp.shutdown() + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("input", type=Path) + args = parser.parse_args() + expected = capture(str(args.input.resolve())) + assert len(expected) >= 2, "Fixture must contain audio and video" + sent, release = threading.Event(), threading.Event() + failures = [] + with socket.socket() as listener: + listener.bind(("127.0.0.1", 0)) + listener.listen(1) + + def publish(): + try: + with listener.accept()[0] as client: + with args.input.open("rb") as source: + while block := source.read(64 * 1024): + client.sendall(block) + sent.set() + assert release.wait(30), "Test did not release publisher" + except Exception as exc: + failures.append(exc) + + worker = threading.Thread(target=publish, daemon=True) + worker.start() + try: + actual = capture(f"tcp://127.0.0.1:{listener.getsockname()[1]}", sent) + assert actual == expected, { + "file_packets": {key: len(value) for key, value in expected.items()}, + "live_packets": {key: len(value) for key, value in actual.items()}, + } + finally: + release.set() + worker.join(5) + assert not failures, failures + print("Live input parser drain matches file EOF:", + {key: len(value) for key, value in expected.items()}, flush=True) + + +if __name__ == "__main__": + main() From 13960f3438fe13da8ae5156cc2725bf768c2ac37 Mon Sep 17 00:00:00 2001 From: Jan Pietek Date: Fri, 9 Oct 2026 14:56:06 +0200 Subject: [PATCH 4/4] Wait for input startup and probe audio in live EOF regression --- tests/python/smoke_input_eof_drain.py | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/tests/python/smoke_input_eof_drain.py b/tests/python/smoke_input_eof_drain.py index 5c2fb98c..c2714896 100644 --- a/tests/python/smoke_input_eof_drain.py +++ b/tests/python/smoke_input_eof_drain.py @@ -25,13 +25,18 @@ def capture(url, sent=None): avp.executeCommandsFromString("queue.plan_capacity packets 2\n") avp.addNode(InputRec({"name": "input", "group": "test", "dst": "packets", "url": url, "format": "mpegts", "timeout": 10, - "options": {"analyzeduration": "250000", "probesize": "32768"}})) + "options": {"analyzeduration": "1000000", "probesize": "4194304"}})) eof_count = 0 requested = False before_drain = None try: avp.group("test").startNodes() edge = avp.getEdge("packets", "Packet") + deadline = time.monotonic() + 10 + while not avp.node("input").isWorking: + assert not errors, errors + assert time.monotonic() < deadline, "Input did not start" + time.sleep(0.01) # Exercise normal bounded-queue backpressure before releasing consumers. time.sleep(0.05) deadline = time.monotonic() + 15