Заблокируйте свободную очередь — единственный производитель, несколько потребителей

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

uint8_t* buf;
unsigned int size; // Actual max. buffer size
unsigned int length; // Actual stored data length (suppose in write prohibited from being > size)
unsigned int offset; // Start of current stored data

При чтении вам нужно сделать следующее (я все равно реализовал это так, вы можете поменять местами некоторые шаги, о которых я расскажу позже):

1. Проверить, прочитано ли длина не превышает сохраненную длину
2. Проверить, не превышают ли смещение + длина чтения границы буфера
3. Считывание данных
4. Увеличить смещение, уменьшить длину

Что вы обязательно должны сделать синхронизированным (таким атомарным), чтобы сделать эта работа? Фактически объедините шаги 1 и 4 в один атомарный шаг, или, чтобы уточнить: сделайте это синхронизировано:

1. проверьте read_length, это может быть что-то вроде read_length = min (read_length, length);
2. уменьшить длину с помощью read_length: length- = read_length
3. получить локальную копию из смещения unsigned int local_offset = offset
4. увеличить смещение с read_length: offset + = read_length

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

Теперь, хотя я могу представить, что вы можете комбинировать 3 и 4 (я предполагаю, что это то, что они делают в случае связанного списка) или даже 1 и 2 в атомарных операциях, я не вижу, чтобы вы выполняли все это за одну атомарную операцию :).

Однако вы можете попробовать отказаться от проверки «длины», если ваши потребители очень умны и всегда будут знать, что читать.Тогда вам также понадобится новая переменная woffset, потому что старый метод (offset + length)% size для определения смещения записи больше не будет работать. Обратите внимание, что это близко к случаю связанного списка, где вы фактически всегда читаете один элемент (= фиксированный, известный размер) из списка. Также здесь, если вы сделаете его круговым связным списком, вы можете много читать или писать в позицию, которую вы читаете в данный момент!

Наконец: мой совет, просто используйте блокировки, я использую класс CircularBuffer, полностью безопасный для чтения и записи) для видеостримера 720p60 в реальном времени, и у меня вообще нет проблем со скоростью из-за блокировки.

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

Когда я реализовал очередь MS на C ++, я построил распределитель без блокировок с использованием стека, который очень легко реализовать.Если у меня есть MSQueue, то во время компиляции я знаю sizeof (MSQueue :: node). Затем делаю стопку из N буферов нужного размера. N может увеличиваться, т.е. если pop () возвращает значение null, легко запросить у кучи больше блоков, и они будут помещены в стек. Помимо возможного блокирующего вызова для дополнительной памяти, это операция без блокировки.

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

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

1. lock free делает меня независимым от планировщика. Программирование на основе блокировок зависит от планировщика, чтобы убедиться, что держатели блокировки работают, чтобы они могли разблокировать блокировку.Это то, что вызывает "инверсию приоритета". В Linux есть некоторые атрибуты блокировки, чтобы убедиться, что это произойдет
2. . Если я не зависим от планировщика, ОС будет намного легче управлять временными интервалами, и я получаю гораздо меньше переключений контекста
3. ] легче писать правильные многопоточные программы, используя безблокирующие методы, поскольку мне не нужно беспокоиться о взаимоблокировке, динамической блокировке, планировании, синхронизации и т. д. Это особенно верно для реализаций с общей памятью, где процесс может умереть, удерживая блокировку в общей памяти, и нет способа освободить блокировку
4. методы освобождения от блокировок намного проще масштабировать. Фактически, я реализовал методы без блокировки, используя обмен сообщениями по сети. Подобные распределенные блокировки - это кошмар

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

1. при обновлении вещей, которые дорого или невозможно скопировать. Большинство методов без блокировок используют своего рода управление версиями, то есть делают копию объекта, обновляют его и проверяют, остается ли общая версия такой же, как при ее копировании, а затем сделайте текущую версию, которую вы обновляете. Els снова скопируйте его, примените обновление и проверьте еще раз. Продолжайте делать это, пока не сработает. Это нормально, когда объекты маленькие, но если они большие, содержат дескрипторы файлов и т. Д., То не рекомендуется
2. К большинству типов невозможно получить доступ без блокировки, например любой контейнер STL. У них есть инварианты, требующие неатомарного доступа, например assert (vector.size () == vector.end () - vector.begin ()). Поэтому, если вы обновляете / читаете вектор, который является общим, вам необходимо заблокировать его.