From 062f203a0cd1a626808f7326b47d81fd2dc0a596 Mon Sep 17 00:00:00 2001 From: Gunnar Liljas Date: Thu, 15 Oct 2026 09:11:24 +0200 Subject: [PATCH] In process synchronization provider --- README.md | 3 +- docs/DistributedLock.InProcess.md | 40 +++ .../AssemblyAttributes.cs | 1 + ...mpositeDistributedSynchronizationHandle.cs | 14 - .../Internal/SemaphoreSlimExtensions.cs | 21 ++ .../Internal/TimeoutTracker.cs | 20 ++ .../AsyncReaderWriterLockState.cs | 174 +++++++++++ .../ConcurrentWeakReferenceDictionary.cs | 93 ++++++ .../DistributedLock.InProcess.csproj | 64 ++++ ...ProcessDistributedLock.IDistributedLock.cs | 81 ++++++ .../InProcessDistributedLock.cs | 32 ++ .../InProcessDistributedLockHandle.cs | 45 +++ ...IDistributedUpgradeableReaderWriterLock.cs | 226 ++++++++++++++ .../InProcessDistributedReaderWriterLock.cs | 42 +++ ...rocessDistributedReaderWriterLockHandle.cs | 175 +++++++++++ ...tributedSemaphore.IDistributedSemaphore.cs | 81 ++++++ .../InProcessDistributedSemaphore.cs | 38 +++ .../InProcessDistributedSemaphoreHandle.cs | 45 +++ ...ocessDistributedSynchronizationProvider.cs | 74 +++++ .../InProcessReaderWriterLockType.cs | 8 + .../PublicAPI.Shipped.txt | 1 + .../PublicAPI.Unshipped.txt | 50 ++++ .../SemaphoreSlimReleaser.cs | 20 ++ .../packages.lock.json | 275 ++++++++++++++++++ .../DistributedLockCoreTestCases.cs | 12 + .../DistributedLock.Tests.csproj | 1 + .../InProcess/TestingInProcessProviders.cs | 36 +++ ...TestingInProcessSynchronizationStrategy.cs | 16 + .../TestingSynchronizationStrategy.cs | 6 + .../Tests/CombinatorialTests.cs | 11 + ...sDistributedSynchronizationProviderTest.cs | 176 +++++++++++ .../InProcess/InProcessPrimitiveCacheTest.cs | 76 +++++ src/DistributedLock.sln | 6 + ...GenerateIDistributedLockImplementations.cs | 2 +- 34 files changed, 1949 insertions(+), 16 deletions(-) create mode 100644 docs/DistributedLock.InProcess.md create mode 100644 src/DistributedLock.Core/Internal/SemaphoreSlimExtensions.cs create mode 100644 src/DistributedLock.Core/Internal/TimeoutTracker.cs create mode 100644 src/DistributedLock.InProcess/AsyncReaderWriterLockState.cs create mode 100644 src/DistributedLock.InProcess/ConcurrentWeakReferenceDictionary.cs create mode 100644 src/DistributedLock.InProcess/DistributedLock.InProcess.csproj create mode 100644 src/DistributedLock.InProcess/InProcessDistributedLock.IDistributedLock.cs create mode 100644 src/DistributedLock.InProcess/InProcessDistributedLock.cs create mode 100644 src/DistributedLock.InProcess/InProcessDistributedLockHandle.cs create mode 100644 src/DistributedLock.InProcess/InProcessDistributedReaderWriterLock.IDistributedUpgradeableReaderWriterLock.cs create mode 100644 src/DistributedLock.InProcess/InProcessDistributedReaderWriterLock.cs create mode 100644 src/DistributedLock.InProcess/InProcessDistributedReaderWriterLockHandle.cs create mode 100644 src/DistributedLock.InProcess/InProcessDistributedSemaphore.IDistributedSemaphore.cs create mode 100644 src/DistributedLock.InProcess/InProcessDistributedSemaphore.cs create mode 100644 src/DistributedLock.InProcess/InProcessDistributedSemaphoreHandle.cs create mode 100644 src/DistributedLock.InProcess/InProcessDistributedSynchronizationProvider.cs create mode 100644 src/DistributedLock.InProcess/InProcessReaderWriterLockType.cs create mode 100644 src/DistributedLock.InProcess/PublicAPI.Shipped.txt create mode 100644 src/DistributedLock.InProcess/PublicAPI.Unshipped.txt create mode 100644 src/DistributedLock.InProcess/SemaphoreSlimReleaser.cs create mode 100644 src/DistributedLock.InProcess/packages.lock.json create mode 100644 src/DistributedLock.Tests/Infrastructure/InProcess/TestingInProcessProviders.cs create mode 100644 src/DistributedLock.Tests/Infrastructure/InProcess/TestingInProcessSynchronizationStrategy.cs create mode 100644 src/DistributedLock.Tests/Tests/InProcess/InProcessDistributedSynchronizationProviderTest.cs create mode 100644 src/DistributedLock.Tests/Tests/InProcess/InProcessPrimitiveCacheTest.cs diff --git a/README.md b/README.md index f2c511f3..3c2d8f10 100644 --- a/README.md +++ b/README.md @@ -23,7 +23,8 @@ DistributedLock contains implementations based on various technologies; you can - **[DistributedLock.MongoDB](docs/DistributedLock.MongoDB.md)** [![NuGet Status](http://img.shields.io/nuget/v/DistributedLock.MongoDB.svg?style=flat)](https://www.nuget.org/packages/DistributedLock.MongoDB/) [![Static Badge](https://img.shields.io/badge/API%20Docs-DNDocs-190088?logo=readme&logoColor=white)](https://dndocs.com/d/distributedlock/api/Medallion.Threading.MongoDB.html) (uses MongoDB) - **[DistributedLock.ZooKeeper](docs/DistributedLock.ZooKeeper.md)** [![NuGet Status](http://img.shields.io/nuget/v/DistributedLock.ZooKeeper.svg?style=flat)](https://www.nuget.org/packages/DistributedLock.ZooKeeper/) [![Static Badge](https://img.shields.io/badge/API%20Docs-DNDocs-190088?logo=readme&logoColor=white)](https://dndocs.com/d/distributedlock/api/Medallion.Threading.ZooKeeper.html) (uses Apache ZooKeeper) - **[DistributedLock.FileSystem](docs/DistributedLock.FileSystem.md)** [![NuGet Status](http://img.shields.io/nuget/v/DistributedLock.FileSystem.svg?style=flat)](https://www.nuget.org/packages/DistributedLock.FileSystem/) [![Static Badge](https://img.shields.io/badge/API%20Docs-DNDocs-190088?logo=readme&logoColor=white)](https://dndocs.com/d/distributedlock/api/Medallion.Threading.FileSystem.html) (uses lock files) -- **[DistributedLock.WaitHandles](docs/DistributedLock.WaitHandles.md)** [![NuGet Status](http://img.shields.io/nuget/v/DistributedLock.WaitHandles.svg?style=flat)](https://www.nuget.org/packages/DistributedLock.WaitHandles/) [![Static Badge](https://img.shields.io/badge/API%20Docs-DNDocs-190088?logo=readme&logoColor=white)](https://dndocs.com/d/distributedlock/api/Medallion.Threading.WaitHandles.html) (*Windows only*: uses operating system global `WaitHandle`s) +(*Windows only*: uses operating system global `WaitHandle`s) +- **[DistributedLock.InProcess](docs/DistributedLock.InProcess.md)** [![NuGet Status](http://img.shields.io/nuget/v/DistributedLock.InProcess.svg?style=flat)](https://www.nuget.org/packages/DistributedLock.InProcess/) (*non-distributed*: uses in-process primitives such as `SemaphoreSlim`; useful for single-instance apps and testing) **Click on the name** of any of the above packages to see the documentation specific to that implementation, or read on for general documentation that applies to all implementations. diff --git a/docs/DistributedLock.InProcess.md b/docs/DistributedLock.InProcess.md new file mode 100644 index 00000000..12186c85 --- /dev/null +++ b/docs/DistributedLock.InProcess.md @@ -0,0 +1,40 @@ +# DistributedLock.InProcess + +[Download the NuGet package](https://www.nuget.org/packages/DistributedLock.InProcess) [![NuGet Status](http://img.shields.io/nuget/v/DistributedLock.InProcess.svg?style=flat)](https://www.nuget.org/packages/DistributedLock.InProcess/) + +The DistributedLock.InProcess package offers **non-distributed** implementations of the DistributedLock abstractions based on in-process primitives such as `SemaphoreSlim`. This is useful for single-instance deployments, local development, and tests, where you want to program against `IDistributedLockProvider` & co. without a backing service. For example: + +```C# +var provider = new InProcessDistributedSynchronizationProvider(); +var @lock = provider.CreateLock("MyLockName"); +await using (await @lock.AcquireAsync()) +{ + // I have the lock! +} +``` + +## APIs + +- The `InProcessDistributedLock` class implements the `IDistributedLock` interface. +- The `InProcessDistributedSemaphore` class implements the `IDistributedSemaphore` interface. +- The `InProcessDistributedReaderWriterLock` class implements the `IDistributedUpgradeableReaderWriterLock` interface. +- The `InProcessDistributedSynchronizationProvider` class implements the `IDistributedLockProvider`, `IDistributedSemaphoreProvider` and `IDistributedUpgradeableReaderWriterLockProvider` interfaces. + +Primitives are created through the provider; there are no public constructors. + +## Implementation notes + +- **These classes only coordinate within a single process.** They provide no protection across processes or machines. +- **Scopes:** primitives are cached process-wide by (scope, name). All providers constructed with the same `scope` (the default is a shared, process-wide scope) hand out coordinating primitives for the same name, even across provider instances. Providers with different scopes are fully isolated: + + ```C# + var tenantA = new InProcessDistributedSynchronizationProvider(scope: "tenant-a"); + var tenantB = new InProcessDistributedSynchronizationProvider(scope: "tenant-b"); + // tenantA.CreateLock("x") and tenantB.CreateLock("x") do not block each other + ``` + +- **Primitive types are independent:** locks, semaphores and reader-writer locks are cached separately, so a lock and a reader-writer lock (or semaphore) with the same name do not coordinate with one another. This mirrors most other providers. +- **Semaphore max count:** the max count of a named semaphore is fixed by whichever caller first creates it while it is alive. +- **Abandonment:** a handle that is never disposed is released only once it has been garbage collected and processed by a background finalizer, which runs roughly every 30 seconds. Until then the primitive remains held. For reader-writer locks, an abandoned read handle blocks writers, and a waiting writer in turn blocks new readers. Always dispose handles. +- **Reader-writer lock fairness:** the reader-writer lock is writer-preferring. Once a writer is waiting, new readers queue behind it. +- `HandleLostToken` is always `CancellationToken.None`, since an in-process handle cannot be lost. diff --git a/src/DistributedLock.Core/AssemblyAttributes.cs b/src/DistributedLock.Core/AssemblyAttributes.cs index 888ed9ef..a1cf3ec3 100644 --- a/src/DistributedLock.Core/AssemblyAttributes.cs +++ b/src/DistributedLock.Core/AssemblyAttributes.cs @@ -16,4 +16,5 @@ [assembly: InternalsVisibleTo("DistributedLock.MySql, PublicKey=0024000004800000940000000602000000240000525341310004000001000100fd3af56ccc8ed94fffe25bfd651e6a5674f8f20a76d37de800dd0f7380e04f0fde2da6fa200380b14fe398605b6f470c87e5e0a0bf39ae871f07536a4994aa7a0057c4d3bcedc8fef3eecb0c88c2024a1b3289305c2393acd9fb9f9a42d0bd7826738ce864d507575ea3a1fe1746ab19823303269f79379d767949807f494be8")] [assembly: InternalsVisibleTo("DistributedLock.Oracle, PublicKey=0024000004800000940000000602000000240000525341310004000001000100fd3af56ccc8ed94fffe25bfd651e6a5674f8f20a76d37de800dd0f7380e04f0fde2da6fa200380b14fe398605b6f470c87e5e0a0bf39ae871f07536a4994aa7a0057c4d3bcedc8fef3eecb0c88c2024a1b3289305c2393acd9fb9f9a42d0bd7826738ce864d507575ea3a1fe1746ab19823303269f79379d767949807f494be8")] [assembly: InternalsVisibleTo("DistributedLock.MongoDB, PublicKey=0024000004800000940000000602000000240000525341310004000001000100fd3af56ccc8ed94fffe25bfd651e6a5674f8f20a76d37de800dd0f7380e04f0fde2da6fa200380b14fe398605b6f470c87e5e0a0bf39ae871f07536a4994aa7a0057c4d3bcedc8fef3eecb0c88c2024a1b3289305c2393acd9fb9f9a42d0bd7826738ce864d507575ea3a1fe1746ab19823303269f79379d767949807f494be8")] +[assembly: InternalsVisibleTo("DistributedLock.InProcess, PublicKey=0024000004800000940000000602000000240000525341310004000001000100fd3af56ccc8ed94fffe25bfd651e6a5674f8f20a76d37de800dd0f7380e04f0fde2da6fa200380b14fe398605b6f470c87e5e0a0bf39ae871f07536a4994aa7a0057c4d3bcedc8fef3eecb0c88c2024a1b3289305c2393acd9fb9f9a42d0bd7826738ce864d507575ea3a1fe1746ab19823303269f79379d767949807f494be8")] #endif diff --git a/src/DistributedLock.Core/CompositeDistributedSynchronizationHandle.cs b/src/DistributedLock.Core/CompositeDistributedSynchronizationHandle.cs index d567f0cd..a92a19d1 100644 --- a/src/DistributedLock.Core/CompositeDistributedSynchronizationHandle.cs +++ b/src/DistributedLock.Core/CompositeDistributedSynchronizationHandle.cs @@ -149,20 +149,6 @@ private static async ValueTask DisposeHandlesAsync(IReadOnlyList - this._stopwatch is { Elapsed: var elapsed } - ? elapsed >= timeout.TimeSpan - ? TimeSpan.Zero - : timeout.TimeSpan - elapsed - : Timeout.InfiniteTimeSpan; - } } internal static class CompositeDistributedLockHandleExtensions diff --git a/src/DistributedLock.Core/Internal/SemaphoreSlimExtensions.cs b/src/DistributedLock.Core/Internal/SemaphoreSlimExtensions.cs new file mode 100644 index 00000000..861a1169 --- /dev/null +++ b/src/DistributedLock.Core/Internal/SemaphoreSlimExtensions.cs @@ -0,0 +1,21 @@ +namespace Medallion.Threading.Internal; + +#if DEBUG +public +#else +internal +#endif +static class SemaphoreSlimExtensions +{ + /// + /// A -compatible implementation of + /// . + /// + public static ValueTask InternalWaitAsync( + this SemaphoreSlim semaphore, + TimeSpan timeout, + CancellationToken cancellationToken) => + SyncViaAsync.IsSynchronous + ? semaphore.Wait(timeout, cancellationToken).AsValueTask() + : semaphore.WaitAsync(timeout, cancellationToken).AsValueTask(); +} diff --git a/src/DistributedLock.Core/Internal/TimeoutTracker.cs b/src/DistributedLock.Core/Internal/TimeoutTracker.cs new file mode 100644 index 00000000..7fccb160 --- /dev/null +++ b/src/DistributedLock.Core/Internal/TimeoutTracker.cs @@ -0,0 +1,20 @@ +namespace Medallion.Threading.Internal; + +#if DEBUG +public +#else +internal +#endif +readonly struct TimeoutTracker(TimeoutValue timeout) +{ + private readonly System.Diagnostics.Stopwatch? _stopwatch = timeout.IsInfinite + ? null + : System.Diagnostics.Stopwatch.StartNew(); + + public TimeSpan Remaining => + this._stopwatch is { Elapsed: var elapsed } + ? elapsed >= timeout.TimeSpan + ? TimeSpan.Zero + : timeout.TimeSpan - elapsed + : Timeout.InfiniteTimeSpan; +} diff --git a/src/DistributedLock.InProcess/AsyncReaderWriterLockState.cs b/src/DistributedLock.InProcess/AsyncReaderWriterLockState.cs new file mode 100644 index 00000000..004ccc29 --- /dev/null +++ b/src/DistributedLock.InProcess/AsyncReaderWriterLockState.cs @@ -0,0 +1,174 @@ +using Medallion.Threading.Internal; + +namespace Medallion.Threading.InProcess; + +internal class AsyncReaderWriterLockState +{ + /// + /// Held by writers for the duration of the lock; blocks incoming readers. + /// + private readonly SemaphoreSlim _entryGate = new(1, 1); + /// + /// Held by a writer, or collectively by the current set of readers (first in/last out). + /// + private readonly SemaphoreSlim _writeSemaphore = new(1, 1); + /// + /// Held by the upgradeable reader and by writers. This makes upgradeable acquisitions mutually + /// exclusive with each other and with writers, while still allowing plain readers. + /// + private readonly SemaphoreSlim _upgradeSemaphore = new(1, 1); + /// + /// Guards . This is a plain monitor rather than a because + /// the release path must not block or go async: it runs from Dispose/DisposeAsync and from the + /// thread. Every critical section it guards is non-blocking. + /// + private readonly object _readerCountLock = new(); + private int _readerCount = 0; + /// + /// The number of writers (including upgrading readers) that are waiting for or holding the write claim. + /// Guarded by . While this is zero, readers may bypass . + /// + private int _writerCount = 0; + + internal ValueTask TryAcquireAsync(TimeoutValue timeout, CancellationToken cancellationToken, bool isWrite) => + this.TryAcquireAsync(timeout, cancellationToken, isWrite ? InProcessReaderWriterLockType.Write : InProcessReaderWriterLockType.Read); + + internal async ValueTask TryAcquireAsync(TimeoutValue timeout, CancellationToken cancellationToken, InProcessReaderWriterLockType type) + { + var tracker = new TimeoutTracker(timeout); + + switch (type) + { + case InProcessReaderWriterLockType.Upgrade: + // an upgradeable reader only excludes other upgradeable readers and writers + return await this._upgradeSemaphore.InternalWaitAsync(tracker.Remaining, cancellationToken).ConfigureAwait(false); + + case InProcessReaderWriterLockType.Write: + if (!await this._upgradeSemaphore.InternalWaitAsync(tracker.Remaining, cancellationToken).ConfigureAwait(false)) + { + return false; + } + var acquiredWrite = false; + try { return acquiredWrite = await this.TryAcquireWriteCoreAsync(tracker.Remaining, cancellationToken).ConfigureAwait(false); } + finally { if (!acquiredWrite) { this._upgradeSemaphore.Release(); } } + + default: + return await this.TryAcquireReadCoreAsync(tracker, cancellationToken).ConfigureAwait(false); + } + } + + /// + /// Acquires the write claim ( + ). Assumes + /// is already held by the caller. The inverse is . + /// + internal async ValueTask TryAcquireWriteCoreAsync(TimeoutValue timeout, CancellationToken cancellationToken) + { + var tracker = new TimeoutTracker(timeout); + + // announce ourselves before touching the write semaphore so that readers stop taking the fast path + lock (this._readerCountLock) { ++this._writerCount; } + + var acquired = false; + try + { + // hold the entry gate for the duration of the write lock so that new readers queue up behind us + if (!await this._entryGate.InternalWaitAsync(tracker.Remaining, cancellationToken).ConfigureAwait(false)) + { + return false; + } + + try { return acquired = await this._writeSemaphore.InternalWaitAsync(tracker.Remaining, cancellationToken).ConfigureAwait(false); } + finally { if (!acquired) { this._entryGate.Release(); } } + } + finally + { + if (!acquired) { lock (this._readerCountLock) { --this._writerCount; } } + } + } + + private async ValueTask TryAcquireReadCoreAsync(TimeoutTracker tracker, CancellationToken cancellationToken) + { + // Fast path: if no writer is waiting or holding, we can join the readers without going through the entry + // gate. Every writer increments _writerCount under _readerCountLock before it tries to take _writeSemaphore, + // so while _writerCount == 0 the write semaphore is either free or held by the current readers. This avoids + // spurious zero-timeout failures caused by other readers briefly holding the entry gate. + lock (this._readerCountLock) + { + if (this._writerCount == 0) + { + if (this._readerCount == 0) + { + var acquiredWriteSemaphore = this._writeSemaphore.Wait(TimeSpan.Zero); + Invariant.Require(acquiredWriteSemaphore, "the write semaphore must be free when there are no readers or writers"); + } + ++this._readerCount; + return true; + } + } + + // Slow path: a writer is pending. Wait + if (!await this._entryGate.InternalWaitAsync(tracker.Remaining, cancellationToken).ConfigureAwait(false)) + { + return false; + } + + try + { + // We hold the entry gate, so no writer can hold _writeSemaphore right now. It is therefore either + // free (_readerCount == 0) or held by the existing readers (_readerCount > 0). In the former case + // the Wait below is guaranteed to succeed immediately, so this critical section never blocks. + lock (this._readerCountLock) + { + if (this._readerCount == 0) + { + var acquiredWriteSemaphore = this._writeSemaphore.Wait(TimeSpan.Zero); + Invariant.Require(acquiredWriteSemaphore, "the write semaphore must be free while the entry gate is held and there are no readers"); + } + ++this._readerCount; + return true; + } + } + finally { this._entryGate.Release(); } + } + + internal void Release(bool isWrite) => + this.Release(isWrite ? InProcessReaderWriterLockType.Write : InProcessReaderWriterLockType.Read, upgraded: false); + + internal void Release(InProcessReaderWriterLockType type, bool upgraded) + { + switch (type) + { + case InProcessReaderWriterLockType.Write: + this.ReleaseWriteCore(); + this._upgradeSemaphore.Release(); + break; + + case InProcessReaderWriterLockType.Upgrade: + if (upgraded) { this.ReleaseWriteCore(); } + this._upgradeSemaphore.Release(); + break; + + default: + // Non-blocking: see the remarks on _readerCountLock. This matters because release runs from + // Dispose(), DisposeAsync() and the ManagedFinalizerQueue thread, none of which may block. + lock (this._readerCountLock) + { + if (--this._readerCount == 0) { this._writeSemaphore.Release(); } + } + break; + } + } + + /// + /// Releases the write claim ( + ) WITHOUT releasing + /// . This is the inverse of . It is used + /// both by normal write release and to compensate for an upgrade that raced with handle disposal. + /// + internal void ReleaseWriteCore() + { + this._writeSemaphore.Release(); + this._entryGate.Release(); + // decrement last so that fast-path readers never observe _writerCount == 0 while we still hold _writeSemaphore + lock (this._readerCountLock) { --this._writerCount; } + } +} \ No newline at end of file diff --git a/src/DistributedLock.InProcess/ConcurrentWeakReferenceDictionary.cs b/src/DistributedLock.InProcess/ConcurrentWeakReferenceDictionary.cs new file mode 100644 index 00000000..82d6808f --- /dev/null +++ b/src/DistributedLock.InProcess/ConcurrentWeakReferenceDictionary.cs @@ -0,0 +1,93 @@ +using System.Collections.Concurrent; + +namespace Medallion.Threading.InProcess; + +/// +/// A thread-safe name-keyed cache which does not keep its values alive. This allows synchronization primitives +/// to be shared by all callers using the same name while still being collected once nothing references them. +/// +internal class ConcurrentWeakReferenceDictionary + where TKey : notnull + where TValue : class +{ + private const int CleanupThreshold = 1000; + + private readonly ConcurrentDictionary> _dictionary = new(); + private int _operationCount; + + public TValue GetOrAdd(TKey key, Func valueFactory) + { + var result = GetOrAddCore(); + + // Opportunistic cleanup + if (Interlocked.Increment(ref this._operationCount) >= CleanupThreshold) + { + Interlocked.Exchange(ref this._operationCount, 0); + this.RemoveDeadEntries(); + } + + return result; + + TValue GetOrAddCore() + { + // Cached across loop iterations so that losing a race does not re-invoke valueFactory. + TValue? newValue = null; + WeakReference? newReference = null; + + while (true) + { + if (this._dictionary.TryGetValue(key, out var existingReference)) + { + // fast path: a live value is already present + if (existingReference.TryGetTarget(out var existingValue)) { return existingValue; } + + // The entry is dead. Replace it, but only if the slot still holds the exact dead reference + // we observed; otherwise another thread has already installed a live replacement. + newValue ??= valueFactory(key); + newReference ??= new WeakReference(newValue); + if (this._dictionary.TryUpdate(key, newReference, comparisonValue: existingReference)) + { + return newValue; + } + } + else + { + newValue ??= valueFactory(key); + newReference ??= new WeakReference(newValue); + if (this._dictionary.TryAdd(key, newReference)) + { + return newValue; + } + } + + // We lost a race (or our own value was collected before we could return it). Re-read and retry. + } + } + } + + private void RemoveDeadEntries() + { + foreach (var kvp in this._dictionary) + { + // Only remove if the entry is STILL the exact dead reference we observed. Removing by key alone would + // race with a concurrent GetOrAdd which may have just installed a live replacement, which would cause + // a subsequent GetOrAdd to hand out a second primitive for a name that is already in use. + if (!kvp.Value.TryGetTarget(out _)) + { + this._dictionary.AsCollection().Remove(kvp); + } + } + } +} + +internal static class ConcurrentDictionaryExtensions +{ + /// + /// Exposes the explicitly-implemented , which removes only if both the + /// key AND the value match. has a public TryRemove overload + /// with these semantics, but not on all of the frameworks we target. + /// + public static ICollection> AsCollection( + this ConcurrentDictionary dictionary) + where TKey : notnull => dictionary; +} diff --git a/src/DistributedLock.InProcess/DistributedLock.InProcess.csproj b/src/DistributedLock.InProcess/DistributedLock.InProcess.csproj new file mode 100644 index 00000000..9540857e --- /dev/null +++ b/src/DistributedLock.InProcess/DistributedLock.InProcess.csproj @@ -0,0 +1,64 @@ + + + + net462;netstandard2.0;netstandard2.1;net8.0 + Medallion.Threading.InProcess + True + 4 + Latest + enable + enable + + + + 1.0.0 + 1.0.0.0 + Michael Adelson + Provides a non-distributed lock implementation based on in process primitives, such as SemaphoreSlim + Copyright © 2026 Michael Adelson + MIT + distributed lock in-process semaphore reader-writer + https://github.com/madelson/DistributedLock + https://github.com/madelson/DistributedLock + 1.0.0.0 + See https://github.com/madelson/DistributedLock#release-notes + true + ..\DistributedLock.snk + + + + True + True + True + + + embedded + + true + true + + + + False + 1591 + TRACE;DEBUG + + + + + + + + + all + + + + + + + + + + + \ No newline at end of file diff --git a/src/DistributedLock.InProcess/InProcessDistributedLock.IDistributedLock.cs b/src/DistributedLock.InProcess/InProcessDistributedLock.IDistributedLock.cs new file mode 100644 index 00000000..7fc49921 --- /dev/null +++ b/src/DistributedLock.InProcess/InProcessDistributedLock.IDistributedLock.cs @@ -0,0 +1,81 @@ +using Medallion.Threading.Internal; + +namespace Medallion.Threading.InProcess; + +public partial class InProcessDistributedLock +{ + // AUTO-GENERATED + + IDistributedSynchronizationHandle? IDistributedLock.TryAcquire(TimeSpan timeout, CancellationToken cancellationToken) => + this.TryAcquire(timeout, cancellationToken); + IDistributedSynchronizationHandle IDistributedLock.Acquire(TimeSpan? timeout, CancellationToken cancellationToken) => + this.Acquire(timeout, cancellationToken); + ValueTask IDistributedLock.TryAcquireAsync(TimeSpan timeout, CancellationToken cancellationToken) => + this.TryAcquireAsync(timeout, cancellationToken).Convert(To.ValueTask); + ValueTask IDistributedLock.AcquireAsync(TimeSpan? timeout, CancellationToken cancellationToken) => + this.AcquireAsync(timeout, cancellationToken).Convert(To.ValueTask); + + /// + /// Attempts to acquire the lock synchronously. Usage: + /// + /// using (var handle = myLock.TryAcquire(...)) + /// { + /// if (handle != null) { /* we have the lock! */ } + /// } + /// // dispose releases the lock if we took it + /// + /// + /// How long to wait before giving up on the acquisition attempt. Defaults to 0 + /// Specifies a token by which the wait can be canceled + /// An which can be used to release the lock or null on failure + public InProcessDistributedLockHandle? TryAcquire(TimeSpan timeout = default, CancellationToken cancellationToken = default) => + DistributedLockHelpers.TryAcquire(this, timeout, cancellationToken); + + /// + /// Acquires the lock synchronously, failing with if the attempt times out. Usage: + /// + /// using (myLock.Acquire(...)) + /// { + /// /* we have the lock! */ + /// } + /// // dispose releases the lock + /// + /// + /// How long to wait before giving up on the acquisition attempt. Defaults to + /// Specifies a token by which the wait can be canceled + /// An which can be used to release the lock + public InProcessDistributedLockHandle Acquire(TimeSpan? timeout = null, CancellationToken cancellationToken = default) => + DistributedLockHelpers.Acquire(this, timeout, cancellationToken); + + /// + /// Attempts to acquire the lock asynchronously. Usage: + /// + /// await using (var handle = await myLock.TryAcquireAsync(...)) + /// { + /// if (handle != null) { /* we have the lock! */ } + /// } + /// // dispose releases the lock if we took it + /// + /// + /// How long to wait before giving up on the acquisition attempt. Defaults to 0 + /// Specifies a token by which the wait can be canceled + /// An which can be used to release the lock or null on failure + public ValueTask TryAcquireAsync(TimeSpan timeout = default, CancellationToken cancellationToken = default) => + this.As>().InternalTryAcquireAsync(timeout, cancellationToken); + + /// + /// Acquires the lock asynchronously, failing with if the attempt times out. Usage: + /// + /// await using (await myLock.AcquireAsync(...)) + /// { + /// /* we have the lock! */ + /// } + /// // dispose releases the lock + /// + /// + /// How long to wait before giving up on the acquisition attempt. Defaults to + /// Specifies a token by which the wait can be canceled + /// An which can be used to release the lock + public ValueTask AcquireAsync(TimeSpan? timeout = null, CancellationToken cancellationToken = default) => + DistributedLockHelpers.AcquireAsync(this, timeout, cancellationToken); +} \ No newline at end of file diff --git a/src/DistributedLock.InProcess/InProcessDistributedLock.cs b/src/DistributedLock.InProcess/InProcessDistributedLock.cs new file mode 100644 index 00000000..68ec1b6e --- /dev/null +++ b/src/DistributedLock.InProcess/InProcessDistributedLock.cs @@ -0,0 +1,32 @@ +using Medallion.Threading.Internal; + +namespace Medallion.Threading.InProcess; + +/// +/// Implements a non-distributed mutex lock based on a shared . Instances are created via +/// , which shares the underlying primitive by scope and name. +/// A handle which is abandoned without being disposed is released only after it has been garbage collected and +/// processed by a background finalizer (roughly 30 seconds); until then the lock remains held. +/// +public sealed partial class InProcessDistributedLock : IInternalDistributedLock +{ + private readonly SemaphoreSlim _semaphore; + + internal InProcessDistributedLock(string name, SemaphoreSlim semaphore) + { + this.Name = name ?? throw new ArgumentNullException(nameof(name)); + this._semaphore = semaphore ?? throw new ArgumentNullException(nameof(semaphore)); + } + + /// + /// Implements + /// + public string Name { get; } + + async ValueTask IInternalDistributedLock.InternalTryAcquireAsync( + TimeoutValue timeout, + CancellationToken cancellationToken) => + await this._semaphore.InternalWaitAsync(timeout.TimeSpan, cancellationToken).ConfigureAwait(false) + ? new InProcessDistributedLockHandle(this._semaphore) + : null; +} diff --git a/src/DistributedLock.InProcess/InProcessDistributedLockHandle.cs b/src/DistributedLock.InProcess/InProcessDistributedLockHandle.cs new file mode 100644 index 00000000..e9dd1388 --- /dev/null +++ b/src/DistributedLock.InProcess/InProcessDistributedLockHandle.cs @@ -0,0 +1,45 @@ +using Medallion.Threading.Internal; + +namespace Medallion.Threading.InProcess; + +/// +/// Implements +/// +public sealed class InProcessDistributedLockHandle : IDistributedSynchronizationHandle +{ + private readonly SemaphoreSlimReleaser _releaser; + private IDisposable? _finalizerRegistration; + + internal InProcessDistributedLockHandle(SemaphoreSlim semaphore) + { + this._releaser = new SemaphoreSlimReleaser(semaphore); + // We need a managed finalizer here because the underlying SemaphoreSlim is shared by all handles for the + // name; an abandoned handle would otherwise leak its ticket for as long as the semaphore stays alive. + this._finalizerRegistration = ManagedFinalizerQueue.Instance.Register(this, this._releaser); + } + + /// + /// Implements . An in-process handle can never + /// be lost, so this is always until the handle is disposed. + /// + public CancellationToken HandleLostToken => + Volatile.Read(ref this._finalizerRegistration) != null ? CancellationToken.None : throw this.ObjectDisposed(); + + /// + /// Releases the lock + /// + public void Dispose() + { + Interlocked.Exchange(ref this._finalizerRegistration, null)?.Dispose(); + this._releaser.Dispose(); + } + + /// + /// Releases the lock + /// + public ValueTask DisposeAsync() + { + this.Dispose(); + return default; + } +} diff --git a/src/DistributedLock.InProcess/InProcessDistributedReaderWriterLock.IDistributedUpgradeableReaderWriterLock.cs b/src/DistributedLock.InProcess/InProcessDistributedReaderWriterLock.IDistributedUpgradeableReaderWriterLock.cs new file mode 100644 index 00000000..2bef2a8c --- /dev/null +++ b/src/DistributedLock.InProcess/InProcessDistributedReaderWriterLock.IDistributedUpgradeableReaderWriterLock.cs @@ -0,0 +1,226 @@ +using Medallion.Threading.Internal; + +namespace Medallion.Threading.InProcess; + +public partial class InProcessDistributedReaderWriterLock +{ + // AUTO-GENERATED + + IDistributedSynchronizationHandle? IDistributedReaderWriterLock.TryAcquireReadLock(TimeSpan timeout, CancellationToken cancellationToken) => + this.TryAcquireReadLock(timeout, cancellationToken); + IDistributedSynchronizationHandle IDistributedReaderWriterLock.AcquireReadLock(TimeSpan? timeout, CancellationToken cancellationToken) => + this.AcquireReadLock(timeout, cancellationToken); + ValueTask IDistributedReaderWriterLock.TryAcquireReadLockAsync(TimeSpan timeout, CancellationToken cancellationToken) => + this.TryAcquireReadLockAsync(timeout, cancellationToken).Convert(To.ValueTask); + ValueTask IDistributedReaderWriterLock.AcquireReadLockAsync(TimeSpan? timeout, CancellationToken cancellationToken) => + this.AcquireReadLockAsync(timeout, cancellationToken).Convert(To.ValueTask); + IDistributedLockUpgradeableHandle? IDistributedUpgradeableReaderWriterLock.TryAcquireUpgradeableReadLock(TimeSpan timeout, CancellationToken cancellationToken) => + this.TryAcquireUpgradeableReadLock(timeout, cancellationToken); + IDistributedLockUpgradeableHandle IDistributedUpgradeableReaderWriterLock.AcquireUpgradeableReadLock(TimeSpan? timeout, CancellationToken cancellationToken) => + this.AcquireUpgradeableReadLock(timeout, cancellationToken); + ValueTask IDistributedUpgradeableReaderWriterLock.TryAcquireUpgradeableReadLockAsync(TimeSpan timeout, CancellationToken cancellationToken) => + this.TryAcquireUpgradeableReadLockAsync(timeout, cancellationToken).Convert(To.ValueTask); + ValueTask IDistributedUpgradeableReaderWriterLock.AcquireUpgradeableReadLockAsync(TimeSpan? timeout, CancellationToken cancellationToken) => + this.AcquireUpgradeableReadLockAsync(timeout, cancellationToken).Convert(To.ValueTask); + IDistributedSynchronizationHandle? IDistributedReaderWriterLock.TryAcquireWriteLock(TimeSpan timeout, CancellationToken cancellationToken) => + this.TryAcquireWriteLock(timeout, cancellationToken); + IDistributedSynchronizationHandle IDistributedReaderWriterLock.AcquireWriteLock(TimeSpan? timeout, CancellationToken cancellationToken) => + this.AcquireWriteLock(timeout, cancellationToken); + ValueTask IDistributedReaderWriterLock.TryAcquireWriteLockAsync(TimeSpan timeout, CancellationToken cancellationToken) => + this.TryAcquireWriteLockAsync(timeout, cancellationToken).Convert(To.ValueTask); + ValueTask IDistributedReaderWriterLock.AcquireWriteLockAsync(TimeSpan? timeout, CancellationToken cancellationToken) => + this.AcquireWriteLockAsync(timeout, cancellationToken).Convert(To.ValueTask); + + /// + /// Attempts to acquire a READ lock synchronously. Multiple readers are allowed. Not compatible with a WRITE lock. Usage: + /// + /// using (var handle = myLock.TryAcquireReadLock(...)) + /// { + /// if (handle != null) { /* we have the lock! */ } + /// } + /// // dispose releases the lock if we took it + /// + /// + /// How long to wait before giving up on the acquisition attempt. Defaults to 0 + /// Specifies a token by which the wait can be canceled + /// An which can be used to release the lock or null on failure + public InProcessDistributedReaderWriterLockHandle? TryAcquireReadLock(TimeSpan timeout = default, CancellationToken cancellationToken = default) => + DistributedLockHelpers.TryAcquire(this, timeout, cancellationToken, isWrite: false); + + /// + /// Acquires a READ lock synchronously, failing with if the attempt times out. Multiple readers are allowed. Not compatible with a WRITE lock. Usage: + /// + /// using (myLock.AcquireReadLock(...)) + /// { + /// /* we have the lock! */ + /// } + /// // dispose releases the lock + /// + /// + /// How long to wait before giving up on the acquisition attempt. Defaults to + /// Specifies a token by which the wait can be canceled + /// An which can be used to release the lock + public InProcessDistributedReaderWriterLockHandle AcquireReadLock(TimeSpan? timeout = null, CancellationToken cancellationToken = default) => + DistributedLockHelpers.Acquire(this, timeout, cancellationToken, isWrite: false); + + /// + /// Attempts to acquire a READ lock asynchronously. Multiple readers are allowed. Not compatible with a WRITE lock. Usage: + /// + /// await using (var handle = await myLock.TryAcquireReadLockAsync(...)) + /// { + /// if (handle != null) { /* we have the lock! */ } + /// } + /// // dispose releases the lock if we took it + /// + /// + /// How long to wait before giving up on the acquisition attempt. Defaults to 0 + /// Specifies a token by which the wait can be canceled + /// An which can be used to release the lock or null on failure + public ValueTask TryAcquireReadLockAsync(TimeSpan timeout = default, CancellationToken cancellationToken = default) => + this.As>().InternalTryAcquireAsync(timeout, cancellationToken, isWrite: false); + + /// + /// Acquires a READ lock asynchronously, failing with if the attempt times out. Multiple readers are allowed. Not compatible with a WRITE lock. Usage: + /// + /// await using (await myLock.AcquireReadLockAsync(...)) + /// { + /// /* we have the lock! */ + /// } + /// // dispose releases the lock + /// + /// + /// How long to wait before giving up on the acquisition attempt. Defaults to + /// Specifies a token by which the wait can be canceled + /// An which can be used to release the lock + public ValueTask AcquireReadLockAsync(TimeSpan? timeout = null, CancellationToken cancellationToken = default) => + DistributedLockHelpers.AcquireAsync(this, timeout, cancellationToken, isWrite: false); + + /// + /// Attempts to acquire an UPGRADE lock synchronously. Not compatible with another UPGRADE lock or a WRITE lock. Usage: + /// + /// using (var handle = myLock.TryAcquireUpgradeableReadLock(...)) + /// { + /// if (handle != null) { /* we have the lock! */ } + /// } + /// // dispose releases the lock if we took it + /// + /// + /// How long to wait before giving up on the acquisition attempt. Defaults to 0 + /// Specifies a token by which the wait can be canceled + /// An which can be used to release the lock or null on failure + public InProcessDistributedReaderWriterLockUpgradeableHandle? TryAcquireUpgradeableReadLock(TimeSpan timeout = default, CancellationToken cancellationToken = default) => + DistributedLockHelpers.TryAcquireUpgradeableReadLock(this, timeout, cancellationToken); + + /// + /// Acquires an UPGRADE lock synchronously, failing with if the attempt times out. Not compatible with another UPGRADE lock or a WRITE lock. Usage: + /// + /// using (myLock.AcquireUpgradeableReadLock(...)) + /// { + /// /* we have the lock! */ + /// } + /// // dispose releases the lock + /// + /// + /// How long to wait before giving up on the acquisition attempt. Defaults to + /// Specifies a token by which the wait can be canceled + /// An which can be used to release the lock + public InProcessDistributedReaderWriterLockUpgradeableHandle AcquireUpgradeableReadLock(TimeSpan? timeout = null, CancellationToken cancellationToken = default) => + DistributedLockHelpers.AcquireUpgradeableReadLock(this, timeout, cancellationToken); + + /// + /// Attempts to acquire an UPGRADE lock asynchronously. Not compatible with another UPGRADE lock or a WRITE lock. Usage: + /// + /// await using (var handle = await myLock.TryAcquireUpgradeableReadLockAsync(...)) + /// { + /// if (handle != null) { /* we have the lock! */ } + /// } + /// // dispose releases the lock if we took it + /// + /// + /// How long to wait before giving up on the acquisition attempt. Defaults to 0 + /// Specifies a token by which the wait can be canceled + /// An which can be used to release the lock or null on failure + public ValueTask TryAcquireUpgradeableReadLockAsync(TimeSpan timeout = default, CancellationToken cancellationToken = default) => + this.As>().InternalTryAcquireUpgradeableReadLockAsync(timeout, cancellationToken); + + /// + /// Acquires an UPGRADE lock asynchronously, failing with if the attempt times out. Not compatible with another UPGRADE lock or a WRITE lock. Usage: + /// + /// await using (await myLock.AcquireUpgradeableReadLockAsync(...)) + /// { + /// /* we have the lock! */ + /// } + /// // dispose releases the lock + /// + /// + /// How long to wait before giving up on the acquisition attempt. Defaults to + /// Specifies a token by which the wait can be canceled + /// An which can be used to release the lock + public ValueTask AcquireUpgradeableReadLockAsync(TimeSpan? timeout = null, CancellationToken cancellationToken = default) => + DistributedLockHelpers.AcquireUpgradeableReadLockAsync(this, timeout, cancellationToken); + + /// + /// Attempts to acquire a WRITE lock synchronously. Not compatible with another WRITE lock or an UPGRADE lock. Usage: + /// + /// using (var handle = myLock.TryAcquireWriteLock(...)) + /// { + /// if (handle != null) { /* we have the lock! */ } + /// } + /// // dispose releases the lock if we took it + /// + /// + /// How long to wait before giving up on the acquisition attempt. Defaults to 0 + /// Specifies a token by which the wait can be canceled + /// An which can be used to release the lock or null on failure + public InProcessDistributedReaderWriterLockHandle? TryAcquireWriteLock(TimeSpan timeout = default, CancellationToken cancellationToken = default) => + DistributedLockHelpers.TryAcquire(this, timeout, cancellationToken, isWrite: true); + + /// + /// Acquires a WRITE lock synchronously, failing with if the attempt times out. Not compatible with another WRITE lock or an UPGRADE lock. Usage: + /// + /// using (myLock.AcquireWriteLock(...)) + /// { + /// /* we have the lock! */ + /// } + /// // dispose releases the lock + /// + /// + /// How long to wait before giving up on the acquisition attempt. Defaults to + /// Specifies a token by which the wait can be canceled + /// An which can be used to release the lock + public InProcessDistributedReaderWriterLockHandle AcquireWriteLock(TimeSpan? timeout = null, CancellationToken cancellationToken = default) => + DistributedLockHelpers.Acquire(this, timeout, cancellationToken, isWrite: true); + + /// + /// Attempts to acquire a WRITE lock asynchronously. Not compatible with another WRITE lock or an UPGRADE lock. Usage: + /// + /// await using (var handle = await myLock.TryAcquireWriteLockAsync(...)) + /// { + /// if (handle != null) { /* we have the lock! */ } + /// } + /// // dispose releases the lock if we took it + /// + /// + /// How long to wait before giving up on the acquisition attempt. Defaults to 0 + /// Specifies a token by which the wait can be canceled + /// An which can be used to release the lock or null on failure + public ValueTask TryAcquireWriteLockAsync(TimeSpan timeout = default, CancellationToken cancellationToken = default) => + this.As>().InternalTryAcquireAsync(timeout, cancellationToken, isWrite: true); + + /// + /// Acquires a WRITE lock asynchronously, failing with if the attempt times out. Not compatible with another WRITE lock or an UPGRADE lock. Usage: + /// + /// await using (await myLock.AcquireWriteLockAsync(...)) + /// { + /// /* we have the lock! */ + /// } + /// // dispose releases the lock + /// + /// + /// How long to wait before giving up on the acquisition attempt. Defaults to + /// Specifies a token by which the wait can be canceled + /// An which can be used to release the lock + public ValueTask AcquireWriteLockAsync(TimeSpan? timeout = null, CancellationToken cancellationToken = default) => + DistributedLockHelpers.AcquireAsync(this, timeout, cancellationToken, isWrite: true); + +} \ No newline at end of file diff --git a/src/DistributedLock.InProcess/InProcessDistributedReaderWriterLock.cs b/src/DistributedLock.InProcess/InProcessDistributedReaderWriterLock.cs new file mode 100644 index 00000000..ec51d3e7 --- /dev/null +++ b/src/DistributedLock.InProcess/InProcessDistributedReaderWriterLock.cs @@ -0,0 +1,42 @@ +using Medallion.Threading.Internal; + +namespace Medallion.Threading.InProcess; + +/// +/// Implements a non-distributed, writer-preferring upgradeable reader-writer lock. Instances are created via +/// , which shares the underlying state by scope and name. +/// A handle which is abandoned without being disposed is released only after it has been garbage collected and +/// processed by a background finalizer (roughly 30 seconds). Until then, an abandoned read handle blocks writers +/// and, once a writer is waiting, all new readers as well. +/// +public sealed partial class InProcessDistributedReaderWriterLock : IInternalDistributedUpgradeableReaderWriterLock +{ + private readonly AsyncReaderWriterLockState _state; + + internal InProcessDistributedReaderWriterLock(string name, AsyncReaderWriterLockState state) + { + this.Name = name ?? throw new ArgumentNullException(nameof(name)); + this._state = state ?? throw new ArgumentNullException(nameof(state)); + } + + /// + /// Implements + /// + public string Name { get; } + + async ValueTask + IInternalDistributedReaderWriterLock.InternalTryAcquireAsync( + TimeoutValue timeout, + CancellationToken cancellationToken, + bool isWrite) => + await this._state.TryAcquireAsync(timeout, cancellationToken, isWrite).ConfigureAwait(false) + ? new InProcessDistributedReaderWriterLockNonUpgradeableHandle(this._state, isWrite) + : null; + + async ValueTask + IInternalDistributedUpgradeableReaderWriterLock + .InternalTryAcquireUpgradeableReadLockAsync(TimeoutValue timeout, CancellationToken cancellationToken) => + await this._state.TryAcquireAsync(timeout, cancellationToken, InProcessReaderWriterLockType.Upgrade).ConfigureAwait(false) + ? new InProcessDistributedReaderWriterLockUpgradeableHandle(this._state) + : null; +} \ No newline at end of file diff --git a/src/DistributedLock.InProcess/InProcessDistributedReaderWriterLockHandle.cs b/src/DistributedLock.InProcess/InProcessDistributedReaderWriterLockHandle.cs new file mode 100644 index 00000000..938dece2 --- /dev/null +++ b/src/DistributedLock.InProcess/InProcessDistributedReaderWriterLockHandle.cs @@ -0,0 +1,175 @@ +using Medallion.Threading.Internal; + +namespace Medallion.Threading.InProcess; + +/// +/// Implements +/// +public abstract class InProcessDistributedReaderWriterLockHandle : IDistributedSynchronizationHandle +{ + private readonly LockReleaser _lockReleaser; + private IDisposable? _finalizerRegistration; + + internal InProcessDistributedReaderWriterLockHandle(AsyncReaderWriterLockState state, InProcessReaderWriterLockType lockType) + { + this._lockReleaser = new LockReleaser(state, lockType); + // We need a managed finalizer here because the underlying state is shared by all handles for the name; + // an abandoned handle would otherwise hold its read/write claim for as long as that state stays alive. + this._finalizerRegistration = ManagedFinalizerQueue.Instance.Register(this, this._lockReleaser); + } + + private protected bool IsDisposed => Volatile.Read(ref this._finalizerRegistration) is null; + + private protected LockReleaser Releaser => this._lockReleaser; + + /// + /// Implements . + /// An in-process handle can never be lost, so this is always + /// until the handle is disposed. + /// + public CancellationToken HandleLostToken => + !this.IsDisposed ? CancellationToken.None : throw this.ObjectDisposed(); + + /// + /// Releases the lock + /// + public void Dispose() + { + Interlocked.Exchange(ref this._finalizerRegistration, null)?.Dispose(); + this._lockReleaser.Dispose(); + } + + /// + /// Releases the lock + /// + public ValueTask DisposeAsync() + { + this.Dispose(); + return default; + } + + private protected class LockReleaser(AsyncReaderWriterLockState state, InProcessReaderWriterLockType lockType) + : IDisposable, IAsyncDisposable + { + private const int NotUpgradedState = 0, UpgradedState = 1, DisposedState = 2; + + private readonly AsyncReaderWriterLockState _state = state; + private int _upgradeState = NotUpgradedState; + + public AsyncReaderWriterLockState State => this._state; + + /// + /// Records that the caller has successfully taken the write claim, transferring responsibility for + /// releasing it to . Returns false if the handle was already disposed, in which + /// case the caller still owns the write claim and must release it via + /// . + /// + public bool TryMarkUpgraded() => + Interlocked.CompareExchange(ref this._upgradeState, UpgradedState, NotUpgradedState) == NotUpgradedState; + + public void Dispose() + { + var previousState = Interlocked.Exchange(ref this._upgradeState, DisposedState); + if (previousState != DisposedState) + { + this._state.Release(lockType, upgraded: previousState == UpgradedState); + } + } + + public ValueTask DisposeAsync() + { + this.Dispose(); + return default; + } + } +} + +internal sealed class InProcessDistributedReaderWriterLockNonUpgradeableHandle(AsyncReaderWriterLockState state, bool isWrite) + : InProcessDistributedReaderWriterLockHandle(state, isWrite ? InProcessReaderWriterLockType.Write : InProcessReaderWriterLockType.Read) +{ +} + +/// +/// Implements +/// +public sealed class InProcessDistributedReaderWriterLockUpgradeableHandle + : InProcessDistributedReaderWriterLockHandle, IInternalDistributedLockUpgradeableHandle +{ + private int _upgradeAttemptState; // 0 = no upgrade in flight or completed, 1 = in flight or upgraded + + internal InProcessDistributedReaderWriterLockUpgradeableHandle(AsyncReaderWriterLockState state) + : base(state, InProcessReaderWriterLockType.Upgrade) + { + } + + /// + /// Implements + /// + public bool TryUpgradeToWriteLock(TimeSpan timeout = default, CancellationToken cancellationToken = default) => + DistributedLockHelpers.TryUpgradeToWriteLock(this, timeout, cancellationToken); + + /// + /// Implements + /// + public ValueTask TryUpgradeToWriteLockAsync(TimeSpan timeout = default, CancellationToken cancellationToken = default) => + this.As().InternalTryUpgradeToWriteLockAsync(timeout, cancellationToken); + + /// + /// Implements + /// + public void UpgradeToWriteLock(TimeSpan? timeout = null, CancellationToken cancellationToken = default) => + DistributedLockHelpers.UpgradeToWriteLock(this, timeout, cancellationToken); + + /// + /// Implements + /// + public ValueTask UpgradeToWriteLockAsync(TimeSpan? timeout = null, CancellationToken cancellationToken = default) => + DistributedLockHelpers.UpgradeToWriteLockAsync(this, timeout, cancellationToken); + + ValueTask IInternalDistributedLockUpgradeableHandle.InternalTryUpgradeToWriteLockAsync(TimeoutValue timeout, CancellationToken cancellationToken) + { + // these validations must throw synchronously + if (this.IsDisposed) { throw this.ObjectDisposed(); } + if (Interlocked.CompareExchange(ref this._upgradeAttemptState, 1, 0) != 0) + { + throw new InvalidOperationException("the lock has already been upgraded"); + } + + return TryPerformUpgradeAsync(); + + async ValueTask TryPerformUpgradeAsync() + { + var releaser = this.Releaser; + var state = releaser.State; + + bool acquiredWriteClaim; + try + { + // we already hold the upgrade semaphore, so we only need the entry gate + write semaphore + acquiredWriteClaim = await state.TryAcquireWriteCoreAsync(timeout, cancellationToken).ConfigureAwait(false); + } + catch + { + // on cancellation/failure the upgrade never happened, so allow a retry + Volatile.Write(ref this._upgradeAttemptState, 0); + throw; + } + + if (!acquiredWriteClaim) + { + Volatile.Write(ref this._upgradeAttemptState, 0); // timed out; allow a retry + return false; + } + + if (!releaser.TryMarkUpgraded()) + { + // The handle was disposed while we were acquiring. Dispose() released the upgrade claim but knows + // nothing about the write claim we just took, so we must unwind it here. + state.ReleaseWriteCore(); + throw this.ObjectDisposed(); + } + + return true; + } + } +} \ No newline at end of file diff --git a/src/DistributedLock.InProcess/InProcessDistributedSemaphore.IDistributedSemaphore.cs b/src/DistributedLock.InProcess/InProcessDistributedSemaphore.IDistributedSemaphore.cs new file mode 100644 index 00000000..231153ad --- /dev/null +++ b/src/DistributedLock.InProcess/InProcessDistributedSemaphore.IDistributedSemaphore.cs @@ -0,0 +1,81 @@ +using Medallion.Threading.Internal; + +namespace Medallion.Threading.InProcess; + +public partial class InProcessDistributedSemaphore +{ + // AUTO-GENERATED + + IDistributedSynchronizationHandle? IDistributedSemaphore.TryAcquire(TimeSpan timeout, CancellationToken cancellationToken) => + this.TryAcquire(timeout, cancellationToken); + IDistributedSynchronizationHandle IDistributedSemaphore.Acquire(TimeSpan? timeout, CancellationToken cancellationToken) => + this.Acquire(timeout, cancellationToken); + ValueTask IDistributedSemaphore.TryAcquireAsync(TimeSpan timeout, CancellationToken cancellationToken) => + this.TryAcquireAsync(timeout, cancellationToken).Convert(To.ValueTask); + ValueTask IDistributedSemaphore.AcquireAsync(TimeSpan? timeout, CancellationToken cancellationToken) => + this.AcquireAsync(timeout, cancellationToken).Convert(To.ValueTask); + + /// + /// Attempts to acquire a semaphore ticket synchronously. Usage: + /// + /// using (var handle = mySemaphore.TryAcquire(...)) + /// { + /// if (handle != null) { /* we have the ticket! */ } + /// } + /// // dispose releases the ticket if we took it + /// + /// + /// How long to wait before giving up on the acquisition attempt. Defaults to 0 + /// Specifies a token by which the wait can be canceled + /// An which can be used to release the ticket or null on failure + public InProcessDistributedSemaphoreHandle? TryAcquire(TimeSpan timeout = default, CancellationToken cancellationToken = default) => + DistributedLockHelpers.TryAcquire(this, timeout, cancellationToken); + + /// + /// Acquires a semaphore ticket synchronously, failing with if the attempt times out. Usage: + /// + /// using (mySemaphore.Acquire(...)) + /// { + /// /* we have the ticket! */ + /// } + /// // dispose releases the ticket + /// + /// + /// How long to wait before giving up on the acquisition attempt. Defaults to + /// Specifies a token by which the wait can be canceled + /// An which can be used to release the ticket + public InProcessDistributedSemaphoreHandle Acquire(TimeSpan? timeout = null, CancellationToken cancellationToken = default) => + DistributedLockHelpers.Acquire(this, timeout, cancellationToken); + + /// + /// Attempts to acquire a semaphore ticket asynchronously. Usage: + /// + /// await using (var handle = await mySemaphore.TryAcquireAsync(...)) + /// { + /// if (handle != null) { /* we have the ticket! */ } + /// } + /// // dispose releases the ticket if we took it + /// + /// + /// How long to wait before giving up on the acquisition attempt. Defaults to 0 + /// Specifies a token by which the wait can be canceled + /// An which can be used to release the ticket or null on failure + public ValueTask TryAcquireAsync(TimeSpan timeout = default, CancellationToken cancellationToken = default) => + this.As>().InternalTryAcquireAsync(timeout, cancellationToken); + + /// + /// Acquires a semaphore ticket asynchronously, failing with if the attempt times out. Usage: + /// + /// await using (await mySemaphore.AcquireAsync(...)) + /// { + /// /* we have the ticket! */ + /// } + /// // dispose releases the ticket + /// + /// + /// How long to wait before giving up on the acquisition attempt. Defaults to + /// Specifies a token by which the wait can be canceled + /// An which can be used to release the ticket + public ValueTask AcquireAsync(TimeSpan? timeout = null, CancellationToken cancellationToken = default) => + DistributedLockHelpers.AcquireAsync(this, timeout, cancellationToken); +} \ No newline at end of file diff --git a/src/DistributedLock.InProcess/InProcessDistributedSemaphore.cs b/src/DistributedLock.InProcess/InProcessDistributedSemaphore.cs new file mode 100644 index 00000000..f23a5f5d --- /dev/null +++ b/src/DistributedLock.InProcess/InProcessDistributedSemaphore.cs @@ -0,0 +1,38 @@ +using Medallion.Threading.Internal; + +namespace Medallion.Threading.InProcess; + +/// +/// Implements a non-distributed semaphore based on a shared . Instances are created via +/// , which shares the underlying primitive by scope and name. +/// A handle which is abandoned without being disposed is released only after it has been garbage collected and +/// processed by a background finalizer (roughly 30 seconds); until then its ticket remains taken. +/// +public sealed partial class InProcessDistributedSemaphore : IInternalDistributedSemaphore +{ + private readonly SemaphoreSlim _semaphore; + + internal InProcessDistributedSemaphore(string name, int maxCount, SemaphoreSlim semaphore) + { + this.Name = name ?? throw new ArgumentNullException(nameof(name)); + this.MaxCount = maxCount; + this._semaphore = semaphore ?? throw new ArgumentNullException(nameof(semaphore)); + } + + /// + /// Implements + /// + public string Name { get; } + + /// + /// Implements + /// + public int MaxCount { get; } + + async ValueTask IInternalDistributedSemaphore.InternalTryAcquireAsync( + TimeoutValue timeout, + CancellationToken cancellationToken) => + await this._semaphore.InternalWaitAsync(timeout.TimeSpan, cancellationToken).ConfigureAwait(false) + ? new InProcessDistributedSemaphoreHandle(this._semaphore) + : null; +} diff --git a/src/DistributedLock.InProcess/InProcessDistributedSemaphoreHandle.cs b/src/DistributedLock.InProcess/InProcessDistributedSemaphoreHandle.cs new file mode 100644 index 00000000..047d0c8f --- /dev/null +++ b/src/DistributedLock.InProcess/InProcessDistributedSemaphoreHandle.cs @@ -0,0 +1,45 @@ +using Medallion.Threading.Internal; + +namespace Medallion.Threading.InProcess; + +/// +/// Implements +/// +public sealed class InProcessDistributedSemaphoreHandle : IDistributedSynchronizationHandle +{ + private readonly SemaphoreSlimReleaser _releaser; + private IDisposable? _finalizerRegistration; + + internal InProcessDistributedSemaphoreHandle(SemaphoreSlim semaphore) + { + this._releaser = new SemaphoreSlimReleaser(semaphore); + // We need a managed finalizer here because the underlying SemaphoreSlim is shared by all handles for the + // name; an abandoned handle would otherwise leak its ticket for as long as the semaphore stays alive. + this._finalizerRegistration = ManagedFinalizerQueue.Instance.Register(this, this._releaser); + } + + /// + /// Implements . An in-process handle can never + /// be lost, so this is always until the handle is disposed. + /// + public CancellationToken HandleLostToken => + Volatile.Read(ref this._finalizerRegistration) != null ? CancellationToken.None : throw this.ObjectDisposed(); + + /// + /// Releases the semaphore ticket + /// + public void Dispose() + { + Interlocked.Exchange(ref this._finalizerRegistration, null)?.Dispose(); + this._releaser.Dispose(); + } + + /// + /// Releases the semaphore ticket + /// + public ValueTask DisposeAsync() + { + this.Dispose(); + return default; + } +} diff --git a/src/DistributedLock.InProcess/InProcessDistributedSynchronizationProvider.cs b/src/DistributedLock.InProcess/InProcessDistributedSynchronizationProvider.cs new file mode 100644 index 00000000..ed42d393 --- /dev/null +++ b/src/DistributedLock.InProcess/InProcessDistributedSynchronizationProvider.cs @@ -0,0 +1,74 @@ +namespace Medallion.Threading.InProcess; + +/// +/// Implements , , and +/// for in-process synchronization primitives. +/// +/// Primitives are cached process-wide by (, name), so all primitives created for the same name +/// by providers with the same coordinate with one another, even across provider instances. +/// Providers with different scopes never coordinate. +/// +/// Locks, semaphores, and reader-writer locks are cached separately: a lock and a reader-writer lock with the +/// same name do NOT coordinate with one another. +/// +/// Note that these primitives only synchronize within the current process. Handles which are never disposed are +/// released by a background finalizer only after the handle has been garbage collected, which can take on the order +/// of 30 seconds; until then, the primitive remains held. +/// +public sealed class InProcessDistributedSynchronizationProvider : + IDistributedLockProvider, IDistributedSemaphoreProvider, IDistributedUpgradeableReaderWriterLockProvider +{ + private static readonly ConcurrentWeakReferenceDictionary<(string Scope, string Name), SemaphoreSlim> Locks = new(); + private static readonly ConcurrentWeakReferenceDictionary<(string Scope, string Name), SemaphoreSlim> Semaphores = new(); + private static readonly ConcurrentWeakReferenceDictionary<(string Scope, string Name), AsyncReaderWriterLockState> ReaderWriterLocks = new(); + + /// + /// Constructs a provider for the given . Providers with the same scope share primitives; + /// providers with different scopes are isolated from one another. When is not specified, + /// a default process-wide scope is used. + /// + public InProcessDistributedSynchronizationProvider(string? scope = null) + { + this.Scope = scope ?? string.Empty; + } + + /// + /// The scope within which primitives created by this provider are shared. + /// + public string Scope { get; } + + /// + /// Creates an with the given . + /// + public InProcessDistributedLock CreateLock(string name) => + new(name, Locks.GetOrAdd(this.GetKey(name), _ => new SemaphoreSlim(1, 1))); + + IDistributedLock IDistributedLockProvider.CreateLock(string name) => this.CreateLock(name); + + /// + /// Creates an with the given . + /// + public InProcessDistributedReaderWriterLock CreateReaderWriterLock(string name) => + new(name, ReaderWriterLocks.GetOrAdd(this.GetKey(name), _ => new AsyncReaderWriterLockState())); + + IDistributedReaderWriterLock IDistributedReaderWriterLockProvider.CreateReaderWriterLock(string name) => + this.CreateReaderWriterLock(name); + + IDistributedUpgradeableReaderWriterLock IDistributedUpgradeableReaderWriterLockProvider.CreateUpgradeableReaderWriterLock(string name) => + this.CreateReaderWriterLock(name); + + /// + /// Creates an with the given and . + /// + public InProcessDistributedSemaphore CreateSemaphore(string name, int maxCount) + { + if (maxCount < 1) { throw new ArgumentOutOfRangeException(nameof(maxCount), maxCount, "must be positive"); } + return new(name, maxCount, Semaphores.GetOrAdd(this.GetKey(name), _ => new SemaphoreSlim(maxCount, maxCount))); + } + + IDistributedSemaphore IDistributedSemaphoreProvider.CreateSemaphore(string name, int maxCount) => + this.CreateSemaphore(name, maxCount); + + private (string Scope, string Name) GetKey(string name) => + (this.Scope, name ?? throw new ArgumentNullException(nameof(name))); +} diff --git a/src/DistributedLock.InProcess/InProcessReaderWriterLockType.cs b/src/DistributedLock.InProcess/InProcessReaderWriterLockType.cs new file mode 100644 index 00000000..feb9ab8f --- /dev/null +++ b/src/DistributedLock.InProcess/InProcessReaderWriterLockType.cs @@ -0,0 +1,8 @@ +namespace Medallion.Threading.InProcess; + +internal enum InProcessReaderWriterLockType +{ + Read, + Upgrade, + Write, +} \ No newline at end of file diff --git a/src/DistributedLock.InProcess/PublicAPI.Shipped.txt b/src/DistributedLock.InProcess/PublicAPI.Shipped.txt new file mode 100644 index 00000000..ab058de6 --- /dev/null +++ b/src/DistributedLock.InProcess/PublicAPI.Shipped.txt @@ -0,0 +1 @@ +#nullable enable diff --git a/src/DistributedLock.InProcess/PublicAPI.Unshipped.txt b/src/DistributedLock.InProcess/PublicAPI.Unshipped.txt new file mode 100644 index 00000000..3851cca2 --- /dev/null +++ b/src/DistributedLock.InProcess/PublicAPI.Unshipped.txt @@ -0,0 +1,50 @@ +Medallion.Threading.InProcess.InProcessDistributedLock +Medallion.Threading.InProcess.InProcessDistributedLock.Acquire(System.TimeSpan? timeout = null, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> Medallion.Threading.InProcess.InProcessDistributedLockHandle! +Medallion.Threading.InProcess.InProcessDistributedLock.AcquireAsync(System.TimeSpan? timeout = null, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.ValueTask +Medallion.Threading.InProcess.InProcessDistributedLock.Name.get -> string! +Medallion.Threading.InProcess.InProcessDistributedLock.TryAcquire(System.TimeSpan timeout = default(System.TimeSpan), System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> Medallion.Threading.InProcess.InProcessDistributedLockHandle? +Medallion.Threading.InProcess.InProcessDistributedLock.TryAcquireAsync(System.TimeSpan timeout = default(System.TimeSpan), System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.ValueTask +Medallion.Threading.InProcess.InProcessDistributedLockHandle +Medallion.Threading.InProcess.InProcessDistributedLockHandle.Dispose() -> void +Medallion.Threading.InProcess.InProcessDistributedLockHandle.DisposeAsync() -> System.Threading.Tasks.ValueTask +Medallion.Threading.InProcess.InProcessDistributedLockHandle.HandleLostToken.get -> System.Threading.CancellationToken +Medallion.Threading.InProcess.InProcessDistributedReaderWriterLock +Medallion.Threading.InProcess.InProcessDistributedReaderWriterLock.AcquireReadLock(System.TimeSpan? timeout = null, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> Medallion.Threading.InProcess.InProcessDistributedReaderWriterLockHandle! +Medallion.Threading.InProcess.InProcessDistributedReaderWriterLock.AcquireReadLockAsync(System.TimeSpan? timeout = null, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.ValueTask +Medallion.Threading.InProcess.InProcessDistributedReaderWriterLock.AcquireUpgradeableReadLock(System.TimeSpan? timeout = null, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> Medallion.Threading.InProcess.InProcessDistributedReaderWriterLockUpgradeableHandle! +Medallion.Threading.InProcess.InProcessDistributedReaderWriterLock.AcquireUpgradeableReadLockAsync(System.TimeSpan? timeout = null, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.ValueTask +Medallion.Threading.InProcess.InProcessDistributedReaderWriterLock.AcquireWriteLock(System.TimeSpan? timeout = null, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> Medallion.Threading.InProcess.InProcessDistributedReaderWriterLockHandle! +Medallion.Threading.InProcess.InProcessDistributedReaderWriterLock.AcquireWriteLockAsync(System.TimeSpan? timeout = null, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.ValueTask +Medallion.Threading.InProcess.InProcessDistributedReaderWriterLock.Name.get -> string! +Medallion.Threading.InProcess.InProcessDistributedReaderWriterLock.TryAcquireReadLock(System.TimeSpan timeout = default(System.TimeSpan), System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> Medallion.Threading.InProcess.InProcessDistributedReaderWriterLockHandle? +Medallion.Threading.InProcess.InProcessDistributedReaderWriterLock.TryAcquireReadLockAsync(System.TimeSpan timeout = default(System.TimeSpan), System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.ValueTask +Medallion.Threading.InProcess.InProcessDistributedReaderWriterLock.TryAcquireUpgradeableReadLock(System.TimeSpan timeout = default(System.TimeSpan), System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> Medallion.Threading.InProcess.InProcessDistributedReaderWriterLockUpgradeableHandle? +Medallion.Threading.InProcess.InProcessDistributedReaderWriterLock.TryAcquireUpgradeableReadLockAsync(System.TimeSpan timeout = default(System.TimeSpan), System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.ValueTask +Medallion.Threading.InProcess.InProcessDistributedReaderWriterLock.TryAcquireWriteLock(System.TimeSpan timeout = default(System.TimeSpan), System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> Medallion.Threading.InProcess.InProcessDistributedReaderWriterLockHandle? +Medallion.Threading.InProcess.InProcessDistributedReaderWriterLock.TryAcquireWriteLockAsync(System.TimeSpan timeout = default(System.TimeSpan), System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.ValueTask +Medallion.Threading.InProcess.InProcessDistributedReaderWriterLockHandle +Medallion.Threading.InProcess.InProcessDistributedReaderWriterLockHandle.Dispose() -> void +Medallion.Threading.InProcess.InProcessDistributedReaderWriterLockHandle.DisposeAsync() -> System.Threading.Tasks.ValueTask +Medallion.Threading.InProcess.InProcessDistributedReaderWriterLockHandle.HandleLostToken.get -> System.Threading.CancellationToken +Medallion.Threading.InProcess.InProcessDistributedReaderWriterLockUpgradeableHandle +Medallion.Threading.InProcess.InProcessDistributedReaderWriterLockUpgradeableHandle.TryUpgradeToWriteLock(System.TimeSpan timeout = default(System.TimeSpan), System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> bool +Medallion.Threading.InProcess.InProcessDistributedReaderWriterLockUpgradeableHandle.TryUpgradeToWriteLockAsync(System.TimeSpan timeout = default(System.TimeSpan), System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.ValueTask +Medallion.Threading.InProcess.InProcessDistributedReaderWriterLockUpgradeableHandle.UpgradeToWriteLock(System.TimeSpan? timeout = null, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> void +Medallion.Threading.InProcess.InProcessDistributedReaderWriterLockUpgradeableHandle.UpgradeToWriteLockAsync(System.TimeSpan? timeout = null, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.ValueTask +Medallion.Threading.InProcess.InProcessDistributedSemaphore +Medallion.Threading.InProcess.InProcessDistributedSemaphore.Acquire(System.TimeSpan? timeout = null, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> Medallion.Threading.InProcess.InProcessDistributedSemaphoreHandle! +Medallion.Threading.InProcess.InProcessDistributedSemaphore.AcquireAsync(System.TimeSpan? timeout = null, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.ValueTask +Medallion.Threading.InProcess.InProcessDistributedSemaphore.MaxCount.get -> int +Medallion.Threading.InProcess.InProcessDistributedSemaphore.Name.get -> string! +Medallion.Threading.InProcess.InProcessDistributedSemaphore.TryAcquire(System.TimeSpan timeout = default(System.TimeSpan), System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> Medallion.Threading.InProcess.InProcessDistributedSemaphoreHandle? +Medallion.Threading.InProcess.InProcessDistributedSemaphore.TryAcquireAsync(System.TimeSpan timeout = default(System.TimeSpan), System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.ValueTask +Medallion.Threading.InProcess.InProcessDistributedSemaphoreHandle +Medallion.Threading.InProcess.InProcessDistributedSemaphoreHandle.Dispose() -> void +Medallion.Threading.InProcess.InProcessDistributedSemaphoreHandle.DisposeAsync() -> System.Threading.Tasks.ValueTask +Medallion.Threading.InProcess.InProcessDistributedSemaphoreHandle.HandleLostToken.get -> System.Threading.CancellationToken +Medallion.Threading.InProcess.InProcessDistributedSynchronizationProvider +Medallion.Threading.InProcess.InProcessDistributedSynchronizationProvider.CreateLock(string! name) -> Medallion.Threading.InProcess.InProcessDistributedLock! +Medallion.Threading.InProcess.InProcessDistributedSynchronizationProvider.CreateReaderWriterLock(string! name) -> Medallion.Threading.InProcess.InProcessDistributedReaderWriterLock! +Medallion.Threading.InProcess.InProcessDistributedSynchronizationProvider.CreateSemaphore(string! name, int maxCount) -> Medallion.Threading.InProcess.InProcessDistributedSemaphore! +Medallion.Threading.InProcess.InProcessDistributedSynchronizationProvider.InProcessDistributedSynchronizationProvider(string? scope = null) -> void +Medallion.Threading.InProcess.InProcessDistributedSynchronizationProvider.Scope.get -> string! diff --git a/src/DistributedLock.InProcess/SemaphoreSlimReleaser.cs b/src/DistributedLock.InProcess/SemaphoreSlimReleaser.cs new file mode 100644 index 00000000..9d514644 --- /dev/null +++ b/src/DistributedLock.InProcess/SemaphoreSlimReleaser.cs @@ -0,0 +1,20 @@ +using Medallion.Threading.Internal; + +namespace Medallion.Threading.InProcess; + +/// +/// Releases a ticket on a shared exactly once. This is separate from the handle so +/// that it can be registered with without keeping the handle alive. +/// +internal sealed class SemaphoreSlimReleaser(SemaphoreSlim semaphore) : IDisposable, IAsyncDisposable +{ + private SemaphoreSlim? _semaphore = semaphore; + + public void Dispose() => Interlocked.Exchange(ref this._semaphore, null)?.Release(); + + public ValueTask DisposeAsync() + { + this.Dispose(); + return default; + } +} diff --git a/src/DistributedLock.InProcess/packages.lock.json b/src/DistributedLock.InProcess/packages.lock.json new file mode 100644 index 00000000..f3b9f1cf --- /dev/null +++ b/src/DistributedLock.InProcess/packages.lock.json @@ -0,0 +1,275 @@ +{ + "version": 2, + "dependencies": { + ".NETFramework,Version=v4.6.2": { + "Microsoft.CodeAnalysis.PublicApiAnalyzers": { + "type": "Direct", + "requested": "[3.3.4, )", + "resolved": "3.3.4", + "contentHash": "kNLTfXtXUWDHVt5iaPkkiPuyHYlMgLI6SOFT4w88bfeI2vqSeGgHunFkdvlaCM8RDfcY0t2+jnesQtidRJJ/DA==" + }, + "Microsoft.SourceLink.GitHub": { + "type": "Direct", + "requested": "[10.0.201, )", + "resolved": "10.0.201", + "contentHash": "qxYAmO4ktzd9L+HMdnqWucxpu7bI9undPyACXOMqPyhaiMtbpbYL/n0ACyWIJlbyEJrXFwxiOaBOSasLtDvsCg==" + }, + "Nullable": { + "type": "Direct", + "requested": "[1.3.1, )", + "resolved": "1.3.1", + "contentHash": "Mk4ZVDfAORTjvckQprCSehi1XgOAAlk5ez06Va/acRYEloN9t6d6zpzJRn5MEq7+RnagyFIq9r+kbWzLGd+6QA==" + }, + "System.Runtime.CompilerServices.Unsafe": { + "type": "Transitive", + "resolved": "4.5.3", + "contentHash": "3TIsJhD1EiiT0w2CcDMN/iSSwnNnsrnbzeVHSKkaEgV85txMprmuO+Yq2AdSbeVGcg28pdNDTPK87tJhX7VFHw==" + }, + "System.Threading.Tasks.Extensions": { + "type": "Transitive", + "resolved": "4.5.4", + "contentHash": "zteT+G8xuGu6mS+mzDzYXbzS7rd3K6Fjb9RiZlYlJPam2/hU7JCBZBVEcywNuR+oZ1ncTvc/cq0faRr3P01OVg==", + "dependencies": { + "System.Runtime.CompilerServices.Unsafe": "4.5.3" + } + }, + "distributedlock.core": { + "type": "Project", + "dependencies": { + "Microsoft.Bcl.AsyncInterfaces": "[8.0.0, )", + "System.ValueTuple": "[4.5.0, )" + } + }, + "Microsoft.Bcl.AsyncInterfaces": { + "type": "CentralTransitive", + "requested": "[8.0.0, )", + "resolved": "8.0.0", + "contentHash": "3WA9q9yVqJp222P3x1wYIGDAkpjAku0TMUaaQV22g6L67AI0LdOIrVS7Ht2vJfLHGSPVuqN94vIr15qn+HEkHw==", + "dependencies": { + "System.Threading.Tasks.Extensions": "4.5.4" + } + }, + "System.ValueTuple": { + "type": "CentralTransitive", + "requested": "[4.5.0, )", + "resolved": "4.5.0", + "contentHash": "okurQJO6NRE/apDIP23ajJ0hpiNmJ+f0BwOlB/cSqTLQlw5upkf+5+96+iG2Jw40G1fCVCyPz/FhIABUjMR+RQ==" + } + }, + ".NETStandard,Version=v2.0": { + "Microsoft.CodeAnalysis.PublicApiAnalyzers": { + "type": "Direct", + "requested": "[3.3.4, )", + "resolved": "3.3.4", + "contentHash": "kNLTfXtXUWDHVt5iaPkkiPuyHYlMgLI6SOFT4w88bfeI2vqSeGgHunFkdvlaCM8RDfcY0t2+jnesQtidRJJ/DA==" + }, + "Microsoft.SourceLink.GitHub": { + "type": "Direct", + "requested": "[10.0.201, )", + "resolved": "10.0.201", + "contentHash": "qxYAmO4ktzd9L+HMdnqWucxpu7bI9undPyACXOMqPyhaiMtbpbYL/n0ACyWIJlbyEJrXFwxiOaBOSasLtDvsCg==", + "dependencies": { + "Microsoft.Build.Tasks.Git": "10.0.201", + "Microsoft.SourceLink.Common": "10.0.201", + "System.IO.Hashing": "10.0.5" + } + }, + "NETStandard.Library": { + "type": "Direct", + "requested": "[2.0.3, )", + "resolved": "2.0.3", + "contentHash": "st47PosZSHrjECdjeIzZQbzivYBJFv6P2nv4cj2ypdI204DO+vZ7l5raGMiX4eXMJ53RfOIg+/s4DHVZ54Nu2A==", + "dependencies": { + "Microsoft.NETCore.Platforms": "1.1.0" + } + }, + "Nullable": { + "type": "Direct", + "requested": "[1.3.1, )", + "resolved": "1.3.1", + "contentHash": "Mk4ZVDfAORTjvckQprCSehi1XgOAAlk5ez06Va/acRYEloN9t6d6zpzJRn5MEq7+RnagyFIq9r+kbWzLGd+6QA==" + }, + "Microsoft.Build.Tasks.Git": { + "type": "Transitive", + "resolved": "10.0.201", + "contentHash": "DMYBnrFZvLnBKn14VavEuuIr31CY6YY2i2L9P8DorS/Qp6ifRR8ZPLdJCFLFfjikNq8DykbYyLd/RP6lSqHcWw==", + "dependencies": { + "System.IO.Hashing": "10.0.5" + } + }, + "Microsoft.NETCore.Platforms": { + "type": "Transitive", + "resolved": "1.1.0", + "contentHash": "kz0PEW2lhqygehI/d6XsPCQzD7ff7gUJaVGPVETX611eadGsA3A877GdSlU0LRVMCTH/+P3o2iDTak+S08V2+A==" + }, + "Microsoft.SourceLink.Common": { + "type": "Transitive", + "resolved": "10.0.201", + "contentHash": "QbBYhkjgL6rCnBfDbzsAJLlsad13TlBHqYCFDIw56OO2g6ix+9RsmY8uxiQGdWwFKbZXaXyAA6jDCzFYVGCZDw==" + }, + "System.Buffers": { + "type": "Transitive", + "resolved": "4.6.1", + "contentHash": "N8GXpmiLMtljq7gwvyS+1QvKT/W2J8sNAvx+HVg4NGmsG/H+2k/y9QI23auLJRterrzCiDH+IWAw4V/GPwsMlw==" + }, + "System.IO.Hashing": { + "type": "Transitive", + "resolved": "10.0.5", + "contentHash": "8IBJWcCT9+e4Bmevm4T7+fQEiAh133KGiz4oiVTgJckd3Q76OFdR1falgn9lpz7+C4HJvogCDJeAa2QmvbeVtg==", + "dependencies": { + "System.Buffers": "4.6.1", + "System.Memory": "4.6.3" + } + }, + "System.Memory": { + "type": "Transitive", + "resolved": "4.6.3", + "contentHash": "qdcDOgnFZY40+Q9876JUHnlHu7bosOHX8XISRoH94fwk6hgaeQGSgfZd8srWRZNt5bV9ZW2TljcegDNxsf+96A==", + "dependencies": { + "System.Buffers": "4.6.1", + "System.Numerics.Vectors": "4.6.1", + "System.Runtime.CompilerServices.Unsafe": "6.1.2" + } + }, + "System.Numerics.Vectors": { + "type": "Transitive", + "resolved": "4.6.1", + "contentHash": "sQxefTnhagrhoq2ReR0D/6K0zJcr9Hrd6kikeXsA1I8kOCboTavcUC4r7TSfpKFeE163uMuxZcyfO1mGO3EN8Q==" + }, + "System.Runtime.CompilerServices.Unsafe": { + "type": "Transitive", + "resolved": "6.1.2", + "contentHash": "2hBr6zdbIBTDE3EhK7NSVNdX58uTK6iHW/P/Axmm9sl1xoGSLqDvMtpecn226TNwHByFokYwJmt/aQQNlO5CRw==" + }, + "System.Threading.Tasks.Extensions": { + "type": "Transitive", + "resolved": "4.5.4", + "contentHash": "zteT+G8xuGu6mS+mzDzYXbzS7rd3K6Fjb9RiZlYlJPam2/hU7JCBZBVEcywNuR+oZ1ncTvc/cq0faRr3P01OVg==", + "dependencies": { + "System.Runtime.CompilerServices.Unsafe": "4.5.3" + } + }, + "distributedlock.core": { + "type": "Project", + "dependencies": { + "Microsoft.Bcl.AsyncInterfaces": "[8.0.0, )" + } + }, + "Microsoft.Bcl.AsyncInterfaces": { + "type": "CentralTransitive", + "requested": "[8.0.0, )", + "resolved": "8.0.0", + "contentHash": "3WA9q9yVqJp222P3x1wYIGDAkpjAku0TMUaaQV22g6L67AI0LdOIrVS7Ht2vJfLHGSPVuqN94vIr15qn+HEkHw==", + "dependencies": { + "System.Threading.Tasks.Extensions": "4.5.4" + } + } + }, + ".NETStandard,Version=v2.1": { + "Microsoft.CodeAnalysis.PublicApiAnalyzers": { + "type": "Direct", + "requested": "[3.3.4, )", + "resolved": "3.3.4", + "contentHash": "kNLTfXtXUWDHVt5iaPkkiPuyHYlMgLI6SOFT4w88bfeI2vqSeGgHunFkdvlaCM8RDfcY0t2+jnesQtidRJJ/DA==" + }, + "Microsoft.SourceLink.GitHub": { + "type": "Direct", + "requested": "[10.0.201, )", + "resolved": "10.0.201", + "contentHash": "qxYAmO4ktzd9L+HMdnqWucxpu7bI9undPyACXOMqPyhaiMtbpbYL/n0ACyWIJlbyEJrXFwxiOaBOSasLtDvsCg==", + "dependencies": { + "Microsoft.Build.Tasks.Git": "10.0.201", + "Microsoft.SourceLink.Common": "10.0.201", + "System.IO.Hashing": "10.0.5" + } + }, + "Microsoft.Build.Tasks.Git": { + "type": "Transitive", + "resolved": "10.0.201", + "contentHash": "DMYBnrFZvLnBKn14VavEuuIr31CY6YY2i2L9P8DorS/Qp6ifRR8ZPLdJCFLFfjikNq8DykbYyLd/RP6lSqHcWw==", + "dependencies": { + "System.IO.Hashing": "10.0.5" + } + }, + "Microsoft.SourceLink.Common": { + "type": "Transitive", + "resolved": "10.0.201", + "contentHash": "QbBYhkjgL6rCnBfDbzsAJLlsad13TlBHqYCFDIw56OO2g6ix+9RsmY8uxiQGdWwFKbZXaXyAA6jDCzFYVGCZDw==" + }, + "System.Buffers": { + "type": "Transitive", + "resolved": "4.6.1", + "contentHash": "N8GXpmiLMtljq7gwvyS+1QvKT/W2J8sNAvx+HVg4NGmsG/H+2k/y9QI23auLJRterrzCiDH+IWAw4V/GPwsMlw==" + }, + "System.IO.Hashing": { + "type": "Transitive", + "resolved": "10.0.5", + "contentHash": "8IBJWcCT9+e4Bmevm4T7+fQEiAh133KGiz4oiVTgJckd3Q76OFdR1falgn9lpz7+C4HJvogCDJeAa2QmvbeVtg==", + "dependencies": { + "System.Buffers": "4.6.1", + "System.Memory": "4.6.3" + } + }, + "System.Memory": { + "type": "Transitive", + "resolved": "4.6.3", + "contentHash": "qdcDOgnFZY40+Q9876JUHnlHu7bosOHX8XISRoH94fwk6hgaeQGSgfZd8srWRZNt5bV9ZW2TljcegDNxsf+96A==" + }, + "distributedlock.core": { + "type": "Project" + } + }, + "net8.0": { + "Microsoft.CodeAnalysis.PublicApiAnalyzers": { + "type": "Direct", + "requested": "[3.3.4, )", + "resolved": "3.3.4", + "contentHash": "kNLTfXtXUWDHVt5iaPkkiPuyHYlMgLI6SOFT4w88bfeI2vqSeGgHunFkdvlaCM8RDfcY0t2+jnesQtidRJJ/DA==" + }, + "Microsoft.NET.ILLink.Tasks": { + "type": "Direct", + "requested": "[8.0.29, )", + "resolved": "8.0.29", + "contentHash": "HSBTfrkIZijz8z3ybLRKB7E8rHk4QQufFwpHa9fc5CMIgRhRzdn4mBGmlyXZqaueiMPtuJcnjresGvSTfaW8Mg==" + }, + "Microsoft.SourceLink.GitHub": { + "type": "Direct", + "requested": "[10.0.201, )", + "resolved": "10.0.201", + "contentHash": "qxYAmO4ktzd9L+HMdnqWucxpu7bI9undPyACXOMqPyhaiMtbpbYL/n0ACyWIJlbyEJrXFwxiOaBOSasLtDvsCg==", + "dependencies": { + "Microsoft.Build.Tasks.Git": "10.0.201", + "Microsoft.SourceLink.Common": "10.0.201", + "System.IO.Hashing": "10.0.5" + } + }, + "Nullable": { + "type": "Direct", + "requested": "[1.3.1, )", + "resolved": "1.3.1", + "contentHash": "Mk4ZVDfAORTjvckQprCSehi1XgOAAlk5ez06Va/acRYEloN9t6d6zpzJRn5MEq7+RnagyFIq9r+kbWzLGd+6QA==" + }, + "Microsoft.Build.Tasks.Git": { + "type": "Transitive", + "resolved": "10.0.201", + "contentHash": "DMYBnrFZvLnBKn14VavEuuIr31CY6YY2i2L9P8DorS/Qp6ifRR8ZPLdJCFLFfjikNq8DykbYyLd/RP6lSqHcWw==", + "dependencies": { + "System.IO.Hashing": "10.0.5" + } + }, + "Microsoft.SourceLink.Common": { + "type": "Transitive", + "resolved": "10.0.201", + "contentHash": "QbBYhkjgL6rCnBfDbzsAJLlsad13TlBHqYCFDIw56OO2g6ix+9RsmY8uxiQGdWwFKbZXaXyAA6jDCzFYVGCZDw==" + }, + "System.IO.Hashing": { + "type": "Transitive", + "resolved": "10.0.5", + "contentHash": "8IBJWcCT9+e4Bmevm4T7+fQEiAh133KGiz4oiVTgJckd3Q76OFdR1falgn9lpz7+C4HJvogCDJeAa2QmvbeVtg==" + }, + "distributedlock.core": { + "type": "Project" + } + } + } +} \ No newline at end of file diff --git a/src/DistributedLock.Tests/AbstractTestCases/DistributedLockCoreTestCases.cs b/src/DistributedLock.Tests/AbstractTestCases/DistributedLockCoreTestCases.cs index 77d791a4..255699c7 100644 --- a/src/DistributedLock.Tests/AbstractTestCases/DistributedLockCoreTestCases.cs +++ b/src/DistributedLock.Tests/AbstractTestCases/DistributedLockCoreTestCases.cs @@ -397,6 +397,8 @@ public async Task TestLockAbandonment() [Test] public void TestCrossProcess() { + this.SkipIfCrossProcessIsNotSupported(); + var lockName = this._lockProvider.GetUniqueSafeName(); var command = this.RunLockTaker(this._lockProvider, this._lockProvider.GetCrossProcessLockType(), lockName); Assert.That(command.StandardOutput.ReadLineAsync().Wait(TimeSpan.FromSeconds(10)), Is.True); @@ -428,6 +430,8 @@ public void TestCrossProcessAbandonmentWithKill() private void CrossProcessAbandonmentHelper(bool asyncWait, bool kill) { + this.SkipIfCrossProcessIsNotSupported(); + var name = this._lockProvider.GetUniqueSafeName($"cpl-{asyncWait}-{kill}"); var command = this.RunLockTaker(this._lockProvider, this._lockProvider.GetCrossProcessLockType(), name); Assert.That(command.StandardOutput.ReadLineAsync().Wait(TimeSpan.FromSeconds(10)), Is.True); @@ -464,6 +468,14 @@ private void CrossProcessAbandonmentHelper(bool asyncWait, bool kill) } } + private void SkipIfCrossProcessIsNotSupported() + { + if (!this._lockProvider.Strategy.SupportsCrossProcess) + { + Assert.Pass($"{this.GetType().Name}: provider does not support cross-process synchronization"); + } + } + private Command RunLockTaker(TLockProvider engine, params string[] args) { const string Configuration = diff --git a/src/DistributedLock.Tests/DistributedLock.Tests.csproj b/src/DistributedLock.Tests/DistributedLock.Tests.csproj index be82219e..fd5700ae 100644 --- a/src/DistributedLock.Tests/DistributedLock.Tests.csproj +++ b/src/DistributedLock.Tests/DistributedLock.Tests.csproj @@ -30,6 +30,7 @@ + diff --git a/src/DistributedLock.Tests/Infrastructure/InProcess/TestingInProcessProviders.cs b/src/DistributedLock.Tests/Infrastructure/InProcess/TestingInProcessProviders.cs new file mode 100644 index 00000000..48b10393 --- /dev/null +++ b/src/DistributedLock.Tests/Infrastructure/InProcess/TestingInProcessProviders.cs @@ -0,0 +1,36 @@ +namespace Medallion.Threading.Tests.InProcess; + +[SupportsContinuousIntegration] +public sealed class TestingInProcessDistributedLockProvider : TestingLockProvider +{ + public override IDistributedLock CreateLockWithExactName(string name) => + this.Strategy.Provider.CreateLock(name); + + public override string GetSafeName(string name) => InProcessNameHelper.GetSafeName(name); +} + +[SupportsContinuousIntegration] +public sealed class TestingInProcessDistributedSemaphoreProvider : TestingSemaphoreProvider +{ + public override IDistributedSemaphore CreateSemaphoreWithExactName(string name, int maxCount) => + this.Strategy.Provider.CreateSemaphore(name, maxCount); + + public override string GetSafeName(string name) => InProcessNameHelper.GetSafeName(name); +} + +[SupportsContinuousIntegration] +public sealed class TestingInProcessDistributedReaderWriterLockProvider : TestingUpgradeableReaderWriterLockProvider +{ + public override IDistributedUpgradeableReaderWriterLock CreateUpgradeableReaderWriterLockWithExactName(string name) => + this.Strategy.Provider.CreateReaderWriterLock(name); + + public override string GetSafeName(string name) => InProcessNameHelper.GetSafeName(name); +} + +internal static class InProcessNameHelper +{ + /// + /// In-process names have no length or character restrictions, so any non-null name is already safe + /// + public static string GetSafeName(string name) => name ?? throw new ArgumentNullException(nameof(name)); +} diff --git a/src/DistributedLock.Tests/Infrastructure/InProcess/TestingInProcessSynchronizationStrategy.cs b/src/DistributedLock.Tests/Infrastructure/InProcess/TestingInProcessSynchronizationStrategy.cs new file mode 100644 index 00000000..23caf937 --- /dev/null +++ b/src/DistributedLock.Tests/Infrastructure/InProcess/TestingInProcessSynchronizationStrategy.cs @@ -0,0 +1,16 @@ +using Medallion.Threading.InProcess; + +namespace Medallion.Threading.Tests.InProcess; + +[SupportsContinuousIntegration] +public sealed class TestingInProcessSynchronizationStrategy : TestingSynchronizationStrategy +{ + /// + /// caches its underlying primitives process-wide by scope, + /// so each strategy uses a unique scope to isolate tests from one another. All synchronization primitives used by + /// a single test must come from this provider. + /// + public InProcessDistributedSynchronizationProvider Provider { get; } = new(scope: Guid.NewGuid().ToString("N")); + + public override bool SupportsCrossProcess => false; +} diff --git a/src/DistributedLock.Tests/Infrastructure/TestingSynchronizationStrategy.cs b/src/DistributedLock.Tests/Infrastructure/TestingSynchronizationStrategy.cs index 8a58ea08..819436f9 100644 --- a/src/DistributedLock.Tests/Infrastructure/TestingSynchronizationStrategy.cs +++ b/src/DistributedLock.Tests/Infrastructure/TestingSynchronizationStrategy.cs @@ -12,6 +12,12 @@ public abstract class TestingSynchronizationStrategy : IDisposable /// public virtual bool SupportsCrossProcessSingleSemaphoreTicketAbandonment => true; + /// + /// Whether or not synchronization is observed across processes. When false, test cases which + /// coordinate with the DistributedLockTaker process are skipped + /// + public virtual bool SupportsCrossProcess => true; + public virtual void PrepareForHandleAbandonment() { } public virtual void PerformAdditionalCleanupForHandleAbandonment() { } public virtual IDisposable? PrepareForHandleLost() => null; diff --git a/src/DistributedLock.Tests/Tests/CombinatorialTests.cs b/src/DistributedLock.Tests/Tests/CombinatorialTests.cs index fdff8c42..6852fef6 100644 --- a/src/DistributedLock.Tests/Tests/CombinatorialTests.cs +++ b/src/DistributedLock.Tests/Tests/CombinatorialTests.cs @@ -13,6 +13,17 @@ namespace Medallion.Threading.Tests.FileSystem [Category("CI")] public class Core_File_FileSynchronizationStrategyTest : DistributedLockCoreTestCases { } } +namespace Medallion.Threading.Tests.InProcess +{ + [Category("CI")] public class Core_InProcess_InProcessSynchronizationStrategyTest : DistributedLockCoreTestCases { } + [Category("CI")] public class Core_Semaphore1AsMutex_InProcessSemaphore_InProcessSynchronizationStrategy_InProcessSynchronizationStrategyTest : DistributedLockCoreTestCases, TestingInProcessSynchronizationStrategy> { } + [Category("CI")] public class Core_Semaphore5AsMutex_InProcessSemaphore_InProcessSynchronizationStrategy_InProcessSynchronizationStrategyTest : DistributedLockCoreTestCases, TestingInProcessSynchronizationStrategy> { } + [Category("CI")] public class ReaderWriterCore_InProcessReaderWriter_InProcessSynchronizationStrategyTest : DistributedReaderWriterLockCoreTestCases { } + [Category("CI")] public class SemaphoreCore_InProcessSemaphore_InProcessSynchronizationStrategyTest : DistributedSemaphoreCoreTestCases { } + [Category("CI")] public class UpgradeableReaderWriterCore_InProcessReaderWriter_InProcessSynchronizationStrategyTest : DistributedUpgradeableReaderWriterLockCoreTestCases { } + public class Core_ReaderWriterAsMutex_InProcessReaderWriter_InProcessSynchronizationStrategy_InProcessSynchronizationStrategyTest : DistributedLockCoreTestCases, TestingInProcessSynchronizationStrategy> { } +} + namespace Medallion.Threading.Tests.MongoDB { public class Core_Mongo_MongoDbSynchronizationStrategyTest : DistributedLockCoreTestCases { } diff --git a/src/DistributedLock.Tests/Tests/InProcess/InProcessDistributedSynchronizationProviderTest.cs b/src/DistributedLock.Tests/Tests/InProcess/InProcessDistributedSynchronizationProviderTest.cs new file mode 100644 index 00000000..4b8c860e --- /dev/null +++ b/src/DistributedLock.Tests/Tests/InProcess/InProcessDistributedSynchronizationProviderTest.cs @@ -0,0 +1,176 @@ +using Medallion.Threading.InProcess; +using NUnit.Framework; + +namespace Medallion.Threading.Tests.InProcess; + +public class InProcessDistributedSynchronizationProviderTest +{ + [Test] + public async Task BasicTest() + { + var provider = new InProcessDistributedSynchronizationProvider(); + const string LockName = TargetFramework.Current + "ProviderBasicTest"; + await using (await provider.AcquireLockAsync(LockName)) + { + await using var handle = await provider.TryAcquireLockAsync(LockName); + Assert.That(handle, Is.Null); + } + + const string ReaderWriterLockName = TargetFramework.Current + "ProviderBasicTest_ReaderWriter"; + await using (await provider.AcquireReadLockAsync(ReaderWriterLockName)) + { + await using var handle = await provider.TryAcquireWriteLockAsync(ReaderWriterLockName); + Assert.That(handle, Is.Null); + } + + await using (await provider.AcquireReadLockAsync(ReaderWriterLockName)) + { + await using var handle = await provider.TryAcquireReadLockAsync(ReaderWriterLockName); + Assert.That(handle, Is.Not.Null); + } + + await using (var upgradeHandle = await provider.AcquireUpgradeableReadLockAsync(ReaderWriterLockName)) + { + await using var handle = await provider.TryAcquireReadLockAsync(ReaderWriterLockName); + Assert.That(handle, Is.Not.Null); + var upgraded = upgradeHandle.TryUpgradeToWriteLockAsync(); + Assert.That(await upgraded, Is.False); + } + + const string SemaphoreName = TargetFramework.Current + "ProviderBasicTest_Semaphore"; + await using (await provider.AcquireSemaphoreAsync(SemaphoreName, 2)) + { + await using var handle = await provider.TryAcquireSemaphoreAsync(SemaphoreName, 2); + Assert.That(handle, Is.Not.Null); + + await using var failedHandle = await provider.TryAcquireSemaphoreAsync(SemaphoreName, 2); + Assert.That(failedHandle, Is.Null); + } + } + + [Test] + public void TestDifferentScopesAreIsolated() + { + var providerA = new InProcessDistributedSynchronizationProvider(NewScope()); + var providerB = new InProcessDistributedSynchronizationProvider(NewScope()); + const string Name = "ScopeIsolation"; + + using var handleA = providerA.AcquireLock(Name); + using var handleB = providerB.TryAcquireLock(Name); + Assert.That(handleB, Is.Not.Null); + } + + [Test] + public void TestSameScopeIsSharedAcrossInstances() + { + var scope = NewScope(); + var provider1 = new InProcessDistributedSynchronizationProvider(scope); + var provider2 = new InProcessDistributedSynchronizationProvider(scope); + Assert.That(provider2.Scope, Is.EqualTo(scope)); + const string Name = "ScopeSharing"; + + using var handle = provider1.AcquireLock(Name); + using var handle2 = provider2.TryAcquireLock(Name); + Assert.That(handle2, Is.Null); + } + + [Test] + public void TestDefaultScopeIsShared() + { + var name = "DefaultScope" + Guid.NewGuid().ToString("N"); + var provider1 = new InProcessDistributedSynchronizationProvider(); + var provider2 = new InProcessDistributedSynchronizationProvider(); + + using var handle = provider1.AcquireLock(name); + using var handle2 = provider2.TryAcquireLock(name); + Assert.That(handle2, Is.Null); + } + + [Test] + public async Task TestZeroTimeoutReadSucceedsUnderReaderChurn() + { + var provider = new InProcessDistributedSynchronizationProvider(NewScope()); + const string Name = "ReaderChurn"; + + using var stop = new CancellationTokenSource(TimeSpan.FromSeconds(1)); + var churners = Enumerable.Range(0, Math.Max(2, Environment.ProcessorCount)) + .Select(_ => Task.Run(() => + { + while (!stop.IsCancellationRequested) + { + using var handle = provider.AcquireReadLock(Name); + } + })) + .ToArray(); + + var failures = 0; + while (!stop.IsCancellationRequested) + { + using var handle = provider.TryAcquireReadLock(Name); + if (handle == null) { ++failures; } + } + await Task.WhenAll(churners); + + Assert.That(failures, Is.EqualTo(0), "readers must never block zero-timeout readers when no writer is present"); + } + + [Test] + public async Task TestWaitingWriterBlocksNewReaders() + { + var provider = new InProcessDistributedSynchronizationProvider(NewScope()); + const string Name = "WriterPriority"; + + var readHandle = provider.AcquireReadLock(Name); + var writeTask = provider.AcquireWriteLockAsync(Name).AsTask(); + await Task.Delay(100); + Assert.That(writeTask.IsCompleted, Is.False); + + using (var blockedRead = provider.TryAcquireReadLock(Name)) + { + Assert.That(blockedRead, Is.Null, "a waiting writer must take priority over new readers"); + } + + readHandle.Dispose(); + await using (await writeTask) { } + + using var readAfter = provider.TryAcquireReadLock(Name); + Assert.That(readAfter, Is.Not.Null); + } + + [Test] + public async Task TestFailedWriterDoesNotBlockReaders() + { + var provider = new InProcessDistributedSynchronizationProvider(NewScope()); + const string Name = "FailedWriter"; + + using (provider.AcquireReadLock(Name)) + { + await using var write = await provider.TryAcquireWriteLockAsync(Name, TimeSpan.FromMilliseconds(50)); + Assert.That(write, Is.Null); + + using var read = provider.TryAcquireReadLock(Name); + Assert.That(read, Is.Not.Null, "a timed-out writer must not leave readers on the slow path"); + } + } + + [Test] + public async Task TestReadFastPathRecoversAfterUpgrade() + { + var provider = new InProcessDistributedSynchronizationProvider(NewScope()); + const string Name = "UpgradeRecovery"; + + await using (var upgradeable = await provider.AcquireUpgradeableReadLockAsync(Name)) + { + await upgradeable.UpgradeToWriteLockAsync(); + using var blocked = provider.TryAcquireReadLock(Name); + Assert.That(blocked, Is.Null); + } + + using var read1 = provider.TryAcquireReadLock(Name); + using var read2 = provider.TryAcquireReadLock(Name); + Assert.That(read1, Is.Not.Null); + Assert.That(read2, Is.Not.Null); + } + + private static string NewScope() => Guid.NewGuid().ToString("N"); +} diff --git a/src/DistributedLock.Tests/Tests/InProcess/InProcessPrimitiveCacheTest.cs b/src/DistributedLock.Tests/Tests/InProcess/InProcessPrimitiveCacheTest.cs new file mode 100644 index 00000000..7fb1671e --- /dev/null +++ b/src/DistributedLock.Tests/Tests/InProcess/InProcessPrimitiveCacheTest.cs @@ -0,0 +1,76 @@ +using Medallion.Threading.InProcess; +using NUnit.Framework; + +namespace Medallion.Threading.Tests.InProcess; + +public class InProcessPrimitiveCacheTest +{ + /// + /// The provider caches its underlying primitives by name using weak references. If that cache ever hands out + /// two different primitives for the same name, mutual exclusion is silently lost. + /// + [Test] + public void TestSameNameAlwaysYieldsMutualExclusionUnderConcurrency() + { + var provider = new InProcessDistributedSynchronizationProvider(); + const string LockName = TargetFramework.Current + "PrimitiveCacheRace"; + + var threadCount = Math.Max(4, Environment.ProcessorCount); + using var barrier = new Barrier(threadCount); + var concurrentHolders = 0; + var maxConcurrentHolders = 0; + + Parallel.For(0, threadCount, new ParallelOptions { MaxDegreeOfParallelism = threadCount }, _ => + { + // maximize the chance of racing inside GetOrAdd + barrier.SignalAndWait(); + + for (var i = 0; i < 200; ++i) + { + using (provider.AcquireLock(LockName)) + { + var holders = Interlocked.Increment(ref concurrentHolders); + InterlockedMax(ref maxConcurrentHolders, holders); + Interlocked.Decrement(ref concurrentHolders); + } + + // give the cache a chance to see collectible entries + if (i % 50 == 0) { GC.Collect(); } + } + }); + + maxConcurrentHolders.ShouldEqual(1, "the same name must always map to the same underlying primitive"); + } + + /// + /// Once all references to a name's primitive are dropped it may be collected, but re-acquiring the same + /// name must still produce a working lock. + /// + [Test] + public void TestNameIsReusableAfterCollection() + { + var provider = new InProcessDistributedSynchronizationProvider(); + const string LockName = TargetFramework.Current + "PrimitiveCacheReuse"; + + provider.AcquireLock(LockName).Dispose(); + + GC.Collect(); + GC.WaitForPendingFinalizers(); + GC.Collect(); + + using var handle = provider.AcquireLock(LockName); + Assert.That(handle, Is.Not.Null); + Assert.That(provider.TryAcquireLock(LockName), Is.Null, "the recreated primitive must still be exclusive"); + } + + private static void InterlockedMax(ref int target, int value) + { + var current = Volatile.Read(ref target); + while (value > current) + { + var found = Interlocked.CompareExchange(ref target, value, current); + if (found == current) { return; } + current = found; + } + } +} diff --git a/src/DistributedLock.sln b/src/DistributedLock.sln index 39196b2d..1acbcdeb 100644 --- a/src/DistributedLock.sln +++ b/src/DistributedLock.sln @@ -49,6 +49,8 @@ Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "Solution Items", "Solution package.readme.md = package.readme.md EndProjectSection EndProject +Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "DistributedLock.InProcess", "DistributedLock.InProcess\DistributedLock.InProcess.csproj", "{95EFF32C-19C4-41AE-93EE-165973AD3076}" +EndProject Global GlobalSection(SolutionConfigurationPlatforms) = preSolution Debug|Any CPU = Debug|Any CPU @@ -115,6 +117,10 @@ Global {92074E6D-99D1-46B1-A0AE-442EA1FEA397}.Debug|Any CPU.Build.0 = Debug|Any CPU {92074E6D-99D1-46B1-A0AE-442EA1FEA397}.Release|Any CPU.ActiveCfg = Release|Any CPU {92074E6D-99D1-46B1-A0AE-442EA1FEA397}.Release|Any CPU.Build.0 = Release|Any CPU + {95EFF32C-19C4-41AE-93EE-165973AD3076}.Debug|Any CPU.ActiveCfg = Debug|Any CPU + {95EFF32C-19C4-41AE-93EE-165973AD3076}.Debug|Any CPU.Build.0 = Debug|Any CPU + {95EFF32C-19C4-41AE-93EE-165973AD3076}.Release|Any CPU.ActiveCfg = Release|Any CPU + {95EFF32C-19C4-41AE-93EE-165973AD3076}.Release|Any CPU.Build.0 = Release|Any CPU EndGlobalSection GlobalSection(SolutionProperties) = preSolution HideSolutionNode = FALSE diff --git a/src/DistributedLockCodeGen/GenerateIDistributedLockImplementations.cs b/src/DistributedLockCodeGen/GenerateIDistributedLockImplementations.cs index f6543897..4dde2dfd 100644 --- a/src/DistributedLockCodeGen/GenerateIDistributedLockImplementations.cs +++ b/src/DistributedLockCodeGen/GenerateIDistributedLockImplementations.cs @@ -16,7 +16,7 @@ public void GenerateForIDistributedLockAndSemaphore([Values("Lock", "Semaphore") { var files = CodeGenHelpers.EnumerateSolutionFiles() .Where(f => !f.Contains($"Distributed{name}.Core", StringComparison.OrdinalIgnoreCase)) - .Where(f => f.EndsWith($"Distributed{name}.cs", StringComparison.OrdinalIgnoreCase) && Path.GetFileName(f)[0] != 'I'); + .Where(f => f.EndsWith($"Distributed{name}.cs", StringComparison.OrdinalIgnoreCase) && !Regex.IsMatch(Path.GetFileName(f), "^I[A-Z]")); var errors = new List(); foreach (var file in files)