A composite splits into four horizontal row strips across four DEDICATED background threads; the calling thread takes a strip too (so four workers never mean five-way oversubscription), and anything under 64 rows runs inline. BlitContentRaw, BlitBakedElement and BlitOverlay now fan out. Output is byte-identical — strips already partitioned by row. The demotion deliberately does NOT come from Parallel.For. The pool is shared, so Thread.CurrentThread.Priority = BelowNormal inside it leaks into every unrelated work item that thread later runs — a full test run after trying it broke AudioPipelineTests.Mix_HonorsProviderGains_AndGameMute_ KillsTheLoopback. Established guidance agrees: use your own producer/consumer queue with your own Threads and set their priority yourself (https://stackoverflow.com/questions/524483/net-backgroundworker-objects-thread-priority, https://stackoverflow.com/questions/1578599/c-sharp-lower-thread-priority-in-thread-pool). Helpers/StripPool.cs is that queue. Four threads is the measured sweet spot: fewer leaves render over budget, more steals cycles from the game and the encoder and made render worse. The scaled blit's column LUT moves from a shared static buffer to ArrayPool<float>.Shared with a finally — a single static buffer is a data race the moment strips run in parallel. Gate: SceneCompositorTests.StripParallel_RepeatedRenders_AreByteIdentical — 8 renders of a scaled round webcam + mirrored image + master-sized flash over a short backdrop must be byte-for-byte identical, and must differ from a backdrop-only render so it cannot pass vacuously. It caught a loop-variable capture (new Thread(() => Worker(i)) saw i == Workers) that crashed the test host on the first run.
204 lines
7.2 KiB
C#
204 lines
7.2 KiB
C#
using System;
|
|
using System.Threading;
|
|
|
|
namespace ytLive.Helpers;
|
|
|
|
/// <summary>
|
|
/// A fixed pool of four DEDICATED background threads at <see cref="ThreadPriority.BelowNormal"/>
|
|
/// that split one composite into horizontal row strips. The compositor needs the split to get
|
|
/// render inside a 16.67ms slot, but it must not compete with the encoder, the WGC capture
|
|
/// callbacks, the WebView2 chat or the game the user is playing — hence the hard cap of four
|
|
/// and the demotion.
|
|
///
|
|
/// Research (2026-10-05, Unit 5): the demotion must NOT be done by poking
|
|
/// <c>Thread.CurrentThread.Priority</c> inside <c>Parallel.For</c> — the pool is shared, so the
|
|
/// demotion leaks into every unrelated work item that thread later runs
|
|
/// (https://stackoverflow.com/questions/524483/net-backgroundworker-objects-thread-priority:
|
|
/// "you have next to no control of which thread pool thread will actually run your task");
|
|
/// https://stackoverflow.com/questions/1578599/c-sharp-lower-thread-priority-in-thread-pool:
|
|
/// "You shouldn't mess with the priority of the regular ThreadPool, since you aren't the only
|
|
/// consumer ... write your own producer/consumer queue, using your own Thread(s) as the
|
|
/// consumer(s) — setting the thread priority yourself"). A real failed attempt confirmed it: a
|
|
/// full test run after demoting the pool threads broke an unrelated audio-mix timing test.
|
|
/// This class IS that producer/consumer queue; the caller thread is the producer and also runs
|
|
/// one strip itself, so four workers never means five-way oversubscription.
|
|
///
|
|
/// Contract: <see cref="For"/> blocks until every strip has run, each strip must write only its
|
|
/// own rows (the compositor's blits already partition by row), and it must not be called from
|
|
/// inside a strip (the blits never nest).
|
|
/// </summary>
|
|
internal static class StripPool
|
|
{
|
|
/// <summary>Four workers: the measured sweet spot on this box. Fewer leaves render over
|
|
/// budget; more steals cycles from the game and the encoder and made render worse, not better.</summary>
|
|
public const int Workers = 4;
|
|
|
|
/// <summary>Below this many rows the fan-out costs more than it saves — run it inline.</summary>
|
|
private const int MinRowsForFanOut = 64;
|
|
|
|
private static readonly object[] _queueLocks = CreateLocks();
|
|
private static readonly Queue<Strip>[] _queues = CreateQueues();
|
|
private static readonly Queue<Strip>[] _free = CreateQueues();
|
|
private static readonly SemaphoreSlim[] _signals = CreateSignals();
|
|
private static Thread[]? _threads;
|
|
private static int _roundRobin = -1;
|
|
|
|
private sealed class Strip
|
|
{
|
|
public int From;
|
|
public int To; // exclusive
|
|
public Action<int>? Body;
|
|
public Batch? Owner;
|
|
}
|
|
|
|
private sealed class Batch
|
|
{
|
|
public int Remaining;
|
|
public Exception? Error;
|
|
}
|
|
|
|
/// <summary>Run <paramref name="body"/> for every row in [from, to), split across the pool.</summary>
|
|
public static void For(int from, int to, Action<int> body)
|
|
{
|
|
if (from >= to) return;
|
|
|
|
var rows = to - from;
|
|
if (rows < MinRowsForFanOut || !EnsureStarted())
|
|
{
|
|
for (var y = from; y < to; y++) body(y);
|
|
return;
|
|
}
|
|
|
|
// Strips are handed to whole workers (not per-row tasks) so the strip boundaries —
|
|
// and therefore the destination row each strip owns — are fixed before any of them run.
|
|
var strips = Math.Min(Workers, rows);
|
|
var chunk = (rows + strips - 1) / strips;
|
|
var batch = new Batch { Remaining = strips - 1 }; // the caller runs the last one
|
|
|
|
var first = from;
|
|
for (var k = 0; k < strips - 1; k++)
|
|
{
|
|
var end = Math.Min(first + chunk, to);
|
|
var w = NextWorker();
|
|
var strip = Rent(w);
|
|
strip.From = first;
|
|
strip.To = end;
|
|
strip.Body = body;
|
|
strip.Owner = batch;
|
|
lock (_queueLocks[w]) _queues[w].Enqueue(strip);
|
|
_signals[w].Release();
|
|
first = end;
|
|
}
|
|
|
|
try
|
|
{
|
|
for (var y = first; y < to; y++) body(y);
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
batch.Error ??= ex;
|
|
}
|
|
|
|
if (batch.Remaining > 0)
|
|
{
|
|
lock (batch)
|
|
{
|
|
while (batch.Remaining > 0) Monitor.Wait(batch);
|
|
}
|
|
}
|
|
|
|
if (batch.Error != null) throw new InvalidOperationException("A compositor strip failed.", batch.Error);
|
|
}
|
|
|
|
private static void Worker(int index)
|
|
{
|
|
var signal = _signals[index];
|
|
while (true)
|
|
{
|
|
signal.Wait();
|
|
Strip? strip = null;
|
|
lock (_queueLocks[index])
|
|
{
|
|
if (_queues[index].Count > 0) strip = _queues[index].Dequeue();
|
|
}
|
|
if (strip == null) continue; // released for an already-drained queue
|
|
var batch = strip.Owner!;
|
|
try
|
|
{
|
|
for (var y = strip.From; y < strip.To; y++) strip.Body!(y);
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
batch.Error ??= ex;
|
|
}
|
|
finally
|
|
{
|
|
strip.Body = null;
|
|
strip.Owner = null;
|
|
lock (_queueLocks[index]) _free[index].Enqueue(strip);
|
|
lock (batch)
|
|
{
|
|
if (--batch.Remaining == 0) Monitor.PulseAll(batch);
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
private static Strip Rent(int worker)
|
|
{
|
|
lock (_queueLocks[worker])
|
|
return _free[worker].Count > 0 ? _free[worker].Dequeue() : new Strip();
|
|
}
|
|
|
|
private static int NextWorker()
|
|
{
|
|
var w = Interlocked.Increment(ref _roundRobin) % Workers;
|
|
return w < 0 ? w + Workers : w;
|
|
}
|
|
|
|
/// <summary>Start the workers on first use, so a WPF cold start pays nothing for them.</summary>
|
|
private static bool EnsureStarted()
|
|
{
|
|
if (_threads != null) return true;
|
|
lock (_signals)
|
|
{
|
|
if (_threads != null) return true;
|
|
var threads = new Thread[Workers];
|
|
for (var i = 0; i < Workers; i++)
|
|
{
|
|
var index = i; // capture per-iteration: the loop variable would be Workers by the time the thread runs
|
|
var t = new Thread(() => Worker(index), 256 * 1024)
|
|
{
|
|
IsBackground = true,
|
|
Name = $"compositor-strip-{i}",
|
|
Priority = ThreadPriority.BelowNormal,
|
|
};
|
|
threads[i] = t;
|
|
t.Start();
|
|
}
|
|
_threads = threads;
|
|
return true;
|
|
}
|
|
}
|
|
|
|
private static object[] CreateLocks()
|
|
{
|
|
var l = new object[Workers];
|
|
for (var i = 0; i < Workers; i++) l[i] = new object();
|
|
return l;
|
|
}
|
|
|
|
private static Queue<Strip>[] CreateQueues()
|
|
{
|
|
var q = new Queue<Strip>[Workers];
|
|
for (var i = 0; i < Workers; i++) q[i] = new Queue<Strip>();
|
|
return q;
|
|
}
|
|
|
|
private static SemaphoreSlim[] CreateSignals()
|
|
{
|
|
var s = new SemaphoreSlim[Workers];
|
|
for (var i = 0; i < Workers; i++) s[i] = new SemaphoreSlim(0, int.MaxValue);
|
|
return s;
|
|
}
|
|
} |