用于单线程异步 async-await 式编程的信号量答案

【问题标题】：Semaphores for single-threaded asynchronous async-await-style programming用于单线程异步 async-await 式编程的信号量
【发布时间】：2017-06-24 17:38:08
【问题描述】：

信号量是一种多线程锁定机制，可确保只有有限数量的线程在给定资源上运行。互斥锁是一种特殊情况，其限制数量为 1。

异步编程与多线程编程有很多共同点（有时也与之相关），尽管它本身并不是多线程的。

以下代码创建了十个任务，它们只需等待一秒钟并记录它们的开始和结束。

所有这些都只在一个线程上执行（我假设适当的同步上下文维护已经到位，例如 WPF 中的情况）。

因此，即使只有一个线程，我们也有“并行”任务，并且可能存在希望将资源访问限制为仅几个或其中一项任务的用例。（例如，限制并行网络请求。）

似乎需要一个“异步信号量”——一个锁定的不是线程而是异步延续的概念。

我已经实现了这样一个信号量来检查它是否真的有意义，并说明我的意思。

我的问题是：这个东西是否已经可用，最好是在 .NET 框架本身中？我找不到任何东西，尽管在我看来它应该存在。

这是代码 (LINQPad share here)：

    async void Main()
    {
        // Necessary in LINQPad to ensure a single thread.
        // Other environments such as WPF do this for you.
        SynchronizationContext.SetSynchronizationContext(
            new DispatcherSynchronizationContext());

        var tasks = Enumerable.Range(1, 10).Select(SampleWork).ToArray();

        await Task.WhenAll(tasks);

        "All done.".Dump();
    }

    AsyncSemaphore commonSemaphore = new AsyncSemaphore(4);

    async Task SampleWork(Int32 i)
    {
        using (await commonSemaphore.Acquire())
        {
            $"Beginning work #{i} {Thread.CurrentThread.ManagedThreadId}".Dump();

            await Task.Delay(TimeSpan.FromSeconds(1));

            $"Finished work #{i} {Thread.CurrentThread.ManagedThreadId}".Dump();
        }
    }

    public class AsyncSemaphore
    {
        Int32 maxTasks;
        Int32 currentTasks;

        ReleasingDisposable release;

        Queue<TaskCompletionSource<Object>> continuations
            = new Queue<TaskCompletionSource<Object>>();

        public AsyncSemaphore(Int32 maxTasks = 1)
        {
            this.maxTasks = maxTasks;
            release = new ReleasingDisposable(this);
        }

        public async Task<IDisposable> Acquire()
        {
            ++currentTasks;

            if (currentTasks > maxTasks)
            {
                var tcs = new TaskCompletionSource<Object>();

                continuations.Enqueue(tcs);

                await tcs.Task;
            }

            return release;
        }

        void Release()
        {
            --currentTasks;

            if (continuations.Count > 0)
            {
                var tcs = continuations.Dequeue();

                tcs.SetResult(null);
            }
        }

        class ReleasingDisposable : IDisposable
        {
            AsyncSemaphore self;

            public ReleasingDisposable(AsyncSemaphore self) => this.self = self;

            public void Dispose() => self.Release();
        }
    }

我得到这个输出：

Beginning work #1 1
Beginning work #2 1
Beginning work #3 1
Beginning work #4 1
Finished work #4 1
Finished work #3 1
Finished work #2 1
Finished work #1 1
Beginning work #5 1
Beginning work #6 1
Beginning work #7 1
Beginning work #8 1
Finished work #5 1
Beginning work #9 1
Finished work #8 1
Finished work #7 1
Finished work #6 1
Beginning work #10 1
Finished work #9 1
Finished work #10 1
All done.

确实，我最多有 4 个任务在运行，而且都在同一个线程上运行。

【问题讨论】：

你看过SemaphoreSlim吗？它有一个 WaitOneAsync 方法。
@ScottChamberlain 不，SemaphoreSlim 类是多线程的信号量。我的问题是关于信号量的类似单线程概念。
SemaphoreSlim 在单线程上同样有效。如果同一个线程连续多次调用WaitAsync，它将继续递减计数器直到它达到0，然后暂停执行直到调用Release。见this answer。
一目了然，是的。我可以看到的主要问题是SemaphoreSlim 不保证 FIFO，尽管在单个线程上使用它可能会提供它。
检查the source 看起来它在内部锁定了所有等待和释放，然后使用链表来跟踪服务员，因此在.net 的当前实现上应该是 FIFO 排序。

标签： .net asynchronous async-await task-parallel-library

【解决方案1】：

所以即使只有一个线程，我们也有“并行”任务

为了避免与Parallel / Parallel LINQ 混淆，我通常更喜欢术语“并发”。

我的问题是：这个东西是否已经可用，最好是在 .NET 框架本身中？

是的。 SemaphoreSlim 是一个可以同步或异步使用的信号量。

我在 NuGet 上也有一个 full suite of asynchronous coordination primitives，其灵感来自 Stephen Toub's blog post series on the subject。我的原语都是同步和异步兼容的（并且是线程安全的），这在例如资源的一个用户是同步的但其他用户是异步的情况下很有用。

【讨论】：