Construir primitivas de coordenação assíncronas

Primitivas de coordenação síncronas como ManualResetEventSlim, CountdownEvent, e Barrier bloqueiam o fio de chamada enquanto se espera. No código assíncrono, bloquear um thread desperdiça um recurso que poderia estar a fazer outro trabalho. Use TaskCompletionSource para construir equivalentes assíncronos que permitam os chamadores await em vez de bloquear.

A TaskCompletionSource produz um Task que se completa manualmente ao chamar SetResult(), SetException, ou SetCanceled. O código que aguarda essa tarefa é suspenso sem bloquear um thread e retoma quando completas a fonte. Este padrão constitui o bloco de construção de cada primitivo neste artigo.

Evento de reinicialização manual assíncrona

Um evento de reset manual começa num estado não sinalizado. Os ouvintes aguardam pelo evento, e todos os empregados retomam quando outra parte sinaliza (marca) o evento. O evento mantém-se sinalizado até o reiniciares explicitamente. O equivalente síncrono é ManualResetEventSlim. O runtime .NET fornece TaskCompletionSource<TResult> diretamente para sinalização de difusão única. Cria uma nova instância a cada ciclo em vez de construir um invólucro de reinicialização à volta dela.

TaskCompletionSource é por si só um evento de reinício manual único: Task está incompleto até chamares o método Set*, e depois todos os aguardadores retomam. Adicione um método Reset que insere um novo TaskCompletionSource, e terá um evento de reset manual assíncrono reutilizável.

// 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

Principais detalhes da implementação:

• O construtor passa TaskCreationOptions.RunContinuationsAsynchronously para impedir Set que as continuações do waiter sejam executadas de forma síncrona no thread que chama. Sem esta bandeira, Set podia ficar bloqueado por um tempo indeterminado.
• Reset Usa CompareExchange para trocar um novo TaskCompletionSource apenas quando o atual já está concluído. Esta troca atómica impede de deixar órfão uma tarefa que um empregado já recebeu.

O exemplo seguinte mostra como duas tarefas se coordenam ao longo do 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 de reinício automático assíncrono

Um evento de reinício automático é semelhante a um evento de reset manual, mas regressa automaticamente ao estado não sinalizado após libertar exatamente um empregado. Se vários chamadores estiverem à espera quando o evento é sinalizado, apenas um empregado recomeça. O equivalente síncrono é AutoResetEvent. O tempo de execução .NET inclui SemaphoreSlim para sinalização assíncrona de espera única. Inicializa com 0 um número máximo de 1 e chama WaitAsync para esperar e Release para sinalizar.

Como cada sinal liberta apenas um empregado, precisa de uma coleção de TaskCompletionSource instâncias — uma por empregado — para poder completá-las individualmente:

// 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

Principais detalhes da implementação:

• O Set método completa o TaskCompletionSource (TCS) fora da eclusa. Completar um TCS dentro do bloqueio executa continuações síncronas enquanto o bloqueio está mantido, o que pode causar bloqueios ou reentrância inesperada.
• Quando Set é chamado e nenhum empregado está na fila, o sinal é armazenado para que a próxima WaitAsync chamada seja concluída imediatamente.

O exemplo seguinte mostra um produtor a sinalizar um consumidor através do 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 de contagem decrescente assíncrono

Um evento de contagem decrescente espera por um número especificado de sinais antes de permitir que os empregados prossigam. Este padrão é útil para cenários de fork/join onde inicias N operações e queres esperar por todas as N conclusões. O equivalente síncrono é CountdownEvent. O runtime .NET oferece WhenAll para coordenação de fork/join com um conjunto fixo de tarefas. Usa-o em vez disso.

Constrói a versão assíncrona compondo a AsyncManualResetEvent partir da secção anterior com um contador atómico:

// 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

O método Signal diminui atomicamente a contagem utilizando Decrement. Quando a contagem chega a zero, inicia-se o evento interno e todos os empregados retomam.

O exemplo seguinte utiliza um evento de contagem decrescente para aguardar três operações simultâneas:

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

Barreira assíncrona

Uma barreira coordena um conjunto fixo de participantes ao longo de várias rondas. Cada participante sinaliza quando termina o seu trabalho para a ronda atual e depois espera que todos os outros participantes terminem. Quando o último participante sinaliza, todos os participantes retomam e a barreira reinicia-se para a ronda seguinte. O equivalente síncrono é Barrier. O runtime .NET fornece WhenAll para sincronização assíncrona de múltiplas etapas. Combina isso com um loop, uma WhenAll chamada por iteração.

// 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

Principais detalhes da implementação:

• Antes de completar o recurso partilhado TaskCompletionSource, o método reinicia a contagem e substitui por um novo TaskCompletionSource para a próxima ronda. Esta ordenação garante que, quando os empregados retomarem, a barreira já esteja pronta para a próxima sessão.
• Todos os participantes partilham o mesmo Task. Como a amostra cria o TaskCompletionSource com RunContinuationsAsynchronously, as continuações retomam de forma assíncrona em vez de serem executadas inline no thread que completa a barreira.

O exemplo seguinte conduz três participantes por duas rondas de uma barreira:

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

Ver também

• Semáforos assíncronos, bloqueios e coordenação leitor/escritor
• Padrão assíncrono baseado em tarefas (TAP)
• Completa as tuas tarefas
• Mantenha os métodos assíncronos vivos