c# - 没有等待时删除信号量

Question

这是一个并发问题:

字符串值用于表示抽象资源，并且给定的字符串值仅允许一个线程工作。但是如果它们的字符串值不同，则多个线程可以并发运行。到目前为止，非常简单:

private static readonly Dictionary<String,Object> _locks = new Dictionary<String,Object>();

public static void DoSomethingMutuallyExclusiveByName(String resourceName) {

    Object resourceLock;
    lock( _locks ) {

        if( !_locks.TryGetValue( resourceName, out resourceLock ) ) {
            _locks.Add( resourceName, resourceLock = new Object() );
        }
    }

    lock( resourceLock ) {

        EnterCriticalSection( resourceName );
    }
}

但这不是次优的:resourceName的域是无界的，并且_locks最终可能包含数千个或更多字符串。因此，在没有更多线程正在使用特定的resourceName值之后，则应从字典中删除其锁定对象。

由于下面的这种情况(相关代码位于下面)，仅在使用了锁定对象后删除该对象将是一个错误。请注意，所有三个线程都具有resourceName = "foo"。

线程1位于labelC，刚刚从resourceName = "foo"字典中删除了它的_locks，因此删除了resourceLock (#1)。

线程2位于labelB上，因为没有其他线程锁定在resourceLock (#1)上，因此它不等待lock( resourceLock而是继续进入EnterCriticalSection。

线程3位于labelA上，因为resourceName = "foo"不在_locks中(因为线程1删除了它)，所以它向_locks中添加了一个新实例，并且由于此resourceLock (#2)是一个新实例，因此线程3中的lock( resourceLock )将不等待线程2，因此两者线程2和3可以在EnterCriticalSection内，具有相同的resourceName值。

代码:

public static void DoSomethingMutuallyExclusiveByName(String resourceName) {

labelA:        
    Object resourceLock;
    lock( _locks ) {

        if( !_locks.TryGetValue( resourceName, out resourceLock ) ) {
            _locks.Add( resourceName, resourceLock = new Object() );
        }
    }

    try {

labelB:
        lock( resourceLock ) {

            EnterCriticalSection( resourceName );
        }
    }
    finally {

        lock( _locks ) {
            _locks.Remove( resourceName );
        }
labelC:
    }

}

我最初用自己的小技巧解决了这个问题:

class CountedLock {
    public Int32 Count;
}

private static readonly Dictionary<String,CountedLock> _locks = new Dictionary<String,CountedLock>();

public static void DoSomethingMutuallyExclusiveByName(String resourceName) {

labelA:        
    CountedLock resourceLock;
    lock( _locks ) {

        if( !_locks.TryGetValue( resourceName, out resourceLock ) ) {
            _locks.Add( resourceName, resourceLock = new CountedLock() { Count = 1 } );
        }
        else {
            resourceLock.Count++; // no need for Interlocked.Increment as we're already in a mutex code block
        }
    }

    try {

labelB:
        lock( resourceLock ) {

            EnterCriticalSection( resourceName );
        }
    }
    finally {

        lock( _locks ) {
labelD:
            if( --resourceLock.Count == 0 ) {
                _locks.Remove( resourceName );
            }
        }
labelC:
    }

}

这样就解决了问题。假设三个线程与以前位于相同的位置:

线程1并未从resourceName中删除其resourceLock _locks值，因为线程2也引用了与线程1相同的resourceLock，因此，当线程1位于2时，关联计数为labelD(当线程1到达时，计数变为1。 labelC)。

线程2继续进入关键部分，因为它到达线程3之前。

如果在线程2仍处于关键部分时线程3(位于labelA)进入TryGetValue，则将看到其resourceName="foo"仍在_locks字典中，因此获得与线程2相同的实例，因此它将在labelB中等待线程2完成。

这样行得通!

但是我想您现在正在思考:

"So you have a lock with an associated count... sounds like you just re-invented semaphores. Don't re-invent the wheel, use System.Threading.Sempahore or SemaphoreSlim.

确实，所以我更改了代码以使用SemaphoreSlim-SemaphoreSlim实例具有内部Count值，该值是允许进入的线程数(与当前“在”信号量内部的线程数相反”)，这与方法相反我的CountedLock示例在上一个示例中有效):

private static readonly Dictionary<String,SemaphoreSlim> _locks = new Dictionary<String,SemaphoreSlim>();

public static void DoSomethingMutuallyExclusiveByName(String resourceName) {

labelA:
    SemaphoreSlim resourceLock;
    lock( _locks ) {
        if( !_locks.TryGetValue( resourceName, out resourceLock ) {
            _locks.Add( resourceName, resourceLock = new SemaphoreLock( initialCount: 1, maxCount: 1 ) );
        }
    }

labelB:
    resourceLock.Wait(); // this decrements the semaphore's count

    try {
        EnterCriticalSection( resourceName );
    }
    finally {

        lock( _locks ) {
            Int32 count = resourceLock.Release(); // this increments the sempahore's count
            if( count > 0 ) {
                _locks.Remove( resourceName );
                resourceLock.Dispose();
            }
labelC:
        }
    }
}

...但是发现错误!

考虑这种情况:

线程1位于labelC处，刚刚删除并处置了其resourceLock (#1)实例。

线程2位于labelB(紧接在调用Wait之前)。它已经获得了对与线程1相同的SemaphoreSlim resourceLock (#1)实例的引用；但是由于Wait方法是在线程离开lock( _locks )下的labelA之后调用的，因此这意味着存在一个很小但存在的机会窗口，线程2随后将调用resourceLock (#1).Wait()(忽略可能的ObjectDisposedException)，而线程3(当前位于labelA )然后将输入TryGetValue并为相同的SemaphoreLock (#2)实例化一个resourceName的新实例，但是由于线程2和线程3具有不同的信号量实例，它们都可能同时进入临界区。

您可能会建议:

You should just find a way to decrement the semaphore while you're inside the lock( _locks ) block under labelA

...除了SemaphoreSlim类不公开任何Decrement方法。您可以调用.Wait(0)使其立即返回，因此我的代码如下所示:

[...]
labelA:
    SemaphoreSlim resourceLock;
    lock( _locks ) {
        if( !_locks.TryGetValue( resourceName, out resourceLock ) {
            _locks.Add( resourceName, resourceLock = new SemaphoreLock( initialCount: 1, maxCount: 1 ) );
        }
        resourceLock.Wait( 0 );
    }

labelB:
    resourceLock.Wait();
[...]

...除非这是行不通的。 Wait(Int32)状态的文档(重点是我的):

If a thread or task is blocked when calling Wait(Int32), and the time-out interval specified by millisecondsTimeout expires, the thread or task doesn’t enter the semaphore, and the CurrentCount property isn’t decremented.

...对于该理论而言如此之多。即使它确实起作用，在同一线程中两次调用Wait可能也会使计数递减两次，而不是一次。

那么有可能有一个由互斥锁保护的临界区，该互斥锁以某种方式“知道”何时不再需要它们？

Answer 1

实际上，我会坚持使用更简单的基于计数的解决方案，而不是SemaphoreSlim，因为无论如何您已经实现了它。虽然被称为“slim”，但SemaphoreSlim却比您的简单计数器轻巧。如您所知，使用信号量实际上会使代码的性能稍差一些，并且更加复杂。如果没有别的，如果花更多时间让自己确信该版本确实有效，那么也许不是更好的版本。

因此，也许您正在重新发明轮子，但是SemaphoreSlim是通用信号灯，具有您不需要的功能。甚至在SemaphoreSlim已经存在的情况下，Microsoft也会通过在BCL中添加Semaphore来重新发明轮子。

另一方面，如果您认为争用可能是全局锁定的问题，则可以尝试使用无锁方法。最有可能的是，您不会遇到此类问题，但是如果您真的认为在整个代码中都会调用数千次，则可以选择以下内容:

private static readonly ConcurrentDictionary<string, CountedLock> _locks 
    = new ConcurrentDictionary<string, CountedLock>();

public static void DoSomethingMutuallyExclusiveByName(string resourceName)
{
    CountedLock resourceLock;

    // we must use a loop to avoid incrementing a stale lock object
    var spinWait = new SpinWait();
    while (true)
    {
        resourceLock = _locks.GetOrAdd(resourceName, i => new CountedLock());
        resourceLock.Increment();

        // check that the instance wasn't removed in the meantime
        if (resourceLock == _locks.GetOrAdd(resourceName, i => new CountedLock()))
            break;

        // otherwise retry
        resourceLock.Decrement();
        spinWait.SpinOnce();
    }

    try
    {
        lock (resourceLock)
        {
            // EnterCriticalSection(resourceName);
        }
    }
    finally
    {
        if (resourceLock.Decrement() <= 0)
            _locks.TryRemove(resourceName, out resourceLock);
    }
}

随着CountedLock也被修改为使用Interlocked类:

class CountedLock
{
    Int32 _count;
    public int Increment() => Interlocked.Increment(ref _count);
    public int Decrement() => Interlocked.Decrement(ref _count);
}

无论哪种方式，我都可能将代码重组为通用代码，并(ab)使用IDisposable接口(interface)来允许您将调用简单地包装在单个using块中。