Мне нужно по причине X написать инкапсуляцию семафора, которая позволила бы мне отменить все ожидающие процессы на моем SemaphoreSlim
.(SemaphoreSlim Cancellation Encapsulation)
Есть мой класс:
public class CancellableSemaphoreSlim
{
readonly Queue<CancellationTokenSource> tokens = new Queue<CancellationTokenSource>();
readonly SemaphoreSlim ss;
/// <summary>
/// Initializes a new instance of the <see cref="T:Eyes.Mobile.Core.Helpers.CancellableSemaphoreSlim"/> class.
/// </summary>
/// <param name="initialCount">Initial count.</param>
public CancellableSemaphoreSlim(int initialCount) { ss = new SemaphoreSlim(initialCount); }
/// <summary>Asynchronously waits to enter the <see cref="T:System.Threading.SemaphoreSlim" />, while observing a <see cref="T:System.Threading.CancellationToken" />. </summary>
/// <returns>A task that will complete when the semaphore has been entered. </returns>
/// <exception cref="T:System.ObjectDisposedException">The current instance has already been disposed.</exception>
/// <exception cref="T:System.OperationCanceledException" />
public Task WaitAsync()
{
CancellationTokenSource cancellationTokenSource = new CancellationTokenSource();
tokens.Enqueue(cancellationTokenSource);
return ss.WaitAsync(cancellationTokenSource.Token);
}
/// <summary>Asynchronously waits to enter the <see cref="T:System.Threading.SemaphoreSlim" />, while observing a <see cref="T:System.Threading.CancellationTokenSource" />. </summary>
/// <returns>A task that will complete when the semaphore has been entered. </returns>
/// <param name="cancellationTokenSource">The <see cref="T:System.Threading.CancellationToken" /> token to observe.</param>
/// <exception cref="T:System.ObjectDisposedException">The current instance has already been disposed.</exception>
/// <exception cref="T:System.OperationCanceledException">
/// <paramref name="cancellationTokenSource" /> was canceled.
/// </exception>
public Task WaitAsync(CancellationToken cancellationToken)
{
CancellationTokenSource cancellationTokenSource = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
tokens.Enqueue(cancellationTokenSource);
return ss.WaitAsync(cancellationTokenSource.Token);
}
/// <summary>
/// Release this instance.
/// </summary>
/// <returns>The released semaphore return.</returns>
public int Release() => ss.Release();
/// <summary>
/// Cancel all processus currently in WaitAsync() state.
/// </summary>
public void CancelAll()
{
while (tokens.Count > 0)
{
CancellationTokenSource token = tokens.Dequeue();
if (!token.IsCancellationRequested)
token.Cancel();
}
}
}
Вы можете использовать его как базовый SemaphoreSlim
, я написал простой пример:
class Program
{
static void Main(string[] args)
{
AsyncContext.Run(() => MainAsync(args));
}
static async void MainAsync(string[] args)
{
for (int i = 0; i < 5; i++)
{
try
{
CancellationTokenSource cancellationTokenSource = new CancellationTokenSource(10000);
await Task.WhenAll(
MakeAnAction(i, cancellationTokenSource),
MakeAnAction(i, cancellationTokenSource),
MakeAnAction(i, cancellationTokenSource),
MakeAnAction(i, cancellationTokenSource),
MakeAnAction(i, cancellationTokenSource)
);
}
catch (OperationCanceledException) { }
}
await Task.Delay(5000);
cancellableSemaphoreSlim.CancelAll();
await Task.Delay(5000);
}
readonly static CancellableSemaphoreSlim cancellableSemaphoreSlim = new CancellableSemaphoreSlim(1);
readonly static Random rnd = new Random();
internal static async Task MakeAnAction(int id, CancellationTokenSource cancellationTokenSource)
{
try
{
await cancellableSemaphoreSlim.WaitAsync(cancellationTokenSource.Token);
int actionTime = rnd.Next(2, 10) * 1000;
Output($"{id} : Start ({actionTime})");
await Task.Delay(actionTime, cancellationTokenSource.Token);
Output($"{id} : OK ({actionTime})");
}
catch (OperationCanceledException)
{
Output($"{id} : Cancelled");
}
finally
{
cancellableSemaphoreSlim.Release();
}
}
private static void Output(string str)
{
Debug.WriteLine(str);
Console.WriteLine(str);
}
}
Однако мне было интересно, может ли использование Queue<CancellationTokenSource>
создать какую-либо асинхронную проблему?Потому что, если у нас есть метод (наподобие makeAnAction), который может вызываться различными потоками / задачами, если CancelAll () вызывается перед новым вызовом задачи / потока makeAnAction, это означает, что этот будет добавленв очередь, которая фактически снимает все свои предметы с очереди ..
Я так думал о попытке создать уникальную связь между всеми моими токенами отмены, используя CancellationTokenSource.CreateLinkedTokenSource(cancellationToken)
.Однако, даже если это логика varargs (params
), создаст ли она ту же проблему?
Я просто пытаюсь достичь ее так, чтобы она не провалилась, но, полагаю, я простов данный момент у меня плохой подход, поэтому я просто хотел бы знать, может ли кто-нибудь дать мне точку зрения об этой инкапсуляции и ее логике?
Не стесняйтесь давать мне какие-либо советы, если вы считаете, что что-то не таклогика:)
Макс
Редактировать 1
Затем я отредактировал код, чтобы следить за обсуждением с @NthDeveloper.Я пытался добавить систему lock
public class CancellableSemaphoreSlim
{
object _syncObj = new object();
readonly Queue<CancellationTokenSource> tokens = new Queue<CancellationTokenSource>();
readonly SemaphoreSlim ss;
/// <summary>
/// Initializes a new instance of the <see cref="T:Eyes.Mobile.Core.Helpers.CancellableSemaphoreSlim"/> class.
/// </summary>
/// <param name="initialCount">Initial count.</param>
public CancellableSemaphoreSlim(int initialCount) { ss = new SemaphoreSlim(initialCount); }
/// <summary>Asynchronously waits to enter the <see cref="T:System.Threading.SemaphoreSlim" />, while observing a <see cref="T:System.Threading.CancellationToken" />. </summary>
/// <returns>A task that will complete when the semaphore has been entered. </returns>
/// <exception cref="T:System.ObjectDisposedException">The current instance has already been disposed.</exception>
/// <exception cref="T:System.OperationCanceledException" />
public Task WaitAsync()
{
CancellationTokenSource cancellationTokenSource = new CancellationTokenSource();
lock (_syncObj)
{
tokens.Enqueue(cancellationTokenSource);
}
return ss.WaitAsync(cancellationTokenSource.Token);
}
/// <summary>Asynchronously waits to enter the <see cref="T:System.Threading.SemaphoreSlim" />, while observing a <see cref="T:System.Threading.CancellationTokenSource" />. </summary>
/// <returns>A task that will complete when the semaphore has been entered. </returns>
/// <param name="cancellationTokenSource">The <see cref="T:System.Threading.CancellationToken" /> token to observe.</param>
/// <exception cref="T:System.ObjectDisposedException">The current instance has already been disposed.</exception>
/// <exception cref="T:System.OperationCanceledException">
/// <paramref name="cancellationTokenSource" /> was canceled.
/// </exception>
public Task WaitAsync(CancellationToken cancellationToken)
{
CancellationTokenSource cancellationTokenSource = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
lock (_syncObj)
{
tokens.Enqueue(cancellationTokenSource);
}
return ss.WaitAsync(cancellationTokenSource.Token);
}
/// <summary>
/// Release this instance.
/// </summary>
/// <returns>The released semaphore return.</returns>
public int Release() => ss.Release();
/// <summary>
/// Cancel all processus currently in WaitAsync() state.
/// </summary>
public void CancelAll()
{
lock (_syncObj)
{
while (tokens.Count > 0)
{
CancellationTokenSource token = tokens.Dequeue();
if (!token.IsCancellationRequested)
token.Cancel();
}
}
}
}