мучаюсь с выбором Concurrent коллекции среди имеющихся в .net, при этом до конца не уверен, что в моем случае потокобезопасная коллекция мне необходима.
Имею следующий сценарий:
1. Не большое количество потоков (2-5) добавляют в коллекцию элементы, на каждом лежит ответственность по проверке уникальности добавляемого элемента.
2. Группа потоков выбирает из коллекции элементы, которые удовлетворяют ряду критериев, могут быть ситуации, при котором набор критериев может быть одинаковый у нескольких потоков. Выбрав элемент, поток производит его клонирование и продолжает работу с клоном. При завершении обработки содержимое элемента в коллекции может быть модифицировано, для предоставления актуализированной информации последующим потокам. Дополнительно, каждый из потоков-обработчиков маркирует элемент коллекции, о том, что он был им обработан.
3. Отдельный поток производит удаление элементов из коллекции. Критерии для удаления могут быть следующими: закончилось время жизни элемента либо все потоки с одинаковыми критериями выбора отметили элемент обработанным.
Помогите пожалуйста разобраться, определиться с выбором.
Спасибо!
Здравствуйте, RQ, Вы писали:
RQ>мучаюсь с выбором Concurrent коллекции среди имеющихся в .net, при этом до конца не уверен, что в моем случае потокобезопасная коллекция мне необходима.
Тут как минимум два подводных камня есть, с которыми сначала определиться надо.
RQ> Выбрав элемент, поток производит его клонирование и продолжает работу с клоном. RQ> При завершении обработки содержимое элемента в коллекции может быть модифицировано, для предоставления актуализированной информации последующим потокам.
Что нужно сделать делать в случае гонки? В смысле, два потока обработали клон одного и того же объекта и собираются скинуть исправления обратно, что делать, если обнаружен конфликт?
RQ>Отдельный поток производит удаление элементов из коллекции. Критерии для удаления могут быть следующими: закончилось время жизни элемента либо все потоки с одинаковыми критериями выбора отметили элемент обработанным.
Снова состояние гонки. В смысле, только что удалённый объект может быть снова добавлен, а может обломиться, если добавлен слишком рано. Это нормально?
Ну и последний — сколько элементов в самой коллекции и что именно является узким местом: собственно обработка, или куча поступающих запросов, которые надо отбрасывать?
Хороших вариантов пока предложить не могу, разве что аналог LMAX / любого другого ring buffer dispatch но он тут смотрится оверкиллом.
Здравствуйте, Sinix, Вы писали:
S>Здравствуйте, RQ, Вы писали:
RQ>>мучаюсь с выбором Concurrent коллекции среди имеющихся в .net, при этом до конца не уверен, что в моем случае потокобезопасная коллекция мне необходима. S>Тут как минимум два подводных камня есть, с которыми сначала определиться надо.
RQ>> Выбрав элемент, поток производит его клонирование и продолжает работу с клоном. RQ>> При завершении обработки содержимое элемента в коллекции может быть модифицировано, для предоставления актуализированной информации последующим потокам. S>Что нужно сделать делать в случае гонки? В смысле, два потока обработали клон одного и того же объекта и собираются скинуть исправления обратно, что делать, если обнаружен конфликт?
Данные актуализируются потоком, который завершил обработку позже. Те многочисленного обновления данных я не боюсь, ровно как и факта, что обработчик возьмет объект с потерявшими актуальность данными, тк в любом случае в результате обработки актуальность будет проверена и восстановлена. Эта часть необходима для отсечения объектов для "сильно отстающих" потоков.
RQ>>Отдельный поток производит удаление элементов из коллекции. Критерии для удаления могут быть следующими: закончилось время жизни элемента либо все потоки с одинаковыми критериями выбора отметили элемент обработанным. S>Снова состояние гонки. В смысле, только что удалённый объект может быть снова добавлен, а может обломиться, если добавлен слишком рано. Это нормально?
Я думал разрешить этот и предыдущий случай следующим образом:
1. поток забирает объект из коллекции, держит ссылку на него
2. создает клона, обрабатывает и модифицирует в процессе обработки клон
3. завершает обработку, проверяет флаг исходного объекта, был ли он помечен как удаленный? Если нет, то модифицирует содержание
Но чую, что так поступать совсем плохо?
S>Ну и последний — сколько элементов в самой коллекции и что именно является узким местом: собственно обработка, или куча поступающих запросов, которые надо отбрасывать?
Обработка, тк в процессе выполняются 5-8 http запросов, это на наиболее долгая по времени операция.
S>Хороших вариантов пока предложить не могу, разве что аналог LMAX / любого другого ring buffer dispatch но он тут смотрится оверкиллом.
И еще, общая (логическая) стратегия выглядит так, лучше пропустить больше сомнительных объектов, тем самым повысить количество обработанных.
Здравствуйте, RQ, Вы писали:
RQ>Обработка, тк в процессе выполняются 5-8 http запросов, это на наиболее долгая по времени операция.
Вот есть у меня подозрение, что при таком раскладе самым дешёвым вариантом будет lock-free буфер для закидывания задач + диспатчер, который периодически будет внутри lock забирать задачи из буфера и распихивать содержимое буфера с учётом текущего набора задач и условий от обработчиков. Обработчики по завершению скидывают задачи всё в тот же буфер.
Самый дубовый и надёжный вариант для "много пришло, мало обработалось".
RQ>Я думал разрешить этот и предыдущий случай следующим образом:
RQ>1. поток забирает объект из коллекции, держит ссылку на него...Но чую, что так поступать совсем плохо?
Не, почему, обычный подход, только вместо п.3 для обновления используют compare and swap loop. Взяли копию текущего состояния, обновили копию, если текущее состояние не поменялось — записали, иначе на новый круг.
RQ>Помогите пожалуйста разобраться, определиться с выбором. RQ>Спасибо!
Выглядит как задачка отлично помещающаяся в парадигму TPL-Dataflow, с маленькой особенностью — "флаги" исходного объекта сделать lock-free и шарить между потоками, "удалятор" сделать как фильтр (в самом начале конвейера или в каждом блоке), и по завершении обработки возвращать элемент в начало конвейера (вдруг по новой информации какой-то ещё обработчик решит что-то сделать).
Здравствуйте, RQ, Вы писали:
RQ>Добрый день,
RQ>мучаюсь с выбором Concurrent коллекции среди имеющихся в .net, при этом до конца не уверен, что в моем случае потокобезопасная коллекция мне необходима.
RQ>Имею следующий сценарий:
RQ>1. Не большое количество потоков (2-5) добавляют в коллекцию элементы, на каждом лежит ответственность по проверке уникальности добавляемого элемента. RQ>2. Группа потоков выбирает из коллекции элементы, которые удовлетворяют ряду критериев, могут быть ситуации, при котором набор критериев может быть одинаковый у нескольких потоков. Выбрав элемент, поток производит его клонирование и продолжает работу с клоном. При завершении обработки содержимое элемента в коллекции может быть модифицировано, для предоставления актуализированной информации последующим потокам. Дополнительно, каждый из потоков-обработчиков маркирует элемент коллекции, о том, что он был им обработан. RQ>3. Отдельный поток производит удаление элементов из коллекции. Критерии для удаления могут быть следующими: закончилось время жизни элемента либо все потоки с одинаковыми критериями выбора отметили элемент обработанным.
RQ>Помогите пожалуйста разобраться, определиться с выбором. RQ>Спасибо!
Вы описали алгоритм, а спрашиваете про структуру данных.
Попробуйте поискать готовые реализации алгоритмов — т.е. переформулировать вопрос
Здравствуйте, Sinix, Вы писали:
S>Здравствуйте, RQ, Вы писали:
S>Не, почему, обычный подход, только вместо п.3 для обновления используют compare and swap loop. Взяли копию текущего состояния, обновили копию, если текущее состояние не поменялось — записали, иначе на новый круг.
Спасибо, то что нужно!
S>Вот есть у меня подозрение, что при таком раскладе самым дешёвым вариантом будет lock-free буфер
Я правильно Вас понял, вы предлагаете создать собственную реализацию lock-free буфер'а? Те, каждый поток-обработчик блокирует доступ к коллекции с данными, при взятии (получении) элемента?
S>+ диспатчер, который периодически будет внутри lock забирать задачи из буфера и распихивать содержимое буфера с учётом текущего набора задач и условий от обработчиков. Обработчики по завершению скидывают задачи всё в тот же буфер.
Эту часть я к сожалению совсем не понимаю, не представляю. Те некий внешний (с точки зрения обрабочика) код, который допустим выполняется с определенным интервалом, будет проходить по очереди внутри lock'а и раздавать обработчикам необходимые элементы?
Здравствуйте, VladCore, Вы писали:
VC>Вы описали алгоритм, а спрашиваете про структуру данных.
Я описал сценарий работы множества потоков с данными и да, спрашиваю какая из стандартных структур данных лучше всего подойдет для описанного сценария.
Здравствуйте, hi_octane, Вы писали:
_>Выглядит как задачка отлично помещающаяся в парадигму TPL-Dataflow, с маленькой особенностью — "флаги" исходного объекта сделать lock-free и шарить между потоками, "удалятор" сделать как фильтр (в самом начале конвейера или в каждом блоке), и по завершении обработки возвращать элемент в начало конвейера (вдруг по новой информации какой-то ещё обработчик решит что-то сделать).
Как мне кажется TPL-Dataflow я не смогу использовать, тк все обработчики фактически являются Job'ами механизма расписаний quartz.net, Каждый внутри себя запускает поток для поиска и обработки элемента общей коллекции. Обработчики внутри довольно сильно отличаются друг от друга, хотя фактически выполняют одни и те же действия. Общую коллекцию заполняют несколько других Job'ов.
Не представляю как в таком разрезе выстроить общий конвейер. Делать свою систему запуска по расписанию тоже не хочется.
RQ>Эту часть я к сожалению совсем не понимаю, не представляю. Те некий внешний (с точки зрения обрабочика) код, который допустим выполняется с определенным интервалом, будет проходить по очереди внутри lock'а и раздавать обработчикам необходимые элементы?
Не, смысл такой:
1. Заводим неблокирующую очередь, в которую скидываются все задачи.
2. Заводим поток, который в цикле (по времени, или по прерыванию из п.1) загребает все задачи из буфера, раскидывает их по обработчикам (с учётом ранее забранных задач), убирает выполненные / отпавшие по таймауту задачи и снова засыпает. Понятно, что логику раскидывания стоит делать как можно более дешёвой, т.к. тут это основное узкое место.
3. Обработчики по завершению закидывают свои задачи в общую очередь, при следующем пробуждении поток в п.2 их обработает. По возможности делать обработку асинхронной, на тасках, чтобы сэкономить на блокировках при IO.
код в п.2 обычно и обзывается диспатчером.
Такой подход хорош для ситуаций, когда сами задачи стоят на порядки дороже, чем диспетчеризация — он максимально дубовый, при необходимости сам диспатчер можно распарралелить и от него относительно легко перейти к любому другому варианту.
Единственное сложное место — поиграться с временем засыпания / количеством задач, которые забираются из очереди.
Лучше всего подбирать эти значения динамическими — уменьшать время ожидания при постоянно забитой очереди и возвращать обратно при свободной, но это уже потом, после того как характер нагрузки будет понятен.
Здравствуйте, RQ, Вы писали:
RQ>Здравствуйте, VladCore, Вы писали:
VC>>Вы описали алгоритм, а спрашиваете про структуру данных.
RQ>Я описал сценарий работы множества потоков с данными и да, спрашиваю какая из стандартных структур данных лучше всего подойдет для описанного сценария.
Ну никаких ограничений в первом посте не написано. Подойдёт и List
Здравствуйте, RQ, Вы писали:
RQ>Здравствуйте, VladCore, Вы писали:
VC>>Вы описали алгоритм, а спрашиваете про структуру данных.
RQ>Я описал сценарий работы множества потоков с данными и да, спрашиваю какая из стандартных структур данных лучше всего подойдет для описанного сценария.
Заведи BlockingCollection<Action> для обработчиков и IList<T> для самих элементов. В начале в the BlockingCollection<Action> добавь все нужные экшны. Работа будет завершена когда в the BlockingCollection<Action> не останется ни одного экшна. Это если в лоб делать. Но обычно одну и туже коллекцию на input и output не используют для "a workflow/a processing". Нет таких задач в которых и на входе workflow и на выходе структуры данных одинаковые. Ну кроме сортировки.
Ты же в курсе в an Action из the BlockingCollection<Action> можно добавлять новые экшны?
Вообразим себе комнату с большой доской, рядом с которой находится группа людей, держащих в руках фрагменты изображения. Процесс начинают добровольцы, которые размещают на доске наиболее "вероятные" фрагменты изображения (предположим, что они прилепляются к доске). Далее каждый участник группы смотрит на оставшиеся у него фрагменты и решает, есть ли такие, которые подходят к уже находящимся на доске. Участник, нашедший соответствие, подходит к доске и прилепляет свой кусок. В результате фрагмент за фрагментом занимают нужное место. При этом не существенно, что один из участников может иметь больше фрагментов, чем другой. Все изображение будет полностью собрано без всякого обмена информацией между членами группы. Каждый участник активизируется самостоятельно и знает, когда ему нужно включиться в процесс. Никакого порядка подхода к доске заранее не устанавливается. Совместное поведение регулируется только информацией на доске. Наблюдение за процессом демонстрирует его последовательность (по одному фрагменту за подход) и произвольность (когда возникает возможность, фрагмент устанавливается). Это существенно отличается от строгой систематичности, например, от прохождения с левого верхнего угла и перебора каждого фрагмента
Потоки-продюсеры помещают данные в ConcurrentBag, поток-контроллёр забирает данные оттуда в ConcurrentQueue для обработки, потоки-обработчики заглядывают соответственно туда, поток-контроллёр потом вынимает из очереди данные и, если они уже обработаны или им пора умирать, выбрасывает их.
Запускать с осторожностью, ибо создание данных ничем не ограничено и обработка отстаёт от него примерно в пять раз. Или можно поместить вызов Thread.Sleep в метод Producer.Execute.
