Как лучше организовать многопоточное добавление данных в словарь (Dictionary)?

Question

[Андрей]

Есть обычный словарь (ключ-значение):
Dictionary<int, string> dict

В словарь с частотой в несколько миллисекунд добавляются строки из трёх-пяти потоков (Threads) простым методом присваивания:
dict[i] = "какая-то строка";

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

Стали терзать сомнения, открыл статью Албахари про параллелизм, и он в данном случае рекомендует использовать тип ConcurrentDictionary для этого вместо обычного Dictionary (но оговаривает что сие будет медленнее - особенно когда периодически вызывается dict.Count), либо использовать вот такую конструкцию добавления/обновления (говорит что она быстрее в 3 раза):

lock (dict)
{
    dict[i] = "что-то";
}

А в третьем источнике Эндрю Лок рекомендует использовать Interlocked.Exchange.

Чей метод будет лучше в плане обеспечения наилучшей целостности данных и скорости для поставленной задачи? Или может для таких случаев подойдут in-memory базы данных (вроде как они лучше всего ориентированы на такие многопоточные добавления/обновления)?

Comments

[Василий Банников]

Как эти данные потом используются?
Как ещё вариант - всем пишущим выдать ChannelWriter и кидать в канал команду типа "записать по ключу x значение y"
И на выделенном потоке звать в цикле ChannelReader и работать с этим словарём.

Но надо смотреть на то, как эти данные потом используются.

[Андрей]

Василий Банников, в словаре данные накапливаются до 1000 значений и как только накопилось, то отдельной асинхронной задачей записываются в PostgreSQL. При этом словарь очищается и снова накапливает значения.

[Василий Банников]

soundie, тогда вариант с каналом выглядит как решение.
Советую попробовать System.Threading.Channels.

Из N потоков будешь писать в канал операции, а в N+1 потоке накапливать в обычный словарь записи и отправлять их в постгрес.

[Василий Банников]

soundie, А в этих других потоках читать нужно данные, которые уже записаны в словарь?

[Андрей]

Василий Банников, нет, 3-5 потоков только записывают данные, а главный поток отслеживает Count и сохраняет в базу при превышении 1000.

[Василий Банников]

soundie, тогда точно канал подходит.
Ещё и никаких цикличных проверок на Count не придётся делать - сразу как набралось можно будет отправлять.

[Андрей]

Василий Банников, у Эндрю Лока (вроде бы у него) тоже про каналы как-то давно вычитывал, но побоялся из-за неопытности реализовывать. Теперь уверенности добавилось - рассмотрю эту тему, благодарю!

[Василий Банников]

soundie, а что там сложного? На отправляющей стороне await ch.Send, а на принимающей - await foreach(var msg = ch.ReadToEnd())

[Андрей]

Василий Банников, нашёл где впервые увидел теорию про каналы (автор не Лок, но Лок как раз этот курс и рекомендовал) - там всё было реализовано в достаточно сложном исполнении с привлечением даже сервиса очередей сообщений Amazon SQS, поэтому вся эта сложность сразу и оттолкнула. Но сейчас уже вижу что да - тема вполне несложная, просто не с той стороны подошёл с ней впервые знакомиться :)

Answer 1

[mayton2019]

Несколько мыслей.

1. In-memory базы данных будут медленнее чем C# структуры данных. Им ведь надо перекидывать данные через границу процесса. Память не шарится соотв - сериализация-сетевые сокеты пускай даже локальные. Все это будет медленнее.

2. Я не знаю ни одной структуры данных которая-бы себя хорошо вела при конкуретнтной вставке. Concurrent - коллекции нужно только в том случае когда результат вставки в ту-же милисекунду нужен вам на чтение как результат из другого потока. Это очень строгое требование и реально очень мало систем им обладают. Я-бы предложил следующее. Если 5 потоков пишут независимо без обратной связи - то пускай пишут в буферы в несколько килобайт. И пускай 6 поток периодически собирает эти буферы. Так можно уменьшить конкуренцию. Лаг можно регулировать. Сколько хотим? 10 мс? 100мс?

Еще вариант - потоки пишут каждый в свой Dictionary. И периодически происходит merge. Пакетом. Тут надо померять performance. Вообще в конкурирующих работах с key-value ищут возможность сделать partitioning. Если удасться то будут конкурировать максимум 2 потока а не 5.

Answer 2

[Василий Банников]

многопоточное добавление данных в словарь

Dictionary не потокобезопасен, по тому при обращении к нему из нескольких потоков нужно синхронизировать потоки (желательно при любых обращениях)

Лучше использовать ConcurrentDictionary - в нём уже решены потенциальные проблемы при конкурентном доступе (например попытке изменить запись, которая уже удалена в другом потоке или удаление удалённой, или создании уже существующей, или чтении несуществующей/удалённой)

Interlocked exchange лучше смотреть по месту, может подойдёт, а может нет.

На счёт инмемори базы тоже следует посмотреть на конкретную задачу.

UPD: ещё как хорошее решение звучит использование System.Threading.Channels, раз задача стоит накапливать записи и батчем их отправлять в СУБД.
Так мы гарантированно решаем проблему гонок, так как фактически менять данные будет только один поток