Add CTE LL-DASH ingest publisher - #15
Conversation
| #endif | ||
| } | ||
|
|
||
| void LiveDashIngestServer::stop() { |
There was a problem hiding this comment.
Noticed that stop() shuts down the listen socket and then joins all worker threads. In a scenario (e.g. a stalled encoder, interrupted network connection etc. ) where worker threads may still be blocked in recv() on already accepted client sockets, this could cause shutdown to hang if a client connects but does not finish sending the request/body.
One possible consideration could be to track active client sockets and shut them down during stop(), or use receive timeouts / non-blocking IO so worker threads can exit cleanly.
There was a problem hiding this comment.
Fixed in dcb2a4b. LiveDashIngestServer::stop() now tracks active accepted client sockets and shuts them down before joining worker threads, so a stalled chunked request blocked in recv() no longer hangs shutdown. Added a regression test that opens a partial chunked PUT, calls stop(), and verifies it returns without waiting for the client to finish the body.
There was a problem hiding this comment.
Follow-up in 6c08575: this shutdown path is hardened further against two fd-reuse races that remained after dcb2a4b.
close_client()now erases the client fd fromactive_client_fdsbefore::close(). Previously it closed first, so the accept loop could reuse the same fd number for a new client, insert it into the set, and then have the finishing worker'serase()drop that new client's registration — leavingstop()unable to::shutdown()it and hanging the join (the exact stalled-recv hang you described).stop()now keeps the listen fd valid (shutdown, not closed) until after the accept thread joins, then closes it. Previously it closed the listen fd before joining, so the number could be reused whileaccept_loopwas still calling::accept()on it.
The stalled-encoder case you flagged is covered by the stalled client stop case in openmoq-publisher-live-dash-tests, which shuts down active client sockets in stop() to unblock the blocked recv().
| close_fd(client_fd); | ||
| break; | ||
| } | ||
| worker_threads.emplace_back([this, client_fd]() { |
There was a problem hiding this comment.
Noticed that each accepted connection appends a new worker thread, and completed workers seem to be joined only during stop(). In a long-running ingest server with many reconnects/requests, it could cause worker_threads to keep growing for the lifetime of the process
One possible consideration could be to periodically reap completed workers.
There was a problem hiding this comment.
Fixed in dcb2a4b. Completed workers now mark themselves done, and the accept loop reaps/join-removes completed worker records before adding another accepted connection. stop() still joins any remaining active workers after shutting down the listen socket and active clients.
There was a problem hiding this comment.
The opportunistic reaping added in dcb2a4b (reap_finished_workers_locked() on each accept) is unchanged by 6c08575, so the worker_threads vector no longer grows unbounded across reconnects. 6c08575 only touches the per-client fd bookkeeping and the listen-fd close ordering in stop(), not the reaping loop.
|
I've run it: |
|
Follow-up for @piersoh's FFmpeg usage: dcb2a4b adds a live DASH ingest regression test based on that shape. The test uses FFmpeg-style representation paths under the ingest prefix ( I also probed the minimized FFmpeg command locally and confirmed that the DASH muxer writes |
Resolve the confirmed findings from the PR #15 review of the CTE LL-DASH ingest publisher: - Guard the ingest worker thread so a malformed fragment (e.g. a moof referencing an unknown track) is dropped instead of escaping the thread and calling std::terminate. - Embed each track's base64 CMAF init segment in the catalog via a reusable track_init_data_base64 helper so subscribers can initialize decoders. - Freeze the announced track set when source() is taken; paths whose init segments arrive afterwards are ignored rather than enqueued against an unknown track (which aborted the whole publish session). - Add a bounded poll plus an is_finished predicate on LiveObjectSource so the publisher services control messages during media gaps and ends only on close, instead of blocking indefinitely. - Give each catalog emission a monotonic group id so a re-published catalog no longer collides with an already-delivered object id. - Erase a client fd from the active set before closing it, and defer closing the listen fd until after the accept thread joins, to close two fd-reuse races on shutdown. - Reject trailing garbage in --dash-listen/--dash-queue-depth with clear messages and require --dash-path/--dash-queue-depth to accompany --live-source dash. Add regression tests for crash-safety, catalog initData, late-path rejection, and the CLI validation.
|
Follow-up hardening in 6c08575, from a correctness/reuse pass over the DASH ingest path. Beyond the shutdown fd-reuse races (replied inline above), it resolves:
Regression tests added for crash-safety, catalog initData, late-path rejection, and the CLI validation. Full suite (10 targets) passes and |
There was a problem hiding this comment.
Pull request overview
This PR adds a new live ingest mode for the publisher: a CTE LL-DASH / CMAF HTTP/1.1 chunked PUT/POST ingest listener that can accept multiple concurrent ingest paths and publish the resulting live objects via the existing live object API.
Changes:
- Added
LiveDashIngestServer/LiveDashIngestSession(with an incremental chunked-transfer decoder) to ingest CMAF and emitLiveObjects + a catalog. - Extended CLI/main wiring to support
--live-source dashwith--dash-listen,--dash-path, and--dash-queue-depth. - Updated live publishing loop to support “transient gaps” via a new
LiveObjectSource::is_finishedpredicate, plus added docs/tests/CMake integration.
Reviewed changes
Copilot reviewed 16 out of 16 changed files in this pull request and generated 3 comments.
Show a summary per file
| File | Description |
|---|---|
| tests/live_dash_ingest_test.cpp | New tests for chunked decoding, session behavior, concurrent ingest paths, and server stop behavior. |
| tests/cli_options_test.cpp | New CLI parsing/validation tests for DASH flags and error messages. |
| src/transport/moqt_session.cpp | Adds support for live sources that can report “not finished yet” during temporary object gaps. |
| src/main.cpp | Wires --live-source dash to start the ingest server and publish from its LiveObjectSource. |
| src/live_dash_ingest.cpp | Implements HTTP/1.1 chunked ingest server, decoder, per-path MP4 parsing, and live object/catalog production. |
| src/cmsf_packager.cpp | Exposes track_init_data_base64() helper for embedding per-track init segments in catalogs. |
| src/cli_options.cpp | Adds DASH live-source parsing, strict integer parsing, host:port parsing, and validations for DASH flags. |
| README.md | Documents the new live DASH ingest mode with curl/FFmpeg examples and platform notes. |
| include/openmoq/publisher/live_object.h | Extends LiveObjectSource with is_finished predicate to distinguish gaps vs EOF. |
| include/openmoq/publisher/live_dash_ingest.h | New public header for DASH ingest config/session/server and chunked decoder. |
| include/openmoq/publisher/cmsf_packager.h | Declares track_init_data_base64() for reuse by live sources assembling catalogs. |
| include/openmoq/publisher/cli_options.h | Adds LiveSourceKind::kDash and new DASH-related CLI option fields. |
| docs/quickstart.md | Adds a quickstart section for DASH ingest usage and operational guidance. |
| docs/ffmpeg.md | Adds FFmpeg guidance for pushing live DASH/CMAF over HTTP chunked ingest. |
| docs/ctedash-implementation-plan.md | Adds implementation plan document for the feature. |
| CMakeLists.txt | Builds src/live_dash_ingest.cpp and registers new DASH test executable. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>
Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>
Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>
shutdown() on a listening socket does not wake a blocked accept() on macOS/BSD (it fails with ENOTCONN), so once stop() started keeping the listener fd open until the accept thread joins, macOS CI hung in the first server stop and the live-dash test binary hit the ctest timeout. Gate accept() behind a 100ms poll() that re-checks stop_requested, matching the client workers' receive cadence, and make the listener non-blocking so a connection aborted between poll() and accept() cannot park the loop. Accepted sockets inherit O_NONBLOCK on BSD-derived systems, so handle_client clears it before applying SO_RCVTIMEO.
publish_live parked its stdin reader thread in istream::read on std::cin, which blocks until a full 16 KiB chunk or EOF. Every error-path join then depended on the feeder closing stdin: a relay drop with a live-but-idle feeder hung the publisher - the same unwakeable-blocking-call shape as the DASH accept() hang fixed in 65c5a4f. Route all stdin consumption in publish_live (both the ftyp+moov discovery phase and the reader thread) through one helper that, for real stdin on POSIX, reads fd 0 directly behind a bounded 100ms poll and re-checks a stop flag that joins now set first. Reading the fd raw in only one phase would lose bytes read ahead into cin's stdio buffer, so both phases share the helper and the istream is never used for cin. Non-cin istreams (unit tests) and Windows keep the original blocking read.
Correct setup option parsing, request-stream polling and framing, and draft-17/18 SUBSCRIBE_OK encoding. Add regressions and archive the draft-19 text for later review.
Summary
Test Plan