Erstellen asynchroner Koordinationsgrundtypen

Synchrone Koordinationsprimitiven wie ManualResetEventSlim, CountdownEvent und Barrier blockieren den aufrufenden Thread während des Wartens. Im asynchronen Code verschwendet das Blockieren eines Threads eine Ressource, die andere Aufgaben ausführen könnte. Verwenden Sie TaskCompletionSource, um asynchrone Entsprechungen zu erstellen, damit Aufrufer await können, anstatt zu blockieren.

Ein TaskCompletionSource erzeugt ein Task, das Sie manuell durch Aufrufen von SetResult(), SetException oder SetCanceled abschließen. Code, der auf die Aufgabe wartet, wird ausgesetzt, ohne einen Thread zu blockieren, und wird fortgesetzt, wenn die Quelle abgeschlossen ist. Dieses Muster bildet den Baustein für jeden Primitiven in diesem Artikel.

Asynchrones Ereignis für manuelles Zurücksetzen

Ein manuell zurückzusetzendes Ereignis beginnt im nicht signalisierten Zustand. Aufrufer warten auf das Ereignis, und alle Wartenden werden fortgesetzt, wenn ein anderer Teilnehmer das Ereignis signalisiert (festlegt). Das Ereignis bleibt signalisiert, bis Sie es explizit zurücksetzen. Das synchrone Äquivalent ist ManualResetEventSlim. Die .NET Laufzeit stellt TaskCompletionSource<TResult> direkt für einmaliges Senden von Signalen bereit. Erstellen Sie für jeden Zyklus eine neue Instanz, anstatt einen Reset-Wrapper darum zu bauen.

TaskCompletionSource ist selbst ein manuell zurückgesetztes Ereignis: Dessen Task ist solange unvollständig, bis Sie eine Methode Set* aufrufen, und dann werden alle Wartenden fortgesetzt. Fügen Sie eine Reset-Methode hinzu, die ein neues TaskCompletionSource-Element austauscht, und Sie erhalten ein wiederverwendbares asynchrones manuelles Reset-Ereignis.

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

Wichtige Implementierungsdetails:

• Der Konstruktor übergibt TaskCreationOptions.RunContinuationsAsynchronously, um Set daran zu hindern, die Fortsetzung von Wartenden synchron im aufrufenden Thread auszuführen. Ohne diese Kennzeichnung könnte Set für einen unvorhersehbaren Zeitraum blockieren.
• Reset verwendet CompareExchange, um ein neues TaskCompletionSource nur dann einzusetzen, wenn das aktuelle bereits abgeschlossen ist. Dieser Atomtausch verhindert, dass eine vom Wartende bereits empfangene Aufgabe verwaist.

Das folgende Beispiel zeigt, wie zwei Aufgaben das Ereignis koordinieren:

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

Asynchrones Auto-Reset-Ereignis

Ein Autoreset-Ereignis ähnelt einem manuellen Reset-Ereignis, kehrt jedoch automatisch in den nicht signalisierten Zustand zurück, nachdem genau ein Wartender freigegeben wurde. Wenn mehrere Aufrufer warten, während das Ereignis signalisiert wird, fährt nur ein Wartender fort. Das synchrone Äquivalent ist AutoResetEvent. Die .NET Laufzeit enthält SemaphoreSlim für die asynchrone Signalisierung mit einem einzelnen Wartenden. Initialisieren Sie es mit 0 bei einer maximalen Anzahl von 1, und rufen Sie WaitAsync auf, um zu warten, und Release, um zu signalisieren.

Da jedes Signal nur einen Waiter loslässt, benötigen Sie eine Sammlung von TaskCompletionSource Instanzen – eine pro Waiter – so dass Sie sie einzeln abschließen können:

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

Wichtige Implementierungsdetails:

• Die Methode Set schließt das TaskCompletionSource (TCS) außerhalb der Sperre ab. Durch das Abschließen eines TCS innerhalb der Sperre werden synchrone Fortsetzungen ausgeführt, während die Sperre gehalten wird, was zu Deadlocks oder unerwartetem Wiedereinstieg führen kann.
• Wenn Set aufgerufen wird und kein Wartender in der Warteschlange ist, wird das Signal gespeichert, sodass der nächste WaitAsync Aufruf sofort ausgeführt wird.

Das folgende Beispiel zeigt einen Produzenten, der einem Verbraucher mittels des Ereignisses ein Signal gibt.

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

Countdown-Ereignis (asynchron)

Ein Countdown-Ereignis wartet auf eine bestimmte Anzahl von Signalen, bevor Wartende fortfahren dürfen. Dieses Muster eignet sich für Verzweigungs-/Zusammenführungsszenarien, in denen Sie N-Vorgänge starten und auf alle N-Abschlüsse warten möchten. Das synchrone Äquivalent ist CountdownEvent. Die .NET Runtime stellt WhenAll für Fork/Join-Koordination mit einer festen Anzahl von Aufgaben bereit. Verwenden Sie sie stattdessen.

Erstellen Sie die asynchrone Version, indem Sie den AsyncManualResetEvent aus dem vorherigen Abschnitt mit einem Atomzähler verfassten:

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

Die Signal Methode verringert die Anzahl atomisch mit Decrement. Wenn die Anzahl Null erreicht, wird das interne Ereignis ausgelöst, und alle Wartenden werden fahren fort.

Im folgenden Beispiel wird ein Countdownereignis verwendet, um drei gleichzeitige Vorgänge zu erwarten:

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

Asynchrone Barriere

Eine Barriere koordiniert einen festen Satz von Teilnehmern über mehrere Runden hinweg. Jeder Teilnehmer signalisiert, wann er seine Arbeit für die aktuelle Runde beendet hat, und wartet dann, bis alle anderen Teilnehmer fertig sind. Sobald der letzte Teilnehmer signalisiert, setzen alle Teilnehmer ihre Aktivitäten fort, und die Barriere wird für die nächste Runde zurückgesetzt. Das synchrone Äquivalent ist Barrier. Die .NET-Laufzeit stellt WhenAll für die asynchrone Multi-Round-Synchronisierung bereit. Kombinieren Sie es mit einer Schleife, ein Aufruf WhenAll pro Runde.

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

Wichtige Implementierungsdetails:

• Bevor die freigegebene TaskCompletionSource abgeschlossen wird, setzt die Methode den Zähler zurück und ersetzt sie durch einen neuen TaskCompletionSource für die nächste Runde. Durch diese Sortierung wird sichergestellt, dass die Barriere für die nächste Runde bereit ist, wenn die Wartenden fortfahren.
• Alle Teilnehmer teilen dasselbe Task. Da das Beispiel die TaskCompletionSource mit RunContinuationsAsynchronously erstellt, werden Fortsetzungen asynchron ausgeführt, anstatt direkt im Thread zu laufen, der die Barriere vervollständigt.

Im folgenden Beispiel durchlaufen drei Teilnehmer zwei Runden einer Barriere:

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

Siehe auch

• Asynchrone Semaphore, Sperren und Lese-/Schreiberkoordination
• Aufgabenbasiertes asynchrones Muster (TAP)
• Ausführen Ihrer Aufgaben
• Asynchrone Methoden lebendig halten