Creare primitive di coordinamento asincrone

Primitive di coordinamento sincrono come ManualResetEventSlim, CountdownEvente Barrier bloccano il thread chiamante durante l'attesa. Nel codice asincrono, il blocco di un thread comporta lo spreco di una risorsa che potrebbe eseguire altre operazioni. Usare TaskCompletionSource per creare equivalenti asincroni che consentono ai chiamanti await invece di bloccare.

Un TaskCompletionSource genera un oggetto Task completato manualmente chiamando SetResult(), SetExceptiono SetCanceled. Codice che attende un'attività si sospende senza bloccare un thread e riprende quando si completa la sorgente. Questo modello costituisce il blocco predefinito per ogni primitiva in questo articolo.

Evento di reset manuale asincrono

Un evento di reimpostazione manuale viene avviato in uno stato non segnalato. I chiamanti attendono l'evento e tutti i camerieri riprendono quando un'altra parte segnala (imposta) l'evento. L'evento rimane segnalato fino a quando non viene reimpostato in modo esplicito. L'equivalente sincrono è ManualResetEventSlim. Il runtime di .NET fornisce TaskCompletionSource<TResult> direttamente per la segnalazione di trasmissione one-shot. Creare una nuova istanza a ogni ciclo anziché creare un wrapper di reimpostazione attorno a essa.

TaskCompletionSource è un evento di reimpostazione manuale una tantum: è Task incompleto fino a quando non si chiama un metodo Set*, e quindi tutti gli attendenti riprendono. Aggiungere un Reset metodo che sostituisce con un nuovo TaskCompletionSource oggetto, ottenendo così un evento di reset manuale asincrono riutilizzabile.

// Educational only — use TaskCompletionSource<T> directly instead of this sample implementation; create a new instance each cycle.
public class AsyncManualResetEvent
{
    private volatile TaskCompletionSource _tcs = new(TaskCreationOptions.RunContinuationsAsynchronously);

    public Task WaitAsync() => _tcs.Task;

    public void Set() => _tcs.TrySetResult();

    public void Reset()
    {
        while (true)
        {
            TaskCompletionSource tcs = _tcs;
            if (!tcs.Task.IsCompleted ||
                Interlocked.CompareExchange(
                    ref _tcs,
                    new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously),
                    tcs) == tcs)
            {
                return;
            }
        }
    }
}

' Educational only — use TaskCompletionSource(Of T) directly instead of this sample implementation; create a new instance each cycle.
Public Class AsyncManualResetEvent
    Private _tcs As TaskCompletionSource = New TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously)

    Public Function WaitAsync() As Task
        Return _tcs.Task
    End Function

    Public Sub [Set]()
        _tcs.TrySetResult()
    End Sub

    Public Sub Reset()
        Do
            Dim tcs As TaskCompletionSource = _tcs
            If Not tcs.Task.IsCompleted OrElse
               Interlocked.CompareExchange(
                   _tcs,
                   New TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously),
                   tcs) Is tcs Then
                Return
            End If
        Loop
    End Sub
End Class

Dettagli principali sull'implementazione:

• Il costruttore passa TaskCreationOptions.RunContinuationsAsynchronously per impedire a Set di eseguire le continuazioni di waiter in modo sincrono sul thread chiamante. Senza questo flag, Set potrebbe bloccarsi per un periodo di tempo imprevedibile.
• Reset utilizza CompareExchange per eseguire lo scambio in un nuovo TaskCompletionSource solo quando quello corrente è già stato completato. Questo scambio atomico impedisce l'orfanizzazione di un'attività già ricevuta da un cameriere.

Nell'esempio seguente viene illustrato il modo in cui due attività si coordinano tramite l'evento :

public static class AsyncManualResetEventDemo
{
    public static async Task RunAsync()
    {
        var gate = new AsyncManualResetEvent();

        Task waiter = Task.Run(async () =>
        {
            Console.WriteLine("Waiter: waiting for signal...");
            await gate.WaitAsync();
            Console.WriteLine("Waiter: signal received!");
        });

        await Task.Delay(100);
        Console.WriteLine("Signaler: setting the event.");
        gate.Set();

        await waiter;
    }
}

Public Module AsyncManualResetEventDemo
    Public Async Function RunAsync() As Task
        Dim gate As New AsyncManualResetEvent()

        Dim waiter As Task = Task.Run(Async Function()
            Console.WriteLine("Waiter: waiting for signal...")
            Await gate.WaitAsync()
            Console.WriteLine("Waiter: signal received!")
        End Function)

        Await Task.Delay(100)
        Console.WriteLine("Signaler: setting the event.")
        gate.Set()

        Await waiter
    End Function
End Module

Evento di reimpostazione automatica asincrona

Un evento di reimpostazione automatica è simile a un evento di reimpostazione manuale, ma torna automaticamente allo stato non segnalato dopo il rilascio di un solo cameriere. Se più chiamanti sono in attesa quando viene segnalato l'evento, solo un chiamante riprende. L'equivalente sincrono è AutoResetEvent. Il runtime di .NET include SemaphoreSlim per la segnalazione asincrona a un solo waiter. Inizializzalo in 0 con un conteggio massimo di 1, chiama WaitAsync per attendere e Release per segnalare.

Poiché ogni segnale rilascia un solo cameriere, è necessaria una raccolta di TaskCompletionSource istanze, una per ogni waiter, in modo da poterle completare singolarmente:

// Educational only — use SemaphoreSlim(0, 1) instead of this sample implementation: call WaitAsync() to wait and Release() to signal.
public class AsyncAutoResetEvent
{
    private readonly Queue<TaskCompletionSource> _waiters = new();
    private bool _signaled;

    public Task WaitAsync()
    {
        lock (_waiters)
        {
            if (_signaled)
            {
                _signaled = false;
                return Task.CompletedTask;
            }
            else
            {
                var tcs = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
                _waiters.Enqueue(tcs);
                return tcs.Task;
            }
        }
    }

    public void Set()
    {
        TaskCompletionSource? toRelease = null;

        lock (_waiters)
        {
            if (_waiters.Count > 0)
            {
                toRelease = _waiters.Dequeue();
            }
            else if (!_signaled)
            {
                _signaled = true;
            }
        }

        toRelease?.TrySetResult();
    }
}

' Educational only — use SemaphoreSlim(0, 1) instead of this sample implementation: call WaitAsync() to wait and Release() to signal.
Public Class AsyncAutoResetEvent
    Private ReadOnly _waiters As New Queue(Of TaskCompletionSource)()
    Private _signaled As Boolean

    Public Function WaitAsync() As Task
        SyncLock _waiters
            If _signaled Then
                _signaled = False
                Return Task.CompletedTask
            Else
                Dim tcs As New TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously)
                _waiters.Enqueue(tcs)
                Return tcs.Task
            End If
        End SyncLock
    End Function

    Public Sub [Set]()
        Dim toRelease As TaskCompletionSource = Nothing

        SyncLock _waiters
            If _waiters.Count > 0 Then
                toRelease = _waiters.Dequeue()
            ElseIf Not _signaled Then
                _signaled = True
            End If
        End SyncLock

        toRelease?.TrySetResult()
    End Sub
End Class

Dettagli principali sull'implementazione:

• Il Set metodo completa il TaskCompletionSource metodo (TCS) all'esterno del blocco. Il completamento di un TCS all'interno del blocco esegue continuazioni sincrone mentre il blocco viene mantenuto, causando deadlock o reentrancy imprevisti.
• Quando Set viene chiamato e non viene accodato alcun cameriere, il segnale viene archiviato in modo che la chiamata successiva WaitAsync venga completata immediatamente.

L'esempio seguente mostra un produttore che segnala a un consumatore tramite l'evento.

public static class AsyncAutoResetEventDemo
{
    public static async Task RunAsync()
    {
        var autoEvent = new AsyncAutoResetEvent();

        Task consumer = Task.Run(async () =>
        {
            for (int i = 0; i < 3; i++)
            {
                await autoEvent.WaitAsync();
                Console.WriteLine($"Consumer: received signal {i + 1}");
            }
        });

        for (int i = 0; i < 3; i++)
        {
            await Task.Delay(50);
            Console.WriteLine($"Producer: sending signal {i + 1}");
            autoEvent.Set();
        }

        await consumer;
    }
}

Public Module AsyncAutoResetEventDemo
    Public Async Function RunAsync() As Task
        Dim autoEvent As New AsyncAutoResetEvent()

        Dim consumer As Task = Task.Run(Async Function()
            For i As Integer = 0 To 2
                Await autoEvent.WaitAsync()
                Console.WriteLine($"Consumer: received signal {i + 1}")
            Next
        End Function)

        For i As Integer = 0 To 2
            Await Task.Delay(50)
            Console.WriteLine($"Producer: sending signal {i + 1}")
            autoEvent.Set()
        Next

        Await consumer
    End Function
End Module

Evento conto alla rovescia asincrono

Un evento conto alla rovescia attende un numero specificato di segnali prima che consenta ai camerieri di procedere. Questo modello è utile per gli scenari di fork/join in cui si avviano le operazioni N e si vuole attendere tutti i completamenti N. L'equivalente sincrono è CountdownEvent. Il runtime di .NET fornisce il coordinamento fork/join per un set fisso di attività. Utilizzalo invece.

Compila la versione asincrona componendo la AsyncManualResetEvent della sezione precedente con un contatore atomico:

// Educational only — use Task.WhenAll() instead of this sample implementation to coordinate a fixed set of tasks.
public class AsyncCountdownEvent
{
    private readonly AsyncManualResetEvent _event = new();
    private int _count;

    public AsyncCountdownEvent(int initialCount)
    {
        ArgumentOutOfRangeException.ThrowIfNegativeOrZero(initialCount, nameof(initialCount));
        _count = initialCount;
    }

    public Task WaitAsync() => _event.WaitAsync();

    public void Signal()
    {
        if (_count <= 0)
            throw new InvalidOperationException("The event is already signaled.");

        int newCount = Interlocked.Decrement(ref _count);

        if (newCount == 0)
            _event.Set();
        else if (newCount < 0)
            throw new InvalidOperationException("Too many signals.");
    }

    public Task SignalAndWait()
    {
        Signal();
        return WaitAsync();
    }
}

' Educational only — use Task.WhenAll() instead of this sample implementation to coordinate a fixed set of tasks.
Public Class AsyncCountdownEvent
    Private ReadOnly _event As New AsyncManualResetEvent()
    Private _count As Integer

    Public Sub New(initialCount As Integer)
        If initialCount <= 0 Then Throw New ArgumentOutOfRangeException(NameOf(initialCount))
        _count = initialCount
    End Sub

    Public Function WaitAsync() As Task
        Return _event.WaitAsync()
    End Function

    Public Sub Signal()
        If _count <= 0 Then
            Throw New InvalidOperationException("The event is already signaled.")
        End If

        Dim newCount As Integer = Interlocked.Decrement(_count)

        If newCount = 0 Then
            _event.Set()
        ElseIf newCount < 0 Then
            Throw New InvalidOperationException("Too many signals.")
        End If
    End Sub

    Public Function SignalAndWait() As Task
        Signal()
        Return WaitAsync()
    End Function
End Class

Il Signal metodo decrementa il conteggio in modo atomico con Decrement. Quando il conteggio raggiunge zero, imposta l'evento interno e tutti i camerieri riprendono.

L'esempio seguente usa un evento conto alla rovescia per attendere tre operazioni simultanee:

public static class AsyncCountdownEventDemo
{
    public static async Task RunAsync()
    {
        var countdown = new AsyncCountdownEvent(3);

        for (int i = 1; i <= 3; i++)
        {
            int id = i;
            _ = Task.Run(async () =>
            {
                await Task.Delay(id * 30);
                Console.WriteLine($"Worker {id}: done.");
                countdown.Signal();
            });
        }

        await countdown.WaitAsync();
        Console.WriteLine("All workers finished.");
    }
}

Public Module AsyncCountdownEventDemo
    Public Async Function RunAsync() As Task
        Dim countdown As New AsyncCountdownEvent(3)

        For i As Integer = 1 To 3
            Dim id As Integer = i
            Dim backgroundTask As Task = Task.Run(Async Function()
                Await Task.Delay(id * 30)
                Console.WriteLine($"Worker {id}: done.")
                countdown.Signal()
            End Function)
        Next

        Await countdown.WaitAsync()
        Console.WriteLine("All workers finished.")
    End Function
End Module

Barriera asincrona

Una barriera coordina un set fisso di partecipanti tra più round. Ogni partecipante segnala quando termina il suo lavoro per il round corrente e quindi attende il completamento di tutti gli altri partecipanti. Quando l'ultimo partecipante segnala, tutti i partecipanti riprendono e la barriera viene reimpostata per il turno successivo. L'equivalente sincrono è Barrier. Il runtime di .NET fornisce WhenAll per la sincronizzazione asincrona a più round. Combinalo con un ciclo, una chiamata WhenAll per ogni round.

// Educational only — use Task.WhenAll() in a loop instead of this sample implementation, one call per round.
public class AsyncBarrier
{
    private readonly int _participantCount;
    private int _remainingParticipants;
    private TaskCompletionSource _tcs = new(TaskCreationOptions.RunContinuationsAsynchronously);

    public AsyncBarrier(int participantCount)
    {
        ArgumentOutOfRangeException.ThrowIfNegativeOrZero(participantCount, nameof(participantCount));
        _remainingParticipants = _participantCount = participantCount;
    }

    public Task SignalAndWait()
    {
        TaskCompletionSource tcs = _tcs;

        if (Interlocked.Decrement(ref _remainingParticipants) == 0)
        {
            _remainingParticipants = _participantCount;
            _tcs = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
            tcs.SetResult();
        }

        return tcs.Task;
    }
}

' Educational only — use Task.WhenAll() in a loop instead of this sample implementation, one call per round.
Public Class AsyncBarrier
    Private ReadOnly _participantCount As Integer
    Private _remainingParticipants As Integer
    Private _tcs As TaskCompletionSource = New TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously)

    Public Sub New(participantCount As Integer)
        If participantCount <= 0 Then Throw New ArgumentOutOfRangeException(NameOf(participantCount))
        _participantCount = participantCount
        _remainingParticipants = participantCount
    End Sub

    Public Function SignalAndWait() As Task
        Dim tcs As TaskCompletionSource = _tcs

        If Interlocked.Decrement(_remainingParticipants) = 0 Then
            _remainingParticipants = _participantCount
            _tcs = New TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously)
            tcs.SetResult()
        End If

        Return tcs.Task
    End Function
End Class

Dettagli principali sull'implementazione:

• Prima di completare il TaskCompletionSource condiviso, il metodo reimposta il conteggio e sostituisce con un nuovo TaskCompletionSource per il round successivo. Questo ordinamento garantisce che quando i camerieri riprendono, la barriera è già pronta per il turno successivo.
• Tutti i partecipanti condividono lo stesso Task. Poiché l'esempio crea TaskCompletionSource con RunContinuationsAsynchronously, le continuazioni riprendono in modo asincrono anziché essere eseguite inline sul thread che completa la barriera.

L'esempio seguente sottopone tre partecipanti a due turni di superamento di una barriera:

public static class AsyncBarrierDemo
{
    public static async Task RunAsync()
    {
        var barrier = new AsyncBarrier(3);
        int rounds = 2;

        Task[] participants = Enumerable.Range(1, 3).Select(id => Task.Run(async () =>
        {
            for (int round = 1; round <= rounds; round++)
            {
                Console.WriteLine($"Participant {id}: working on round {round}");
                await Task.Delay(id * 20);
                Console.WriteLine($"Participant {id}: finished round {round}, waiting at barrier");
                await barrier.SignalAndWait();
            }
        })).ToArray();

        await Task.WhenAll(participants);
        Console.WriteLine("All rounds complete.");
    }
}

Public Module AsyncBarrierDemo
    Public Async Function RunAsync() As Task
        Dim barrier As New AsyncBarrier(3)
        Dim rounds As Integer = 2

        Dim participants As Task() = Enumerable.Range(1, 3).Select(
            Function(id) Task.Run(Async Function()
                For round As Integer = 1 To rounds
                    Console.WriteLine($"Participant {id}: working on round {round}")
                    Await Task.Delay(id * 20)
                    Console.WriteLine($"Participant {id}: finished round {round}, waiting at barrier")
                    Await barrier.SignalAndWait()
                Next
            End Function)).ToArray()

        Await Task.WhenAll(participants)
        Console.WriteLine("All rounds complete.")
    End Function
End Module

Vedere anche

• Semafori asincroni, lucchetti e coordinamento lettore/scrittore
• modello asincrono basato su attività (TAP)
• Completare le attività
• Mantenere attivi i metodi asincroni