信号量的多线程问题

Question

我需要有一段代码，该代码只允许基于参数键同时由 1 个线程执行：

    private static readonly ConcurrentDictionary<string, SemaphoreSlim> Semaphores = new();

    private async Task<TModel> GetValueWithBlockAsync<TModel>(string valueKey, Func<Task<TModel>> valueAction)
    {
        var semaphore = Semaphores.GetOrAdd(valueKey, s => new SemaphoreSlim(1, 1));

        try
        {
            await semaphore.WaitAsync();

            return await valueAction();
        }
        finally
        {
            semaphore.Release(); // Exception here - System.ObjectDisposedException
            if (semaphore.CurrentCount > 0 && Semaphores.TryRemove(valueKey, out semaphore))
            {
                semaphore?.Dispose();
            }
        }
    }

有时我会收到错误：

The semaphore has been disposed. : System.ObjectDisposedException: The semaphore has been disposed.
   at System.Threading.SemaphoreSlim.CheckDispose()
   at System.Threading.SemaphoreSlim.Release(Int32 releaseCount)
   at Project.GetValueWithBlockAsync[TModel](String valueKey, Func`1 valueAction)

我在这里可以想象的所有情况都是线程安全的。请帮忙，我错过了什么案例？

Answer 1

您在这里有一个线程竞赛，另一个任务正在尝试获取相同的信号量，并在您版本 - ie另一个线程等待semaphore.waitasync（）时获取它。 。针对CurrentCount的检查是一种竞赛条件，并且可以根据时间来进行任何任务。检查TryRemove的检查是无关紧要的，因为竞争性线程已经 删除了信号量 - 毕竟，等待waitaSync（）。

Answer 2

如评论中所述，您在这里有几个种族条件。

1. 线程1保持锁定，线程2正在等待waitasync（）。线程1释放锁，然后在Thread 2之前检查SmemAphore.CurrentCount 。
2. 线程1握住锁，将其释放，然后检查Semaphore.currentCount通过。线程2输入getValueWithBlockAsync，呼叫semaphores.getoradd并获取了信号量。线程1然后调用信号量。tryremove并递给信号量。

您确实需要锁定从Semaphores中删除条目的决定 - 这是没有办法的。您也没有一种方法来跟踪是否有任何线程从semaphores中获取了信号量（并且当前正在等待它，或者还没有达到这一点）。

一种方法是做这样的事情：拥有一个在每个人之间共享的锁，但是只有在获取/创建信号量并决定是否处置时才需要。我们手动跟踪目前有多少线程对特定信号量有兴趣。当线程发布信号量时，它将获取共享锁，以检查其他人当前是否对该信号量有兴趣，并只有在没有其他人的情况下才能将其处置。

private static readonly object semaphoresLock = new();
private static readonly Dictionary<string, State> semaphores = new();

private async Task<TModel> GetValueWithBlockAsync<TModel>(string valueKey, Func<Task<TModel>> valueAction)
{
    State state;
    lock (semaphoresLock)
    {
        if (!semaphores.TryGetValue(valueKey, out state))
        {
            state = new();
            semaphores[valueKey] = state;
        }
        
        state.Count++;
    }

    try
    {
        await state.Semaphore.WaitAsync();

        return await valueAction();
    }
    finally
    {
        state.Semaphore.Release();
        lock (semaphoresLock)
        {
            state.Count--;
            if (state.Count == 0)
            {
                semaphores.Remove(valueKey);
                state.Semaphore.Dispose();
            }
        }
    }
}

private class State
{
    public int Count { get; set; }
    public SemaphoreSlim Semaphore { get; } = new(1, 1);
}

当然，另一个选择是让信号量增长。也许您有定期操作可以进行并清除任何未使用的东西，但是这当然需要受到保护，以确保线程不会突然对被清除的信号量感兴趣。