Многопоточное единственное устройство записи единственного читателя первым прибыл, первым обслужен очередь

Вот как написать очередь без блокировок на C ++:

http://www.ddj.com/hpc-high-performance-computing/210604448

Но когда вы скажите «поток A не должен блокироваться», вы уверены, что это требование? Windows не является операционной системой реального времени (как и Linux при обычном использовании). Если вы хотите, чтобы поток A мог использовать всю доступную системную память, ему необходимо выделить память (или подождать, пока это сделает кто-то другой). Сама операционная система не может предоставить гарантии синхронизации лучше, чем те, которые были бы у вас, если бы и читатель, и писатель взяли внутрипроцессную блокировку (то есть не разделяемый мьютекс) для управления списком. И в худшем случае добавления сообщения придется обращаться к ОС за памятью.

Короче говоря, есть причина, по которой те очереди, которые вам не нравятся, имеют фиксированную емкость - это '

• Почему бы не использовать STL < list > или < deque > с мьютексом вокруг добавления / удаления? Является ли потокобезопасность STL недостаточной?

• Почему бы не создать свой собственный (одиночный / двойной) класс узла связанного списка, который содержит указатель, и унаследовать элементы, которые нужно добавить / удалить, от который? Таким образом, отпадает необходимость в дополнительном распределении. Вы просто добавляете несколько указателей в threadA :: add () и threadB :: remove () , и все готово. (Хотя вы хотели бы сделать это под мьютексом, эффект блокировки для threadA будет незначительным, если вы не сделали что-то действительно неправильно ...)

• Если вы используете pthreads, проверьте sem_post () и sem_wait () . Идея состоит в том, что threadB может блокироваться на неопределенное время через sem_wait () , пока threadA не поместит что-то в очередь. Затем threadA вызывает sem_post () . Что пробуждает ThreadB для выполнения своей работы. После чего threadB может вернуться в спящий режим. Это эффективный способ обработки асинхронной сигнализации, поддерживающий такие вещи, как несколько threadA :: add () до завершения threadB :: remove () .

Возможно, вы захотите учесть свои требования - действительно ли А не может отбросить какие-либо элементы очереди? Или вы не хотите, чтобы B вытаскивал из очереди два последовательных элемента, которые не входили последовательно, потому что это каким-то образом искажает последовательность событий?

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

В этом случае одним из решений является иметь какой-то особый элемент, который может быть помещен в очередь, что представляет собой случай, когда A обнаруживает, что ему нужно отбросить элементы. По сути, вы храните один дополнительный элемент, который большую часть времени имеет значение NULL. Каждый раз, когда A добавляет элементы в очередь, если этот дополнительный элемент не равен нулю, это происходит. Если A обнаруживает, что в очереди нет места, то он настраивает этот дополнительный элемент так, чтобы он сказал: «Привет, очередь заполнена». .

Таким образом, A никогда не блокируется, вы можете отбрасывать элементы, когда система очень занята, но вы не упускаете из виду тот факт, что элементы были отброшены, потому что, как только пространство в очереди становится доступным, эта метка появляется чтобы указать, где произошла потеря данных. Затем процесс B делает все, что ему нужно, когда обнаруживает, что вытащил этот элемент метки переполнения из очереди.

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

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

Visual Studio 2010 добавляет 2 новые библиотеки, которые очень хорошо поддерживают этот сценарий: Библиотека асинхронных агентов и Библиотека параллельных шаблонов.

Библиотека агентов поддерживает асинхронный режим. передача сообщений и содержит блоки сообщений для отправки сообщений «целям» и для получения сообщений из «источников»

unbounded_buffer - это шаблонный класс, который предлагает то, что, я думаю, вы ищете:

#include <agents.h>
#include <ppl.h>
#include <iostream>

using namespace ::Concurrency;
using namespace ::std;

int main()
{
   //to hold our messages, the buffer is unbounded...
   unbounded_buffer<int> buf1;
   task_group tasks;

   //thread 1 sends messages to the unbounded_buffer
   //without blocking
   tasks.run([&buf1](){
      for(int i = 0 ; i < 10000; ++i)
         send(&buf1,i)
     //signal exit 
     send(&buf1,-1);
   });

   //thread 2 receives messages and blocks if there are none

   tasks.run([&buf1](){
      int result;
      while(result = receive(&buf1)!=-1)
      {
           cout << "I got a " << result << endl;
      }
   });

   //wait for the threads to end
   tasks.wait();
}