From f2f9f7fbd8e4ab9a49f0e1631382bea3b1352bb8 Mon Sep 17 00:00:00 2001 From: "J. Ritchie Carroll" Date: Fri, 22 Aug 2025 20:54:57 -0500 Subject: [PATCH] Added KeyedSignal that represents a wait-by-key async signaling operationy --- src/Gemstone.Threading/KeyedSignal.cs | 406 ++++++++++++++++++++++++++ 1 file changed, 406 insertions(+) create mode 100644 src/Gemstone.Threading/KeyedSignal.cs diff --git a/src/Gemstone.Threading/KeyedSignal.cs b/src/Gemstone.Threading/KeyedSignal.cs new file mode 100644 index 000000000..2aced087b --- /dev/null +++ b/src/Gemstone.Threading/KeyedSignal.cs @@ -0,0 +1,406 @@ +//****************************************************************************************************** +// KeyedSignal.cs - Gbtc +// +// Copyright © 2025, Grid Protection Alliance. All Rights Reserved. +// +// Licensed to the Grid Protection Alliance (GPA) under one or more contributor license agreements. See +// the NOTICE file distributed with this work for additional information regarding copyright ownership. +// The GPA licenses this file to you under the MIT License (MIT), the "License"; you may not use this +// file except in compliance with the License. You may obtain a copy of the License at: +// +// http://opensource.org/licenses/MIT +// +// Unless agreed to in writing, the subject software distributed under the License is distributed on an +// "AS-IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. Refer to the +// License for the specific language governing permissions and limitations. +// +// Code Modification History: +// ---------------------------------------------------------------------------------------------------- +// 08/22/2025 - J. Ritchie Carroll +// Generated original version of source code with assistance from ChatGPT. +// +//****************************************************************************************************** + +using System; +using System.Collections.Concurrent; +using System.Collections.Generic; +using System.Threading; +using System.Threading.Tasks; + +namespace Gemstone.Threading; + +/// +/// Represents a wait-by-key async signaling operation that can work with multiple concurrent waiters +/// per key. Multiple callers await a task associated with a key; another single thread then signals +/// completion with a result or failure with an exception for that key. +/// +/// +/// Pruning happens periodically to avoid orphaned waiters when all callers timeout from no signal. +/// +/// Key type, e.g., a . +/// Result type delivered to all waiters when signaled. +public sealed class KeyedSignal : IDisposable, IAsyncDisposable where TKey : notnull +{ + #region [ Members ] + + // Nested Types + private sealed class Waiters + { + public readonly Lock Gate = new(); + public readonly List> List = []; + + // Atomically captures the current completion sources and clears the list + public TaskCompletionSource[] Drain() + { + lock (Gate) + { + TaskCompletionSource[] sources = List.ToArray(); + List.Clear(); + return sources; + } + } + } + + // Fields + private readonly ConcurrentDictionary m_waiters; + + // Background prune used to handle possible orphaned waiters (off the hot path) + private readonly CancellationTokenSource m_pruneCompletionSource; + private readonly Task m_pruneTask; + private readonly PeriodicTimer? m_pruneTimer; + + private bool m_disposed; + + #endregion + + #region [ Constructors ] + + /// + /// Creates a new instance of the class. + /// + /// + /// Internal for pruning orphaned waiters.
+ /// Use null for default pruning interval of 1 minute.
+ /// Use for no pruning. + /// + /// Key comparer; use null for default. + public KeyedSignal(TimeSpan? pruneInterval = null, IEqualityComparer? keyComparer = null) + { + TimeSpan interval = pruneInterval ?? TimeSpan.FromMinutes(1); + + m_waiters = new ConcurrentDictionary(keyComparer ?? EqualityComparer.Default); + m_pruneCompletionSource = new CancellationTokenSource(); + + if (interval == Timeout.InfiniteTimeSpan || interval <= TimeSpan.Zero) + { + m_pruneTask = Task.CompletedTask; + } + else + { + m_pruneTimer = new PeriodicTimer(interval); + m_pruneTask = Task.Run(() => PruneLoopAsync(m_pruneCompletionSource.Token)); + } + } + + #endregion + + #region [ Properties ] + + /// + /// Gets the number of pending keys that have waiters. + /// + public int PendingKeysCount => m_waiters.Count; + + #endregion + + #region [ Methods ] + + /// + /// Releases all the resources used by the object. + /// + public void Dispose() + { + if (m_disposed) + return; + + try + { + m_pruneCompletionSource.Cancel(); + + try + { + m_pruneTask.Wait(); + } + catch (Exception ex) + { + LibraryEvents.OnSuppressedException(this, ex); + } + + m_pruneTimer?.Dispose(); + m_pruneCompletionSource.Dispose(); + } + finally + { + m_disposed = true; // Prevent duplicate dispose. + } + } + + /// + public async ValueTask DisposeAsync() + { + if (m_disposed) + return; + + try + { + await m_pruneCompletionSource.CancelAsync(); + + try + { + await m_pruneTask.ConfigureAwait(false); + } + catch (Exception ex) + { + LibraryEvents.OnSuppressedException(this, ex); + } + + m_pruneTimer?.Dispose(); + m_pruneCompletionSource.Dispose(); + } + finally + { + m_disposed = true; // Prevent duplicate dispose. + } + } + + /// + /// Waits synchronously for the specified to be signaled or failed. + /// Multiple concurrent waiters on the same key complete together. + /// + /// Key to wait on. + /// Optional token to cancel or timeout the wait. + /// + /// A tuple where: + /// + /// + /// result is the signaled result when successful; otherwise default. + /// + /// + /// ex is null on success; otherwise the exception that caused the failure, + /// including when canceled. + /// + /// + /// + /// Thrown if the instance has been disposed. + /// + /// + /// A failure triggered by returns (default, ex) where ex is the + /// exception provided to .Cancellation is treated like a failure where returned + /// ex is an . + /// + /// + /// For asynchronous waiting, use . + /// + /// + public (TResult result, Exception? ex) Wait(TKey key, CancellationToken cancellationToken = default) + { + try + { + TResult result = WaitAsync(key, cancellationToken).ConfigureAwait(false).GetAwaiter().GetResult(); + return (result, null); + } + catch (Exception ex) + { + return (default!, ex); + } + } + + /// + /// Asynchronously waits for to be signaled (or failed). + /// Multiple concurrent waiters on the same key complete together. + /// + /// Key to wait on. + /// Optional token to cancel or timeout the wait. + /// A task that completes when signaled, failed, or canceled. + /// Thrown if the instance has been disposed. + /// + /// We never remove the per-key bucket on cancellation—only individual waiters. + /// Buckets are removed on / or by periodic pruning. + /// A mapping-stability check ensures we never add to an orphaned bucket. + /// + public Task WaitAsync(TKey key, CancellationToken cancellationToken = default) + { + ObjectDisposedException.ThrowIf(m_disposed, this); + + if (cancellationToken.IsCancellationRequested) + return Task.FromCanceled(cancellationToken); + + TaskCompletionSource source = new(TaskCreationOptions.RunContinuationsAsynchronously); + Waiters bucket; + + // Handle possible race condition where mapping changes between GetOrAdd and lock + while (true) + { + bucket = m_waiters.GetOrAdd(key, static _ => new Waiters()); + + // Check mapping-stability under the bucket lock + lock (bucket.Gate) + { + // Check if mapping changed, e.g., signaled/failed and removed, in which case continue and retry + if (!m_waiters.TryGetValue(key, out Waiters? current) || !ReferenceEquals(current, bucket)) + continue; + + bucket.List.Add(source); + break; + } + } + + // Cancellation removes only this waiter, not the bucket + CancellationTokenRegistration tokenRegistration = default; + + if (cancellationToken.CanBeCanceled) + { + // Track disposable registration to cancel this waiter + tokenRegistration = cancellationToken.Register(static state => + { + (Waiters bucket, TaskCompletionSource source, CancellationToken cancellationToken) = + ((Waiters, TaskCompletionSource, CancellationToken))state!; + + bool removed; + + lock (bucket.Gate) + removed = bucket.List.Remove(source); + + if (removed) + source.TrySetCanceled(cancellationToken); + }, + (bucket, source, cancellationToken)); + } + + // Ensure registration is disposed on all completion paths + _ = source.Task.ContinueWith(static (_, state) => + { + ((CancellationTokenRegistration)state!).Dispose(); + }, + tokenRegistration, CancellationToken.None, TaskContinuationOptions.ExecuteSynchronously, TaskScheduler.Default); + + return source.Task; + } + + /// + /// Tries to signal all current waiters for with . + /// New waiters added after this call for same key will await a subsequent signal/fail. + /// + /// Key to signal. + /// Result to signal to all waiters. + /// + /// true if the signal was successful; otherwise, + /// false if there were no waiters for the key. + /// + /// Thrown if the instance has been disposed. + public bool TrySignal(TKey key, TResult result) + { + return Signal(key, result) > 0; + } + + /// + /// Signals all current waiters for with . + /// New waiters added after this call for same key will await a subsequent signal/fail. + /// + /// Key to signal. + /// Result to signal to all waiters. + /// The number of waiters released. + /// Thrown if the instance has been disposed. + public int Signal(TKey key, TResult result) + { + ObjectDisposedException.ThrowIf(m_disposed, this); + + if (!m_waiters.TryRemove(key, out Waiters? bucket)) + return 0; + + TaskCompletionSource[] sources = bucket.Drain(); + + foreach (TaskCompletionSource source in sources) + source.TrySetResult(result); + + return sources.Length; + } + + /// + /// Tries to fail all current waiters for with . + /// New waiters added after this call for same key will await a subsequent signal/fail. + /// + /// Key to fail. + /// Exception to signal to all waiters. + /// + /// true if the fail was successful; otherwise, + /// false if there were no waiters for the key. + /// + /// Thrown if the instance has been disposed. + public bool TryFail(TKey key, Exception ex) + { + return Fail(key, ex) > 0; + } + + /// + /// Fails all current waiters for with . + /// New waiters added after this call for same key will await a subsequent signal/fail. + /// + /// Key to fail. + /// Exception to signal to all waiters. + /// The number of waiters released. + /// Thrown if the instance has been disposed. + public int Fail(TKey key, Exception ex) + { + ObjectDisposedException.ThrowIf(m_disposed, this); + + if (!m_waiters.TryRemove(key, out Waiters? bucket)) + return 0; + + TaskCompletionSource[] sources = bucket.Drain(); + + foreach (TaskCompletionSource source in sources) + source.TrySetException(ex); + + return sources.Length; + } + + // Background prune loop scans and prunes empty buckets at configured interval. + private async Task PruneLoopAsync(CancellationToken token) + { + try + { + while (await m_pruneTimer!.WaitForNextTickAsync(token).ConfigureAwait(false)) + { + foreach ((TKey key, Waiters value) in m_waiters) + TryPruneEmpty(key, value); + } + } + catch (OperationCanceledException) + { + // Normal shutdown + } + } + + // Attempts to remove the mapping, key to bucket, if bucket is still empty + // and the dictionary still maps to exact instance. This avoids race condition + // where a newer bucket of waiters could have been created for the same key. + private void TryPruneEmpty(TKey key, Waiters bucket) + { + lock (bucket.Gate) + { + if (bucket.List.Count != 0) + return; + + // Ensure mapping still points to this bucket + if (!m_waiters.TryGetValue(key, out Waiters? current) || !ReferenceEquals(current, bucket)) + return; + + // Value-sensitive remove (compare-and-remove), checks key AND value + ((ICollection>)m_waiters) + .Remove(new KeyValuePair(key, bucket)); + } + } + + #endregion +}