diff --git a/src/libraries/Microsoft.Extensions.Primitives/ref/Microsoft.Extensions.Primitives.cs b/src/libraries/Microsoft.Extensions.Primitives/ref/Microsoft.Extensions.Primitives.cs index a5f5f188981cd6..2db31d9fb182a7 100644 --- a/src/libraries/Microsoft.Extensions.Primitives/ref/Microsoft.Extensions.Primitives.cs +++ b/src/libraries/Microsoft.Extensions.Primitives/ref/Microsoft.Extensions.Primitives.cs @@ -16,7 +16,9 @@ public CancellationChangeToken(System.Threading.CancellationToken cancellationTo public static partial class ChangeToken { public static System.IDisposable OnChange(System.Func changeTokenProducer, System.Action changeTokenConsumer) { throw null; } + public static System.IDisposable OnChange(System.Func changeTokenProducer, System.Func changeTokenConsumer) { throw null; } public static System.IDisposable OnChange(System.Func changeTokenProducer, System.Action changeTokenConsumer, TState state) { throw null; } + public static System.IDisposable OnChange(System.Func changeTokenProducer, System.Func changeTokenConsumer, TState state) { throw null; } } public partial class CompositeChangeToken : Microsoft.Extensions.Primitives.IChangeToken { diff --git a/src/libraries/Microsoft.Extensions.Primitives/src/ChangeToken.cs b/src/libraries/Microsoft.Extensions.Primitives/src/ChangeToken.cs index 250fff01ef9c5d..6ad3938e5947c8 100644 --- a/src/libraries/Microsoft.Extensions.Primitives/src/ChangeToken.cs +++ b/src/libraries/Microsoft.Extensions.Primitives/src/ChangeToken.cs @@ -3,6 +3,7 @@ using System; using System.Threading; +using System.Threading.Tasks; namespace Microsoft.Extensions.Primitives { @@ -15,8 +16,12 @@ public static class ChangeToken /// Registers the action to be called whenever the token produced changes. /// /// Produces the change token. - /// Action called when the token changes. - /// + /// Action called when the token changes. The token is re-registered once the action returns. + /// An that, when disposed, unregisters the consumer. + /// + /// Exceptions from are propagated to the caller of this method or to the code that triggers the change token. + /// Exceptions from are propagated to the code that triggers the change token. + /// public static IDisposable OnChange(Func changeTokenProducer, Action changeTokenConsumer) { if (changeTokenProducer is null) @@ -28,16 +33,20 @@ public static IDisposable OnChange(Func changeTokenProducer, Acti ThrowHelper.ThrowArgumentNullException(ExceptionArgument.changeTokenConsumer); } - return new ChangeTokenRegistration(changeTokenProducer, callback => callback(), changeTokenConsumer); + return new SyncChangeTokenRegistration(changeTokenProducer, static callback => callback(), changeTokenConsumer); } /// /// Registers the action to be called whenever the token produced changes. /// /// Produces the change token. - /// Action called when the token changes. + /// Action called when the token changes. The token is re-registered once the action returns. /// state for the consumer. - /// + /// An that, when disposed, unregisters the consumer. + /// + /// Exceptions from are propagated to the caller of this method or to the code that triggers the change token. + /// Exceptions from are propagated to the code that triggers the change token. + /// public static IDisposable OnChange(Func changeTokenProducer, Action changeTokenConsumer, TState state) { if (changeTokenProducer is null) @@ -49,65 +58,97 @@ public static IDisposable OnChange(Func changeTokenProduc ThrowHelper.ThrowArgumentNullException(ExceptionArgument.changeTokenConsumer); } - return new ChangeTokenRegistration(changeTokenProducer, changeTokenConsumer, state); + return new SyncChangeTokenRegistration(changeTokenProducer, changeTokenConsumer, state); } - private sealed class ChangeTokenRegistration : IDisposable + /// + /// Registers the function to be called whenever the token produced changes. + /// + /// Produces the change token. + /// Function called when the token changes. The token is only re-registered once the returned completes. + /// An that, when disposed, unregisters the consumer. + /// + /// Exceptions from are propagated to the caller of this method or to the code that triggers the change token. + /// Synchronous exceptions from are propagated to the code that triggers the change token. + /// Asynchronous exceptions from are left unobserved. + /// + public static IDisposable OnChange(Func changeTokenProducer, Func changeTokenConsumer) + { + if (changeTokenProducer is null) + { + ThrowHelper.ThrowArgumentNullException(ExceptionArgument.changeTokenProducer); + } + if (changeTokenConsumer is null) + { + ThrowHelper.ThrowArgumentNullException(ExceptionArgument.changeTokenConsumer); + } + + return new AsyncChangeTokenRegistration>(changeTokenProducer, static callback => callback(), changeTokenConsumer); + } + + /// + /// Registers the function to be called whenever the token produced changes. + /// + /// Produces the change token. + /// Function called when the token changes. The token is only re-registered once the returned completes. + /// state for the consumer. + /// An that, when disposed, unregisters the consumer. + /// + /// Exceptions from are propagated to the caller of this method or to the code that triggers the change token. + /// Synchronous exceptions from are propagated to the code that triggers the change token. + /// Asynchronous exceptions from are left unobserved. + /// + public static IDisposable OnChange(Func changeTokenProducer, Func changeTokenConsumer, TState state) + { + if (changeTokenProducer is null) + { + ThrowHelper.ThrowArgumentNullException(ExceptionArgument.changeTokenProducer); + } + if (changeTokenConsumer is null) + { + ThrowHelper.ThrowArgumentNullException(ExceptionArgument.changeTokenConsumer); + } + + return new AsyncChangeTokenRegistration(changeTokenProducer, changeTokenConsumer, state); + } + + private abstract class ChangeTokenRegistration(Func changeTokenProducer, TState state) : IDisposable { - private readonly Func _changeTokenProducer; - private readonly Action _changeTokenConsumer; - private readonly TState _state; private IDisposable? _disposable; private static readonly NoopDisposable _disposedSentinel = new NoopDisposable(); - public ChangeTokenRegistration(Func changeTokenProducer, Action changeTokenConsumer, TState state) - { - _changeTokenProducer = changeTokenProducer; - _changeTokenConsumer = changeTokenConsumer; - _state = state; + protected TState State { get; } = state; - IChangeToken? token = changeTokenProducer(); + protected Func ChangeTokenProducer { get; } = changeTokenProducer; - RegisterChangeTokenCallback(token); - } + protected abstract void OnChangeTokenFired(); - private void OnChangeTokenFired() + protected void RegisterChangeTokenCallback(IChangeToken? token) { - // The order here is important. We need to take the token and then apply our changes BEFORE - // registering. This prevents us from possible having two change updates to process concurrently. - // - // If the token changes after we take the token, then we'll process the update immediately upon - // registering the callback. - IChangeToken? token = _changeTokenProducer(); - - try - { - _changeTokenConsumer(_state); - } - finally + if (token is null) { - // We always want to ensure the callback is registered - RegisterChangeTokenCallback(token); + return; } - } - private void RegisterChangeTokenCallback(IChangeToken? token) - { - if (token is null) + // If the registration has already been disposed, don't register again. This guards re-registration + // after disposal: registering on a token that has already changed invokes the callback synchronously, which + // would re-run the consumer after disposal. + if (Volatile.Read(ref _disposable) == _disposedSentinel) { return; } - IDisposable registraton = token.RegisterChangeCallback(s => ((ChangeTokenRegistration?)s)!.OnChangeTokenFired(), this); + + IDisposable? registration = token.RegisterChangeCallback(static s => ((ChangeTokenRegistration?)s)!.OnChangeTokenFired(), this); if (token.HasChanged && token.ActiveChangeCallbacks) { - registraton?.Dispose(); + registration?.Dispose(); return; } - SetDisposable(registraton); + SetDisposable(registration); } - private void SetDisposable(IDisposable disposable) + private void SetDisposable(IDisposable? disposable) { // We don't want to transition from _disposedSentinel => anything since it's terminal // but we want to allow going from previously assigned disposable, to another @@ -117,7 +158,7 @@ private void SetDisposable(IDisposable disposable) // If Dispose was called, then immediately dispose the disposable if (current == _disposedSentinel) { - disposable.Dispose(); + disposable?.Dispose(); return; } @@ -127,7 +168,7 @@ private void SetDisposable(IDisposable disposable) if (previous == _disposedSentinel) { // The subscription was disposed so we dispose immediately and return - disposable.Dispose(); + disposable?.Dispose(); } else if (previous == current) { @@ -136,7 +177,7 @@ private void SetDisposable(IDisposable disposable) else { // Sets can never overlap with other SetDisposable calls so we should never get into this situation - throw new InvalidOperationException("Somebody else set the _disposable field"); + ThrowHelper.ThrowInvalidOperationException(ExceptionResource.InvalidOperation_ConcurrentDisposableSet); } } @@ -154,5 +195,106 @@ public void Dispose() } } } + + private sealed class SyncChangeTokenRegistration : ChangeTokenRegistration + { + private readonly Action _changeTokenConsumer; + + public SyncChangeTokenRegistration(Func changeTokenProducer, Action changeTokenConsumer, TState state) + : base(changeTokenProducer, state) + { + _changeTokenConsumer = changeTokenConsumer; + + RegisterChangeTokenCallback(changeTokenProducer()); + } + + protected override void OnChangeTokenFired() + { + // The order here is important. We need to take the token and then apply our changes BEFORE + // registering. This prevents us from possible having two change updates to process concurrently. + // + // If the token changes after we take the token, then we'll process the update immediately upon + // registering the callback. + IChangeToken? token = ChangeTokenProducer(); + + try + { + _changeTokenConsumer(State); + } + finally + { + // We always want to ensure the callback is registered + RegisterChangeTokenCallback(token); + } + } + } + + private sealed class AsyncChangeTokenRegistration : ChangeTokenRegistration + { + private readonly Func _changeTokenConsumer; + + public AsyncChangeTokenRegistration(Func changeTokenProducer, Func changeTokenConsumer, TState state) + : base(changeTokenProducer, state) + { + _changeTokenConsumer = changeTokenConsumer; + + RegisterChangeTokenCallback(changeTokenProducer()); + } + + protected override void OnChangeTokenFired() + { + // The order here is important. We need to take the token and then apply our changes BEFORE + // registering. This prevents us from possible having two change updates to process concurrently. + // + // If the token changes after we take the token, then we'll process the update immediately upon + // registering the callback once the consumer's task completes. + IChangeToken? token = ChangeTokenProducer(); + + Task consumerTask; + try + { + // The consumer is invoked synchronously here, so that synchronous exceptions from it are propagated + // to the code that triggers the change token, just like the sync overload does. + consumerTask = _changeTokenConsumer(State); + } + catch + { + // We always want to ensure the callback is registered, even when the consumer throws synchronously. + RegisterChangeTokenCallback(token); + throw; + } + + if (consumerTask is null) + { + RegisterChangeTokenCallback(token); + ThrowHelper.ThrowInvalidOperationException(ExceptionResource.InvalidOperation_NullConsumerTask); + } + + if (consumerTask.Status == TaskStatus.RanToCompletion) + { + // The common case where the consumer completes synchronously: re-register without allocations. + RegisterChangeTokenCallback(token); + } + else + { + // Asynchronous exceptions can't be propagated without blocking, so they are left unobserved + // (meaning they can be observed only through TaskScheduler.UnobservedTaskException). + _ = AwaitConsumerAndRegisterCallback(consumerTask, token); + } + } + + private async Task AwaitConsumerAndRegisterCallback(Task consumerTask, IChangeToken? token) + { + try + { + await consumerTask.ConfigureAwait(false); + } + finally + { + // We always want to ensure the callback is registered + RegisterChangeTokenCallback(token); + } + } + } } } diff --git a/src/libraries/Microsoft.Extensions.Primitives/src/Resources/Strings.resx b/src/libraries/Microsoft.Extensions.Primitives/src/Resources/Strings.resx index e27880cfaa5858..a5b559fc49ca95 100644 --- a/src/libraries/Microsoft.Extensions.Primitives/src/Resources/Strings.resx +++ b/src/libraries/Microsoft.Extensions.Primitives/src/Resources/Strings.resx @@ -132,4 +132,10 @@ Entire reserved capacity was not used. Capacity: '{0}', written '{1}'. + + The change token callback registration was set concurrently, which is not supported. + + + The change token consumer returned a null task. + \ No newline at end of file diff --git a/src/libraries/Microsoft.Extensions.Primitives/src/ThrowHelper.cs b/src/libraries/Microsoft.Extensions.Primitives/src/ThrowHelper.cs index 06fa14a31cf0c0..88fff02adf6c19 100644 --- a/src/libraries/Microsoft.Extensions.Primitives/src/ThrowHelper.cs +++ b/src/libraries/Microsoft.Extensions.Primitives/src/ThrowHelper.cs @@ -68,6 +68,8 @@ private static string GetResourceText(ExceptionResource resource) case ExceptionResource.Capacity_CannotChangeAfterWriteStarted: return SR.Capacity_CannotChangeAfterWriteStarted; case ExceptionResource.Capacity_NotEnough: return SR.Capacity_NotEnough; case ExceptionResource.Capacity_NotUsedEntirely: return SR.Capacity_NotUsedEntirely; + case ExceptionResource.InvalidOperation_ConcurrentDisposableSet: return SR.InvalidOperation_ConcurrentDisposableSet; + case ExceptionResource.InvalidOperation_NullConsumerTask: return SR.InvalidOperation_NullConsumerTask; default: Debug.Fail($"Unexpected resource {resource}"); return ""; @@ -108,6 +110,8 @@ internal enum ExceptionResource Argument_InvalidOffsetLengthStringSegment, Capacity_CannotChangeAfterWriteStarted, Capacity_NotEnough, - Capacity_NotUsedEntirely + Capacity_NotUsedEntirely, + InvalidOperation_ConcurrentDisposableSet, + InvalidOperation_NullConsumerTask } } diff --git a/src/libraries/Microsoft.Extensions.Primitives/tests/ChangeTokenTest.cs b/src/libraries/Microsoft.Extensions.Primitives/tests/ChangeTokenTest.cs index d42a4a41b6f666..7c677df338a2f1 100644 --- a/src/libraries/Microsoft.Extensions.Primitives/tests/ChangeTokenTest.cs +++ b/src/libraries/Microsoft.Extensions.Primitives/tests/ChangeTokenTest.cs @@ -3,6 +3,7 @@ using System; using System.Threading; +using System.Threading.Tasks; using Xunit; namespace Microsoft.Extensions.Primitives @@ -29,6 +30,22 @@ public void Changed() } } + // A change token whose registration succeeds but whose HasChanged throws. + private sealed class ThrowOnHasChangedChangeToken : IChangeToken + { + public int RegisterCalls { get; private set; } + + public bool ActiveChangeCallbacks => true; + + public bool HasChanged => throw new InvalidTimeZoneException(); + + public IDisposable RegisterChangeCallback(Action callback, object state) + { + RegisterCalls++; + return null; + } + } + [Fact] public void HasChangeFiresChange() { @@ -45,11 +62,12 @@ public void ChangesFireAfterExceptions() { TestChangeToken token = null; var count = 0; - ChangeToken.OnChange(() => token = new TestChangeToken(), () => + // The lambda is explicitly typed as Action so it binds to the synchronous overload rather than the Func one. + ChangeToken.OnChange(() => token = new TestChangeToken(), (Action)(() => { count++; throw new Exception(); - }); + })); Assert.Throws(() => token.Changed()); Assert.Equal(1, count); Assert.Throws(() => token.Changed()); @@ -75,12 +93,13 @@ public void ChangesFireAfterExceptionsWithState() var count = 0; object state = new object(); object callbackState = null; - ChangeToken.OnChange(() => token = new TestChangeToken(), s => + // The lambda is explicitly typed as Action so it binds to the synchronous overload rather than the Func one. + ChangeToken.OnChange(() => token = new TestChangeToken(), (Action)(s => { callbackState = s; count++; throw new Exception(); - }, state); + }), state); Assert.Throws(() => token.Changed()); Assert.Equal(1, count); Assert.NotNull(callbackState); @@ -251,8 +270,8 @@ public void DoubleDisposeDisposesOnce() provider.Changed(); Assert.Equal(1, count); - Assert.Equal(2, provider.RegistrationCalls); - Assert.Equal(2, provider.DisposeCalls); + Assert.Equal(1, provider.RegistrationCalls); + Assert.Equal(1, provider.DisposeCalls); } [Fact] @@ -261,6 +280,521 @@ public void NullTokenDisposeShouldNotThrow() ChangeToken.OnChange(() => null, () => Assert.Fail()).Dispose(); } + [Fact] + public void AsyncOnChangeThrowsForNullArguments() + { + Assert.Throws(() => ChangeToken.OnChange(null, () => Task.CompletedTask)); + Assert.Throws(() => ChangeToken.OnChange(() => new TestChangeToken(), (Func)null)); + Assert.Throws(() => ChangeToken.OnChange(null, _ => Task.CompletedTask, null)); + Assert.Throws(() => ChangeToken.OnChange(() => new TestChangeToken(), (Func)null, null)); + } + + [Fact] + public void AsyncHasChangeFiresChange() + { + var provider = new ResettableChangeTokenProvider(); + int count = 0; + ChangeToken.OnChange(provider.GetChangeToken, () => + { + count++; + return Task.CompletedTask; + }); + + Assert.Equal(0, count); + provider.Changed(); + Assert.Equal(1, count); + provider.Changed(); + Assert.Equal(2, count); + } + + [Fact] + public void AsyncHasChangeFiresChangeWithState() + { + var provider = new ResettableChangeTokenProvider(); + object state = new object(); + object callbackState = null; + ChangeToken.OnChange(provider.GetChangeToken, s => + { + callbackState = s; + return Task.CompletedTask; + }, state); + + Assert.Null(callbackState); + provider.Changed(); + Assert.Same(state, callbackState); + } + + [Fact] + public void AsyncDisposingChangeTokenRegistrationDoesNotRaiseConsumerCallback() + { + var provider = new ResettableChangeTokenProvider(); + int count = 0; + IDisposable reg = ChangeToken.OnChange(provider.GetChangeToken, () => + { + count++; + return Task.CompletedTask; + }); + + for (int i = 0; i < 5; i++) + { + provider.Changed(); + } + + Assert.Equal(5, count); + + reg.Dispose(); + + for (int i = 0; i < 5; i++) + { + provider.Changed(); + } + + Assert.Equal(5, count); + } + + [Fact] + public async Task AsyncDoesNotReregisterUntilConsumerTaskCompletes() + { + var provider = new ResettableChangeTokenProvider(); + int invocations = 0; + TaskCompletionSource started = NewTcs(); + TaskCompletionSource gate = NewTcs(); + + // Plain Task-returning consumer (no async lambda) so no continuation is captured on + // xunit's SynchronizationContext, keeping the test isolated from other tests. + IDisposable reg = ChangeToken.OnChange(provider.GetChangeToken, () => + { + Interlocked.Increment(ref invocations); + Volatile.Read(ref started).SetResult(true); + return Volatile.Read(ref gate).Task; + }); + + // First change starts the consumer synchronously; it then blocks awaiting its gate. + TaskCompletionSource firstGate = gate; + provider.Changed(); + await started.Task.WaitAsync(TimeSpan.FromSeconds(30)); + Assert.Equal(1, invocations); + + // Arm gates for the next invocation before triggering another change. The consumer re-runs on a + // different thread (the awaited gate's continuation), so publish these with Volatile.Write. + Volatile.Write(ref started, NewTcs()); + TaskCompletionSource secondGate = NewTcs(); + Volatile.Write(ref gate, secondGate); + + // A change while the consumer is running must not start another invocation, because the + // token is only re-registered once the consumer's task completes. + provider.Changed(); + Assert.Equal(1, invocations); + + // Completing the first consumer re-registers and coalesces the pending change into a single invocation. + firstGate.SetResult(true); + await started.Task.WaitAsync(TimeSpan.FromSeconds(30)); + Assert.Equal(2, invocations); + + // Drain the second invocation and unsubscribe. + secondGate.SetResult(true); + reg.Dispose(); + } + + [Theory] + [InlineData(false)] + [InlineData(true)] + public async Task AsyncDisposeDuringConsumerSuppressesSynchronousReinvocationWhenTokenAlreadyChanged(bool useStateOverload) + { + var provider = new ResettableChangeTokenProvider(); + int invocations = 0; + + // Plain (synchronous-continuation) TCS: completing it runs the deferred re-registration inline on the + // completing thread. The gate is completed from a thread-pool thread (below) so that, with the product + // code's ConfigureAwait(false), that continuation runs synchronously there rather than being posted to + // xunit's SynchronizationContext, keeping the test deterministic without arbitrary delays. + var gate = new TaskCompletionSource(); + + IDisposable reg; + if (useStateOverload) + { + reg = ChangeToken.OnChange(provider.GetChangeToken, _ => + { + Interlocked.Increment(ref invocations); + return gate.Task; + }, new object()); + } + else + { + reg = ChangeToken.OnChange(provider.GetChangeToken, () => + { + Interlocked.Increment(ref invocations); + return gate.Task; + }); + } + + // The first change starts the consumer; it stays in flight awaiting the gate, so the change token + // captured for re-registration is held until the consumer completes. + provider.Changed(); + Assert.Equal(1, invocations); + + // Dispose while the consumer is still in flight. + reg.Dispose(); + + await Task.Run(() => + { + // A change while disposed cancels the captured token, so it is now already changed. Registering a + // callback on an already-changed CancellationChangeToken would invoke the callback synchronously. + provider.Changed(); + + // Completing the consumer drives the deferred re-registration inline on this thread. It must observe + // the disposal and skip registering; otherwise registering on the already-changed token invokes the + // callback synchronously and re-runs the consumer after Dispose. Completing from a thread-pool thread + // (rather than directly here) is required for determinism: with the product code's + // ConfigureAwait(false), the continuation runs synchronously on this thread instead of being posted to + // xunit's SynchronizationContext, so the (buggy) re-invocation would be observed before the assert. + gate.SetResult(true); + }); + + Assert.Equal(1, invocations); + } + + [Theory] + [InlineData(false)] + [InlineData(true)] + public void SyncDisposeAndChangeDuringConsumerSuppressesSynchronousReinvocation(bool useStateOverload) + { + var provider = new ResettableChangeTokenProvider(); + int invocations = 0; + + IDisposable reg = null; + void Consume() + { + if (Interlocked.Increment(ref invocations) == 1) + { + // Dispose, then change the captured token, so the synchronous re-registration in the consumer's + // finally block would target an already-changed token (which invokes the callback synchronously). + reg.Dispose(); + provider.Changed(); + } + } + + reg = useStateOverload + ? ChangeToken.OnChange(provider.GetChangeToken, _ => Consume(), new object()) + : ChangeToken.OnChange(provider.GetChangeToken, Consume); + + provider.Changed(); + + // The registration was disposed during the consumer, so the synchronous re-registration must be + // suppressed; the consumer must not be invoked again even though the token it would re-register on + // has already changed. + Assert.Equal(1, invocations); + } + + [Theory] + [InlineData(false)] + [InlineData(true)] + public void AsyncReRegistrationFailureIsNotRetriedAndPropagates(bool useStateOverload) + { + var firstToken = new TestChangeToken(); + var throwingToken = new ThrowOnHasChangedChangeToken(); + int produced = 0; + Func producer = () => produced++ == 0 ? firstToken : throwingToken; + + if (useStateOverload) + { + ChangeToken.OnChange(producer, _ => Task.CompletedTask, new object()); + } + else + { + ChangeToken.OnChange(producer, () => Task.CompletedTask); + } + + // When the token fires, the synchronously-completed consumer triggers re-registration on the next token, + // whose HasChanged throws during registration. That exception must propagate to the code that triggers + // the change, and re-registration must be attempted exactly once. + Assert.Throws(firstToken.Changed); + Assert.Equal(1, throwingToken.RegisterCalls); + } + + [Theory] + [InlineData(false)] + [InlineData(true)] + public async Task DisposingDuringConsumerSuppressesReRegistration(bool useAsyncConsumer) + { + var provider = new ResettableChangeTokenProvider(); + int invocations = 0; + + // Plain (synchronous-continuation) TCS so completing it from a thread-pool thread (below) drives the + // deferred re-registration inline there rather than posting to xunit's SynchronizationContext. + var gate = new TaskCompletionSource(); + + // The consumer disposes its own registration mid-invocation. The re-registration that follows the + // consumer (synchronously for the sync overload, after the returned task completes for the async one) + // must observe the disposal and not arm a live callback. Disposal happens on the thread running the + // consumer, so it doesn't block on the in-progress change callback. + IDisposable reg = null; + if (useAsyncConsumer) + { + reg = ChangeToken.OnChange(provider.GetChangeToken, () => + { + Interlocked.Increment(ref invocations); + reg.Dispose(); + + // Stay in flight by returning an incomplete task; re-registration is deferred until it completes. + return gate.Task; + }); + } + else + { + reg = ChangeToken.OnChange(provider.GetChangeToken, () => + { + Interlocked.Increment(ref invocations); + reg.Dispose(); + }); + } + + // Fire the change. The consumer runs synchronously and disposes itself. + provider.Changed(); + + if (useAsyncConsumer) + { + // Complete the in-flight consumer from a thread-pool thread so the deferred re-registration runs + // synchronously and completes before the subsequent changes below (the product code awaits with + // ConfigureAwait(false)), keeping the test deterministic without arbitrary delays. + await Task.Run(() => gate.SetResult(true)); + } + + Assert.Equal(1, invocations); + + // Subsequent changes must not invoke the consumer again, because disposal suppressed re-registration. + for (int i = 0; i < 5; i++) + { + provider.Changed(); + } + + Assert.Equal(1, invocations); + } + + [ConditionalTheory(typeof(PlatformDetection), nameof(PlatformDetection.IsMultithreadingSupported))] + [InlineData(false)] + [InlineData(true)] + public async Task DisposingFromAnotherThreadWhileConsumerExecutingSuppressesReRegistration(bool useAsyncConsumer) + { + var provider = new ResettableChangeTokenProvider(); + int invocations = 0; + TaskCompletionSource started = NewTcs(); + + // Plain (synchronous-continuation) TCS so completing it from a thread-pool thread drives the deferred + // re-registration inline there (the product code awaits with ConfigureAwait(false)). + var gate = new TaskCompletionSource(); + + IDisposable reg; + if (useAsyncConsumer) + { + reg = ChangeToken.OnChange(provider.GetChangeToken, () => + { + Interlocked.Increment(ref invocations); + started.SetResult(true); + + // Stay in flight by returning an incomplete task; re-registration is deferred until it completes. + return gate.Task; + }); + } + else + { + reg = ChangeToken.OnChange(provider.GetChangeToken, () => + { + Interlocked.Increment(ref invocations); + started.SetResult(true); + + // Stay in flight by blocking the triggering thread until the gate is released. + gate.Task.GetAwaiter().GetResult(); + }); + } + + // A synchronous consumer blocks the triggering thread, so fire the change on a background thread. + Task trigger = Task.Run(provider.Changed); + await started.Task.WaitAsync(TimeSpan.FromSeconds(30)); + Assert.Equal(1, invocations); + + if (useAsyncConsumer) + { + // The change callback already returned (the consumer is in flight via its task), so Dispose does + // not block. Dispose from another thread, then complete the consumer from a thread-pool thread so + // the deferred re-registration runs synchronously there and must observe the disposal. + await Task.Run(reg.Dispose).WaitAsync(TimeSpan.FromSeconds(30)); + await Task.Run(() => gate.SetResult(true)); + await trigger.WaitAsync(TimeSpan.FromSeconds(30)); + } + else + { + // The consumer is running on the triggering thread, so Dispose blocks until it returns + // (CancellationTokenRegistration.Dispose waits for the in-progress callback). Run it on a + // separate thread, then release the consumer so both can complete. + Task dispose = Task.Run(reg.Dispose); + gate.SetResult(true); + await dispose.WaitAsync(TimeSpan.FromSeconds(30)); + await trigger.WaitAsync(TimeSpan.FromSeconds(30)); + } + + Assert.Equal(1, invocations); + + // Subsequent changes must not invoke the consumer again, because disposal suppressed re-registration. + for (int i = 0; i < 5; i++) + { + provider.Changed(); + } + + Assert.Equal(1, invocations); + } + + private static TaskCompletionSource NewTcs() => + new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + + [Theory] + [InlineData(false)] + [InlineData(true)] + public void AsyncConsumerSynchronousExceptionPropagatesToProducerAndSubscriptionSurvives(bool useStateOverload) + { + var provider = new ResettableChangeTokenProvider(); + int count = 0; + object expectedState = new object(); + object observedState = null; + + Func faulting = () => + { + count++; + throw new InvalidTimeZoneException(); + }; + + if (useStateOverload) + { + ChangeToken.OnChange(provider.GetChangeToken, s => + { + observedState = s; + return faulting(); + }, expectedState); + } + else + { + ChangeToken.OnChange(provider.GetChangeToken, faulting); + } + + // A synchronous exception from the consumer is propagated to the code that triggers the change token, + // just like the synchronous overload. CancellationTokenSource.Cancel wraps it in an AggregateException. + AggregateException ex = Assert.Throws(provider.Changed); + Assert.IsType(ex.InnerException); + Assert.Equal(1, count); + + // The subscription survives a throwing consumer, so a later change invokes (and throws from) it again. + Assert.Throws(provider.Changed); + Assert.Equal(2, count); + + if (useStateOverload) + { + Assert.Same(expectedState, observedState); + } + } + + [Theory] + [InlineData(false)] + [InlineData(true)] + public void AsyncConsumerAsynchronousFaultIsNotPropagatedToProducerAndSubscriptionSurvives(bool useStateOverload) + { + var provider = new ResettableChangeTokenProvider(); + int count = 0; + object expectedState = new object(); + object observedState = null; + + Func faulting = () => + { + count++; + return Task.FromException(new InvalidTimeZoneException()); + }; + + if (useStateOverload) + { + ChangeToken.OnChange(provider.GetChangeToken, s => + { + observedState = s; + return faulting(); + }, expectedState); + } + else + { + ChangeToken.OnChange(provider.GetChangeToken, faulting); + } + + // An asynchronous fault from the consumer's task cannot be propagated without blocking, so it is left + // unobserved (observable only through TaskScheduler.UnobservedTaskException) and is not surfaced to the + // code that triggers the change token. + provider.Changed(); + Assert.Equal(1, count); + + // The subscription survives a faulted consumer, so later changes still invoke it. + provider.Changed(); + Assert.Equal(2, count); + + if (useStateOverload) + { + Assert.Same(expectedState, observedState); + } + } + + // Subscribes a no-op consumer using each of the four OnChange overloads, selected by index. + private static void Subscribe(int overload, Func producer) + { + switch (overload) + { + case 0: + ChangeToken.OnChange(producer, () => { }); + break; + case 1: + ChangeToken.OnChange(producer, _ => { }, new object()); + break; + case 2: + ChangeToken.OnChange(producer, () => Task.CompletedTask); + break; + case 3: + ChangeToken.OnChange(producer, _ => Task.CompletedTask, new object()); + break; + } + } + + [Theory] + [InlineData(0)] + [InlineData(1)] + [InlineData(2)] + [InlineData(3)] + public void ProducerExceptionDuringInitialRegistrationPropagatesToCaller(int overload) + { + Func producer = () => throw new InvalidTimeZoneException(); + + // The producer is invoked while registering, so its exception is propagated to the caller of OnChange. + Assert.Throws(() => Subscribe(overload, producer)); + } + + [Theory] + [InlineData(0)] + [InlineData(1)] + [InlineData(2)] + [InlineData(3)] + public void ProducerExceptionWhenTokenFiresPropagatesToTrigger(int overload) + { + TestChangeToken token = null; + int calls = 0; + Func producer = () => + { + if (calls++ == 0) + { + return token = new TestChangeToken(); + } + throw new InvalidTimeZoneException(); + }; + + Subscribe(overload, producer); + + // When the token fires, the producer is invoked again to obtain the next token; its exception is + // propagated to the code that triggers the change token. + Assert.Throws(() => token.Changed()); + } + public class TrackableChangeTokenProvider { private TrackableChangeToken _cts = new TrackableChangeToken();