diff --git a/ErsatzTV.Application/Streaming/HlsSessionWorker.cs b/ErsatzTV.Application/Streaming/HlsSessionWorker.cs index cc00e1b36..c5acce6d4 100644 --- a/ErsatzTV.Application/Streaming/HlsSessionWorker.cs +++ b/ErsatzTV.Application/Streaming/HlsSessionWorker.cs @@ -26,7 +26,9 @@ namespace ErsatzTV.Application.Streaming; public class HlsSessionWorker : IHlsSessionWorker { - private static int _workAheadCount; + // process-wide, shared by every session — the work-ahead limit is a global resource budget + private static readonly WorkAheadSlots _workAheadSlots = new(); + private readonly OutputFormatKind _outputFormatKind; private readonly IHlsInitSegmentCache _hlsInitSegmentCache; private readonly Dictionary _discontinuityMap = []; @@ -243,10 +245,12 @@ public class HlsSessionWorker : IHlsSessionWorker cancellationToken); } - bool initialWorkAhead = Volatile.Read(ref _workAheadCount) < await GetWorkAheadLimit(cancellationToken); + // claim the slot here rather than checking here and claiming inside Transcode: the check + // and the claim have to be one atomic step or every simultaneous tune-in wins (#536) + bool initialWorkAhead = _workAheadSlots.TryAcquire(await GetWorkAheadLimit(cancellationToken)); _state = initialWorkAhead ? HlsSessionState.SeekAndWorkAhead : HlsSessionState.SeekAndRealtime; - if (!await Transcode(!initialWorkAhead, cancellationToken)) + if (!await Transcode(initialWorkAhead, cancellationToken)) { return; } @@ -269,8 +273,8 @@ public class HlsSessionWorker : IHlsSessionWorker // only use realtime encoding when we're at least 30 seconds ahead bool realtime = transcodedBuffer >= TimeSpan.FromSeconds(30); bool subsequentWorkAhead = - !realtime && Volatile.Read(ref _workAheadCount) < await GetWorkAheadLimit(cancellationToken); - if (!await Transcode(!subsequentWorkAhead, cancellationToken)) + !realtime && _workAheadSlots.TryAcquire(await GetWorkAheadLimit(cancellationToken)); + if (!await Transcode(subsequentWorkAhead, cancellationToken)) { return; } @@ -456,15 +460,23 @@ public class HlsSessionWorker : IHlsSessionWorker return result; } - private async Task Transcode(bool realtime, CancellationToken cancellationToken) + /// + /// Runs one transcode. The caller is the one that races for a work-ahead slot, so ownership is + /// passed IN: means the caller already claimed a slot from + /// , and this method releases it in its finally — acquire + /// and release stay one-for-one (#536). + /// + private async Task Transcode(bool ownsWorkAheadSlot, CancellationToken cancellationToken) { + // a session works ahead exactly when it holds a slot; everything else runs realtime (throttled) + bool realtime = !ownsWorkAheadSlot; + try { bool wasSeekAndWorkAhead = _state is HlsSessionState.SeekAndWorkAhead; if (!realtime) { - Interlocked.Increment(ref _workAheadCount); _logger.LogDebug("HLS segmenter will work ahead for channel {Channel}", _channelNumber); HlsSessionState nextState = _state switch @@ -747,9 +759,9 @@ public class HlsSessionWorker : IHlsSessionWorker // do nothing } - if (!realtime) + if (ownsWorkAheadSlot) { - Interlocked.Decrement(ref _workAheadCount); + _workAheadSlots.Release(); } } diff --git a/ErsatzTV.Application/Streaming/WorkAheadSlots.cs b/ErsatzTV.Application/Streaming/WorkAheadSlots.cs new file mode 100644 index 000000000..cbddb355b --- /dev/null +++ b/ErsatzTV.Application/Streaming/WorkAheadSlots.cs @@ -0,0 +1,91 @@ +namespace ErsatzTV.Application.Streaming; + +/// +/// The process-wide pool of work-ahead slots shared by every HLS session (ersatztv#536). +/// +/// +/// +/// workAheadSegmenterLimit is a resource guarantee, not a tuning knob: it bounds how many +/// transcodes may run unthrottled (no -readrate) at once, and the QSV hardware-frame pool +/// sizing from ersatztv#529 assumes that bound holds. +/// +/// +/// Acquisition must therefore be atomic. The previous shape — Volatile.Read(count) < limit +/// in the caller, Interlocked.Increment later inside the transcode — is a check-then-act +/// TOCTOU separated by at least one await (the limit is a DB-backed config read), so N +/// simultaneous tune-ins all observed 0 < limit and all ran unthrottled. Same class as +/// ersatztv#231 / #250. +/// +/// +public sealed class WorkAheadSlots +{ + private int _count; + private int _unbalancedReleases; + + /// + /// Gets the number of slots currently held. For diagnostics and tests only — never branch on + /// this to decide whether to work ahead; that is exactly the race exists to close. + /// + public int Count => Volatile.Read(ref _count); + + /// + /// Atomically claims one slot if fewer than are held. + /// + /// true when a slot was claimed; the caller then owns it and MUST + /// it exactly once. + public bool TryAcquire(int limit) + { + while (true) + { + int current = Volatile.Read(ref _count); + if (current >= limit) + { + return false; + } + + // only the thread whose compare-exchange observes the value it read wins the slot, so + // the count can never transiently exceed the limit and two racers can never both claim + if (Interlocked.CompareExchange(ref _count, current + 1, current) == current) + { + return true; + } + } + } + + /// + /// Gets the number of releases that were not matched by a successful acquire. Always zero in a + /// correct program; a non-zero value means the ownership contract was broken somewhere. + /// + public int UnbalancedReleases => Volatile.Read(ref _unbalancedReleases); + + /// + /// Returns a slot claimed by . Only ever called by the owner of that slot. + /// + /// + /// Ownership is a discipline, not a token — the same call-once contract as `EntityLocker` (#231). + /// The one failure this does defend against is an unbalanced release driving the count negative: + /// this pool is process-wide and lives for the life of the app, so a count of -1 would silently + /// and permanently admit one extra unthrottled transcode, re-opening the #529 QSV pool + /// exhaustion with nothing in the logs to find it by. Clamping keeps the resource guarantee + /// intact and records the breakage in instead of throwing — + /// the single caller releases from a `finally`, where a throw would swallow the real exception. + /// + public void Release() + { + if (Interlocked.Decrement(ref _count) >= 0) + { + return; + } + + Interlocked.Increment(ref _unbalancedReleases); + + while (true) + { + int current = Volatile.Read(ref _count); + if (current >= 0 || Interlocked.CompareExchange(ref _count, 0, current) == current) + { + return; + } + } + } +} diff --git a/ErsatzTV.Core.Tests/Streaming/WorkAheadSlotsTests.cs b/ErsatzTV.Core.Tests/Streaming/WorkAheadSlotsTests.cs new file mode 100644 index 000000000..eda97f1d6 --- /dev/null +++ b/ErsatzTV.Core.Tests/Streaming/WorkAheadSlotsTests.cs @@ -0,0 +1,200 @@ +using ErsatzTV.Application.Streaming; +using NUnit.Framework; +using Shouldly; + +namespace ErsatzTV.Core.Tests.Streaming; + +// Verifies the atomic-claim contract of the work-ahead slot pool (ersatztv#536): at most +// `limit` sessions may hold a slot at once, no matter how many race for one simultaneously. +// +// NEGATIVE CONTROL (per the ersatztv#231/#250 lesson — a one-shot Barrier + Task.WhenAll does NOT +// catch this class on this hardware): to prove these tests are non-vacuous, temporarily replace the +// compare-exchange in WorkAheadSlots.TryAcquire with the check-then-act shape this issue fixed — +// +// int current = Volatile.Read(ref _count); +// if (current >= limit) return false; +// Interlocked.Increment(ref _count); +// return true; +// +// — and TryAcquire_ParallelCallers_NeverExceedsLimit must FAIL (badRounds > 0). Do NOT "break" it by +// stubbing `if (true)`: that leaves values assigned-but-never-read, and CS0219 under +// warnings-as-errors fails the build silently, so `dotnet test --no-build` then runs the STALE +// (fixed) dll and the control falsely passes. Always grep the build output for `error CS` first. +[TestFixture] +public class WorkAheadSlotsTests +{ + [Test] + public void TryAcquire_BelowLimit_Succeeds() + { + var slots = new WorkAheadSlots(); + + slots.TryAcquire(2).ShouldBeTrue(); + slots.TryAcquire(2).ShouldBeTrue(); + slots.TryAcquire(2).ShouldBeFalse(); + slots.Count.ShouldBe(2); + } + + [Test] + public void TryAcquire_DoesNotConsumeASlotWhenItFails() + { + var slots = new WorkAheadSlots(); + + slots.TryAcquire(1).ShouldBeTrue(); + slots.TryAcquire(1).ShouldBeFalse(); + slots.Count.ShouldBe(1); + + // the failed attempt must not have leaked a slot: releasing the one real holder frees the pool + slots.Release(); + slots.Count.ShouldBe(0); + slots.TryAcquire(1).ShouldBeTrue(); + } + + [Test] + public void TryAcquire_WithNonPositiveLimit_NeverSucceeds() + { + var slots = new WorkAheadSlots(); + + slots.TryAcquire(0).ShouldBeFalse(); + slots.TryAcquire(-1).ShouldBeFalse(); + slots.Count.ShouldBe(0); + } + + // The pool is process-wide and never recreated, so a negative count would not self-heal: it would + // permanently admit more than `limit` unthrottled transcodes, with nothing in the logs to find it by. + [Test] + public void Release_WithoutAcquire_ClampsAtZeroAndIsRecorded() + { + var slots = new WorkAheadSlots(); + + slots.Release(); + + slots.Count.ShouldBe(0); + slots.UnbalancedReleases.ShouldBe(1); + + // the budget is intact: a limit of 1 still admits exactly one holder, not two + slots.TryAcquire(1).ShouldBeTrue(); + slots.TryAcquire(1).ShouldBeFalse(); + } + + [Test] + public void UnbalancedReleases_IsZeroUnderCorrectUse() + { + var slots = new WorkAheadSlots(); + + slots.TryAcquire(1).ShouldBeTrue(); + slots.Release(); + + slots.UnbalancedReleases.ShouldBe(0); + } + + // Hammers the acquire race over many rounds rather than a single simultaneous burst: the + // read->increment window is far too narrow to collide reliably when threads release once. + [Test] + [TestCase(1)] + [TestCase(2)] + [TestCase(3)] + public void TryAcquire_ParallelCallers_NeverExceedsLimit(int limit) + { + const int threads = 8; + const int rounds = 20_000; + + var slots = new WorkAheadSlots(); + var winners = 0; + var badRounds = 0; + + using var startRound = new Barrier(threads); + using var endRound = new Barrier( + threads, + _ => + { + int held = Volatile.Read(ref winners); + if (held > limit || held != slots.Count) + { + Interlocked.Increment(ref badRounds); + } + + // reset for the next round: release every slot claimed this round + for (var i = 0; i < held; i++) + { + slots.Release(); + } + + Volatile.Write(ref winners, 0); + }); + + var workers = new Thread[threads]; + for (var t = 0; t < threads; t++) + { + workers[t] = new Thread(() => + { + for (var r = 0; r < rounds; r++) + { + startRound.SignalAndWait(); + if (slots.TryAcquire(limit)) + { + Interlocked.Increment(ref winners); + } + + endRound.SignalAndWait(); + } + }); + workers[t].Start(); + } + + foreach (Thread worker in workers) + { + worker.Join(); + } + + badRounds.ShouldBe(0); + slots.Count.ShouldBe(0); + slots.UnbalancedReleases.ShouldBe(0); + } + + // The release path is the fiddly half: a slot freed by its owner must become available again, + // and a burst of acquire/release cycles must not drift the count in either direction. + [Test] + public void AcquireAndRelease_UnderContention_LeavesNoLeakedOrDoubleFreedSlots() + { + const int threads = 8; + const int rounds = 20_000; + const int limit = 3; + + var slots = new WorkAheadSlots(); + var overLimit = 0; + var live = 0; + + var workers = new Thread[threads]; + for (var t = 0; t < threads; t++) + { + workers[t] = new Thread(() => + { + for (var r = 0; r < rounds; r++) + { + if (!slots.TryAcquire(limit)) + { + continue; + } + + if (Interlocked.Increment(ref live) > limit) + { + Interlocked.Increment(ref overLimit); + } + + Interlocked.Decrement(ref live); + slots.Release(); + } + }); + workers[t].Start(); + } + + foreach (Thread worker in workers) + { + worker.Join(); + } + + overLimit.ShouldBe(0); + slots.Count.ShouldBe(0); + slots.UnbalancedReleases.ShouldBe(0); + } +} diff --git a/docs/decisions.md b/docs/decisions.md index d4cab234e..9df647f52 100644 --- a/docs/decisions.md +++ b/docs/decisions.md @@ -132,6 +132,7 @@ in-file entries. - [2026-07-21 — Parallel orientation + selection is the startup protocol; #237 retired (#520)](#2026-07-21--parallel-orientation--selection-is-the-startup-protocol-237-retired-520) - [2026-07-21 — External channel-logo URLs are downloaded and cached at save time; the render path never fetches a logo (#525)](#2026-07-21--external-channel-logo-urls-are-downloaded-and-cached-at-save-time-the-render-path-never-fetches-a-logo-525) - [2026-07-21 — Check the worked issue before the decision corpus; a closed tracker's comments need no retrofit (#524)](#2026-07-21--check-the-worked-issue-before-the-decision-corpus-a-closed-trackers-comments-need-no-retrofit-524) +- [2026-07-21 — Work-ahead slots are claimed atomically by the caller, released by the transcode it hands them to (#536)](#2026-07-21--work-ahead-slots-are-claimed-atomically-by-the-caller-released-by-the-transcode-it-hands-them-to-536) --- @@ -2855,6 +2856,14 @@ again — the duplication is what let the defect exist in triplicate. Covered by > made the failure near-deterministic. See `ffmpeg.qsv-extra-hw-frames-floor`; the decision recorded > here still stands, with that floor in place. +> **Correction (2026-07-21, #536):** the bullet below stating that "every concurrent tune-in falls +> back to the throttled path" described the *intent*, not the behaviour. The slot check and its +> increment straddled an `await`, so N simultaneous tune-ins all read `0 < limit` and all started +> unthrottled; the "one winner, two throttled" observation held only because those tunes were +> effectively staggered. The measurement and the conclusions drawn from it stand — slot availability +> *was* the variable behind the bimodality — but the guarantee itself was not enforced until +> `ffmpeg.work-ahead-slot-atomic`. + - **`-readrate` throttles from the first read, so it sets a floor on time-to-first-segment.** The realtime playback path pins input reading to 1.05× wall clock so a channel behaves like live TV. Since `OutputFormatHls.SegmentSeconds` is 4 and the segmenter serves the playlist only once the @@ -3316,3 +3325,40 @@ that skips the lore bucket will leave real knowledge on the floor. **A zero result is a legitimate outcome**, not evidence the triage was done wrong. What would change the answer is a tracker whose sessions did *not* also write per-issue closing records — one where the tracker genuinely was the primary rather than the narration layer. + +## 2026-07-21 — Work-ahead slots are claimed atomically by the caller, released by the transcode it hands them to (#536) + +`key: ffmpeg.work-ahead-slot-atomic` · `status: active` · `since: 2026-07-21` · `supersedes: none` · `superseded-by: none` +**Rule:** `workAheadSegmenterLimit` is enforced by a single compare-exchange claim on a shared `WorkAheadSlots` pool taken by the *caller* of `Transcode`, which then passes ownership in and gets the release in `Transcode`'s `finally` — never a `Volatile.Read` compare in one place and an `Interlocked.Increment` in another. +**Signals:** work-ahead slot, workAheadSegmenterLimit, check-then-act across an await, unthrottled tune-in, TOCTOU · paths: `ErsatzTV.Application/Streaming/WorkAheadSlots.cs`, `HlsSessionWorker.Run`/`Transcode`, `ErsatzTV.Core.Tests/Streaming/WorkAheadSlotsTests.cs` · issues: #536, #350, #529, #231, #250 +**Mechanics:** `WorkAheadSlots.TryAcquire(limit)` CAS loop; `Transcode(bool ownsWorkAheadSlot, …)` + +**The defect was structural, not a missing `Interlocked`.** The write side already used +`Interlocked.Increment`, which is why the code read as thread-safe. The read side was a separate, +earlier `Volatile.Read(ref _workAheadCount) < await GetWorkAheadLimit(...)` in `Run`, and the +increment happened later inside `Transcode` — with at least one `await` (a DB-backed config read) in +between. `Interlocked` on one half of a check-then-act buys nothing. Three simultaneous tune-ins on +prod with `workAheadSegmenterLimit: 1` all observed `0 < 1` and all ran with no `-readrate`. + +**Why the caller claims and the callee releases.** Moving the whole acquire/release pair inside +`Transcode` would be more symmetric, but `Run` needs the outcome *before* the call: it sets +`_state = SeekAndWorkAhead | SeekAndRealtime`, and `Transcode` reads that state on entry +(`wasSeekAndWorkAhead`) to decide whether the item starts at `DateTimeOffset.Now` or at +`_transcodedUntil`. Folding acquisition inward would have silently changed that branch. Instead the +parameter was inverted — `Transcode(bool ownsWorkAheadSlot, …)` with `realtime = !ownsWorkAheadSlot` +derived on the first line — so the contract is stated in the signature rather than implied by a +double negative, and there is exactly one release site guarded by the same flag. + +**Compare-exchange rather than increment-then-back-off.** The issue proposed +`Interlocked.Increment` followed by a decrement when the post-increment value exceeds the limit. +That is correct on holder count, but the counter transiently overshoots, so a concurrent reader of +`Count` can observe a value above the limit. The CAS loop never publishes a state that violates the +invariant, which matters because the QSV hardware-frame pool sizing (`ffmpeg.qsv-extra-hw-frames-floor`) +is derived from that bound. + +**Testing shape is inherited from #231/#250.** A single `Barrier(N)` + `Task.WhenAll` round does not +reliably collide on this hardware; the tests hammer 8 threads × 20 000 rounds with a per-round +barrier that validates the winner count and resets the pool. The negative control is documented in +the test file: reinstate the check-then-act body (**not** `if (true)`, which trips CS0219 under +warnings-as-errors and leaves `--no-build` running a stale, still-fixed dll). Verified: with the +pre-fix shape, 15 912 of 20 000 rounds over-claimed. diff --git a/docs/decisions/README.md b/docs/decisions/README.md index 9744bdc7d..52332d07d 100644 --- a/docs/decisions/README.md +++ b/docs/decisions/README.md @@ -53,6 +53,7 @@ the link for rationale. Superseded/retired history lives in `archive/`. Regenera | `ffmpeg.qsv-decode-encode-split` | QSV decode is decoupled from QSV encode via a single `FFmpegProfile.QsvPreferNativeDecoder` bool (default ON, Linux-only), so a QSV encode profile can decode with the more tolerant native VA-API decoder instead of the QSV decoder, mirroring Jellyfin's hybrid decode/encode toggle instead of a general decode-family enum. | 2026-07-20 | [link](../decisions.md#2026-07-20-498--qsv-decode-is-split-from-qsv-encode-via-a-single-qsvprefernativedecoder-bool) | | `ffmpeg.qsv-extra-hw-frames-floor` | a QSV upload never emits `extra_hw_frames` below `FFmpegState.MinimumQsvExtraHardwareFrames` (64); a stored `0` or negative value is treated as "no pool configured" rather than honored literally, because with no headroom any unthrottled read exhausts the pool and the transcode writes nothing at all. | 2026-07-21 | [link](../decisions.md#2026-07-21--qsv-hardware-frame-headroom-is-a-floor-not-an-operator-preference-529) | | `ffmpeg.remote-image-fetcher-bounded` | remote graphics-engine images are fetched through `IRemoteImageFetcher` with a pooled `HttpClientFactory` client, a body-covering deadline, a wire-transfer size cap, and a decoder-enforced `DecoderOptions.MaxFrames` bound re-verified post-decode — never cached, re-fetched per element init. | 2026-07-20 | [link](../decisions.md#2026-07-20--remote-graphics-engine-images-are-fetched-through-a-bounded-pooled-iremoteimagefetcher-re-fetched-per-element-init-not-cached-511) | +| `ffmpeg.work-ahead-slot-atomic` | `workAheadSegmenterLimit` is enforced by a single compare-exchange claim on a shared `WorkAheadSlots` pool taken by the *caller* of `Transcode`, which then passes ownership in and gets the release in `Transcode`'s `finally` — never a `Volatile.Read` compare in one place and an `Interlocked.Increment` in another. | 2026-07-21 | [link](../decisions.md#2026-07-21--work-ahead-slots-are-claimed-atomically-by-the-caller-released-by-the-transcode-it-hands-them-to-536) | | `graphics.channel-logo-caching` | An external `http(s)` channel-logo URL is fetched, decode-budget-validated, and stored in the image cache under a content-hash name at SAVE time — becoming byte-identical to an uploaded logo — so the render path never fetches a logo over HTTP; a bad URL fails the save with a 422 (BaseError → ValidationProblemDetails). | 2026-07-21 | [link](../decisions.md#2026-07-21--external-channel-logo-urls-are-downloaded-and-cached-at-save-time-the-render-path-never-fetches-a-logo-525) | | `iptv.base-url` | An optional advertised base URL (`iptv.base_url`) is resolved centrally via a pure Core helper (`AdvertisedBaseUrl`) inside the two IPTV generation handlers (M3U + XMLTV); unset/malformed values fall back byte-identical to the request-derived host, and it's a new `iptv` settings group distinct from `ETV_BASE_URL` and out of scope for HDHomeRun. | 2026-07-16 | [link](../decisions.md#2026-07-16--optional-advertised-iptv-base-url-iptvbase_url-resolved-centrally-in-the-two-generators-340) | | `iptv.logo-drives-bug-preset` | One uploaded channel logo drives both the listing logo and the on-screen bug via a shared, seeded `ChannelLogo`-sourced watermark preset (`Channel Bug`), not new per-channel schema. | 2026-07-20 | [link](../decisions.md#2026-07-20--one-logo-drives-the-bug-via-a-shared-channellogo-preset-not-new-schema-67) |