非同期セマフォ、ロック、およびリーダー/ライター間の協調

非同期コードでスロットリング、相互排他、またはリーダー/ライターの同期が必要な場合は、独自の型を構築するのではなく、組み込みの.NET型を使用します。 この記事では、これらの型を適用する方法について説明し、カスタム実装を使用して内部的にどのように動作するかを説明します。

非同期セマフォ — 同時アクセスを抑える

セマフォは、リソースに同時にアクセスできる呼び出し元の数を制限します。 SemaphoreSlim には、スレッドをブロックせずにエントリを待機できる WaitAsync メソッドが用意されています。

public static class SemaphoreSlimDemo
{
    public static async Task RunAsync()
    {
        using var semaphore = new SemaphoreSlim(3);

        Task[] tasks = Enumerable.Range(1, 6).Select(id => Task.Run(async () =>
        {
            await semaphore.WaitAsync();
            try
            {
                Console.WriteLine($"Task {id}: entered (count = {semaphore.CurrentCount})");
                await Task.Delay(100);
            }
            finally
            {
                semaphore.Release();
                Console.WriteLine($"Task {id}: released");
            }
        })).ToArray();

        await Task.WhenAll(tasks);
    }
}

Public Module SemaphoreSlimDemo
    Public Async Function RunAsync() As Task
        Using semaphore As New SemaphoreSlim(3)
            Dim tasks As Task() = Enumerable.Range(1, 6).Select(
                Function(id) Task.Run(Async Function()
                    Await semaphore.WaitAsync()
                    Try
                        Console.WriteLine($"Task {id}: entered (count = {semaphore.CurrentCount})")
                        Await Task.Delay(100)
                    Finally
                        semaphore.Release()
                        Console.WriteLine($"Task {id}: released")
                    End Try
                End Function)).ToArray()

            Await Task.WhenAll(tasks)
        End Using
    End Function
End Module

WaitAsync Release try ブロック内の/と常にfinallyをペアリングします。 解放するのを忘れると、セマフォカウントは増えず、他の呼び出し元は無期限に待機することになります。

非同期セマフォのしくみ

内部的には、非同期セマフォは待機者のカウントとキューを保持します。 カウントが 0 より大きい場合、 WaitAsync はカウントをデクリメントし、直ちに返します。 カウントが 0 の場合、 WaitAsync は TaskCompletionSource をエンキューし、そのタスクを返します。 Release ウェイターをキューから削除して完了するか、またはカウントをインクリメントします。

// Educational only — use SemaphoreSlim instead of this sample implementation.
public class AsyncSemaphore
{
    private readonly Queue<TaskCompletionSource> _waiters = new();
    private int _currentCount;

    public AsyncSemaphore(int initialCount)
    {
        ArgumentOutOfRangeException.ThrowIfNegative(initialCount, nameof(initialCount));
        _currentCount = initialCount;
    }

    public Task WaitAsync()
    {
        lock (_waiters)
        {
            if (_currentCount > 0)
            {
                _currentCount--;
                return Task.CompletedTask;
            }
            else
            {
                var waiter = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
                _waiters.Enqueue(waiter);
                return waiter.Task;
            }
        }
    }

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

        lock (_waiters)
        {
            if (_waiters.Count > 0)
                toRelease = _waiters.Dequeue();
            else
                _currentCount++;
        }

        toRelease?.TrySetResult();
    }
}

' Educational only — use SemaphoreSlim instead of this sample implementation.
Public Class AsyncSemaphore
    Private ReadOnly _waiters As New Queue(Of TaskCompletionSource)()
    Private _currentCount As Integer

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

    Public Function WaitAsync() As Task
        SyncLock _waiters
            If _currentCount > 0 Then
                _currentCount -= 1
                Return Task.CompletedTask
            Else
                Dim waiter As New TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously)
                _waiters.Enqueue(waiter)
                Return waiter.Task
            End If
        End SyncLock
    End Function

    Public Sub Release()
        Dim toRelease As TaskCompletionSource = Nothing

        SyncLock _waiters
            If _waiters.Count > 0 Then
                toRelease = _waiters.Dequeue()
            Else
                _currentCount += 1
            End If
        End SyncLock

        toRelease?.TrySetResult()
    End Sub
End Class

Release メソッドは、TaskCompletionSourceのAsyncAutoResetEventと同様に、ロック外のを完了します。 この方法では、ロックが保持されている間に同期継続が実行されないようにします。

非同期ロック: awaits 間の相互除外

カウントが 1 のロックでは、相互除外が提供されます。 C# lock ステートメントと Lock (.NET 9 以降) は、スレッド アフィンであるため、await 境界を越えて機能しません。 thread-affine ロックでは、ロックを取得したスレッドと解放するスレッドは同一でなければなりません。 await全体で、継続を再開するスレッドがロックを取得したスレッドではなく、その要件に違反している可能性があります。 代わりに、カウントが 1 の SemaphoreSlim を使用します。

public static class SemaphoreSlimAsLockDemo
{
    private static readonly SemaphoreSlim s_lock = new(1, 1);
    private static int s_sharedCounter;

    public static async Task RunAsync()
    {
        Task[] tasks = Enumerable.Range(1, 5).Select(_ => Task.Run(async () =>
        {
            await s_lock.WaitAsync();
            try
            {
                int before = s_sharedCounter;
                await Task.Delay(10);
                s_sharedCounter = before + 1;
            }
            finally
            {
                s_lock.Release();
            }
        })).ToArray();

        await Task.WhenAll(tasks);
        Console.WriteLine($"Counter = {s_sharedCounter} (expected 5)");
    }
}

Public Module SemaphoreSlimAsLockDemo
    Private ReadOnly s_lock As New SemaphoreSlim(1, 1)
    Private s_sharedCounter As Integer

    Public Async Function RunAsync() As Task
        Dim tasks As Task() = Enumerable.Range(1, 5).Select(
            Function(unused) Task.Run(Async Function()
                Await s_lock.WaitAsync()
                Try
                    Dim before As Integer = s_sharedCounter
                    Await Task.Delay(10)
                    s_sharedCounter = before + 1
                Finally
                    s_lock.Release()
                End Try
            End Function)).ToArray()

        Await Task.WhenAll(tasks)
        Console.WriteLine($"Counter = {s_sharedCounter} (expected 5)")
    End Function
End Module

非同期ロックのしくみ

セマフォ パターンは、自動リリースの using をサポートする型でラップできます。 LockAsync メソッドは破棄可能なReleaserを返します。Releaserが破棄されると、セマフォが解放されます。

// Educational only — use SemaphoreSlim(1, 1) with try/finally instead of this sample implementation.
public class AsyncLock : IDisposable
{
    private readonly SemaphoreSlim _semaphore = new(1, 1);
    private readonly Task<Releaser> _releaser;

    public AsyncLock()
    {
        _releaser = Task.FromResult(new Releaser(this));
    }

    public Task<Releaser> LockAsync()
    {
        Task wait = _semaphore.WaitAsync();
        return wait.IsCompleted
            ? _releaser
            : wait.ContinueWith(
                  (_, state) => new Releaser((AsyncLock)state!),
                  this,
                  CancellationToken.None,
                  TaskContinuationOptions.ExecuteSynchronously,
                  TaskScheduler.Default);
    }

    public struct Releaser : IDisposable
    {
        private readonly AsyncLock? _toRelease;

        internal Releaser(AsyncLock toRelease) => _toRelease = toRelease;

        public void Dispose() => _toRelease?._semaphore.Release();
    }

    public void Dispose() => _semaphore.Dispose();
}

' Educational only — use SemaphoreSlim(1, 1) with Try/Finally instead of this sample implementation.
Public Class AsyncLock
    Implements IDisposable

    Private ReadOnly _semaphore As New SemaphoreSlim(1, 1)
    Private ReadOnly _releaser As Task(Of Releaser)

    Public Sub New()
        _releaser = Task.FromResult(New Releaser(Me))
    End Sub

    Public Function LockAsync() As Task(Of Releaser)
        Dim wait As Task = _semaphore.WaitAsync()
        If wait.IsCompleted Then
            Return _releaser
        Else
            Return wait.ContinueWith(
                Function(unused, state) New Releaser(DirectCast(state, AsyncLock)),
                Me,
                CancellationToken.None,
                TaskContinuationOptions.ExecuteSynchronously,
                TaskScheduler.Default)
        End If
    End Function

    Public Structure Releaser
        Implements IDisposable

        Private ReadOnly _toRelease As AsyncLock

        Friend Sub New(toRelease As AsyncLock)
            _toRelease = toRelease
        End Sub

        Public Sub Dispose() Implements IDisposable.Dispose
            _toRelease?._semaphore.Release()
        End Sub
    End Structure

    Public Sub Dispose() Implements IDisposable.Dispose
        _semaphore.Dispose()
    End Sub
End Class

使用法は簡潔で安全です。

public static class AsyncLockDemo
{
    private static readonly AsyncLock s_lock = new();
    private static int s_sharedValue;

    public static async Task RunAsync()
    {
        Task[] tasks = Enumerable.Range(1, 5).Select(id => Task.Run(async () =>
        {
            using (await s_lock.LockAsync())
            {
                int before = s_sharedValue;
                await Task.Delay(10);
                s_sharedValue = before + 1;
                Console.WriteLine($"Task {id}: incremented to {s_sharedValue}");
            }
        })).ToArray();

        await Task.WhenAll(tasks);
        Console.WriteLine($"Final value = {s_sharedValue} (expected 5)");
    }
}

Public Module AsyncLockDemo
    Private ReadOnly s_lock As New AsyncLock()
    Private s_sharedValue As Integer

    Public Async Function RunAsync() As Task
        Dim tasks As Task() = Enumerable.Range(1, 5).Select(
            Function(id) Task.Run(Async Function()
                Using Await s_lock.LockAsync()
                    Dim before As Integer = s_sharedValue
                    Await Task.Delay(10)
                    s_sharedValue = before + 1
                    Console.WriteLine($"Task {id}: incremented to {s_sharedValue}")
                End Using
            End Function)).ToArray()

        Await Task.WhenAll(tasks)
        Console.WriteLine($"Final value = {s_sharedValue} (expected 5)")
    End Function
End Module

非同期読み取り/書き込みの調整

リーダー/ライター ロックでは、複数の同時リーダーが許可されますが、排他的ライターは 1 つだけです。 .NET は、ConcurrentExclusiveSchedulerPair を提供します。これにより、2 つの TaskScheduler インスタンスを介してタスクのリーダー/ライター スケジューリングが提供されます。

• ConcurrentScheduler — 排他的タスクがアクティブでない限り、タスクを同時に実行します (リーダーなど)。
• ExclusiveScheduler — 他のタスクは実行せず、(ライターなど) 排他的にタスクを実行します。

public static class ConcurrentExclusiveDemo
{
    public static async Task RunAsync()
    {
        var pair = new ConcurrentExclusiveSchedulerPair();
        var factory = new TaskFactory(pair.ExclusiveScheduler);

        int sharedValue = 0;

        Task writerTask = factory.StartNew(() =>
        {
            sharedValue = 42;
            Console.WriteLine($"Writer: set value to {sharedValue}");
        });

        var readerFactory = new TaskFactory(pair.ConcurrentScheduler);

        Task[] readerTasks = Enumerable.Range(1, 3).Select(id =>
            readerFactory.StartNew(() =>
            {
                Console.WriteLine($"Reader {id}: value = {sharedValue}");
            })).ToArray();

        await writerTask;
        await Task.WhenAll(readerTasks);
    }
}

Public Module ConcurrentExclusiveDemo
    Public Async Function RunAsync() As Task
        Dim pair As New ConcurrentExclusiveSchedulerPair()
        Dim exclusiveFactory As New TaskFactory(pair.ExclusiveScheduler)

        Dim sharedValue As Integer = 0

        Dim writerTask As Task = exclusiveFactory.StartNew(Sub()
            sharedValue = 42
            Console.WriteLine($"Writer: set value to {sharedValue}")
        End Sub)

        Dim readerFactory As New TaskFactory(pair.ConcurrentScheduler)

        Dim readerTasks As Task() = Enumerable.Range(1, 3).Select(
            Function(id) readerFactory.StartNew(Sub()
                Console.WriteLine($"Reader {id}: value = {sharedValue}")
            End Sub)).ToArray()

        Await writerTask
        Await Task.WhenAll(readerTasks)
    End Function
End Module

カスタム非同期リーダー/ライター ロック

次の実装では、ライターがリーダーよりも優先されます。 書き手が待っている場合、新しい読み手はその背後に列を作ります。 ライター処理が完了し、他に待機中のライターがない場合は、キューに入っているすべてのリーダーがまとめて実行されます:

public class AsyncReaderWriterLock
{
    private readonly Queue<TaskCompletionSource<Releaser>> _waitingWriters = new();
    private TaskCompletionSource<Releaser> _waitingReader =
        new(TaskCreationOptions.RunContinuationsAsynchronously);
    private int _readersWaiting;
    private int _status; // 0 = free, -1 = writer active, >0 = reader count

    private readonly Task<Releaser> _readerReleaser;
    private readonly Task<Releaser> _writerReleaser;

    public AsyncReaderWriterLock()
    {
        _readerReleaser = Task.FromResult(new Releaser(this, isWriter: false));
        _writerReleaser = Task.FromResult(new Releaser(this, isWriter: true));
    }

    public Task<Releaser> ReaderLockAsync()
    {
        lock (_waitingWriters)
        {
            if (_status >= 0 && _waitingWriters.Count == 0)
            {
                _status++;
                return _readerReleaser;
            }
            else
            {
                _readersWaiting++;
                return _waitingReader.Task;
            }
        }
    }

    public Task<Releaser> WriterLockAsync()
    {
        lock (_waitingWriters)
        {
            if (_status == 0)
            {
                _status = -1;
                return _writerReleaser;
            }
            else
            {
                var waiter = new TaskCompletionSource<Releaser>(TaskCreationOptions.RunContinuationsAsynchronously);
                _waitingWriters.Enqueue(waiter);
                return waiter.Task;
            }
        }
    }

    private void ReaderRelease()
    {
        TaskCompletionSource<Releaser>? toWake = null;

        lock (_waitingWriters)
        {
            _status--;
            if (_status == 0 && _waitingWriters.Count > 0)
            {
                _status = -1;
                toWake = _waitingWriters.Dequeue();
            }
        }

        toWake?.SetResult(new Releaser(this, isWriter: true));
    }

    private void WriterRelease()
    {
        TaskCompletionSource<Releaser>? toWake = null;
        bool toWakeIsWriter = false;

        lock (_waitingWriters)
        {
            if (_waitingWriters.Count > 0)
            {
                toWake = _waitingWriters.Dequeue();
                toWakeIsWriter = true;
            }
            else if (_readersWaiting > 0)
            {
                toWake = _waitingReader;
                _status = _readersWaiting;
                _readersWaiting = 0;
                _waitingReader = new TaskCompletionSource<Releaser>(TaskCreationOptions.RunContinuationsAsynchronously);
            }
            else
            {
                _status = 0;
            }
        }

        toWake?.SetResult(new Releaser(this, toWakeIsWriter));
    }

    public struct Releaser : IDisposable
    {
        private readonly AsyncReaderWriterLock? _lock;
        private readonly bool _isWriter;

        internal Releaser(AsyncReaderWriterLock lockObj, bool isWriter)
        {
            _lock = lockObj;
            _isWriter = isWriter;
        }

        public void Dispose()
        {
            if (_lock is not null)
            {
                if (_isWriter) _lock.WriterRelease();
                else _lock.ReaderRelease();
            }
        }
    }
}

Public Class AsyncReaderWriterLock
    Private ReadOnly _waitingWriters As New Queue(Of TaskCompletionSource(Of Releaser))()
    Private _waitingReader As New TaskCompletionSource(Of Releaser)(TaskCreationOptions.RunContinuationsAsynchronously)
    Private _readersWaiting As Integer
    Private _status As Integer ' 0 = free, -1 = writer active, >0 = reader count

    Private ReadOnly _readerReleaser As Task(Of Releaser)
    Private ReadOnly _writerReleaser As Task(Of Releaser)

    Public Sub New()
        _readerReleaser = Task.FromResult(New Releaser(Me, isWriter:=False))
        _writerReleaser = Task.FromResult(New Releaser(Me, isWriter:=True))
    End Sub

    Public Function ReaderLockAsync() As Task(Of Releaser)
        SyncLock _waitingWriters
            If _status >= 0 AndAlso _waitingWriters.Count = 0 Then
                _status += 1
                Return _readerReleaser
            Else
                _readersWaiting += 1
                Return _waitingReader.Task
            End If
        End SyncLock
    End Function

    Public Function WriterLockAsync() As Task(Of Releaser)
        SyncLock _waitingWriters
            If _status = 0 Then
                _status = -1
                Return _writerReleaser
            Else
                Dim waiter As New TaskCompletionSource(Of Releaser)(
                    System.Threading.Tasks.TaskCreationOptions.RunContinuationsAsynchronously)
                _waitingWriters.Enqueue(waiter)
                Return waiter.Task
            End If
        End SyncLock
    End Function

    Private Sub ReaderRelease()
        Dim toWake As TaskCompletionSource(Of Releaser) = Nothing

        SyncLock _waitingWriters
            _status -= 1
            If _status = 0 AndAlso _waitingWriters.Count > 0 Then
                _status = -1
                toWake = _waitingWriters.Dequeue()
            End If
        End SyncLock

        toWake?.SetResult(New Releaser(Me, isWriter:=True))
    End Sub

    Private Sub WriterRelease()
        Dim toWake As TaskCompletionSource(Of Releaser) = Nothing
        Dim toWakeIsWriter As Boolean = False

        SyncLock _waitingWriters
            If _waitingWriters.Count > 0 Then
                toWake = _waitingWriters.Dequeue()
                toWakeIsWriter = True
            ElseIf _readersWaiting > 0 Then
                toWake = _waitingReader
                _status = _readersWaiting
                _readersWaiting = 0
                _waitingReader = New TaskCompletionSource(Of Releaser)(TaskCreationOptions.RunContinuationsAsynchronously)
            Else
                _status = 0
            End If
        End SyncLock

        toWake?.SetResult(New Releaser(Me, toWakeIsWriter))
    End Sub

    Public Structure Releaser
        Implements IDisposable

        Private ReadOnly _lock As AsyncReaderWriterLock
        Private ReadOnly _isWriter As Boolean

        Friend Sub New(lockObj As AsyncReaderWriterLock, isWriter As Boolean)
            _lock = lockObj
            _isWriter = isWriter
        End Sub

        Public Sub Dispose() Implements IDisposable.Dispose
            If _lock IsNot Nothing Then
                If _isWriter Then
                    _lock.WriterRelease()
                Else
                    _lock.ReaderRelease()
                End If
            End If
        End Sub
    End Structure
End Class

使用法は、 AsyncLockと同じ破棄可能リリース パターンに従います。

public static class AsyncReaderWriterLockDemo
{
    private static readonly AsyncReaderWriterLock s_rwLock = new();
    private static string s_data = "initial";

    public static async Task RunAsync()
    {
        Task writer = Task.Run(async () =>
        {
            using (await s_rwLock.WriterLockAsync())
            {
                Console.WriteLine("Writer: acquired exclusive lock");
                await Task.Delay(50);
                s_data = "updated";
                Console.WriteLine("Writer: data updated");
            }
        });

        Task[] readers = Enumerable.Range(1, 3).Select(id => Task.Run(async () =>
        {
            await Task.Delay(10);
            using (await s_rwLock.ReaderLockAsync())
            {
                Console.WriteLine($"Reader {id}: data = {s_data}");
            }
        })).ToArray();

        await writer;
        await Task.WhenAll(readers);
    }
}

Public Module AsyncReaderWriterLockDemo
    Private ReadOnly s_rwLock As New AsyncReaderWriterLock()
    Private s_data As String = "initial"

    Public Async Function RunAsync() As Task
        Dim writer As Task = Task.Run(Async Function()
            Using Await s_rwLock.WriterLockAsync()
                Console.WriteLine("Writer: acquired exclusive lock")
                Await Task.Delay(50)
                s_data = "updated"
                Console.WriteLine("Writer: data updated")
            End Using
        End Function)

        Dim readers As Task() = Enumerable.Range(1, 3).Select(
            Function(id) Task.Run(Async Function()
                Await Task.Delay(10)
                Using Await s_rwLock.ReaderLockAsync()
                    Console.WriteLine($"Reader {id}: data = {s_data}")
                End Using
            End Function)).ToArray()

        Await writer
        Await Task.WhenAll(readers)
    End Function
End Module

代替調整パターンとしてのチャネル

Channel<T> には、 async の読み取りと書き込みをサポートするスレッド セーフなプロデューサー コンシューマー キューが用意されています。 境界付きチャネル (CreateBounded) は、自然なバックプレッシャーを提供するため、本来であればスロットリング目的でセマフォを使用する一部のしなりを置き換えることができます。

詳細については、「 System.Threading.Channels」を参照してください。

こちらも参照ください

• 非同期調整プリミティブを構築する
• 非同期メソッドを維持する
• タスクを完了する
• タスク ベースの非同期パターンの使用