Asenkron koordinasyon primitiflerini oluşturma

ManualResetEventSlim, CountdownEvent ve Barrier gibi eşzamanlı koordinasyon öncülleri beklerken çağıran iş parçacığını engeller. Asenkron kodda, bir iş parçacığının engellenmesi, başka işler üstlenebilecek bir kaynağı israf eder. Zaman uyumsuz eşdeğerler oluşturmak için TaskCompletionSource, engellemek yerine çağıranlara await izin verin.

Bir TaskCompletionSource, SetResult(), SetException veya SetCanceled çağırarak manuel olarak tamamladığınız bir Task üretir. Bu görevi bekleyen kod, bir iş parçacığını engellemeden askıya alır ve kaynağı tamamladığınızda devam eder. Bu desen, bu makaledeki her temel öğe için yapı taşını oluşturur.

Zaman uyumsuz manuel sıfırlama olayı

El ile sıfırlama olayı sinyalsiz bir durumda başlar. Arayanlar olayı bekler ve başka bir taraf olayı işaret ettiğinde (ayarladığında) tüm garsonlar devam eder. Siz açıkça sıfırlayana kadar etkinlik sinyalli durumda kalır. Eşzamanlı eşdeğer ManualResetEventSlim. .NET çalışma zamanı, tek seferlik yayın sinyallemesi için doğrudan TaskCompletionSource<TResult> sağlar. Her döngünün çevresinde bir sıfırlama sarmalayıcısı oluşturmak yerine yeni bir örnek oluşturun.

TaskCompletionSource kendisi tek seferlik bir manuel sıfırlama olaydır: Task siz bir Set* yöntemi çağırana kadar tamamlanmaz ve ardından tüm bekleme mekanizmaları devam eder. Yeni bir Reset ekleyin, TaskCompletionSource ile değiştiren bir yöntem ve böylece yeniden kullanılabilir bir asenkron manuel sıfırlama olayına sahip olursunuz.

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

Önemli uygulama ayrıntıları:

• Oluşturucu, garsonun devamlılıklarını denetleyici iş parçacığında senkron olarak çalıştırmasını önlemek için TaskCreationOptions.RunContinuationsAsynchronously geçirir Set. Bu bayrak olmadan, Set belirsiz bir süre boyunca tıkayabilir.
• Reset, geçerli TaskCompletionSource zaten tamamlandığında, sadece yeni bir TaskCompletionSource ile değiştirmek için CompareExchange kullanır. Bu atomik değiştirme, bir garson tarafından zaten alınan bir görevin yalnız bırakılmasını önler.

Aşağıdaki örnek, iki görevi olay boyunca nasıl koordine olduğunu gösterir:

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

Eş zamansız otomatik sıfırlama etkinliği

Otomatik sıfırlama durumu, manuel sıfırlama durumuna benzer, ancak tam olarak bir bekleyici serbest bırakıldıktan sonra otomatik olarak sinyalsiz duruma geri döner. Olay sinyallendiğinde birden çok arayan bekliyorsa, yalnızca bir garson devam eder. Senkron eşdeğeridir AutoResetEvent. .NET çalışma zamanı, tek bekleyici asenkron sinyal için SemaphoreSlim içerir. Bunu 0 ile başlatın ve maksimum 1 sayısına ayarlayın, beklemek için WaitAsync'yi çağırın ve sinyal vermek için Release'yi çağırın.

Her sinyal yalnızca bir garson serbest bıraktığından TaskCompletionSource , bunları tek tek tamamlayabilmeniz için bir örnek koleksiyonuna (garson başına bir tane) ihtiyacınız vardır:

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

Önemli uygulama ayrıntıları:

• Set yöntemi, TaskCompletionSource (TCS) kilidinin dışında tamamlanır. Kilidin içinde bir TCS'nin tamamlanması, kilit tutulduğu sürece eşzamanlı devamlar çalıştırır, bu da kilitlenmelere veya beklenmedik yeniden girişe neden olabilir.
• Set çağrıldığında ve hiçbir bekleyici kuyruğa alınmadığında, sinyal saklanır, böylece sonraki WaitAsync çağrısı hemen tamamlanır.

Aşağıdaki örnek, bir üreticinin olay aracılığıyla bir tüketiciye sinyal gönderdiğini göstermektedir.

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

Async geri sayım olayı

Geri sayım olayı, garsonların devam etmesine izin vermeden önce belirtilen sayıda sinyali bekler. Bu desen, N işlemlerini başlatıp tüm N'in tamamlanmasını beklemek istediğiniz fork/join senaryoları için faydalıdır. Eşzamanlı eşdeğeridir CountdownEvent. .NET çalışma zamanı, sabit bir görev kümesiyle fork/join işlemleri için WhenAll sağlar. Onu bunun yerine kullanın.

Önceki bölümden AsyncManualResetEvent ve bir atom sayacı kullanarak zaman uyumsuz sürümü oluşturun.

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

Signal yöntemi, sayıyı Decrement ile atomik olarak azaltır. Sayı sıfıra ulaştığında, iç olayı ayarlar ve tüm garsonlar devam eder.

Aşağıdaki örnek, üç eşzamanlı işlemi beklemek için bir geri sayım olayı kullanır:

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

Asenkron bariyer

Bariyer, birden çok turda sabit bir katılımcı kümesini koordine eder. Her katılımcı, geçerli tur için çalışmasını tamamladığında sinyal gönderir ve ardından diğer tüm katılımcıların bitmesini bekler. Son katılımcı sinyaller verince tüm katılımcılar devam eder ve bir sonraki tur için engel sıfırlanır. Eşzamanlı eşdeğeridir Barrier. .NET çalışma zamanı, çok aşamalı asenkron senkronizasyon için WhenAll sağlar. Döngüyle birleştirin ve her turda bir WhenAll çağrı yapın.

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

Önemli uygulama ayrıntıları:

• Paylaşılan TaskCompletionSourcetamamlanmadan önce yöntemi sayıyı sıfırlar ve bir sonraki tur için yeni TaskCompletionSource bir değer değiştirir. Bu sıralama, garsonlar devam ettiğinde bariyerin bir sonraki tur için zaten hazır olmasını sağlar.
• Tüm katılımcılar aynı Tasköğesini paylaşır. Örnek TaskCompletionSource'yi RunContinuationsAsynchronously ile oluşturduğundan, devamlılıklar engeli tamamlayan iş parçacığında satır içi çalıştırmak yerine zaman uyumsuz olarak devam ettirilir.

Aşağıdaki örnek, üç katılımcıyı iki tur boyunca bir engelden geçirir:

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

Ayrıca bakınız

• Asenkron semaforlar, kilitler ve okuyucu/yazıcı koordinasyonu
• Görev tabanlı eşzamanlı olmayan model (TAP)
• Görevlerinizi tamamlama
• Zaman uyumsuz yöntemleri çalışır durumda tutun