Сoncurrent collections за 10 минут

^{Photo by Robert V. Ruggiero}

Тема не новая. Но задавая вопрос “что такое concurrent collections и когда их использовать?” на собеседовании или code review, я почти всегда получаю ответ, состоящий из одного предложения: “они полностью защищают нас от race conditions” (что невозможно даже в теории). Или: “это как обычные коллекции, но там внутри все на lock-ах”, что тоже не совсем соответствует действительности.

Цель данной статьи — разобрать тему за 10 минут. Будет полезно для быстрого знакомства с некоторыми тонкостями. Или чтобы освежить память перед собеседованием.

Прежде всего мы бегло посмотрим на содержимое пространства имен System.Collections.Concurrent. Затем обсудим основные отличия concurrent и классических коллекций, отметим некоторые неочевидные моменты. В завершение обсудим возможные подводные камни и когда какие типы коллекций стоит использовать.

Что находится в System.Collections.Concurrent

Intellisense подсказывает, что немного:



Давайте кратко обсудим назначение каждого класса.

ConcurrentDictionary: потокобезопасная коллекция общего назначения, применимая в самом широком спектре сценариев.

ConcurrentBag, ConcurrentStack, ConcurrentQueue: коллекции специального назначения. “Специальность” заключается в следующих моментах:

• Отсутствие API для доступа к произвольному элементу
• Stack и Queue (как все мы знаем) имеют заданный порядок добавления и извлечения элементов
• ConcurrentBag для каждого потока поддерживает собственную коллекцию для добавления элементов. При извлечении он “крадет” элементы из соседнего потока, если у текущего потока коллекция пуста

IProducerConsumerCollection — контракт используемый классом BlockingCollection (см. ниже). Реализован коллекциями ConcurrentStack, ConcurrentQueue и ConcurrentBag.

BlockingCollection — используется в сценариях, когда одни потоки заполняют коллекцию, а другие извлекают из нее элементы. Типичный пример — пополняемая очередь задач. Если в момент запроса очередного элемента коллекция пуста, то читающая сторона переходит в состояние ожидания нового элемента (polling). Вызвав метод CompleteAdding() мы можем указать, что коллекция больше не будет пополняться, тогда при чтении polling выполняться не будет. Проверить состояние коллекции можно с помощью свойств IsAddingCompleted (true если данные больше не будут добавляться) и IsCompleted (true если данные больше не будут добавляться и коллекция пуста).

Partitioner, OrderablePartitioner, EnumerablePartitionerOptions — базовые конструкции для реализации сегментирования коллекций. Используется методом Parallel.ForEach для указания, каким образом распределять элементы по обрабатывающим потокам.

Далее в статье мы сосредоточимся именно на коллекциях: ConcurrentDictionary и ConcurrentBag/Stack/Queue.

Отличия concurrent и классических коллекций

Защита внутреннего состояния

Классические коллекции спроектированы с прицелом на максимальную производительность, поэтому их экземплярные методы не гарантируют потокобезопасность.

Например, взглянем на исходный код метода Dictionary.Add.
Мы можем увидеть следующие строки (код упрощен для читаемости):

if (this._buckets == null)
{
    int prime = HashHelpers.GetPrime(capacity);
    this._buckets = new int[prime];
    this._entries = new Dictionary<TKey, TValue>.Entry[prime];
}

Как мы видим, внутреннее состояние словаря не защищено. При добавлении элементов из множества потоков, возможен следующий сценарий:

1. Поток 1 вызвал Add, исполнение остановилось сразу после входа в условие if
2. Поток 2 вызвал Add, инициализировал коллекцию, добавил элемент
3. Поток 1 вернулся к работе, заново проинициализировал коллекцию, тем самым уничтожив добавленные потоком 2 данные.

То есть для записи из нескольких потоков классические коллекции непригодны.

API толерантно к текущему состоянию коллекции

Как мы знаем, в Dictionary нельзя добавлять повторяющиеся ключи. Если мы дважды вызовем Add с одинаковым ключом, то второй вызов бросит ArgumentException.

Подобная защита полезна в однопоточных сценариях. Но при многопоточности мы не можем быть уверены в текущем состоянии коллекции. Естественно, проверки вроде следующей спасают нас только при постоянном оборачивании в lock:

if (!dictionary.ContainsKey(key))
{
    dictionary.Add(key, “Hello”);
}

API, основанное на исключениях является неудачным вариантом и не позволит реализовать стабильное, предсказуемое поведение при многопоточных сценариях. Взамен необходимо API не делающее предположений о текущем состоянии коллекции, не выбрасывающее исключений и оставляющее решение о допустимости того или иного состояния вызывающей стороне.

В concurrent коллекциях API построено на паттерне TryXXX. Вместо привычных Add, Get и Remove мы используем методы TryAdd, TryGetValue и TryRemove. И, если данные методы возвращают false, то мы сами решаем, является это исключительной ситуацией или нет.

Стоит отметить, что классические коллекции сейчас тоже имеют толерантные к состоянию методы. Но в классических коллекциях подобное API это приятное дополнение, а в concurrent коллекциях — необходимость.

API, минимизирующее race conditions

Рассмотрим простейшую операцию обновления элемента:

dictionary[key] += 1;

При всей простоте, код выполняет целых три действия: получает значение из коллекции, прибавляет 1, записывает новое значение. При многопоточном исполнении возможна ситуация, когда код извлек значение, выполнил инкремент, после чего благополучно затер значение, которое было записано другим потоком, пока выполнялся инкремент.

Для решения подобных проблем в API concurrent коллекций содержится целый ряд вспомогательных методов. Например, метод TryUpdate, который принимает три параметра: ключ, новое значение и ожидаемое текущее значение. Если значение в коллекции не будет соответствовать ожидаемому, то обновление не будет выполнено и метод вернет false.

Рассмотрим еще пример. Буквально каждая строка следующего кода (включая Console.WriteLine) может стать причиной проблем при многопоточном исполнении:

if (dictionary.ContainsKey(key))
{
    dictionary[key] += 1;
}
else
{
    dictionary.Add(key, 1);
}
Console.WriteLine(dictionary[key]);

Добавить или обновить значение, а затем провести с результатом какую-то операцию это довольно типичная задача. Поэтому у concurrent dictionary реализован метод AddOrUpdate, выполняющий последовательность действий за один вызов и потокобезопасно:

var result = dictionary.AddOrUpdate(key, 1, (itemKey, itemValue) => itemValue + 1);
Console.WriteLine(result);

Тут есть один момент, о котором стоит знать.

Реализация метода AddOrUpdate вызывает метод TryUpdate, описанный выше, и передает ему актуальное значение из коллекции. Если обновление не удалось (соседний поток уже изменил значение), то попытка повторяется и переданный update-делегат вызывается еще раз с обновленным текущим значением. То есть update-делегат может быть вызван многократно, поэтому он не должен содержать никаких сайд-эффектов.

Lock free алгоритмы и гранулированные блокировки

Microsoft отлично поработал над производительностью concurrent collections, а не просто обернул все операции lock-ами. Изучая исходники можно увидеть множество примеров использования гранулированных блокировок, использования грамотных алгоритмов вместо блокировок, а также использование специальных инструкций и более “легких” примитивов синхронизации, чем Monitor.

Чего не дают concurrent коллекции

Из примеров выше очевидно, что concurrent коллекции не дают полной защиты от race conditions и мы должны проектировать свой код соответствующим образом. Но это не все, есть еще пара моментов, о которых стоит знать.

Полиморфизм с классическими коллекциями

Concurrent-коллекции, как и классические, реализуют интерфейсы IDictionary, ICollection и IEnumerable. Но часть API этих интерфейсов не может быть потокобезопасной по определению. Например, метод Add, который мы разбирали выше.

Concurrent коллекции реализуют такие контракты без потокобезопасности. А чтобы “спрятать” небезопасное API, они используют явную реализацию интерфейсов. Об этом стоит помнить, когда мы передаем concurrent коллекции в методы принимающие на вход, например, ICollection.

Также concurrent коллекции не соблюдают Liskov substitution principle в отношении классических коллекций.

Например, содержимое классической коллекции нельзя модифицировать во время перебора, следующий код выбросит InvalidOperationException для класса List:

foreach (var element in list)
{
    list.Remove(element);
}

Если же говорить о concurrent коллекциях, то модификация в момент перебора не приводит к исключению, чтобы мы могли выполнять одновременные чтение и запись из разных потоков.

Более того, concurrent коллекции по-разному реализуют возможность модификации во время перебора. ConcurrentDictionary просто не выполняет никаких проверок и не гарантирует результат перебора, а ConcurrentStack/Queue/Bag накладывают блокировку и создают копию текущего состояния, по которой и выполняется перебор.

Возможные проблемы с performance

Выше мы упоминали что ConcurrentBag может “красть” элементы у соседних потоков. Это может обернуться проблемами с performance, если писать и читать в ConcurrentBag из разных потоков.

Также concurrent коллекции накладывают полные блокировки при запросе состояния всей коллекции (Count, IsEmpty, GetEnumerator, ToArray и т.п.) и потому существенно медленнее своих классических собратьев.

Вывод: использовать concurrent коллекции стоит только в том случае, если они действительно необходимы, поскольку этот выбор не “бесплатный”.

Когда какие типы коллекций использовать

• Однопоточные сценарии: только классические коллекции с лучшим performance.
• Запись из множества потоков: только concurrent коллекции, защищающие внутреннее состояние и имеющие подходящее для конкурентной записи API.
• Чтение из множества потоков: однозначных рекомендаций нет. Concurrent коллекции, могут создать проблемы с performance при интенсивных запросах состояния всей коллекции. Однако для классических коллекций Microsoft не гарантирует работоспособность даже для операций чтения. Например, внутренняя реализация коллекции может иметь lazy-свойства, инициируемые при чтении данных и, следовательно, возможно разрушение внутреннего состояния при чтении из множества потоков. Хорошим усредненным вариантом является использование immutable коллекций.
• И чтение и запись из нескольких потоков: однозначно concurrent коллекции, как реализующие защиту состояния и безопасное API.

Выводы

В статье мы бегло изучили concurrent collections, когда их стоит использовать и какую они имеют специфику. Разумеется, статья не исчерпывает тему и при серьезной работе с многопоточными коллекциями стоит копать глубже. Простейший способ сделать это — посмотреть исходники используемых коллекций. Это познавательно и совсем не сложно, код очень и очень читабелен.