Concurrency структуры в .net. ConcurrentDictionary изнутри

Все началось с одного собеседования, которое и натолкнуло меня к написанию данной статьи. Довольно большая часть разработчиков на платформе .Net не понимает базовые вещи, хотя и использует их повседневно, например lock-ом оборачивают все методы, использующие ConcurrentDictionary, хотя можно было бы обойтись обычным Dictionary<>.

В науке существуют 3 основных способа реализации конкурентных структур данных:
• Lock-free структуры данных;
• Fine-grained блокировка;
• Transactional memory implementation(транзакционная память);

ConcurrentDictionary<TKey, TValue> — это thread-safe аналог Dictionary<TKey, TValue>. В его основе лежит, так называемый Fine-grained блокировка.

Общая информация

Для настройки есть 2 основных параметра:
• сapacity — первоначальное кол-во элементов. По умолчанию — 31.
• concurrencyLevel – предполагаемое число потоков на запись. По умолчанию выставляется в =4 * число процессоров

Предлагаю рассмотреть эти параметры более подробно для того что бы понять их истинную суть.

Сapacity

Данные параметр аналогичен тому, что используется в обычном «словаре» Dictionary<TKey, TValue> — это первоначальный размер массива для хранения элементов.
Если мы посмотрим исходники, то увидим строчку:

ConcurrentDictionary<TKey, TValue>.Node[] buckets = new ConcurrentDictionary<TKey, TValue>.Node[capacity];

Как известно, Dictionary – это классическая хеш-таблица, поэтому для хранения элементов используется хеш-значение, которое в .Net вычисляется с помощью функции GetHashCode() и имеет тип int. Значения хешей как раз и лежат в бакетах выше, нужный бакет вычисляется:

bucketNo = (hashcode & int.MaxValue) % bucketCount;

Отсюда и получается скорость доступа — O(1).
При переполнении, происходит увеличение массива с бакетами (newLength = oldLength * 2 + 1) и заново перераспределяются все элементы.

ConcurrencyLevel

Предполагаемое число потоков на запись. Из определения сразу возникает вопрос – а зачем «словарю» знать о количеств потоков. На самом деле все просто. Это не что иное, как количество независимых блокировок (пул блокировок), т.е. это максимальное число потоков, которые могут «писать» в «словарь» одновременно. По сути, каждой блокировке «отдается» примерно одинаковый набор бакетов, в которых она и обеспечивает синхронизацию потоков на запись.

Операции над ConcurrentDictionary.

Основные операции над словарем можно разделить на 3 группы:

• полностью неблокируемые;
• блокировка одного элемента из пула блокировок;
• блокировка всего словаря;

К полностью не блокируемым операциям можно отнести:

• ContainsKey
• TryGet
• this [ ]
• GetEnumerator – операция не обеспечивает целостность данных (не использует снепшоты), т.е. данные за время работы функции могут поменяться.

Все операции чтения (Get/ContainsKey) имеют примерно одинаковый алгоритм работы:

• вычисление хеша ключа через GetHashCode()
• вычисление бакета, в котором лежит наш элемент
• сравнения значения ключа в бакете с тем, который у нас
• чтение значения с использованием Volatile.Read

К операциям с блокировкой одного элемента из пула блокировок можно отнести:

• TryAdd
• TryUpdate
• TryRemove

Ниже примерный алгоритм работы:

1. Вычисление хеша ключа нового элемента
2. Вычисление бакета bucketNo, в который будет добавлен элемент, и номера блокировки из пула

   bucketNo = (hashcode & int.MaxValue) % bucketCount;
   lockNo = bucketNo % lockCount;
   

3. Блокировка bucketNo через Monitor.Enter
4. Запись элемента с использованием Volatile.Write
5. Освобождение блокировки Monitor.Exit

К самым неэффективным операциям, которые блокируют весь словарь, относятся:

• Count, IsEmpty. Да, эти операции требуют полной блокировки словаря. Если вам необходимо сохранить в лог-файл число элементов, то можно использовать GetEnumerator и LINQ.
• Keys, Values – получение списка ключей и списка значений соответственно. Кстати, тут получаются целостные данные – снепшоты.
• CopyTo – explicit ICollection
• Clear, ToArray

Методы AddOrUpdate, GetOrAdd

Эти методы довольно интересны тем, что они используют атомарные операции Add/Get/Update, но сами не являются атомарными, они не использую блокировку на всю операцию. Вот так выглядит реализация AddOrUpdate:

      do
      {
        while (!TryGetValue(key, out comparisonValue))
        {
          TValue obj = addValueFactory(key);
          TValue resultingValue;
          if (TryAddInternal(key, obj, false, true, out resultingValue))
            return resultingValue;
        }
        newValue = updateValueFactory(key, comparisonValue);
      }
      while (!TryUpdate(key, newValue, comparisonValue));
      return newValue;

MSDN:

Also, although all methods of ConcurrentDictionary are thread-safe, not all methods are atomic, specifically GetOrAdd and AddOrUpdate. The user delegate that is passed to these methods is invoked outside of the dictionary's internal lock. (This is done to prevent unknown code from blocking all threads.)

Пример использования:

 items.AddOrUpdate(srcKey, srcValue,
                (key, existingVal) =>
                {
                    // сравниваем добавляемое значение и значение из словаря
                    if (srcValue != existingVal)
                        throw new ArgumentException("...");
 

В конце, хочу отметить сложности методов:

1. TryAdd, TryUpdate, TryRemove, — O(1)
2. Get/Contains, Item[] по ключу — O(1)
3. ContainsValue, ToArray(), Keys, Values — O(n)

P.s. В данной статье я хотел обратить внимание на тонкости использования ConcurrentDictionary и на основные моменты в его реализации.

Автор: avsevolodov

Источник