ConcurrentDictionary.GetOrAdd 真的是线程安全的吗? [英] Is ConcurrentDictionary.GetOrAdd truly thread-safe?

问题描述

我有这段代码，如果该任务是为相同的输入创建的，我想在其中等待正在进行的任务.这是我正在做的事情的最小复制.

I have this piece of code where I want to await on a ongoing task if that task was created for the same input. Here is minimal reproduction of what I'm doing.

private static ConcurrentDictionary<int, Task<int>> _tasks = new ConcurrentDictionary<int, Task<int>>();

private readonly ExternalService _service;


public async Task SampleTask(){
  var result = await _service.DoSomething();
  await Task.Delay(1000) //this task takes some time do finish
  return result;
}

public async Task<int> DoTask(int key) {
   var task = _tasks.GetOrAdd(key, _ => SampleTask());
   var taskResult = await task;
   _tasks.TryRemove(key, out task);
   return taskResult;
}

我正在编写一个测试，以确保当多个请求想要(大致)同时执行任务时等待相同的任务.我通过模拟 _service 并计算 _service.DoSomething() 被调用的次数来做到这一点.如果对 DoTask(int key) 的调用大致在同一时间进行，则应该只有一次.

I'm writing a test to ensure the same task is awaited when multiple requests want to perform the task at (roughly) the same time. I'm doing that by mocking _service and counting how many times _service.DoSomething() is being called. It should be only once if the calls to DoTask(int key) where made at roughly the same time.

然而，结果告诉我，如果我多次调用 DoTask(int key) 并且调用之间的延迟小于 1~2ms，两个任务都会在SampleTask() 用第二个替换字典中的第一个.

However, the results show me that if I call DoTask(int key) more than once with a delay between calls of less than 1~2ms, both tasks will create and execute its on instance of SampleTask() with the second one replacing the first one in the dictionary.

考虑到这一点，我们能说这个方法是真正线程安全的吗?或者我的问题本身不是线程安全的问题吗?

Considering this, can we say that this method is truly thread-safe? Or isn't my problem a case of thread-safety per se?

推荐答案

引用 文档(重点是我的):

To quote the documentation (emphasis mine):

对于字典的修改和写入操作，ConcurrentDictionary 使用细粒度锁来确保线程安全.(对字典的读取操作以无锁方式执行.)但是，valueFactory 委托是在锁外调用的，以避免在锁下执行未知代码时可能出现的问题.因此，GetOrAdd 对于 ConcurrentDictionary 类上的所有其他操作不是原子的.

For modifications and write operations to the dictionary, ConcurrentDictionary<TKey,TValue> uses fine-grained locking to ensure thread safety. (Read operations on the dictionary are performed in a lock-free manner.) However, the valueFactory delegate is called outside the locks to avoid the problems that can arise from executing unknown code under a lock. Therefore, GetOrAdd is not atomic with regards to all other operations on the ConcurrentDictionary<TKey,TValue> class.

由于当 valueFactory 生成值时，另一个线程可以插入一个键/值，你不能相信仅仅因为 valueFactory 执行，它产生的值会被插入进入字典并返回.如果您在不同线程上同时调用GetOrAdd，valueFactory 可能会被多次调用，但只会将一个键/值对添加到字典中.

Since a key/value can be inserted by another thread while valueFactory is generating a value, you cannot trust that just because valueFactory executed, its produced value will be inserted into the dictionary and returned. If you call GetOrAdd simultaneously on different threads, valueFactory may be called multiple times, but only one key/value pair will be added to the dictionary.

因此，虽然字典是线程安全的，但调用 valueFactory 或 _ =>SampleTask() 在您的情况下，不能保证是唯一的.所以你的工厂函数应该能够接受这个事实.

So while the dictionary is properly thread-safe, calls to the valueFactory, or _ => SampleTask() in your case, are not guaranteed to be unique. So your factory function should be able to live with that fact.

您可以从源头确认这一点:

public TValue GetOrAdd(TKey key, Func<TKey, TValue> valueFactory)
{
    if (key == null) throw new ArgumentNullException("key");
    if (valueFactory == null) throw new ArgumentNullException("valueFactory");

    TValue resultingValue;
    if (TryGetValue(key, out resultingValue))
    {
        return resultingValue;
    }
    TryAddInternal(key, valueFactory(key), false, true, out resultingValue);
    return resultingValue;
}

如您所见，valueFactory 是在负责正确锁定字典的 TryAddInternal 之外调用的.

As you can see, valueFactory is being called outside of TryAddInternal which is responsible of locking the dictionary properly.

但是，由于 valueFactory 是一个 lambda 函数，它在您的情况下返回一个任务 (_ => SampleTask())，并且字典不会等待该任务本身，该函数将快速完成，并在遇到第一个await(异步状态机设置时)后返回未完成的Task.所以除非调用非常快，否则应该非常快地将任务添加到字典中，随后的调用将重复使用相同的任务.

However, since valueFactory is a lambda function that returns a task in your case (_ => SampleTask()), and the dictionary will not await that task itself, the function will finish quickly and just return the incomplete Task after encountering the first await (when the async state machine is set up). So unless the calls are very quickly after another, the task should be added very quickly to the dictionary and subsequent calls will reuse the same task.

如果您要求在所有情况下只发生一次，您应该考虑自己锁定任务创建.因为它会很快完成(不管你的任务实际需要多长时间来解决)，锁定不会有太大的伤害.

If you require this to happen just once in all cases, you should consider locking on the task creation yourself. Since it will finish quickly (regardless of how long your task actually takes to resolve), locking will not hurt that much.

这篇关于ConcurrentDictionary.GetOrAdd 真的是线程安全的吗?的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！