Tworzenie asynchronicznych elementów pierwotnych koordynacji

Synchroniczne elementy pierwotne koordynacji, takie jak ManualResetEventSlim, CountdownEventi Barrier blokują wątek wywołujący podczas oczekiwania. W asynchronicznym kodzie blokowanie wątku jest marnowaniem zasobu, który mógłby wykonać inne zadanie. Użyj TaskCompletionSource do utworzenia asynchronicznych odpowiedników, które pozwalają użytkownikom await zamiast blokować.

Element TaskCompletionSource generuje Task, który kończysz ręcznie poprzez wywołanie metody SetResult(), SetException, lub SetCanceled. Kod, który oczekuje na to zadanie zawiesza się bez blokowania wątku i wznawia działanie po zakończeniu źródła. Ten wzorzec stanowi blok konstrukcyjny dla każdego elementu pierwotnego w tym artykule.

Zdarzenie asynchronicznego resetowania ręcznego

Zdarzenie resetowania ręcznego rozpoczyna się w stanie niesygnalizowym. Osoby wywołujące czekają na zdarzenie i wszystkie kelnerki wznawiają, gdy inny podmiot sygnalizuje (ustawia) zdarzenie. Zdarzenie pozostaje zasygnalizowane, dopóki nie zostanie jawnie zresetowane. Synchroniczny odpowiednik to ManualResetEventSlim. Środowisko wykonawcze .NET zapewnia TaskCompletionSource<TResult> do jednorazowego nadawania sygnałów. Utwórz nowe wystąpienie w każdym cyklu zamiast tworzyć opakowanie resetujące wokół niego.

TaskCompletionSource jest jednorazowym zdarzeniem z ręcznym resetowaniem: jego Task jest niekompletna, dopóki nie wywołasz metody Set*, a następnie wszyscy oczekujący wznowią. Dodaj metodę Reset do zamiany na nowy TaskCompletionSource, a otrzymasz wielokrotnego użytku asynchroniczne zdarzenie resetu ręcznego.

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

Szczegóły kluczowych elementów implementacji

• Konstruktor przekazuje TaskCreationOptions.RunContinuationsAsynchronously, aby zapobiec uruchamianiu kontynuacji Set synchronicznie na wątku wywołującym. Bez tej flagi Set może zablokować na nieprzewidywalny czas.
• Reset używa CompareExchange do zamiany na nowy TaskCompletionSource tylko wtedy, gdy obecny jest już ukończony. Ta zamiana niepodzielna uniemożliwia oddzielone zadanie, które zostało już odebrane przez kelnera.

W poniższym przykładzie pokazano, jak dwa zadania koordynują się przez zdarzenie:

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

Zdarzenie automatycznego resetowania asynchronicznego

Zdarzenie automatycznego resetowania jest podobne do zdarzenia resetowania ręcznego, ale automatycznie powraca do stanu niesygnalizowanego po zwolnieniu dokładnie jednego kelnera. Jeśli wielu rozmówców czeka, gdy zdarzenie jest zasygnalizowane, tylko jeden kelner wznawia. Synchroniczny odpowiednik to AutoResetEvent. Środowisko uruchomieniowe .NET zawiera SemaphoreSlim do sygnalizacji asynchronicznej dla jednego oczekującego. Zainicjuj ją za pomocą maksymalnej liczby 1, wywołaj WaitAsync do oczekiwania i Release do wysyłania sygnału.

Ponieważ każdy sygnał zwalnia tylko jednego kelnera, potrzebujesz kolekcji TaskCompletionSource wystąpień — jednej na kelnera — aby można było je ukończyć indywidualnie:

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

Szczegóły kluczowych elementów implementacji

• Metoda Set kończy TaskCompletionSource (TCS) poza blokadą. Ukończenie TCS wewnątrz blokady uruchamia synchroniczne kontynuacje, gdy blokada jest utrzymywana, co może spowodować zakleszczenia lub nieoczekiwaną ponowną reentrancyjność.
• Gdy Set jest wywoływany i nie ma oczekujących, sygnał jest przechowywany, więc następne wywołanie WaitAsync zostanie zakończone natychmiast.

W poniższym przykładzie pokazano producenta sygnalizujące odbiorcę za pośrednictwem zdarzenia:

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

Zdarzenie odliczania asynchronicznego

Zdarzenie odliczania czeka na określoną liczbę sygnałów, zanim pozwoli ona kelnerom kontynuować. Ten schemat jest przydatny w scenariuszach rozwidlenia/łączenia, w których uruchamiasz N operacji i chcesz oczekać zakończenia wszystkich N operacji. Synchroniczny odpowiednik to CountdownEvent. Środowisko uruchomieniowe .NET zapewnia koordynację typu fork/join z ustalonym zestawem zadań. Użyj go zamiast tego.

Skompiluj wersję asynchroniczną, komponując AsyncManualResetEvent z poprzedniej sekcji z licznikiem atomowym.

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

Metoda Signal dekrementuje liczbę atomowo za pomocą Decrement metody. Gdy liczba osiągnie zero, ustawia zdarzenie wewnętrzne i wszystkie kelnerów wznawiają działanie.

W poniższym przykładzie użyto zdarzenia odliczania w celu oczekiwania na trzy współbieżne operacje:

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

Bariera asynchroniczna

Bariera koordynuje stały zestaw uczestników w wielu rundach. Każdy uczestnik sygnalizuje, kiedy zakończy swoją pracę w bieżącej rundzie, a następnie czeka, aż wszyscy pozostali uczestnicy zakończą swoją pracę. Gdy ostatni uczestnik sygnalizuje, wszyscy uczestnicy wznawiają swoje działania, a bariera resetuje się do następnej rundy. Synchroniczny odpowiednik to Barrier. Środowisko wykonawcze .NET zapewnia WhenAll na potrzeby wielorundowej asynchronicznej synchronizacji. Połącz to z pętlą, jedno WhenAll wywołanie na rundę.

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

Szczegóły kluczowych elementów implementacji

• Przed zakończeniem wspólnej TaskCompletionSource, metoda resetuje licznik i zamienia go na nowy TaskCompletionSource na następną rundę. Dzięki temu kolejność gwarantuje, że po wznowieniu kelnerów bariera jest już gotowa do następnej rundy.
• Wszyscy uczestnicy dzielą ten sam Task. Ponieważ przykład tworzy element TaskCompletionSource z elementem RunContinuationsAsynchronously, kontynuacje są wznawiane asynchronicznie, zamiast działać w tym samym wątku, który kończy działanie bariery.

Poniższy przykład prowadzi trzech uczestników przez dwie rundy bariery:

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

Zobacz także

• Asynchroniczne semafory, zamki i koordynacja czytelnika/piszącego
• Wzorzec asynchroniczny oparty na zadaniach (TAP)
• Wykonywanie zadań
• Utrzymywanie aktywności metod asynchronicznych