diff --git a/src/AcDream.Headless/Hosting/HeadlessProcessHost.cs b/src/AcDream.Headless/Hosting/HeadlessProcessHost.cs index db260f7d..9cb0dd73 100644 --- a/src/AcDream.Headless/Hosting/HeadlessProcessHost.cs +++ b/src/AcDream.Headless/Hosting/HeadlessProcessHost.cs @@ -156,10 +156,44 @@ internal sealed class HeadlessProcessHost : IDisposable Credential = source.Credential, }; - internal async Task RunAsync( + internal Task RunAsync( CancellationToken cancellationToken) { ObjectDisposedException.ThrowIf(_disposed, this); + // #368: Runtime's gameplay owners require ONE update thread for a + // session's whole lifetime — collision generations bind to the + // first mutating thread and refuse migration. The graphical host + // satisfies that with its game-loop thread; this dedicated thread + // is the headless equivalent. Start (the live connect + // transaction), every scheduler turn, and the post-loop captures + // all execute here. Only disposal stays on the lifecycle thread, + // which the Runtime teardown path explicitly supports (see + // ResetSessionPhysics's own doc comment). + var completion = new TaskCompletionSource( + TaskCreationOptions.RunContinuationsAsynchronously); + var thread = new Thread(() => + { + try + { + completion.SetResult( + RunOnUpdateThread(cancellationToken)); + } + catch (Exception error) + { + completion.SetException(error); + } + }) + { + IsBackground = true, + Name = "acdream-headless-update", + }; + thread.Start(); + return completion.Task; + } + + private HeadlessExitCode RunOnUpdateThread( + CancellationToken cancellationToken) + { foreach (HeadlessSessionHost session in _sessions) { RuntimeSessionStartResult started = session.Start(); @@ -189,8 +223,7 @@ internal sealed class HeadlessProcessHost : IDisposable try { - await _scheduler.RunAsync(cancellationToken) - .ConfigureAwait(false); + _scheduler.Run(cancellationToken); } catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) diff --git a/src/AcDream.Headless/Hosting/HeadlessProcessScheduler.cs b/src/AcDream.Headless/Hosting/HeadlessProcessScheduler.cs index 525a1197..19a3530e 100644 --- a/src/AcDream.Headless/Hosting/HeadlessProcessScheduler.cs +++ b/src/AcDream.Headless/Hosting/HeadlessProcessScheduler.cs @@ -147,8 +147,31 @@ internal sealed class HeadlessProcessScheduler _observationPeriodTicks); } - internal async Task RunAsync(CancellationToken cancellationToken) + /// + /// Drives every session's deadlines on the CALLING thread until + /// cancellation or policy completion, returning normally in both + /// cases. The caller must dedicate one thread for a process's whole + /// run: Runtime's gameplay owners (collision generations foremost, + /// via RuntimePhysicsState.EnsureCollisionMutationThread) bind + /// to the first mutating thread and refuse migration, and an awaited + /// timer loop in a SynchronizationContext-free host resumes on + /// arbitrary ThreadPool workers — which tripped that guard whenever a + /// collision generation spanned two waits (#368). The waits below go + /// through one rearmed timer signalling an + /// event, so the loop never leaves its thread. A stale timer callback + /// from an abandoned wait can set the event early; that only costs + /// one extra pass over the deadline math, which re-sleeps. + /// + internal void Run(CancellationToken cancellationToken) { + using var wake = new ManualResetEventSlim(false); + using ITimer timer = _timeProvider.CreateTimer( + static state => ((ManualResetEventSlim)state!).Set(), + wake, + Timeout.InfiniteTimeSpan, + Timeout.InfiniteTimeSpan); + WaitHandle[] waitHandles = + [wake.WaitHandle, cancellationToken.WaitHandle]; while (!cancellationToken.IsCancellationRequested && HasActiveSession()) { @@ -169,11 +192,9 @@ internal sealed class HeadlessProcessScheduler : _timeProvider.GetElapsedTime(now, deadline); delay = NormalizeTimerDelay(delay); Interlocked.Increment(ref _waitCount); - await Task.Delay( - delay, - _timeProvider, - cancellationToken) - .ConfigureAwait(false); + wake.Reset(); + timer.Change(delay, Timeout.InfiniteTimeSpan); + WaitHandle.WaitAny(waitHandles); } } diff --git a/tests/AcDream.Headless.Tests/HeadlessProcessSchedulerTests.cs b/tests/AcDream.Headless.Tests/HeadlessProcessSchedulerTests.cs index 532251a7..87a768a2 100644 --- a/tests/AcDream.Headless.Tests/HeadlessProcessSchedulerTests.cs +++ b/tests/AcDream.Headless.Tests/HeadlessProcessSchedulerTests.cs @@ -1,3 +1,5 @@ +using System.Collections.Concurrent; +using System.Diagnostics; using System.Net; using AcDream.Core.Net; using AcDream.Core.Net.Messages; @@ -198,7 +200,7 @@ public sealed class HeadlessProcessSchedulerTests } [Fact] - public async Task SystemTimerCadenceDoesNotBusyLoopBetweenTurns() + public void SystemTimerCadenceDoesNotBusyLoopBetweenTurns() { var operations = new FixtureSessionOperations(); using HeadlessSessionHost session = @@ -214,14 +216,10 @@ public sealed class HeadlessProcessSchedulerTests using var cancellation = new CancellationTokenSource(TimeSpan.FromMilliseconds(250)); - try - { - await scheduler.RunAsync(cancellation.Token); - } - catch (OperationCanceledException) - when (cancellation.IsCancellationRequested) - { - } + // Run executes on the calling thread and returns normally when + // the token cancels (#368 moved thread ownership to the process + // host; the scheduler seam itself is synchronous). + scheduler.Run(cancellation.Token); HeadlessSchedulerSnapshot snapshot = scheduler.CaptureSnapshot(); @@ -440,6 +438,63 @@ public sealed class HeadlessProcessSchedulerTests } } + [Fact] + public async Task ProcessHostRunsStartAndEveryTickOnOneDedicatedUpdateThread() + { + // #368: Runtime's gameplay owners (RuntimePhysicsState's collision + // generations foremost) require one update thread for a session's + // whole lifetime — the graphical host provides its game-loop thread; + // the headless host must provide an equivalent. Start() (the live + // connect transaction) and every subsequent tick must share one + // thread that is NOT the RunAsync caller's, across real timer waits. + var configuration = new HeadlessConfiguration + { + Version = 1, + Sessions = + [ + Descriptor( + "update-thread", + "update-thread-stdin"), + ], + }; + var operations = new ThreadRecordingSessionOperations(); + using var diagnostics = new StringWriter(); + using var host = new HeadlessProcessHost( + configuration, + HeadlessPathSet.Resolve(new HeadlessPathOverrides()), + new System.IO.StringReader( + "update-thread-password" + Environment.NewLine), + diagnostics, + operations); + using var cancellation = new CancellationTokenSource(); + + int callerThread = Environment.CurrentManagedThreadId; + Task run = host.RunAsync(cancellation.Token); + // Occupy the calling thread until ticks have crossed several timer + // waits — pre-fix, resumptions migrate to other pool threads while + // this thread is provably unavailable to them. The await happens + // only after the measurement window closes. + var stopwatch = Stopwatch.StartNew(); + while (operations.TickCount < 5 + && stopwatch.Elapsed < TimeSpan.FromSeconds(10)) + { + Thread.Sleep(1); + } + cancellation.Cancel(); + HeadlessExitCode result = await run; + + Assert.Equal(HeadlessExitCode.Success, result); + Assert.True( + operations.TickCount >= 5, + $"Expected at least 5 ticks, observed {operations.TickCount}."); + int updateThread = operations.ConnectThreadId; + Assert.NotEqual(0, updateThread); + Assert.NotEqual(callerThread, updateThread); + Assert.All( + operations.TickThreadIds, + id => Assert.Equal(updateThread, id)); + } + [Fact] public void ThrowingPolicyQuarantinesOnlyItsOwnSession() { @@ -609,6 +664,60 @@ public sealed class HeadlessProcessSchedulerTests } } + private sealed class ThreadRecordingSessionOperations + : ILiveSessionOperations + { + private int _connectThreadId; + private readonly ConcurrentQueue _tickThreadIds = new(); + + internal int ConnectThreadId => + Volatile.Read(ref _connectThreadId); + internal int TickCount => _tickThreadIds.Count; + internal IReadOnlyCollection TickThreadIds => _tickThreadIds; + + public IPEndPoint ResolveEndpoint(string host, int port) => + new(IPAddress.Loopback, port); + + public WorldSession CreateSession(IPEndPoint endpoint) => + new(endpoint); + + public void Connect( + WorldSession session, + string user, + string password) => + Volatile.Write( + ref _connectThreadId, + Environment.CurrentManagedThreadId); + + public CharacterList.Parsed GetCharacters( + WorldSession session) => + new( + 0u, + [ + new CharacterList.Character( + 0x50000001u, + "Headless", + 0u), + ], + [], + 11, + "account", + true, + true); + + public void EnterWorld( + WorldSession session, + int activeCharacterIndex) + { + } + + public void Tick(WorldSession session) => + _tickThreadIds.Enqueue(Environment.CurrentManagedThreadId); + + public void DisposeSession(WorldSession session) => + session.Dispose(); + } + private abstract class FixturePolicy : IHeadlessBotPolicy { public virtual bool IsComplete => false;