Многопоточная сигнализация

Question

Я хочу создать многопоточное приложение, которое выполняет следующие действия: Один поток записывается последовательно в круговой буфер. Тогда будет n-число потоков считывателей, которые ждут некоторого сигнала, инициированного потоком записи, чтобы проснуться и прочитать из кругового буфера. Сигнал должен каким-то образом содержать целочисленное значение, представляющее смещение кругового буфера для чтения. Можно ли это сделать в C++? Любая помощь будет оценена по достоинству.

Поскольку я хочу, чтобы дизайн, способный обрабатывать как можно более быстрый высокоскоростной трафик реального времени, я хочу устранить любое распределение/освобождение памяти. Таким образом, круговая очередь будет смежным куском памяти, выделенной при запуске. Я не уверен, что очередность, о которой вы говорите, соответствует этому.

Изготовителю нужно будет только отслеживать, с чего начать писать в круглый буферный массив байтов каждый раз, когда ему есть что писать. Так что все, что я действительно прошу, - это способ продюсера распространить «сигнал», когда он завершил событие записи, которое содержит местоположение (смещение) последнего байта, записанного в круговой буфер. Это позволит избежать необходимости блокировки механизма.

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

Answer 1

Основываясь на вашем последнем редактировании, я бы предложил использовать ограниченную очередь.

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

Со стороны производителя: если очередь не заполнена, push_back - элемент в очередь.

Со стороны потребителя: Если очередь не пуста, вытащите элемент из очереди и обработайте его.

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

Сама очередь должна быть потокобезопасной, и есть примеры на SO о том, как создать поточную безопасную ограниченную очередь на C++.

Edit:

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

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

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

Если вы хотите, чтобы несколько потоков обращались к данным сразу, вы можете подумать о создании нескольких очередей или буферов. Возможно, один круговой буфер на пару производителей/потребителей или одну очередь на входной поток ... что бы это ни случилось. Трудно сказать без более конкретного примера.

Edit 2

Here's a link to a thread-safe queue. Я не уверен, что это поможет, но выглядит многообещающим.

Answer 2

Это ИМО, плохой способ сделать что-то. Попросите продюсера просто добавить элементы в круглый буфер. Пусть каждый читатель ожидает, что круговой буфер будет непустым. Когда он не пуст, поток читателя просто удаляет следующий элемент из буфера и обрабатывает его. Сам буфер должен отслеживать такие вещи, как смещения.

Что касается того, почему это лучше: во многом потому, что он позволяет каждой части системы выполнять свою собственную работу с минимальным взаимодействием с другими частями системы.

Как вы описываете систему, производитель должен знать о внутренних деталях очереди, а также, по существу, все детали потребительских потоков (которые пробуждаются в любой момент времени, которые простаивают в любое заданное время, которое следует запланировать для выполнения какой-либо конкретной задачи и т. д.)

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

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

Сама очередь отвечает за всю необходимую синхронизацию потоков. Синхронизация необходима только тогда, когда задача ставится/удаляется из очереди. Сама очередь довольно многократно используется (может использоваться для почти такой ситуации с производителем-потребителем) и сменяется (например, довольно тривиальное переключение между блокировкой и блокировкой).

Планирование потоков остается в ОС - незанятые потребительские потоки просто ждут очереди, и ОС решает, какой из них следует выполнять для выполнения конкретной задачи. Если ни один из них в настоящее время не работает, ОС уже «знает» это тоже, и оставляет их для выполнения текущей обработки до тех пор, пока не закончится и не будет ждать очереди.

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

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