可以将其当前计数减少 N (N>=1) 的信号量？

Question

我正在实现一个限制可以发送的最大请求数的流量控制组件。每个工作线程都可以发送单个请求或一批请求，但任何时候待处理请求的总数都不应超过最大数量。

我最初想用 SemaphoreSlim 实现： 将信号量初始化为最大请求计数，然后当工作线程要调用服务时，它必须获取足够的令牌计数，但是我发现实际上 SemaphoreSlim 和 Semaphore 只允许线程将信号量计数减少 1，在我的情况下，我想通过工作线程携带的请求数来减少计数。

我应该在这里使用什么同步原语？

澄清一下，该服务支持批处理，因此一个线程可以在一次服务调用中发送 N 个请求，但相应地它应该能够将信号量的当前计数减少 N。

Answer 1

下面是一个自定义的SemaphoreManyFifo 类，它提供了Wait(int acquireCount) 方法和Release(int releaseCount) 方法。它的行为是严格的先进先出。它具有相当不错的性能（在我的 PC 的 8 个线程上每秒进行约 500,000 次操作）。

public class SemaphoreManyFifo : IDisposable
{
    private readonly object _locker = new object();
    private readonly Queue<(ManualResetEventSlim, int AcquireCount)> _queue;
    private readonly ThreadLocal<ManualResetEventSlim> _pool;
    private readonly int _maxCount;
    private int _currentCount;

    public int CurrentCount => Volatile.Read(ref _currentCount);

    public SemaphoreManyFifo(int initialCount, int maxCount)
    {
        // Proper arguments validation omitted
        Debug.Assert(initialCount >= 0);
        Debug.Assert(maxCount > 0 && maxCount >= initialCount);
        _queue = new Queue<(ManualResetEventSlim, int)>();
        _pool = new ThreadLocal<ManualResetEventSlim>(
            () => new ManualResetEventSlim(false), trackAllValues: true);
        _currentCount = initialCount;
        _maxCount = maxCount;
    }
    public SemaphoreManyFifo(int initialCount) : this(initialCount, Int32.MaxValue) { }

    public void Wait(int acquireCount)
    {
        Debug.Assert(acquireCount > 0 && acquireCount <= _maxCount);
        ManualResetEventSlim gate;
        lock (_locker)
        {
            Debug.Assert(_currentCount >= 0 && _currentCount <= _maxCount);
            if (acquireCount <= _currentCount && _queue.Count == 0)
            {
                _currentCount -= acquireCount; return; // Fast path
            }
            gate = _pool.Value;
            gate.Reset(); // Important, because the gate is reused
            _queue.Enqueue((gate, acquireCount));
        }
        gate.Wait();
    }

    public void Release(int releaseCount)
    {
        Debug.Assert(releaseCount > 0);
        lock (_locker)
        {
            Debug.Assert(_currentCount >= 0 && _currentCount <= _maxCount);
            if (releaseCount > _maxCount - _currentCount)
                throw new SemaphoreFullException();
            _currentCount += releaseCount;
            while (_queue.Count > 0 && _queue.Peek().AcquireCount <= _currentCount)
            {
                var (gate, acquireCount) = _queue.Dequeue();
                _currentCount -= acquireCount;
                gate.Set();
            }
        }
    }

    public void Dispose()
    {
        foreach (var gate in _pool.Values) gate.Dispose();
        _pool.Dispose();
    }
}

在上述实现中添加对超时和取消的支持并非易事。它需要一个不同的（可更新的）数据结构而不是Queue<T>。

---

原始的Wait+Pulse 实现可以在这个答案的1st revision 中找到。它很简单，但缺少理想的 FIFO 行为。

Answer 2

在我看来你想要类似的东西

using System;
using System.Collections.Generic;
using System.Threading;

namespace Sema
{
    class Program
    {
        // do a little bit of timing magic
        static ManualResetEvent go = new ManualResetEvent(false);

        static void Main()
        {
            // limit the resources
            var s = new SemaphoreSlim(30, 30);

            // start up some threads
            var threads = new List<Thread>();
            for (int i = 0; i < 20; i++)
            {
                var start = new ParameterizedThreadStart(dowork);
                Thread t = new Thread(start);
                threads.Add(t);
                t.Start(s);
            }

            go.Set();

            // Wait until all threads finished
            foreach (Thread thread in threads)
            {
                thread.Join();
            }
            Console.WriteLine();
        }

        private static void dowork(object obj)
        {
            go.WaitOne();
            var s = (SemaphoreSlim) obj;
            var batchsize = 3;

            // acquire tokens
            for (int i = 0; i < batchsize; i++)
            {
                s.Wait();
            }

            // send the request
            Console.WriteLine("Working on a batch of size " + batchsize);
            Thread.Sleep(200);

            s.Release(batchsize);
        }
    }
}

但是，您很快就会发现这会导致死锁。您还需要在信号量上进行一些同步，以保证一个线程要么获取其所有令牌，要么不获取任何令牌。

        var trylater = true;
        while (trylater)
        {
            lock (s)
            {
                if (s.CurrentCount >= batchsize)
                {
                    for (int i = 0; i < batchsize; i++)
                    {
                        s.Wait();
                    }

                    trylater = false;
                }
            }

            if (trylater)
            {
                Thread.Sleep(20);
            }
        }

现在，这可能会遭受饥饿的折磨。在发出数百个单个请求时，大量（例如 29 个）可能永远无法获得足够的资源。