Буфер без круговой блокировки

Question

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

Теперь два вопроса:

1. Является ли круговой буфер без блокировки ответом?

2. Если да, то, прежде чем я сделаю свою собственную, знаете ли вы какую-либо публичную реализацию, которая будет соответствовать моим потребностям?

Любые указатели в реализации циклического буфера без блокировки всегда приветствуются.

Кстати, делать это на C ++ в Linux.

Дополнительная информация:

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

Идея дизайна, к которой я склоняюсь, - это циклический буфер без полузакрытия, в котором поток производителя помещает данные в буфер настолько быстро, насколько это возможно, давайте назовем заголовок буфера A без блокировки, если буфер не full, когда A встречает конец буфера Z. Каждый из потоков-потребителей будет содержать два указателя на кольцевой буфер, P и P _n , где P - локальная головка буфера потока, а P _n - это n-ый элемент после P. Каждый потребительский поток будет продвигать свои P и P _n , как только он завершит обработку текущего P, а указатель конца буфера Z продвигается с самым медленным P _n . , Когда P догоняет A, что означает, что больше нет новых обновлений для обработки, потребитель вращается и действительно занят, ожидая, пока A снова продвинется. Если потребительский поток вращается слишком долго, его можно перевести в спящий режим и ждать переменную условия, но я согласен с тем, что потребитель принимает цикл ЦП в ожидании обновления, потому что это не увеличивает мою задержку (у меня будет больше ядер ЦП чем темы). Представьте, что у вас есть круговая дорожка, и производитель работает перед группой потребителей, ключ в том, чтобы настроить систему так, чтобы производитель, как правило, работал на несколько шагов впереди потребителей, и большинство из этих операций могут быть сделано с использованием техники без блокировки. Я понимаю, что правильно понять детали реализации нелегко ... хорошо, очень сложно, поэтому я хочу учиться на чужих ошибках, прежде чем делать несколько своих.

Answer 1

Это старый поток, но, поскольку он еще не был упомянут, есть свободный от блокировки, циклический, 1 производитель -> 1 потребитель, FIFO, доступный в инфраструктуре JUCE C ++.

https://www.juce.com/doc/classAbstractFifo#details

Answer 2

Извлечение Разрушитель ( Как его использовать ), представляющий собой кольцевой буфер, на который могут подписаться несколько потоков:

Answer 3

Просто для полноты: в OtlContainers имеется хорошо протестированный кольцевой буфер без блокировки, но он написан на Delphi (TOmniBaseBoundedQueue - кольцевой буфер, а TOmniBaseBoundedStack - ограниченный стек). В том же модуле есть неограниченная очередь (TOmniBaseQueue). Неограниченная очередь описана в Динамическая очередь без блокировки - все правильно . Начальная реализация ограниченной очереди (кольцевой буфер) была описана в Очередь без блокировки, наконец! , но с тех пор код был обновлен.

Answer 4

Вот как бы я это сделал:

• отобразить очередь в массив
• сохранить состояние при следующих индексах чтения и следующей записи
• оставить пустой полный битовый вектор вокруг

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

Для удаления требуется проверка бита перед тем, как проверять нижние значения, но, кроме этого, они такие же, как и для записи, но с использованием индекса чтения и очистки пустого / полного бита.

Будьте осторожны,

1. Я не эксперт в этих вещах
2. атомарные операции ASM кажутся очень медленными, когда я их использую, поэтому, если у вас получится больше, чем несколько, вы можете быстрее использовать блокировки, встроенные в функции вставки / удаления. Теория состоит в том, что одна атомная операция по захвату блокировки с последующим (очень) небольшим числом неатомных операций ASM может быть быстрее, чем та же самая операция, выполняемая несколькими атомными операциями. Но для выполнения этой работы потребуется ручное или автоматическое встраивание, так что это всего один короткий блок ASM.

Answer 5

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

Рассмотрим этот абзац из LDD3:

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

Answer 6

Вы можете попробовать lfqueue

Прост в использовании, круглая конструкция без блокировки

int *ret;

lfqueue_t results;

lfqueue_init(&results);

/** Wrap This scope in multithread testing **/
int_data = (int*) malloc(sizeof(int));
assert(int_data != NULL);
*int_data = i++;
/*Enqueue*/
while (lfqueue_enq(&results, int_data) != 1) ;

/*Dequeue*/
while ( (ret = lfqueue_deq(&results)) == NULL);

// printf("%d\n", *(int*) ret );
free(ret);
/** End **/

lfqueue_clear(&results);