diff --git a/CLAUDE.md b/CLAUDE.md index 291351a..bfa6507 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -59,8 +59,8 @@ Single OBS MODULE shared library. All source is C11. audio: resample, write to jitter buffer video: keyframe gate, push decoded frame (PTS in ns) onto video queue -[video thread]: pop video queue, HW frame transfer, format conversion, - OBS async video output +[video thread]: pop video queue, HW frame transfer, hold until due, + format conversion, OBS async video output [audio thread]: drain jitter buffer, speed correction, concealment, OBS audio output @@ -85,7 +85,7 @@ Buffer regulation happens through playback speed only, asymmetric like IRLToolki - **`src/receiver-stream.c`**: stream open/close, demuxer options, reconnection, disconnect fade out, periodic stats logging. - **`src/receiver-decode.c`**: packet to decoder plumbing with corruption burst handling and throttled decoder flushes. - **`src/receiver-audio.c`**: the audio core. Intake side (receiver thread): PTS repair, resample to interleaved float, write to the PTS aware jitter buffer. Pre-keyframe audio is discarded (not staged) to avoid decoder warm-up artifacts. Output side (audio thread): sample counter output clock, constant rate submission, swr based speed correction, dropout concealment, hidden backlog trims. -- **`src/receiver-video.c`**: decoded video frame handling, keyframe gate, resolution change detection. Also owns `irl_video_request_clear`: the receiver thread drops the queue and raises a flag, and the *video* thread is what actually calls `obs_source_output_video(source, NULL)`. Clearing from the receiver thread instead would race a frame already inside the format conversion, which would repaint the frozen frame right after the clear. +- **`src/receiver-video.c`**: decoded video frame handling, keyframe gate, resolution change detection, and the video output pacing loop. Frames are copied out of the hardware pool as soon as they arrive (which returns the decoder's surface) and then held in a video-thread-private pacing queue until their mapped timestamp is due, the way OBS's own media source paces in `mp_media_sleep`. This is what keeps libobs's async queue about one frame deep: handing it a frame early makes it hold that frame, and past `MAX_ASYNC_FRAMES` (30) held frames `cache_video` silently discards the entire queue. Also owns `irl_video_request_clear`: the receiver thread drops the queue and raises a flag, and the *video* thread is what actually calls `obs_source_output_video(source, NULL)`. Clearing from the receiver thread instead would race a frame already inside the format conversion, which would repaint the frozen frame right after the clear. - **`src/audio-buffer.c`**: thread safe ring buffer sized in milliseconds with a parallel PTS chunk queue. Mutex protected. Supports fade-out reads. - **`src/video-handler.c`**: converts AVFrames to OBS video. Maps pixel formats (I420, NV12, I010, P010, etc.), handles HW frame transfer, falls back to swscale for unsupported formats. Maps video PTS through the audio playout offset for lip sync. - **`src/pts-repair.c`**: three tier PTS discontinuity repair. Small gaps interpolated, medium gaps get silence, large gaps trigger full reset. @@ -103,7 +103,7 @@ Buffer regulation happens through playback speed only, asymmetric like IRLToolki - **Main/OBS thread**: calls create, destroy, update, tick, get_properties, and the activate/deactivate/show/hide callbacks (used only when "Close Stream When Inactive" is on) - **Receiver thread**: owns demux/decode FFmpeg state. Writes to the audio buffer (mutex protected) and pushes decoded video frames (PTS pre-converted to nanoseconds) onto the video queue. Never blocks on GPU or OBS video delivery. -- **Video thread**: pops the video queue, does the HW frame transfer and format conversion (owns sws_ctx), and calls `obs_source_output_video`. Queue overflow drops the oldest frame (`video_queue_drops`). +- **Video thread**: pops the video queue, does the HW frame transfer, paces each frame to its due time, then converts (owns sws_ctx) and calls `obs_source_output_video`. Queue overflow drops the oldest frame (`video_queue_drops`). The pacing queue it holds those frames in needs no lock — the receiver thread never touches it, and a clear is routed through `video_clear_pending` — but its counters are mirrored under `video_queue_lock` for the stats line. - **Audio thread**: drains the jitter buffer and submits audio to OBS via `obs_source_output_audio`, paced against the sample counter output clock. Shared timing state is protected by `audio_state_lock` (lock order: `audio_state_lock` before the buffer mutex). Config fields marked `/* hot */` in `struct irl_config` are written by `irl_source_update` while the worker threads run, so every cross-thread read goes through `os_atomic_load_long` / `os_atomic_load_bool` (not C11 `_Atomic`, which MSVC does not support without an experimental flag). The remaining fields are only written while the threads are stopped, where `irl_thread_create` and `irl_thread_join` supply the happens-before edge. diff --git a/README.md b/README.md index 0fbe129..748b072 100644 --- a/README.md +++ b/README.md @@ -348,6 +348,8 @@ Stats are exposed through OBS's `proc_handler` API under the `get_stats` call, a | `obs_lead_ms` | int | How far ahead of real time audio is queued inside OBS (healthy is roughly 60 to 100ms) | | `audio_decoder_flushes` | int | Number of audio decoder flushes after repeated decode errors | | `video_decoder_flushes` | int | Number of video decoder flushes after repeated decode errors | +| `video_lead_ms` | int | How far ahead of real time the last video frame was timestamped. Tracks the audio buffer; a value climbing well past Target Buffer and staying there means concealment has inflated the A/V mapping | +| `video_lead_excess` | int | Frames whose lead exceeded what OBS's async queue can absorb. Harmless while the lead is steady; sustained growth is what makes OBS drop queued video | | `stream_delay_ms` | int | End-to-end stream delay (SRT latency + decode + buffering) | | `low_latency_audio` | bool | Whether OBS async unbuffered low-latency mode is enabled | | `reconnect_count` | int | Number of reconnect attempts since the source was created | diff --git a/deps/build-deps.sh b/deps/build-deps.sh index 70ec620..99b8964 100755 --- a/deps/build-deps.sh +++ b/deps/build-deps.sh @@ -125,7 +125,12 @@ fetch() { fi echo "download: ${url}" - curl -fsSL --retry 3 --retry-delay 2 -o "${path}.tmp" "${url}" + # --retry alone only covers timeouts and 5xx; a refused or reset + # connection is not "transient" to curl and fails on the first try. + # ffmpeg.org does both often enough to have cost a CI run. + curl -fsSL --retry 5 --retry-delay 3 --retry-all-errors \ + --retry-connrefused --connect-timeout 30 \ + -o "${path}.tmp" "${url}" local got got="$(sha256_of "${path}.tmp")" @@ -272,15 +277,49 @@ build_zlib() { "zlib-${ZLIB_VERSION}.tar.gz" "${ZLIB_SHA256}" extract "zlib-${ZLIB_VERSION}.tar.gz" "zlib-${ZLIB_VERSION}" + # ZLIB_BUILD_TESTING is what 1.3.2 renamed ZLIB_BUILD_EXAMPLES to; both + # are passed so either version builds only the library. + # + # ZLIB_BUILD_SHARED/STATIC are zlib's own switches and both default ON; + # it does not honour BUILD_SHARED_LIBS from cmake_common. Left alone it + # installs z.dll plus an *import* z.lib alongside the static library, + # and since that import lib already occupies the name FFmpeg links + # against, ensure_msvc_lib_name below would accept it and quietly give + # the plugin a runtime DLL dependency the bundled stack exists to avoid. cmake -S "$(npath "${src}/zlib-${ZLIB_VERSION}")" \ -B "$(npath "${src}/zlib-${ZLIB_VERSION}/build")" \ "${cmake_common[@]}" \ + -DZLIB_BUILD_SHARED=OFF \ + -DZLIB_BUILD_STATIC=ON \ + -DZLIB_BUILD_TESTING=OFF \ -DZLIB_BUILD_EXAMPLES=OFF cmake --build "$(npath "${src}/zlib-${ZLIB_VERSION}/build")" --parallel "${jobs}" cmake --install "$(npath "${src}/zlib-${ZLIB_VERSION}/build")" - # zlib's CMake calls its static output zlibstatic; FFmpeg asks for z.lib. - ensure_msvc_lib_name z zlibstatic zlib + # Belt and braces: if a future zlib renames those switches the way 1.3.2 + # renamed its static target, fail here rather than link a DLL. + if [[ -f ${prefix}/bin/z.dll || -f ${prefix}/bin/zlib1.dll ]]; then + echo "zlib installed a DLL; the bundled stack must be static" >&2 + exit 1 + fi + + # FFmpeg's MSVC flag translator hardcodes -lz to zlib.lib rather than + # z.lib like every other -l name: + # + # -lz) echo zlib.lib ;; + # -l*) echo ${flag#-l}.lib ;; + # + # Under zlib 1.3.1 that name existed by accident, as the *shared* + # import library (1.3.1 named the DLL target zlib; 1.3.2 renamed it to + # z). So the Windows build has been linking zlib dynamically all along, + # and turning the DLL off is what finally made the missing name visible + # as LNK1181: cannot open input file 'zlib.lib'. + # + # Provide both spellings from the static archive: zlib.lib is what + # FFmpeg links, z.lib is what the generic -l handling in the CMake + # description below resolves. + ensure_msvc_lib_name z zs zlibstatic zlib + ensure_msvc_lib_name zlib zs zlibstatic z # zconf.h.cmakein still carries an autoconf-era block: # @@ -604,9 +643,24 @@ build_ffmpeg() { # the only thing that distinguishes a missing library from one whose # name or link order the toolchain got wrong. if ! (cd "${ff}" && ./configure "${args[@]}"); then + local cfglog="${ff}/ffbuild/config.log" echo + # A tail alone is not enough. Autodetected libraries are probed + # early and configure only dies about them in a sweep at the very + # end ("$lib requested but not found"), so by the time it fails + # the probe that actually explains it is a thousand lines above + # the tail and nothing in the visible output names a cause. + echo "---- probes for the libraries we require ----" >&2 + local l + for l in zlib mbedtls libsrt librist ffnvcodec; do + echo "== ${l} ==" >&2 + grep -n -B2 -A25 \ + -e "check_pkg_config ${l} " \ + -e "check_lib ${l} " \ + "${cfglog}" >&2 || echo "(no probe logged)" >&2 + done echo "---- tail of ffbuild/config.log ----" >&2 - tail -60 "${ff}/ffbuild/config.log" >&2 || true + tail -60 "${cfglog}" >&2 || true exit 1 fi diff --git a/deps/versions.env b/deps/versions.env index e196645..5c0f688 100644 --- a/deps/versions.env +++ b/deps/versions.env @@ -6,8 +6,8 @@ # Bumping a version here also requires updating the matching SHA256. # Windows only. Linux and macOS use the system zlib; Windows has none. -ZLIB_VERSION=1.3.1 -ZLIB_SHA256=9a93b2b7dfdac77ceba5a558a580e74667dd6fede4585b91eefb60f03b72df23 +ZLIB_VERSION=1.3.2 +ZLIB_SHA256=bb329a0a2cd0274d05519d61c667c062e06990d72e125ee2dfa8de64f0119d16 FFMPEG_VERSION=9.0 FFMPEG_SHA256=7f607a00dd0d28a729d5a4811205812eef01cf6ef6155025febb6f36a9062d52 @@ -17,14 +17,14 @@ FFMPEG_SHA256=7f607a00dd0d28a729d5a4811205812eef01cf6ef6155025febb6f36a9062d52 SRT_VERSION=1.5.6 SRT_SHA256=2c4980c2c4cfd142d21b829d939dc51db9c6628af5967fff62fd7290769569c7 -MBEDTLS_VERSION=3.6.4 -MBEDTLS_SHA256=ec35b18a6c593cf98c3e30db8b98ff93e8940a8c4e690e66b41dfc011d678110 +MBEDTLS_VERSION=3.6.7 +MBEDTLS_SHA256=a7e8bcbec0e6f761b4af24f25677626b35f762f68eef79c08677a363212d11f6 # RIST ingest. obs-deps pins 0.2.7; this is the same story as libsrt. -LIBRIST_VERSION=0.2.18 -LIBRIST_SHA256=9a2d16dcdb9fb067b7ba4259a3976ff6f8df9a62dbec7f32f19a0b60ec0c114a +LIBRIST_VERSION=0.2.20 +LIBRIST_SHA256=9e40eeb87f014790531ad41326cc271b930a65962e4b15231b301fc59b29fe31 # Headers only. FFmpeg loads nvcuda/nvcuvid at runtime, so this adds no # build-time or load-time dependency on a CUDA install. -NVCODEC_VERSION=13.0.19.0 -NVCODEC_SHA256=86d15d1a7c0ac73a0eafdfc57bebfeba7da8264595bf531cf4d8db1c22940116 +NVCODEC_VERSION=13.1.15.0 +NVCODEC_SHA256=2255bc74d038b95aa4be30f5f66322c2176acbdb90ada1851db6993536fbeaf7 diff --git a/include/irl-source.h b/include/irl-source.h index a591173..a54c52a 100644 --- a/include/irl-source.h +++ b/include/irl-source.h @@ -90,12 +90,91 @@ struct irl_source; * buffer capacity (4x buffer_max_ms) or writes would drop old data. */ #define IRL_BLEED_PACE_FILL_MS 1000 +/* Concealment inflates the audio->OBS playout offset with no bounded + * recovery once primed (see irl_audio_maybe_reanchor_offset). This far + * past the primed baseline the accumulated latency is treated as + * unrecoverable by the speed-drain and reclaimed with one declared + * re-anchor. Set above the worst normal buffer swing (buffer_max is + * only ~200ms over target) so ordinary adaptive-speed excursions + * never trip it; only a real outage's worth of concealment does. + * + * Lives here rather than next to its use in receiver-audio.c because the + * video lead threshold below is expressed in terms of it. */ +#define AUDIO_OFFSET_REANCHOR_MARGIN_MS 400 + +/* Reporting threshold for the video output lead. + * + * libobs schedules async video itself: obs_source_output_video() queues the + * frame and ready_async_frame() releases it once the queue's play head + * (last_frame_ts, which advances at wall-clock rate) reaches its timestamp. + * At MAX_ASYNC_FRAMES (30) queued frames cache_video() drops the incoming + * frame, throws the whole queue away and resets last_frame_ts, silently. + * + * What lands in that queue is the *growth* in lead since the play head last + * anchored, not the lead itself — a large steady lead queues nothing. This + * threshold is therefore a reporting aid, not a limit that gets enforced: an + * earlier version clamped the lead against it and, because it compared the + * absolute lead to the configured target rather than measuring growth, held + * video half a second ahead of audio on a stream whose lead was merely large + * and steady. + * + * The budget is expressed in frames because that is what libobs counts: the + * same 400ms is 12 frames at 30fps and 48 at 120fps. Floored at the audio + * re-anchor margin, below which a lead is still within what concealment can + * legitimately have added. + */ +#define IRL_OBS_ASYNC_FRAME_BUDGET 24 +#define IRL_VIDEO_LEAD_WARN_INTERVAL_NS 10000000000ULL + +/* Bounds on the measured frame interval (250fps..10fps) and the estimate + * used before enough frames have arrived to measure one. */ +#define IRL_VIDEO_INTERVAL_MIN_NS 4000000LL +#define IRL_VIDEO_INTERVAL_MAX_NS 100000000LL +#define IRL_VIDEO_INTERVAL_DEFAULT_NS 33333333LL + +/* Video output pacing. + * + * Video PTS is mapped through the audio playout offset for lip sync, which + * puts each frame's correct display moment roughly one audio-buffer ahead of + * now. Handing libobs a frame that early makes libobs hold it, and libobs + * throws its whole async queue away past 30 held frames. So the plugin keeps + * the frame itself and hands it over when it is due, the way OBS's own media + * source paces (mp_media_sleep in shared/media-playback). libobs then holds + * about one frame, and its 30-frame limit stops being reachable at any lead + * or frame rate. + * + * Holding the frames is the cost: an N-millisecond lead means N milliseconds + * of decoded video in memory, which is unavoidable — libobs was storing the + * same thing, just capped and silently discarded. Bounded two ways, since + * frame count alone means nothing across resolutions: whichever of the frame + * and byte ceilings binds first. Past either, frames are emitted before they + * are due, which is exactly the old behaviour, and counted so it is visible. + */ +#define IRL_VIDEO_PACING_MAX_FRAMES 512 +#define IRL_VIDEO_PACING_MAX_BYTES (192u * 1024u * 1024u) +/* Emit rather than sleep again when this close to due: another wakeup costs + * more than the timing error it would remove. */ +#define IRL_VIDEO_PACING_SLACK_NS 1000000LL +/* Ceiling on a single pacing sleep, so a clear or a shutdown is never left + * waiting on a frame that is due far in the future. */ +#define IRL_VIDEO_PACING_MAX_WAIT_MS 50 + /* Abort a blocking read/connect through the FFmpeg interrupt callback * after this long without progress. A dead-but-open connection (uplink * loss in a dead zone) otherwise hangs av_read_frame forever with no * reconnect. Connect plus stream probe normally completes in under 3s. */ #define IRL_IO_STALL_TIMEOUT_US 10000000ULL +/* One frame waiting for its moment. `due_ns` is the OBS-clock timestamp the + * PTS mapping produced, sampled once when the frame was decoded — the same + * sampling point the un-paced path used — so pacing does not change what + * timestamp a frame gets, only when it is handed over. */ +struct irl_pacing_frame { + AVFrame *frame; + uint64_t due_ns; + size_t bytes; +}; + /* ── Source configuration ─────────────────────────────────── */ /* Fields marked hot are swapped in place by irl_source_update() while the @@ -150,8 +229,9 @@ struct irl_source { * Decouples the GPU→CPU frame transfer and format conversion * from the receiver thread so a GPU stall cannot starve audio * decode. Depth stays small because queued HW frames pin - * decoder surface-pool entries (matched by extra_hw_frames at - * decoder open). Queued frame->pts is in nanoseconds; the + * decoder surface-pool entries (covered by extra_hw_frames at + * decoder open, which budgets this queue plus the two frames + * in flight around it). Queued frame->pts is in nanoseconds; the * receiver converts before queueing because it may close * fmt_ctx while frames are still in flight. */ #define IRL_VIDEO_QUEUE_SIZE 4 @@ -161,6 +241,40 @@ struct irl_source { int video_queue_head; int video_queue_count; uint64_t video_queue_drops; + /* Decoder surfaces this plugin pins at once, for checking the + * extra_hw_frames budget against reality rather than against a + * reading of the code: frames sitting in the queue plus the one the + * video thread has popped and is converting. The frame the decoder + * has just handed the receiver thread is not counted (it lives on + * the other thread and is unref'd immediately), so the pool + * requirement is this peak plus one. Cumulative for the source — + * a two-hour stream's high-water mark is the interesting number. + * Both guarded by video_queue_lock. */ + int video_in_flight; + int video_pinned_peak; + + /* Pacing queue: decoded frames in system memory, waiting for their + * due time. Video-thread-private — the receiver thread never touches + * it, and a clear is routed through video_clear_pending, which the + * video thread consumes — so unlike video_queue above it needs no + * lock. Entries hold no decoder surfaces: irl_video_to_sysmem() + * copies out of the hardware pool precisely so this queue can be + * deep. video_pacing_* are read for stats without the lock, like + * video_queue_drops. */ + struct irl_pacing_frame pacing_queue[IRL_VIDEO_PACING_MAX_FRAMES]; + int pacing_head; + int pacing_count; + size_t pacing_bytes; + int pacing_peak; + uint64_t pacing_overflows; + + /* Published copies of the four above, mirrored under + * video_queue_lock once per pacing cycle so the stats line on the + * receiver thread has something synchronised to read. */ + int video_pacing_now; + int video_pacing_peak; + size_t video_pacing_bytes; + uint64_t video_pacing_overflows; /* Set by the receiver thread on disconnect, consumed by the video * thread. Guarded by video_queue_lock. The clear has to run on the * video thread so it cannot be undone by a frame that was already @@ -208,6 +322,11 @@ struct irl_source { bool video_ts_init; uint64_t video_sys_base; /* os_gettime_ns() at first frame */ int64_t video_pts_base; /* stream PTS at first frame (in ns) */ + /* Previous decoded PTS, receiver-thread-owned; feeds the frame + * interval EMA. */ + int64_t video_prev_pts_ns; + /* Throttle for the lead-cap warning, video-thread-owned. */ + uint64_t video_lead_warn_time_ns; /* Audio output clock. OBS timestamps are a pure sample * counter anchored once at prime time: @@ -252,6 +371,19 @@ struct irl_source { int64_t latest_audio_stream_pts_ns; int64_t latest_video_stream_pts_ns; + /* Video lead diagnostics, all guarded by audio_state_lock. + * + * video_frame_interval_ns is an EMA of decoded PTS deltas, written + * by the receiver thread and read by the video thread to estimate + * how many frames a given lead parks in the libobs async queue. + * video_lead_ns is the lead the PTS mapping asked for before the + * cap (the uncapped value is the diagnostic: it shows the ratchet), + * written by the video thread and read by the OBS thread. */ + int64_t video_frame_interval_ns; + int64_t video_lead_ns; + int64_t video_lead_peak_ns; + uint64_t video_lead_excess; + /* Latest audio already queued to OBS, in OBS clock domain. * Used to align video to actual audio playout instead of * approximating from the plugin-side jitter-buffer fill. */ @@ -297,6 +429,26 @@ struct irl_source { * not reset the decoder state (losing reference frames). */ int audio_decode_errors; int video_decode_errors; + /* Detection of a drain that cannot win: the audio thread owns these. */ + uint64_t audio_drain_stuck_since_us; + int audio_drain_stuck_fill_ms; + uint64_t audio_drain_warn_time_us; + /* Jitter-buffer high-water mark, same sampling argument as + * video_lead_peak_ns: the backlog excursion that drives everything + * here is transient, and `buf` at log time usually misses it. */ + int audio_fill_peak_ms; + /* avcodec_send_packet() returned EAGAIN, meaning the decoder did not + * accept the packet and it must be resent after draining output. + * Rare when the frame pool is adequately sized, which is exactly why + * a non-zero count is worth seeing: it is the signal that decoder + * surfaces are exhausted. */ + uint64_t video_pkt_eagain; + uint64_t audio_pkt_eagain; + /* Packets still refused after the drain-and-resend retry, and so + * genuinely lost. This is the number that costs picture quality; + * the eagain counters above only say the condition was hit. */ + uint64_t video_pkt_dropped; + uint64_t audio_pkt_dropped; uint64_t audio_decoder_flushes; uint64_t video_decoder_flushes; uint64_t audio_last_decoder_flush_time_us; @@ -376,7 +528,10 @@ void irl_receiver_stop(struct irl_source *ctx); /* ── Video handler (video-handler.c) ──────────────────────── */ -void irl_video_output_frame(struct irl_source *ctx, AVFrame *frame); +void irl_video_output_frame(struct irl_source *ctx, AVFrame *frame, + uint64_t timestamp); +AVFrame *irl_video_to_sysmem(struct irl_source *ctx, AVFrame *frame); +uint64_t irl_video_due_time(struct irl_source *ctx, const AVFrame *frame); bool irl_video_is_keyframe(const AVFrame *frame); /* ── PTS repair (pts-repair.c) ────────────────────────────── */ diff --git a/include/irl-threading.h b/include/irl-threading.h index 4bbd9dc..367fe4d 100644 --- a/include/irl-threading.h +++ b/include/irl-threading.h @@ -88,6 +88,16 @@ static inline void irl_cond_wait(irl_cond_t *c, irl_mutex_t *m) SleepConditionVariableCS(c, m, INFINITE); } +/* Wait until signalled or `timeout_ms` elapses. Spurious and early wakeups + * are permitted on every backend, so callers must re-check their predicate + * and recompute the remaining time rather than assume the full interval + * passed. */ +static inline void irl_cond_timedwait(irl_cond_t *c, irl_mutex_t *m, + uint32_t timeout_ms) +{ + SleepConditionVariableCS(c, m, (DWORD)timeout_ms); +} + static inline void irl_cond_signal(irl_cond_t *c) { WakeConditionVariable(c); @@ -150,6 +160,8 @@ static inline void irl_thread_join(irl_thread_t *t) #else /* !_WIN32 */ #include +#include +#include typedef pthread_mutex_t irl_mutex_t; typedef pthread_cond_t irl_cond_t; @@ -177,7 +189,24 @@ static inline void irl_mutex_unlock(irl_mutex_t *m) static inline int irl_cond_init(irl_cond_t *c) { +#if defined(__APPLE__) + /* macOS has no pthread_condattr_setclock. Its timed wait below is + * pthread_cond_timedwait_relative_np, which takes an interval rather + * than a deadline and so is already immune to clock changes. */ return pthread_cond_init(c, NULL); +#else + /* Bind the condvar to CLOCK_MONOTONIC. pthread_cond_timedwait takes an + * absolute deadline against the condvar's clock, and the default is + * CLOCK_REALTIME — an NTP step or a manual clock change would then + * stretch or collapse a video pacing wait. */ + pthread_condattr_t attr; + if (pthread_condattr_init(&attr) != 0) + return pthread_cond_init(c, NULL); + pthread_condattr_setclock(&attr, CLOCK_MONOTONIC); + int ret = pthread_cond_init(c, &attr); + pthread_condattr_destroy(&attr); + return ret; +#endif } static inline void irl_cond_destroy(irl_cond_t *c) @@ -190,6 +219,31 @@ static inline void irl_cond_wait(irl_cond_t *c, irl_mutex_t *m) pthread_cond_wait(c, m); } +/* Wait until signalled or `timeout_ms` elapses. Spurious and early wakeups + * are permitted on every backend, so callers must re-check their predicate + * and recompute the remaining time rather than assume the full interval + * passed. */ +static inline void irl_cond_timedwait(irl_cond_t *c, irl_mutex_t *m, + uint32_t timeout_ms) +{ +#if defined(__APPLE__) + struct timespec rel; + rel.tv_sec = (time_t)(timeout_ms / 1000u); + rel.tv_nsec = (long)(timeout_ms % 1000u) * 1000000L; + pthread_cond_timedwait_relative_np(c, m, &rel); +#else + struct timespec ts; + clock_gettime(CLOCK_MONOTONIC, &ts); + ts.tv_sec += (time_t)(timeout_ms / 1000u); + ts.tv_nsec += (long)(timeout_ms % 1000u) * 1000000L; + if (ts.tv_nsec >= 1000000000L) { + ts.tv_sec += 1; + ts.tv_nsec -= 1000000000L; + } + pthread_cond_timedwait(c, m, &ts); +#endif +} + static inline void irl_cond_signal(irl_cond_t *c) { pthread_cond_signal(c); diff --git a/src/irl-source.c b/src/irl-source.c index 4966bec..7b49a4b 100644 --- a/src/irl-source.c +++ b/src/irl-source.c @@ -325,6 +325,8 @@ static void irl_source_get_stats(void *data, calldata_t *cd) uint64_t video_sys_base = ctx->video_sys_base; int64_t video_pts_base = ctx->video_pts_base; int64_t latest_video_stream_pts_ns = ctx->latest_video_stream_pts_ns; + int64_t video_lead_ms = ctx->video_lead_ns / 1000000LL; + uint64_t video_lead_excess = ctx->video_lead_excess; irl_mutex_unlock(&ctx->audio_state_lock); calldata_set_int(cd, "buffer_fill_ms", buffer_fill_ms); @@ -362,6 +364,9 @@ static void irl_source_get_stats(void *data, calldata_t *cd) (long long)audio_decoder_flushes); calldata_set_int(cd, "video_decoder_flushes", (long long)video_decoder_flushes); + calldata_set_int(cd, "video_lead_ms", (long long)video_lead_ms); + calldata_set_int(cd, "video_lead_excess", + (long long)video_lead_excess); /* Stream delay: how far behind real-time the video output is. * Computed as wall_clock - anchored_video_PTS. Includes SRT @@ -429,6 +434,7 @@ void *irl_source_create(obs_data_t *settings, obs_source_t *source) "out int audio_output_restarts, out int obs_lead_ms, " "out int audio_decoder_flushes, " "out int video_decoder_flushes, " + "out int video_lead_ms, out int video_lead_excess, " "out int stream_delay_ms, out bool low_latency_audio, " "out int reconnect_count)", irl_source_get_stats, ctx); diff --git a/src/receiver-audio.c b/src/receiver-audio.c index 789a2d6..270a7a8 100644 --- a/src/receiver-audio.c +++ b/src/receiver-audio.c @@ -58,14 +58,8 @@ * the clock line instead of letting OBS add permanent buffering. */ #define AUDIO_OUT_MAX_LAG_MS 150 -/* Concealment inflates the audio->OBS playout offset with no bounded - * recovery once primed (see the offset-reanchor logic below). This - * far past the primed baseline the accumulated latency is treated as - * unrecoverable by the speed-drain and reclaimed with one declared - * re-anchor. Set above the worst normal buffer swing (buffer_max is - * only ~200ms over target) so ordinary adaptive-speed excursions - * never trip it; only a real outage's worth of concealment does. */ -#define AUDIO_OFFSET_REANCHOR_MARGIN_MS 400 +/* AUDIO_OFFSET_REANCHOR_MARGIN_MS lives in irl-source.h, where the video + * side also uses it to decide when a lead is worth reporting. */ /* Playback speed authority for buffer regulation. Asymmetric, * IRLToolkit-style: draining a post-stall backlog runs up to +5% @@ -81,6 +75,21 @@ * skipping chunks when fill runs away. */ #define AUDIO_LL_MAX_FILL_MS 100 +/* The drain is bounded at +5%, so a sender whose media clock runs faster + * than that can never be caught up with: the buffer rises to the read + * loop's bleed ceiling and parks there, and latency parks with it. Nothing + * the plugin may do fixes that — draining harder would mean skipping audio, + * which this design does not do once primed — but it should not look like + * normal operation either, so detect it and say so. + * + * Twenty seconds is chosen to sit clear of a legitimate burst: even a + * backlog filling the ceiling drains back under buffer_max in about 13s at + * the default target, after which the speed ramp backs off on its own. */ +#define AUDIO_DRAIN_STUCK_US 20000000ULL +/* Treat the drain as making progress if fill has come down by this much + * since the window opened, so a slow but real recovery is not reported. */ +#define AUDIO_DRAIN_STUCK_PROGRESS_MS 100 + /* Grow a per-thread scratch buffer to at least `need` bytes. Returns * the buffer or NULL on OOM. The buffer is owned by the caller's * thread; no synchronisation here. */ @@ -127,6 +136,9 @@ void irl_reset_audio_timing_state(struct irl_source *ctx) ctx->audio_decode_errors = 0; ctx->audio_last_decoder_flush_time_us = 0; ctx->audio_last_decoder_warning_time_us = 0; + ctx->audio_drain_stuck_since_us = 0; + ctx->audio_drain_stuck_fill_ms = 0; + ctx->audio_drain_warn_time_us = 0; } void irl_reset_stream_timing_state(struct irl_source *ctx) @@ -134,6 +146,13 @@ void irl_reset_stream_timing_state(struct irl_source *ctx) irl_reset_audio_timing_state(ctx); ctx->video_ts_init = false; ctx->latest_video_stream_pts_ns = 0; + /* State, not counters: the interval has to be re-measured for the + * new stream, and a stale lead would be reported until the first + * frame arrives. video_lead_excess is cumulative for the source, + * like the other quality counters. */ + ctx->video_prev_pts_ns = 0; + ctx->video_frame_interval_ns = 0; + ctx->video_lead_ns = 0; ctx->video_decode_errors = 0; ctx->video_last_decoder_flush_time_us = 0; ctx->video_last_decoder_warning_time_us = 0; @@ -637,6 +656,52 @@ static void irl_audio_maybe_reanchor_offset(struct irl_source *ctx, AUDIO_OFFSET_REANCHOR_MARGIN_MS); } +/* ── Unwinnable drain detection ───────────────────────────── */ + +/* Called once per emitted chunk with the fill and speed that produced it. */ +static void audio_check_drain_progress(struct irl_source *ctx, int fill_ms, + float speed) +{ + int target_ms = (int)os_atomic_load_long(&ctx->config.buffer_target_ms); + bool at_full_authority = speed >= AUDIO_SPEED_MAX - 0.0005f && + fill_ms > target_ms + AUDIO_SPEED_DEADBAND_MS; + + if (!at_full_authority) { + ctx->audio_drain_stuck_since_us = 0; + return; + } + + uint64_t now_us = (uint64_t)av_gettime(); + if (ctx->audio_drain_stuck_since_us == 0) { + ctx->audio_drain_stuck_since_us = now_us; + ctx->audio_drain_stuck_fill_ms = fill_ms; + return; + } + + /* Coming down, just slowly: not stuck. */ + if (fill_ms <= ctx->audio_drain_stuck_fill_ms - + AUDIO_DRAIN_STUCK_PROGRESS_MS) { + ctx->audio_drain_stuck_since_us = now_us; + ctx->audio_drain_stuck_fill_ms = fill_ms; + return; + } + + if (now_us - ctx->audio_drain_stuck_since_us < AUDIO_DRAIN_STUCK_US) + return; + if (ctx->audio_drain_warn_time_us != 0 && + now_us - ctx->audio_drain_warn_time_us < AUDIO_DRAIN_STUCK_US) + return; + ctx->audio_drain_warn_time_us = now_us; + + blog(LOG_WARNING, + "[irl-source] Audio buffer stuck at %dms (target %dms) with playback at +%.0f%% for %llus: " + "the sender is delivering faster than real time, so the buffer cannot drain and latency stays here. " + "Video stays in sync with it; check the sender's frame rate and clock", + fill_ms, target_ms, (double)((speed - 1.0f) * 100.0f), + (unsigned long long)((now_us - ctx->audio_drain_stuck_since_us) / + 1000000ULL)); +} + /* ── Pump ─────────────────────────────────────────────────── */ bool irl_pump_audio_once(struct irl_source *ctx) @@ -693,6 +758,15 @@ bool irl_pump_audio_once(struct irl_source *ctx) int chunk_count = 0; bool has_audio = audio_buffer_peek_state(&ctx->audio_buf, &peek, &fill_ms, &chunk_count); + /* The receiver thread reads this for the stats line, so publish it + * under the shared timing lock like the rest of the cross-thread + * state. peek_state released the buffer mutex before returning, so + * nothing is nested here and the documented order (audio_state_lock + * before the buffer mutex) still holds. */ + irl_mutex_lock(&ctx->audio_state_lock); + if (fill_ms > ctx->audio_fill_peak_ms) + ctx->audio_fill_peak_ms = fill_ms; + irl_mutex_unlock(&ctx->audio_state_lock); if (has_audio && maybe_trim_hidden_audio_backlog(ctx, fill_ms, chunk_count)) @@ -768,6 +842,7 @@ bool irl_pump_audio_once(struct irl_source *ctx) (uint64_t)in_frames * 1000000000ULL / (uint64_t)out_rate; float speed = compute_buffered_output_speed(ctx, fill_ms); + audio_check_drain_progress(ctx, fill_ms, speed); uint8_t *emit_buf = in_buf; uint32_t frames_out = (uint32_t)in_frames; diff --git a/src/receiver-decode.c b/src/receiver-decode.c index c68f257..9474891 100644 --- a/src/receiver-decode.c +++ b/src/receiver-decode.c @@ -43,10 +43,67 @@ static void reinit_audio_pts_repair(struct irl_source *ctx) } } +/* Drain everything the decoder has ready. Lifted verbatim out of + * irl_handle_audio_packet() so the EAGAIN retry below can drain without + * duplicating the error handling; the state machine is unchanged. */ +static void drain_audio_frames(struct irl_source *ctx, AVFrame *frame) +{ + for (;;) { + int ret = avcodec_receive_frame(ctx->audio_dec_ctx, frame); + if (ret == AVERROR(EAGAIN) || ret == AVERROR_EOF) + break; + if (ret < 0) { + ctx->audio_decode_errors++; + if (ctx->audio_decode_errors >= 3) { + uint64_t now_us = (uint64_t)av_gettime(); + bool do_flush = should_flush_decoder( + &ctx->audio_last_decoder_flush_time_us, + now_us); + if (should_log_decoder_warning( + &ctx->audio_last_decoder_warning_time_us, + now_us)) { + blog(LOG_WARNING, + "[irl-source] Audio decoder receive: corruption burst (%d consecutive errors)%s", + ctx->audio_decode_errors, + do_flush ? ", resetting audio state" + : ", reset cooldown active"); + } + if (do_flush) { + avcodec_flush_buffers(ctx->audio_dec_ctx); + ctx->audio_decoder_flushes++; + ctx->audio_quality_events++; + irl_mutex_lock(&ctx->audio_state_lock); + audio_buffer_flush(&ctx->audio_buf); + irl_reset_audio_timing_state(ctx); + irl_mark_audio_recovery(ctx, 2500000ULL); + irl_mutex_unlock(&ctx->audio_state_lock); + reinit_audio_pts_repair(ctx); + } + ctx->audio_decode_errors = 0; + } + break; + } + + ctx->audio_decode_errors = 0; + irl_handle_audio_frame(ctx, frame); + av_frame_unref(frame); + } +} + void irl_handle_audio_packet(struct irl_source *ctx, AVPacket *pkt, AVFrame *frame) { int ret = avcodec_send_packet(ctx->audio_dec_ctx, pkt); + if (ret == AVERROR(EAGAIN)) { + /* The decoder did not take the packet. FFmpeg's contract is + * to read output and resend the same packet; returning here + * would silently discard it. */ + ctx->audio_pkt_eagain++; + drain_audio_frames(ctx, frame); + ret = avcodec_send_packet(ctx->audio_dec_ctx, pkt); + if (ret == AVERROR(EAGAIN)) + ctx->audio_pkt_dropped++; + } if (ret < 0 && ret != AVERROR(EAGAIN) && ret != AVERROR_EOF) { ctx->audio_decode_errors++; if (ctx->audio_decode_errors >= 3) { @@ -73,45 +130,45 @@ void irl_handle_audio_packet(struct irl_source *ctx, AVPacket *pkt, ctx->audio_decode_errors = 0; } + drain_audio_frames(ctx, frame); +} + +/* Video counterpart of drain_audio_frames(): same loop as before, moved so + * the EAGAIN retry can reuse it. */ +static void drain_video_frames(struct irl_source *ctx, AVFrame *frame) +{ for (;;) { - ret = avcodec_receive_frame(ctx->audio_dec_ctx, frame); + int ret = avcodec_receive_frame(ctx->video_dec_ctx, frame); if (ret == AVERROR(EAGAIN) || ret == AVERROR_EOF) break; if (ret < 0) { - ctx->audio_decode_errors++; - if (ctx->audio_decode_errors >= 3) { + ctx->video_decode_errors++; + ctx->video_corrupted = true; + if (ctx->video_decode_errors >= 3) { uint64_t now_us = (uint64_t)av_gettime(); bool do_flush = should_flush_decoder( - &ctx->audio_last_decoder_flush_time_us, + &ctx->video_last_decoder_flush_time_us, now_us); if (should_log_decoder_warning( - &ctx->audio_last_decoder_warning_time_us, + &ctx->video_last_decoder_warning_time_us, now_us)) { blog(LOG_WARNING, - "[irl-source] Audio decoder receive: corruption burst (%d consecutive errors)%s", - ctx->audio_decode_errors, - do_flush ? ", resetting audio state" - : ", reset cooldown active"); + "[irl-source] Video decoder receive: corruption burst (%d consecutive errors)%s", + ctx->video_decode_errors, + do_flush ? ", flushing" + : ", flush cooldown active"); } if (do_flush) { - avcodec_flush_buffers(ctx->audio_dec_ctx); - ctx->audio_decoder_flushes++; - ctx->audio_quality_events++; - irl_mutex_lock(&ctx->audio_state_lock); - audio_buffer_flush(&ctx->audio_buf); - irl_reset_audio_timing_state(ctx); - irl_mark_audio_recovery( - ctx, 2500000ULL); - irl_mutex_unlock(&ctx->audio_state_lock); - reinit_audio_pts_repair(ctx); + avcodec_flush_buffers(ctx->video_dec_ctx); + ctx->video_decoder_flushes++; } - ctx->audio_decode_errors = 0; + ctx->video_decode_errors = 0; } break; } - ctx->audio_decode_errors = 0; - irl_handle_audio_frame(ctx, frame); + ctx->video_decode_errors = 0; + irl_handle_video_frame(ctx, frame); av_frame_unref(frame); } } @@ -139,6 +196,27 @@ void irl_handle_video_packet(struct irl_source *ctx, AVPacket *pkt, } int ret = avcodec_send_packet(ctx->video_dec_ctx, pkt); + if (ret == AVERROR(EAGAIN)) { + /* The decoder refused the packet: it has output waiting and, + * on fixed-pool hardware decoders, no free surface until we + * take it. FFmpeg's contract is to read the output and resend + * the same packet — falling through would discard it, and + * with a reference frame that costs artifacts until the next + * keyframe rather than one dropped frame. + * + * One retry, deliberately not a loop. A single drain frees + * every surface the decoder was waiting on, and this runs on + * the receiver thread, which also feeds audio intake: a + * decoder that returned EAGAIN without producing frames would + * spin here and starve the jitter buffer, trading a dropped + * packet for the underrun cascade. If the retry fails too, + * count it and behave as before. */ + ctx->video_pkt_eagain++; + drain_video_frames(ctx, frame); + ret = avcodec_send_packet(ctx->video_dec_ctx, pkt); + if (ret == AVERROR(EAGAIN)) + ctx->video_pkt_dropped++; + } if (ret < 0 && ret != AVERROR(EAGAIN) && ret != AVERROR_EOF) { ctx->video_decode_errors++; ctx->video_corrupted = true; @@ -165,38 +243,5 @@ void irl_handle_video_packet(struct irl_source *ctx, AVPacket *pkt, ctx->video_decode_errors = 0; } - for (;;) { - ret = avcodec_receive_frame(ctx->video_dec_ctx, frame); - if (ret == AVERROR(EAGAIN) || ret == AVERROR_EOF) - break; - if (ret < 0) { - ctx->video_decode_errors++; - ctx->video_corrupted = true; - if (ctx->video_decode_errors >= 3) { - uint64_t now_us = (uint64_t)av_gettime(); - bool do_flush = should_flush_decoder( - &ctx->video_last_decoder_flush_time_us, - now_us); - if (should_log_decoder_warning( - &ctx->video_last_decoder_warning_time_us, - now_us)) { - blog(LOG_WARNING, - "[irl-source] Video decoder receive: corruption burst (%d consecutive errors)%s", - ctx->video_decode_errors, - do_flush ? ", flushing" - : ", flush cooldown active"); - } - if (do_flush) { - avcodec_flush_buffers(ctx->video_dec_ctx); - ctx->video_decoder_flushes++; - } - ctx->video_decode_errors = 0; - } - break; - } - - ctx->video_decode_errors = 0; - irl_handle_video_frame(ctx, frame); - av_frame_unref(frame); - } + drain_video_frames(ctx, frame); } diff --git a/src/receiver-stream.c b/src/receiver-stream.c index 457176e..84e3ce2 100644 --- a/src/receiver-stream.c +++ b/src/receiver-stream.c @@ -158,8 +158,18 @@ static AVCodecContext *open_decoder(struct irl_source *src, AVStream *stream, /* The video output queue holds decoded HW frames, each * pinning a decoder surface; give the pool matching * headroom or the decoder can stall waiting for a - * surface the queue is sitting on. */ - ctx->extra_hw_frames = IRL_VIDEO_QUEUE_SIZE; + * surface the queue is sitting on. Fixed-pool decoders + * (D3D11VA, VAAPI) are where that bites, and it looks + * like frozen video with clean audio. + * + * Count the surfaces this plugin can pin at once, not + * just the queue: IRL_VIDEO_QUEUE_SIZE queued, plus the + * one the video thread has popped and is transferring in + * irl_video_output_frame(), plus the one just returned by + * avcodec_receive_frame() and not yet unref'd. The clone + * irl_video_queue_push() takes references that same + * surface, so it does not add a third. */ + ctx->extra_hw_frames = IRL_VIDEO_QUEUE_SIZE + 2; } if (try_hw && stream->codecpar->codec_type == AVMEDIA_TYPE_VIDEO) { @@ -499,6 +509,21 @@ void irl_log_receiver_stats(struct irl_source *ctx) ctx->last_stats_time = now; + /* Snapshot what other threads write before formatting any of it. + * The audio thread owns the playout offset and the buffer + * high-water mark; the video thread owns the lead figures; the + * pinned-surface peak belongs to video_queue_lock. + * + * Two separate acquisitions, never nested: no path in the plugin + * holds video_queue_lock and audio_state_lock at once (the video + * thread drops the queue lock before irl_video_output_frame, and + * irl_handle_video_frame drops the state lock before pushing), and + * a stats line is the last place that edge should be introduced. + * + * av_drift is computed inside the lock rather than from separate + * reads: its three inputs are only meaningful against each other, + * and the audio thread updates them together. */ + irl_mutex_lock(&ctx->audio_state_lock); /* Drift of the audio->OBS playout offset from its primed baseline. * Stays near 0 when healthy; a climbing value is concealment * inflating the video lip-sync mapping (see receiver-audio.c). */ @@ -511,20 +536,39 @@ void irl_log_receiver_stats(struct irl_source *ctx) ctx->audio_playout_offset_baseline_ns) / 1000000LL; } + int audio_fill_peak_ms = ctx->audio_fill_peak_ms; + int64_t video_lead_ms = ctx->video_lead_ns / 1000000LL; + int64_t video_lead_peak_ms = ctx->video_lead_peak_ns / 1000000LL; + uint64_t video_lead_excess = ctx->video_lead_excess; + int64_t video_frame_interval_ns = ctx->video_frame_interval_ns; + irl_mutex_unlock(&ctx->audio_state_lock); + + irl_mutex_lock(&ctx->video_queue_lock); + int video_pinned_peak = ctx->video_pinned_peak; + int video_pacing_now = ctx->video_pacing_now; + int video_pacing_peak = ctx->video_pacing_peak; + size_t video_pacing_bytes = ctx->video_pacing_bytes; + uint64_t video_pacing_overflows = ctx->video_pacing_overflows; + irl_mutex_unlock(&ctx->video_queue_lock); + + int buffer_fill_ms = audio_buffer_fill_ms_locked(&ctx->audio_buf); blog(LOG_INFO, "[irl-source] Stats: video=%llu audio=%llu " - "buf=%dms target=%dms speed=%.3f ctrl=%s pts_repairs=%llu " + "buf=%dms peak=%dms target=%dms speed=%.3f ctrl=%s pts_repairs=%llu " "norm=%llu interp=%llu silence=%llu resets=%llu " "last_gap=%dms max_gap=%dms underruns=%llu resync_skips=%llu " "hidden_trims=%llu quality_events=%llu " "audio_flushes=%llu video_flushes=%llu vq_drops=%llu " "obs_lead=%lldms chunk=%u@%u " "stream_chunk=%llums obs_chunk=%llums " - "restarts=%llu av_drift=%lldms reanchors=%llu res=%dx%d", + "restarts=%llu av_drift=%lldms reanchors=%llu " + "vlead=%lldms peak=%lldms excess=%llu vfps=%.1f " + "pinned_peak=%d/%d paced=%d/%d(%zuMB) early=%llu eagain=%llu/%llu pktdrop=%llu/%llu res=%dx%d", (unsigned long long)ctx->total_video_frames, (unsigned long long)ctx->total_audio_frames, - audio_buffer_fill_ms_locked(&ctx->audio_buf), + buffer_fill_ms, + audio_fill_peak_ms, (int)os_atomic_load_long(&ctx->config.buffer_target_ms), (double)ctx->current_speed, os_atomic_load_bool(&ctx->config.adaptive_speed) ? "on" : "off", @@ -550,5 +594,21 @@ void irl_log_receiver_stats(struct irl_source *ctx) (unsigned long long)ctx->audio_output_restarts, (long long)av_drift_ms, (unsigned long long)ctx->audio_offset_reanchors, + (long long)video_lead_ms, + (long long)video_lead_peak_ms, + (unsigned long long)video_lead_excess, + video_frame_interval_ns > 0 + ? 1000000000.0 / (double)video_frame_interval_ns + : 0.0, + /* peak pinned surfaces vs what extra_hw_frames budgeted; + * the pool must cover peak + the decoder's own frame. */ + video_pinned_peak, IRL_VIDEO_QUEUE_SIZE + 2, + video_pacing_now, video_pacing_peak, + video_pacing_bytes / (1024u * 1024u), + (unsigned long long)video_pacing_overflows, + (unsigned long long)ctx->video_pkt_eagain, + (unsigned long long)ctx->audio_pkt_eagain, + (unsigned long long)ctx->video_pkt_dropped, + (unsigned long long)ctx->audio_pkt_dropped, ctx->last_video_width, ctx->last_video_height); } diff --git a/src/receiver-video.c b/src/receiver-video.c index 20c38c8..c594ce4 100644 --- a/src/receiver-video.c +++ b/src/receiver-video.c @@ -6,6 +6,8 @@ * SPDX-License-Identifier: AGPL-3.0-or-later */ +#include + #include "receiver-internal.h" /* ── Video output queue ───────────────────────────────────── */ @@ -22,6 +24,13 @@ static void video_queue_drain_locked(struct irl_source *ctx) } } +static void video_pinned_update_locked(struct irl_source *ctx) +{ + int pinned = ctx->video_queue_count + ctx->video_in_flight; + if (pinned > ctx->video_pinned_peak) + ctx->video_pinned_peak = pinned; +} + /* Ask the video thread to blank the source. Queued frames are dropped * here so nothing decoded before the disconnect can repaint after the * clear; a frame already being converted is handled by the ordering in @@ -63,40 +72,171 @@ void irl_video_queue_push(struct irl_source *ctx, AVFrame *frame, IRL_VIDEO_QUEUE_SIZE; ctx->video_queue[tail] = clone; ctx->video_queue_count++; + video_pinned_update_locked(ctx); irl_cond_signal(&ctx->video_queue_cond); irl_mutex_unlock(&ctx->video_queue_lock); } -void *irl_video_thread(void *data) +/* ── Pacing queue (video thread only) ─────────────────────── */ + +static size_t pacing_frame_bytes(const AVFrame *f) { - struct irl_source *ctx = data; + int size = av_image_get_buffer_size(f->format, f->width, f->height, 1); + return size > 0 ? (size_t)size : 0; +} - irl_mutex_lock(&ctx->video_queue_lock); - while (os_atomic_load_bool(&ctx->thread_active)) { - if (ctx->video_clear_pending) { - ctx->video_clear_pending = false; +static bool pacing_has_room(const struct irl_source *ctx) +{ + return ctx->pacing_count < IRL_VIDEO_PACING_MAX_FRAMES && + ctx->pacing_bytes < IRL_VIDEO_PACING_MAX_BYTES; +} + +static void pacing_push(struct irl_source *ctx, AVFrame *frame, uint64_t due_ns) +{ + int tail = (ctx->pacing_head + ctx->pacing_count) % + IRL_VIDEO_PACING_MAX_FRAMES; + ctx->pacing_queue[tail].frame = frame; + ctx->pacing_queue[tail].due_ns = due_ns; + ctx->pacing_queue[tail].bytes = pacing_frame_bytes(frame); + ctx->pacing_bytes += ctx->pacing_queue[tail].bytes; + ctx->pacing_count++; + if (ctx->pacing_count > ctx->pacing_peak) + ctx->pacing_peak = ctx->pacing_count; +} + +static struct irl_pacing_frame pacing_pop(struct irl_source *ctx) +{ + struct irl_pacing_frame e = ctx->pacing_queue[ctx->pacing_head]; + ctx->pacing_queue[ctx->pacing_head].frame = NULL; + ctx->pacing_head = (ctx->pacing_head + 1) % IRL_VIDEO_PACING_MAX_FRAMES; + ctx->pacing_count--; + ctx->pacing_bytes -= e.bytes; + return e; +} + +static void pacing_drain(struct irl_source *ctx) +{ + while (ctx->pacing_count > 0) { + struct irl_pacing_frame e = pacing_pop(ctx); + av_frame_free(&e.frame); + } + ctx->pacing_bytes = 0; +} + +/* Move everything the receiver has decoded into the pacing queue, copying it + * out of the hardware frame pool on the way so the decoder gets its surfaces + * back. Runs before every emit and wait, because holding a decoded frame in + * video_queue is what stalls the decoder. */ +static void pacing_intake(struct irl_source *ctx) +{ + for (;;) { + irl_mutex_lock(&ctx->video_queue_lock); + if (ctx->video_queue_count == 0 || !pacing_has_room(ctx)) { irl_mutex_unlock(&ctx->video_queue_lock); - obs_source_output_video(ctx->source, NULL); - irl_mutex_lock(&ctx->video_queue_lock); - continue; - } - if (ctx->video_queue_count == 0) { - irl_cond_wait(&ctx->video_queue_cond, - &ctx->video_queue_lock); - continue; + return; } AVFrame *f = ctx->video_queue[ctx->video_queue_head]; ctx->video_queue[ctx->video_queue_head] = NULL; ctx->video_queue_head = (ctx->video_queue_head + 1) % IRL_VIDEO_QUEUE_SIZE; ctx->video_queue_count--; + ctx->video_in_flight = 1; + video_pinned_update_locked(ctx); irl_mutex_unlock(&ctx->video_queue_lock); - irl_video_output_frame(ctx, f); + AVFrame *sw = irl_video_to_sysmem(ctx, f); + uint64_t due = sw ? irl_video_due_time(ctx, sw) : 0; av_frame_free(&f); irl_mutex_lock(&ctx->video_queue_lock); + ctx->video_in_flight = 0; + irl_mutex_unlock(&ctx->video_queue_lock); + + if (sw) + pacing_push(ctx, sw, due); + } +} + +/* Emit every frame whose moment has arrived. Over the ceilings the head goes + * out early rather than being dropped: too-early video is what the un-paced + * path did all the time, and it beats a hole in the picture. */ +static void pacing_emit_due(struct irl_source *ctx, uint64_t now) +{ + while (ctx->pacing_count > 0) { + bool over = !pacing_has_room(ctx); + uint64_t due = ctx->pacing_queue[ctx->pacing_head].due_ns; + + if (!over && (int64_t)(due - now) > IRL_VIDEO_PACING_SLACK_NS) + return; + if (over) + ctx->pacing_overflows++; + + struct irl_pacing_frame e = pacing_pop(ctx); + irl_video_output_frame(ctx, e.frame, e.due_ns); + av_frame_free(&e.frame); + } +} + +void *irl_video_thread(void *data) +{ + struct irl_source *ctx = data; + + while (os_atomic_load_bool(&ctx->thread_active)) { + bool clear; + irl_mutex_lock(&ctx->video_queue_lock); + clear = ctx->video_clear_pending; + ctx->video_clear_pending = false; + irl_mutex_unlock(&ctx->video_queue_lock); + + if (clear) { + /* video_queue was already dropped by the requester; + * the paced frames behind it must go too, or the + * blank would be repainted a lead later. */ + pacing_drain(ctx); + obs_source_output_video(ctx->source, NULL); + continue; + } + + pacing_intake(ctx); + pacing_emit_due(ctx, os_gettime_ns()); + + irl_mutex_lock(&ctx->video_queue_lock); + ctx->video_pacing_now = ctx->pacing_count; + ctx->video_pacing_peak = ctx->pacing_peak; + ctx->video_pacing_bytes = ctx->pacing_bytes; + ctx->video_pacing_overflows = ctx->pacing_overflows; + irl_mutex_unlock(&ctx->video_queue_lock); + + /* Sleep until the next frame is due, or until the receiver + * pushes, a clear arrives, or the thread is stopped. */ + uint32_t wait_ms = IRL_VIDEO_PACING_MAX_WAIT_MS; + if (ctx->pacing_count > 0) { + uint64_t now = os_gettime_ns(); + int64_t until = (int64_t)ctx->pacing_queue[ctx->pacing_head] + .due_ns - + (int64_t)now; + if (until <= IRL_VIDEO_PACING_SLACK_NS) + continue; /* due already; go round again */ + uint32_t ms = (uint32_t)(until / 1000000LL); + if (ms < wait_ms) + wait_ms = ms; + } + + irl_mutex_lock(&ctx->video_queue_lock); + /* Re-check under the lock: a push or clear between the work + * above and here would otherwise be slept through. */ + if (!ctx->video_clear_pending && ctx->video_queue_count == 0 && + os_atomic_load_bool(&ctx->thread_active)) { + if (wait_ms == 0) + wait_ms = 1; + irl_cond_timedwait(&ctx->video_queue_cond, + &ctx->video_queue_lock, wait_ms); + } + irl_mutex_unlock(&ctx->video_queue_lock); } + + pacing_drain(ctx); + irl_mutex_lock(&ctx->video_queue_lock); video_queue_drain_locked(ctx); irl_mutex_unlock(&ctx->video_queue_lock); return NULL; @@ -181,8 +321,29 @@ void irl_handle_video_frame(struct irl_source *ctx, AVFrame *frame) ctx->fmt_ctx->streams[ctx->video_stream_idx]; pts_ns = av_rescale_q(frame->pts, vs->time_base, (AVRational){1, 1000000000}); + + /* Frame interval EMA, for the video thread's estimate of how + * many frames a given output lead parks in the libobs async + * queue. Measured rather than taken from avg_frame_rate, + * which live SRT/RTMP demuxers routinely leave unset or + * wrong. Out-of-range deltas (PTS repair, discontinuities, + * reordering) are skipped rather than smoothed in. */ + int64_t delta = pts_ns - ctx->video_prev_pts_ns; + bool usable_delta = ctx->video_prev_pts_ns != 0 && + delta >= IRL_VIDEO_INTERVAL_MIN_NS && + delta <= IRL_VIDEO_INTERVAL_MAX_NS; + ctx->video_prev_pts_ns = pts_ns; + irl_mutex_lock(&ctx->audio_state_lock); ctx->latest_video_stream_pts_ns = pts_ns; + if (usable_delta) { + if (ctx->video_frame_interval_ns == 0) + ctx->video_frame_interval_ns = delta; + else + ctx->video_frame_interval_ns += + (delta - ctx->video_frame_interval_ns) / + 8; + } irl_mutex_unlock(&ctx->audio_state_lock); } diff --git a/src/video-handler.c b/src/video-handler.c index 33f5ad8..f88c4d2 100644 --- a/src/video-handler.c +++ b/src/video-handler.c @@ -265,6 +265,65 @@ static void setup_color_params(struct obs_source_frame *obs_frame, #define VIDEO_TS_CLAMP_NS 500000000LL /* 500ms */ #define VIDEO_TS_CAP_NS 200000000ULL /* 200ms forward cap */ +/* Record how far ahead of wall clock the PTS mapping placed this frame. + * + * This used to clamp the lead as well, to keep libobs's async queue under its + * 30-frame wipe threshold. That clamp is gone: it acted on the lead's + * distance from the configured target, but what libobs actually queues is the + * lead's *growth* since its play head last anchored. A large but steady lead + * — a jitter buffer parked against the bleed ceiling because the sender + * over-delivers, say — queues nothing, and clamping it only shifted video + * ahead of audio. A 720p120 stream showed the cost plainly: a steady 1032ms + * lead clamped to 520ms, so half a second of permanent desync bought against + * a queue that was very likely one frame deep. + * + * The measurement stays, because it is the signal for whether video pacing + * (which removes libobs's scheduler from the path entirely, and with it this + * whole threshold) is doing its job. */ +static void video_record_lead(struct irl_source *ctx, int64_t ts, uint64_t now, + int64_t frame_interval_ns) +{ + int64_t lead_ns = ts - (int64_t)now; + + if (frame_interval_ns <= 0) + frame_interval_ns = IRL_VIDEO_INTERVAL_DEFAULT_NS; + + /* The lead libobs could absorb if the whole of it were growth. */ + int64_t budget_ns = IRL_OBS_ASYNC_FRAME_BUDGET * frame_interval_ns; + int64_t floor_ns = AUDIO_OFFSET_REANCHOR_MARGIN_MS * 1000000LL; + int64_t queue_safe_ns = + os_atomic_load_long(&ctx->config.buffer_target_ms) * 1000000LL + + (budget_ns < floor_ns ? floor_ns : budget_ns); + + irl_mutex_lock(&ctx->audio_state_lock); + ctx->video_lead_ns = lead_ns; + /* Keep the high-water mark too: stats are sampled every 30s, and an + * excursion that drains in ~17s is very likely to fall between two + * samples. The instantaneous value alone would read healthy on + * exactly the streams this is meant to diagnose. */ + if (lead_ns > ctx->video_lead_peak_ns) + ctx->video_lead_peak_ns = lead_ns; + if (lead_ns > queue_safe_ns) + ctx->video_lead_excess++; + irl_mutex_unlock(&ctx->audio_state_lock); + + if (lead_ns <= queue_safe_ns) + return; + + /* Only a risk while the lead is still climbing — a steady lead of any + * size is free — so this is a "watch this" line, not a fault. */ + if (now - ctx->video_lead_warn_time_ns >= + IRL_VIDEO_LEAD_WARN_INTERVAL_NS) { + ctx->video_lead_warn_time_ns = now; + blog(LOG_INFO, + "[irl-source] Video lead %lldms is beyond what OBS can queue (%lldms at %.0ffps); " + "harmless while it holds steady, but a rise of that size would make OBS drop queued video", + (long long)(lead_ns / 1000000LL), + (long long)(queue_safe_ns / 1000000LL), + 1000000000.0 / (double)frame_interval_ns); + } +} + /* Convert stream PTS to OBS nanosecond timestamp. * * When audio is active, treat queued audio as the master playout @@ -275,7 +334,7 @@ static void setup_color_params(struct obs_source_frame *obs_frame, * * If no audio playout mapping exists yet, fall back to the older * video-only wall-clock anchor. */ -static uint64_t frame_timestamp(struct irl_source *ctx, const AVFrame *frame) +uint64_t irl_video_due_time(struct irl_source *ctx, const AVFrame *frame) { /* frame->pts is pre-converted to nanoseconds by the receiver * thread (see irl_video_queue_push); fmt_ctx must not be @@ -287,10 +346,12 @@ static uint64_t frame_timestamp(struct irl_source *ctx, const AVFrame *frame) uint64_t audio_obs_end_ts_ns; int64_t audio_buffered_end_pts_ns; int startup_warmup_ms; + int64_t frame_interval_ns; irl_mutex_lock(&ctx->audio_state_lock); audio_obs_end_ts_ns = ctx->latest_audio_obs_end_ts_ns; audio_buffered_end_pts_ns = ctx->latest_audio_buffered_end_pts_ns; startup_warmup_ms = ctx->startup_audio_warmup_remaining_ms; + frame_interval_ns = ctx->video_frame_interval_ns; irl_mutex_unlock(&ctx->audio_state_lock); if (ctx->audio_stream_idx >= 0 && audio_obs_end_ts_ns != 0 && @@ -300,6 +361,7 @@ static uint64_t frame_timestamp(struct irl_source *ctx, const AVFrame *frame) audio_buffered_end_pts_ns); if (mapped < 0) mapped = 0; + video_record_lead(ctx, mapped, now, frame_interval_ns); return (uint64_t)mapped; } @@ -337,71 +399,62 @@ static uint64_t frame_timestamp(struct irl_source *ctx, const AVFrame *frame) computed += (uint64_t)audio_lead_ns; } + video_record_lead(ctx, (int64_t)computed, now, frame_interval_ns); return computed; } /* ── Video output ─────────────────────────────────────────── */ -void irl_video_output_frame(struct irl_source *ctx, AVFrame *frame) +/* Bring a decoded frame into system memory, releasing any decoder surface it + * held. Returns a new reference the caller owns, or NULL. + * + * This deliberately does not use av_hwframe_map(AV_HWFRAME_MAP_READ), which + * earlier gave VAAPI and VideoToolbox a zero-copy CPU view. A mapped frame + * still pins the surface it maps, and pacing holds frames for the whole + * output lead — hundreds of them at a high frame rate — so mapping would + * exhaust the decoder pool within a few frames of the buffer filling. The + * copy is what makes the surface reusable, so it is not optional here. + * + * On those two backends this costs one extra copy against the old path; on + * D3D11VA and CUDA, where the map always fell back to a copy anyway, nothing + * changes. */ +AVFrame *irl_video_to_sysmem(struct irl_source *ctx, AVFrame *frame) { - /* Hardware-decoded frames (NVDEC/D3D11VA/VAAPI/VideoToolbox) come - * out on the GPU; we have to expose them to OBS as system memory. - * - * Try av_hwframe_map(AV_HWFRAME_MAP_READ) first: on backends that - * can produce a CPU-readable view without a full download (VAAPI - * vaDeriveImage, VideoToolbox IOSurface), this skips the - * gpu->cpu copy entirely. On D3D11VA / CUDA the map call falls - * back to a copy internally or fails, so we fall back to - * av_hwframe_transfer_data which is the historical path. - * - * hw_map_ok caches the outcome so we don't keep paying for a - * doomed map attempt every frame on platforms that can't map. */ - AVFrame *sw_frame = NULL; - if (frame->hw_frames_ctx) { - sw_frame = av_frame_alloc(); - if (!sw_frame) - return; - - bool used_map = false; - if (ctx->hw_map_ok != 0) { - int ret = av_hwframe_map(sw_frame, frame, - AV_HWFRAME_MAP_READ); - if (ret == 0) { - used_map = true; - if (ctx->hw_map_ok != 1) { - ctx->hw_map_ok = 1; - blog(LOG_INFO, - "[irl-source] HW frame path: av_hwframe_map (zero-copy)"); - } - } else { - av_frame_unref(sw_frame); - if (ctx->hw_map_ok != 0) { - ctx->hw_map_ok = 0; - char errbuf[AV_ERROR_MAX_STRING_SIZE]; - av_strerror(ret, errbuf, sizeof(errbuf)); - blog(LOG_INFO, - "[irl-source] HW frame path: av_hwframe_transfer_data (map unsupported: %s)", - errbuf); - } - } - } - - if (!used_map) { - if (av_hwframe_transfer_data(sw_frame, frame, 0) < 0) { - av_frame_free(&sw_frame); - return; - } + UNUSED_PARAMETER(ctx); + + AVFrame *out = av_frame_alloc(); + if (!out) + return NULL; + + if (!frame->hw_frames_ctx) { + /* Already system memory: take a reference so the pacing queue + * owns its entries uniformly. */ + if (av_frame_ref(out, frame) < 0) { + av_frame_free(&out); + return NULL; } + return out; + } - sw_frame->pts = frame->pts; - sw_frame->colorspace = frame->colorspace; - sw_frame->color_range = frame->color_range; - sw_frame->color_trc = frame->color_trc; - sw_frame->color_primaries = frame->color_primaries; - sw_frame->flags = frame->flags; - frame = sw_frame; + if (av_hwframe_transfer_data(out, frame, 0) < 0) { + av_frame_free(&out); + return NULL; } + out->pts = frame->pts; + out->colorspace = frame->colorspace; + out->color_range = frame->color_range; + out->color_trc = frame->color_trc; + out->color_primaries = frame->color_primaries; + out->flags = frame->flags; + return out; +} + +/* Hand a system-memory frame to OBS with the timestamp pacing scheduled it + * for. `frame` must already have been through irl_video_to_sysmem(). */ +void irl_video_output_frame(struct irl_source *ctx, AVFrame *frame, + uint64_t timestamp) +{ enum video_format obs_fmt = avpixfmt_to_obs(frame->format); /* Negative linesize means the frame is laid out bottom-up. OBS's @@ -438,11 +491,8 @@ void irl_video_output_frame(struct irl_source *ctx, AVFrame *frame) size_t need = y_size + uv_size; if (need > ctx->sws_nv12_buf_capacity) { uint8_t *next = realloc(ctx->sws_nv12_buf, need); - if (!next) { - if (sw_frame) - av_frame_free(&sw_frame); + if (!next) return; - } ctx->sws_nv12_buf = next; ctx->sws_nv12_buf_capacity = need; } @@ -451,11 +501,8 @@ void irl_video_output_frame(struct irl_source *ctx, AVFrame *frame) ctx->sws_nv12_buf + y_size}; int dst_strides[2] = {frame->width, frame->width}; - if (!irl_convert_to_nv12(ctx, frame, dst_planes, dst_strides)) { - if (sw_frame) - av_frame_free(&sw_frame); + if (!irl_convert_to_nv12(ctx, frame, dst_planes, dst_strides)) return; - } struct obs_source_frame obs_frame = {0}; obs_frame.width = frame->width; @@ -465,12 +512,10 @@ void irl_video_output_frame(struct irl_source *ctx, AVFrame *frame) obs_frame.data[1] = dst_planes[1]; obs_frame.linesize[0] = dst_strides[0]; obs_frame.linesize[1] = dst_strides[1]; - obs_frame.timestamp = frame_timestamp(ctx, frame); + obs_frame.timestamp = timestamp; setup_color_params(&obs_frame, frame, VIDEO_FORMAT_NV12); obs_source_output_video(ctx->source, &obs_frame); - if (sw_frame) - av_frame_free(&sw_frame); return; } @@ -479,7 +524,7 @@ void irl_video_output_frame(struct irl_source *ctx, AVFrame *frame) obs_frame.width = frame->width; obs_frame.height = frame->height; obs_frame.format = obs_fmt; - obs_frame.timestamp = frame_timestamp(ctx, frame); + obs_frame.timestamp = timestamp; setup_color_params(&obs_frame, frame, obs_fmt); for (int i = 0; i < AV_NUM_DATA_POINTERS; i++) { @@ -491,6 +536,4 @@ void irl_video_output_frame(struct irl_source *ctx, AVFrame *frame) } obs_source_output_video(ctx->source, &obs_frame); - if (sw_frame) - av_frame_free(&sw_frame); } diff --git a/src/websocket-vendor.c b/src/websocket-vendor.c index d43daa4..4ff38f4 100644 --- a/src/websocket-vendor.c +++ b/src/websocket-vendor.c @@ -82,6 +82,8 @@ static const struct irl_stat_field irl_stat_fields[] = { {"obs_lead_ms", IRL_STAT_INT}, {"audio_decoder_flushes", IRL_STAT_INT}, {"video_decoder_flushes", IRL_STAT_INT}, + {"video_lead_ms", IRL_STAT_INT}, + {"video_lead_excess", IRL_STAT_INT}, {"stream_delay_ms", IRL_STAT_INT}, {"low_latency_audio", IRL_STAT_BOOL}, {"reconnect_count", IRL_STAT_INT},