Я очень много искал для бесплатной очереди на С++. Я нашел некоторый код и некоторые испытания, но ничего, что я смог скомпилировать. Также приветствуется хеш-блокировка.
РЕЗЮМЕ: Пока у меня нет положительного ответа. Нет никакой "готовой к производству" библиотеки, и, как ни удивительно, ни одна из существующих библиотек не соответствует API контейнеров STL.
Начиная с версии 1.53, boost предоставляет набор блокировок без блокировки данных, включая очереди, стеки и очереди одного производителя/одного потребителя ( т.е. кольцевые буферы).
Отправной точкой было бы либо из статей Herb Sutter DDJ, либо для одного производителя и потребителя или несколько экземпляров. Код, который он дает (в строке, начинающемся со второй страницы каждой статьи), использует атомный <T> тип шаблона; которые вы можете имитировать, используя библиотеку Interprocess Boost.
Код ускорения похоронен в глубинах библиотеки interprocess, но, прочитав соответствующий файл заголовка (atomic.hpp), реализации для необходимых операций сравнения и свопинга в системах, которые мне знакомы, выглядят звуковыми.
Facebook Folly, похоже, имеет блокированные структуры данных на основе С++ 11 <atomic>:
ProducerConsumerQueue с документами и примером кода здесь.
AtomicHashMap с документами и кодом примера здесь
Я бы осмелился сказать, что в настоящее время они используются в производстве, поэтому я предполагаю, что их можно безопасно использовать в других проектах.
Ура!
I написал свободную очередь. Он имеет функции ™:
Он доступен на GitHub в соответствии с упрощенной лицензией BSD (не стесняйтесь его разветвлять!).
Предостережения:
Существует такая библиотека, но она в C.
Обертка на С++ должна быть простой.
После проверки большинства ответов, я могу только указать:
Ответ НЕТ.
Нет такого права, которое можно было бы использовать прямо из коробки.
boost.lockfree - это попытка создания реализаций С++ без стекового стека и классов fifo.
Ближайшая вещь, о которой я знаю, - это Связанные с Windows списки с блокировкой Windows. Конечно, это только Windows.
Если у вас многопользовательская/однопользовательская очередь /FIFO, вы можете легко сделать один LockFree с помощью SLIST или тривиального стека LIFO Lock Free. У вас есть второй "private" стек для пользователя (который также можно сделать как SLIST для простоты или любой другой выбранной вами модели стека). Пользователь выталкивает предметы из частного стека. Всякий раз, когда частный LIFO exhasted, вы делаете Flush, а не Pop из общего совместного SLIST (захватывая всю цепочку SLIST), а затем пройдите по Flushed-списку, нажимая элементы в отдельный стек.
Это работает для одного производителя/одного потребителя и для нескольких производителей/для одного потребителя.
Однако он не работает для случаев с несколькими потребителями (с одним производителем или с несколькими производителями).
Кроме того, что касается хеш-таблиц, они являются идеальным кандидатом на "разделение", который просто делит хеш на сегменты, имеющие блокировку на сегменты кеша. Так работает параллельная библиотека Java (с использованием 32-полос). Если у вас есть легкая блокировка чтения-записи, хэш-таблицу можно одновременно получить для одновременного чтения, и вы будете останавливаться только тогда, когда запись происходит на оспариваемых полосах (и, возможно, если вы разрешаете выращивать хеш-таблицу).
Если вы катитесь самостоятельно, убедитесь, что вы чередовали блокировки с элементами хэша, а не помещали все блокировки в массив рядом друг с другом, поэтому у вас меньше шансов иметь ложный доступ.
И затем появился Intel Threading Building Blocks. И какое-то время это было хорошо.
PS: вы ищете concurrent_queue и concurrent_hash_map
Я могу немного опоздать.
Отсутствие решений (при заданном вопросе) в основном связано с важной проблемой в С++ (до С++ 0x/11): у С++ нет (параллельной модели памяти).
Теперь, используя std:: atomic, вы можете управлять проблемами упорядочения памяти и иметь правильные операции сравнения и свопинга. Я написал себе реализацию Micheal & Scott lock-free queue (PODC96) с использованием С++ 11 и Micheal Hazard Pointers (IEEE TPDS 2004), чтобы избежать ранних бесплатных проблем и проблем с ABA. Он отлично работает, но это быстро и грязно, и я не удовлетворен действительными выступлениями. Код доступен на битбакете: LockFreeExperiment
Также возможно реализовать блокировку без указателей опасности с использованием двухсловных CAS (но 64-битные версии будут доступны только на x86-64 с использованием cmpxchg16b), у меня есть сообщение в блоге об этом (с непроверенным кодом для queue) здесь: Внедрение совместного сравнения и обмена двойным словом для x86/x86-64 (блог LSE.)
Мой собственный тест показывает мне, что очередь с двойной блокировкой (также на бумаге Micheal & Scott, 1996) работает так же, как и без блокировки (я не дотягиваю до такой степени, что блокированные структуры данных имеют проблемы с производительностью, но мои скамейка слишком светлая), а параллельная очередь от Intel TBB кажется еще лучше (на два раза быстрее) для относительно небольшого числа (в зависимости от операционной системы, в соответствии с FreeBSD 9, самой низкой оценкой, которую я нашел до сих пор, это число это 8 потоков на i7 с 4-дюймовым ядром и, следовательно, 8 логических процессоров) потоков и имеют очень странное поведение (время выполнения моего простого теста перемещается с секунд до часа!)
Другие ограничения на блокировку очередей после стиля STL: наличие итераторов в незаблокированной очереди не имеет смысла.
Насколько я знаю, такой публики пока нет. Одна из проблем, которую должен решить разработчик, заключается в том, что вам нужен блокировщик памяти без блокировки, который существует, хотя я не могу найти ссылку прямо сейчас.
Ниже приведена статья Херба Саттера о параллельной блокировке без очереди http://www.drdobbs.com/parallel/writing-a-generalized-concurrent-queue/211601363?pgno=1. Я внес некоторые изменения, такие как переупорядочивание компилятора. Для компиляции этого кода требуется GCC v4.4 +.
#include <atomic>
#include <iostream>
using namespace std;
//compile with g++ setting -std=c++0x
#define CACHE_LINE_SIZE 64
template <typename T>
struct LowLockQueue {
private:
struct Node {
Node( T* val ) : value(val), next(nullptr) { }
T* value;
atomic<Node*> next;
char pad[CACHE_LINE_SIZE - sizeof(T*)- sizeof(atomic<Node*>)];
};
char pad0[CACHE_LINE_SIZE];
// for one consumer at a time
Node* first;
char pad1[CACHE_LINE_SIZE
- sizeof(Node*)];
// shared among consumers
atomic<bool> consumerLock;
char pad2[CACHE_LINE_SIZE
- sizeof(atomic<bool>)];
// for one producer at a time
Node* last;
char pad3[CACHE_LINE_SIZE
- sizeof(Node*)];
// shared among producers
atomic<bool> producerLock;
char pad4[CACHE_LINE_SIZE
- sizeof(atomic<bool>)];
public:
LowLockQueue() {
first = last = new Node( nullptr );
producerLock = consumerLock = false;
}
~LowLockQueue() {
while( first != nullptr ) { // release the list
Node* tmp = first;
first = tmp->next;
delete tmp->value; // no-op if null
delete tmp;
}
}
void Produce( const T& t ) {
Node* tmp = new Node( new T(t) );
asm volatile("" ::: "memory"); // prevent compiler reordering
while( producerLock.exchange(true) )
{ } // acquire exclusivity
last->next = tmp; // publish to consumers
last = tmp; // swing last forward
producerLock = false; // release exclusivity
}
bool Consume( T& result ) {
while( consumerLock.exchange(true) )
{ } // acquire exclusivity
Node* theFirst = first;
Node* theNext = first-> next;
if( theNext != nullptr ) { // if queue is nonempty
T* val = theNext->value; // take it out
asm volatile("" ::: "memory"); // prevent compiler reordering
theNext->value = nullptr; // of the Node
first = theNext; // swing first forward
consumerLock = false; // release exclusivity
result = *val; // now copy it back
delete val; // clean up the value
delete theFirst; // and the old dummy
return true; // and report success
}
consumerLock = false; // release exclusivity
return false; // report queue was empty
}
};
int main(int argc, char* argv[])
{
//Instead of this Mambo Jambo one can use pthreads in Linux to test comprehensively
LowLockQueue<int> Q;
Q.Produce(2);
Q.Produce(6);
int a;
Q.Consume(a);
cout<< a << endl;
Q.Consume(a);
cout<< a << endl;
return 0;
}
Я нашел другое решение, написанное в c:
Я написал это в какой-то момент, вероятно, в 2010 году, я уверен с помощью разных ссылок. Это мультипроцессор Single Consumer.
template <typename T>
class MPSCLockFreeQueue
{
private:
struct Node
{
Node( T val ) : value(val), next(NULL) { }
T value;
Node* next;
};
Node * Head;
__declspec(align(4)) Node * InsertionPoint; //__declspec(align(4)) forces 32bit alignment this must be changed for 64bit when appropriate.
public:
MPSCLockFreeQueue()
{
InsertionPoint = new Node( T() );
Head = InsertionPoint;
}
~MPSCLockFreeQueue()
{
// release the list
T result;
while( Consume(result) )
{
//The list should be cleaned up before the destructor is called as there is no way to know whether or not to delete the value.
//So we just do our best.
}
}
void Produce( const T& t )
{
Node * node = new Node(t);
Node * oldInsertionPoint = (Node *) InterLockedxChange((volatile void **)&InsertionPoint,node);
oldInsertionPoint->next = node;
}
bool Consume( T& result )
{
if (Head->next)
{
Node * oldHead = Head;
Head = Head->next;
delete oldHead;
result = Head->value;
return true;
}
return false; // else report empty
}
};