diff --git a/src/Runtime/Internal/Execution/MutationExecutionConcurrencyGate.cs b/src/Runtime/Internal/Execution/MutationExecutionConcurrencyGate.cs index 74dd141..fbb0a5e 100644 --- a/src/Runtime/Internal/Execution/MutationExecutionConcurrencyGate.cs +++ b/src/Runtime/Internal/Execution/MutationExecutionConcurrencyGate.cs @@ -1,30 +1,130 @@ using System.Collections.Concurrent; +using System.Runtime.CompilerServices; namespace ModularityKit.Mutator.Runtime.Internal.Execution; /// -/// Coordinates core mutation execution concurrency across the engine. +/// Coordinates concurrent mutation execution using global concurrency limit +/// and per state serialization. /// -internal sealed class MutationExecutionConcurrencyGate(int maxConcurrentMutations) +/// +/// When state identifier is provided, an additional per state gate ensures that +/// mutations targeting the same state are executed sequentially, while mutations +/// targeting different states may execute concurrently up to the global limit. +/// +/// The maximum number of mutations that may execute concurrently across all states. +internal sealed class MutationExecutionConcurrencyGate( + int maxConcurrentMutations) { - private readonly SemaphoreSlim _globalGate = new(maxConcurrentMutations, maxConcurrentMutations); - private readonly ConcurrentDictionary _stateGates = new(StringComparer.Ordinal); + private readonly SemaphoreSlim _globalGate = + new(maxConcurrentMutations, maxConcurrentMutations); - public async ValueTask EnterAsync(string? stateId, CancellationToken cancellationToken) + private readonly ConcurrentDictionary _stateGates = + new(StringComparer.Ordinal); + + /// + /// Enters the concurrency gate for mutation execution. + /// + /// The identifier of the state being mutated. + /// Token that can be used to cancel waiting for the required concurrency gates. + /// An asynchronous lease that releases the acquired concurrency gates when disposed. + [MethodImpl(MethodImplOptions.AggressiveInlining)] + public ValueTask EnterAsync( + string? stateId, + CancellationToken cancellationToken) { - await _globalGate.WaitAsync(cancellationToken).ConfigureAwait(false); + if (stateId is null) + return EnterGlobalOnlyAsync(cancellationToken); - var stateGate = default(SemaphoreSlim); + return EnterWithStateAsync(stateId, cancellationToken); + } - try + /// + /// Attempts to acquire the global concurrency gate without asynchronous waiting + /// when capacity is immediately available. + /// + private ValueTask EnterGlobalOnlyAsync( + CancellationToken cancellationToken) + { + if (_globalGate.Wait(0, cancellationToken)) + { + return new ValueTask( + new Lease(_globalGate, null)); + } + + return SlowEnterGlobalOnlyAsync(cancellationToken); + } + + /// + /// Asynchronously waits for the global concurrency gate when it could not be + /// acquired immediately. + /// + private async ValueTask SlowEnterGlobalOnlyAsync( + CancellationToken cancellationToken) + { + await _globalGate + .WaitAsync(cancellationToken) + .ConfigureAwait(false); + + return new Lease(_globalGate, null); + } + + /// + /// Attempts to acquire both the global concurrency gate and the gate associated + /// with the specified state. + /// + /// + /// The global gate is always acquired before the per state gate to maintain + /// consistent acquisition order and avoid lock order inversion. + /// + private ValueTask EnterWithStateAsync( + string stateId, + CancellationToken cancellationToken) + { + if (_stateGates.TryGetValue(stateId, out var existing)) { - if (!string.IsNullOrWhiteSpace(stateId)) + if (_globalGate.Wait(0, cancellationToken)) { - stateGate = _stateGates.GetOrAdd(stateId, static _ => new SemaphoreSlim(1, 1)); - await stateGate.WaitAsync(cancellationToken).ConfigureAwait(false); + if (existing.Wait(0, cancellationToken)) + { + return new ValueTask( + new Lease(_globalGate, existing)); + } + + _globalGate.Release(); } + } + + return SlowEnterWithStateAsync( + stateId, + cancellationToken); + } + + /// + /// Asynchronously acquires the global and per state gates when the fast path + /// could not acquire them immediately. + /// + private async ValueTask SlowEnterWithStateAsync( + string stateId, + CancellationToken cancellationToken) + { + await _globalGate + .WaitAsync(cancellationToken) + .ConfigureAwait(false); - return new Lease(_globalGate, stateGate); + try + { + var stateGate = _stateGates.GetOrAdd( + stateId, + static _ => new SemaphoreSlim(1, 1)); + + await stateGate + .WaitAsync(cancellationToken) + .ConfigureAwait(false); + + return new Lease( + _globalGate, + stateGate); } catch { @@ -34,24 +134,23 @@ public async ValueTask EnterAsync(string? stateId, CancellationToken canc } /// - /// Represents an acquired execution slot. + /// Represents an acquired concurrency lease that releases the associated gates + /// when disposed. /// - internal readonly struct Lease : IAsyncDisposable + internal readonly struct Lease( + SemaphoreSlim globalGate, + SemaphoreSlim? stateGate) : IAsyncDisposable { - private readonly SemaphoreSlim _globalGate; - private readonly SemaphoreSlim? _stateGate; - - public Lease(SemaphoreSlim globalGate, SemaphoreSlim? stateGate) - { - _globalGate = globalGate; - _stateGate = stateGate; - } - + /// + /// Releases the per state gate, when present, followed by the global gate. + /// + [MethodImpl(MethodImplOptions.AggressiveInlining)] public ValueTask DisposeAsync() { - _stateGate?.Release(); - _globalGate.Release(); + stateGate?.Release(); + globalGate.Release(); + return ValueTask.CompletedTask; } } -} +} \ No newline at end of file