diff --git a/AsyncRx.NET/System.Reactive.Async/Linq/Operators/Synchronize.cs b/AsyncRx.NET/System.Reactive.Async/Linq/Operators/Synchronize.cs index 27d016799..3cce87127 100644 --- a/AsyncRx.NET/System.Reactive.Async/Linq/Operators/Synchronize.cs +++ b/AsyncRx.NET/System.Reactive.Async/Linq/Operators/Synchronize.cs @@ -2,6 +2,7 @@ // The .NET Foundation licenses this file to you under the MIT License. // See the LICENSE file in the project root for more information. +using System.Reactive.Threading; using System.Threading; namespace System.Reactive.Linq @@ -16,7 +17,7 @@ public static IAsyncObservable Synchronize(this IAsyncObservab return Create(source, static (source, observer) => source.SubscribeSafeAsync(AsyncObserver.Synchronize(observer))); } - public static IAsyncObservable Synchronize(this IAsyncObservable source, AsyncGate gate) + public static IAsyncObservable Synchronize(this IAsyncObservable source, IAsyncGate gate) { if (source == null) throw new ArgumentNullException(nameof(source)); @@ -40,7 +41,7 @@ public static IAsyncObserver Synchronize(IAsyncObserver Synchronize(IAsyncObserver observer, AsyncGate gate) + public static IAsyncObserver Synchronize(IAsyncObserver observer, IAsyncGate gate) { if (observer == null) throw new ArgumentNullException(nameof(observer)); diff --git a/AsyncRx.NET/System.Reactive.Async/Threading/AsyncGate.cs b/AsyncRx.NET/System.Reactive.Async/Threading/AsyncGate.cs index 507aa676f..8d63aa820 100644 --- a/AsyncRx.NET/System.Reactive.Async/Threading/AsyncGate.cs +++ b/AsyncRx.NET/System.Reactive.Async/Threading/AsyncGate.cs @@ -3,17 +3,18 @@ // See the LICENSE file in the project root for more information. using System.Diagnostics; +using System.Reactive.Threading; using System.Threading.Tasks; namespace System.Threading { - public sealed class AsyncGate + public sealed class AsyncGate : IAsyncGate { private readonly object _gate = new(); private readonly SemaphoreSlim _semaphore = new(1, 1); private readonly AsyncLocal _recursionCount = new(); - public ValueTask LockAsync() + public ValueTask LockAsync() { var shouldAcquire = false; @@ -32,13 +33,13 @@ public ValueTask LockAsync() if (shouldAcquire) { - return new ValueTask(_semaphore.WaitAsync().ContinueWith(_ => new Releaser(this))); + return new ValueTask(_semaphore.WaitAsync().ContinueWith(_ => new AsyncGateReleaser(this))); } - return new ValueTask(new Releaser(this)); + return new ValueTask(new AsyncGateReleaser(this)); } - private void Release() + void IAsyncGate.Release() { lock (_gate) { @@ -50,14 +51,5 @@ private void Release() } } } - - public readonly struct Releaser : IDisposable - { - private readonly AsyncGate _parent; - - public Releaser(AsyncGate parent) => _parent = parent; - - public void Dispose() => _parent.Release(); - } } } diff --git a/AsyncRx.NET/System.Reactive.Async/Threading/AsyncGateReleaser.cs b/AsyncRx.NET/System.Reactive.Async/Threading/AsyncGateReleaser.cs new file mode 100644 index 000000000..7ae06528d --- /dev/null +++ b/AsyncRx.NET/System.Reactive.Async/Threading/AsyncGateReleaser.cs @@ -0,0 +1,15 @@ +// Licensed to the .NET Foundation under one or more agreements. +// The .NET Foundation licenses this file to you under the MIT License. +// See the LICENSE file in the project root for more information. + +namespace System.Reactive.Threading +{ + public readonly struct AsyncGateReleaser : IDisposable + { + private readonly IAsyncGate _parent; + + public AsyncGateReleaser(IAsyncGate parent) => _parent = parent; + + public void Dispose() => _parent.Release(); + } +} diff --git a/AsyncRx.NET/System.Reactive.Async/Threading/IAsyncGate.cs b/AsyncRx.NET/System.Reactive.Async/Threading/IAsyncGate.cs new file mode 100644 index 000000000..5d34f8bf8 --- /dev/null +++ b/AsyncRx.NET/System.Reactive.Async/Threading/IAsyncGate.cs @@ -0,0 +1,73 @@ +// Licensed to the .NET Foundation under one or more agreements. +// The .NET Foundation licenses this file to you under the MIT License. +// See the LICENSE file in the project root for more information. + +using System.Reactive.Linq; +using System.Threading; +using System.Threading.Tasks; + +namespace System.Reactive.Threading +{ + /// + /// Synchronization primitive that provides -style + /// exclusive access semantics, but with an asynchronous API. + /// + /// + /// + /// This enables + /// and + /// to be used to synchronize access to an observer with a custom synchronization primitive. + /// + /// + /// These methods model the equivalents for and + /// in System.Reactive. Those offer overloads accepting a 'gate' parameter, and if you pass + /// the same object to multiple calls to these methods, they will all synchronize their operation + /// through that same gate object. The gate parameter in those methods is of type + /// , which works because all .NET objects have an associated monitor. + /// (It's created on demand when you first use lock or something equivalent.) + /// + /// + /// That approach is problematic in an async world, because this built-in monitor blocks the + /// calling thread when contention occurs. The basic idea of AsyncRx.NET is to avoid such + /// blocking. It can't always be avoided, and in cases where we can be certain that lock + /// acquisition times will be short, the conventional .NET monitor is still a good choice. + /// But since these Synchronize operators allow the caller to pass a gate which the + /// application code itself might lock, we have no control over how long the lock might be + /// held. So it would be inappropriate to use a monitor here. + /// + /// + /// Since the .NET runtime does not currently offer any asynchronous direct equivalent to + /// monitor, this interface defines the required API. The class + /// provide a basic implementation. If applications require additional features, (e.g. + /// if they want cancellation support when the application tries to acquire the lock) + /// they can provide their own implementation. + /// + /// + public interface IAsyncGate + { + /// + /// Acquires the lock. + /// + /// + /// A task that completes when the lock has been acquired, returning an + /// which can be disposed to release the lock. + /// + /// + /// + /// Applications release the lock by disposing the returned by this + /// method. Typically this is done with a using statement or declaration. + /// + /// + public ValueTask LockAsync(); + + /// + /// Releases the lock. Applications typically won't call this directly, and will use + /// the returned by instead. + /// + /// + /// This method needs to be publicly accessible so that a single + /// can be shared by all implementations of this interface. + /// + public void Release(); + } +}