Является ли boost :: lockfree :: queue (в многопоточной программе) блокируемой? - PullRequest
Новая
       61

Является ли boost :: lockfree :: queue (в многопоточной программе) блокируемой?

1 голос
/ 15 марта 2019

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

Очередь является очередью СОБЫТИЯ: где СОБЫТИЕ является структурой 

typedef struct EVENT{
    EVENT_TYPE Ev_Type;  // EVENT_TYPE is enum of Events
    EVENT_DATA Ev_Data;  // EVENT_DATA is union of data to be stored for that event
}Event_;

После поиска в Google я наткнулся на эти 2 варианта очереди: lockfree :: queue & lockfree :: spsc_queue , предполагая, что lockfree::queues используются для многопоточных приложений.

КОНФУЗИЯ: Почему имя lockFREE? это предполагает, что нельзя (мьютекс) заблокировать?

Также посмотрите этот пример , там написано "boost :: lockfree :: queue is not lockfree"

Ум = взорван ...

Ну, тогда я попытался следовать примеру (ссылка выше) и реализовать эту очередь 

class Foo {
protected:
    boost::lockfree::queue<EVENT> mqSttEventQueue;
public:
    unsigned int SetEventIntoQueue(EVENT *psEvent);
};

И его определение как: 

unsigned int Foo::SetEventIntoQueue(EVENT *psEvent) {
    if(mqSttEventQueue.push(*psEvent)){
         //notify that event is in queue;
    }
}

Это успешно скомпилировано. Но я полностью бегаю здесь в темноте.

ВОПРОСЫ:

  • Почему пример объявляет очередь как

    boost::lockfree::queue<int> queue(128);

для чего это 128 ? Это говорит о том, что размер очереди составляет 128 (байтов / элементов)? queue<int> объявляет тип данных в очереди?

  • Почему это не работает для моей программы

    boost::lockfree::queue<EVENT> mqSttEventQueue(128);

Если я объявлю это так, он получит ошибку компиляции как 

error: expected identifier before numeric constant

boost::lockfree::queue<EVENT> mqSttEventQueue(128);
                                              ^~~

PS: - Я действительно не уверен, какой заголовок поставить здесь ... Отредактируйте его, если можете.

1 Ответ

2 голосов
/ 15 марта 2019

Почему имя lockFREE?это говорит о том, что нельзя (мьютекс) заблокировать?

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

boost::lockfree::queue обеспечивает unsynchronized_pop и unsynchronized_push для использования в случаях, когда вы гарантировали, что только один потокможет получать доступ к очереди.

Но основная цель lockfree::queue и алгоритмов / структур данных без блокировки состоит в том, что они не должны быть заблокированы: несколько потоков могутбезопасно писать и / или читать одновременно.

«без блокировки» имеет 2 значения в программировании , что приводит к потенциально запутанным, но истинным утверждениям, таким как «без блокировки»Алгоритм не свободен от блокировки ".

  • Случайное использование: синоним для блокировки без блокировки - реализовано без мьютексов, с использованием атомарных загрузок, хранилищ и операций RMW, таких как CAS или std::atomic::atomic_fetch_add.См. Например Введение в программирование без блокировки (Джефф Прешинг).И, возможно, Что должен знать каждый системный программист о параллелизме .

    std::shared_ptr использует атомарные операции без блокировки для управления своим блоком управления.C ++ 11 std::atomic<> предоставляет примитивы без блокировки для пользовательских алгоритмов.См. .Обычно в C ++ 11 несинхронизированный доступ к одной и той же переменной несколькими потоками является неопределенным поведением.(Если только они не предназначены только для чтения.) Но std::atomic дает вам четко определенное поведение с вашим выбором последовательной согласованности, получения / освобождения или упорядоченного упорядочения памяти.

  • Техническое значение для информатики: нить, спящая вечно или убитая, не блокирует остальные нити .т.е. гарантированный прогресс вперед для программы в целом (хотя бы один поток).(Ожидание - это когда потоки никогда не должны повторяться).См. https://en.wikipedia.org/wiki/Non-blocking_algorithm. Цикл повторения CAS является классическим примером без блокировки, но не без ожидания.Без ожидания это такие вещи, как потоки чтения RCU (чтение-копирование-обновление) или, в зависимости от определений, atomic_fetch_add на оборудовании, которое реализует его как примитив (например, x86 xadd), а не в терминах LL / SCили цикл повторных попыток CAS.

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

См. Гарантии прогресса без блокировки для примера с анализом возможной блокировкиповедение.Средство записи атомарно увеличивает индекс записи, затем записывает данные в запись массива.Если писатель спит между этими действиями, другие писатели могут заполнить более поздние записи, пока читатели застряли в ожидании этой заявленной, но не письменной записи.(Я не смотрел на boost::lockfree::queue<T>, но, по-видимому, это похоже 1 .)

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

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

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

boost::lockfree::queue<int> queue(128); Почему 128?

Это размер очереди (максимальный) в записях.int в данном случае, потому что вы использовали queue<int>, хм.Как упомянуто выше, большинство очередей без блокировки использует кольцевой буфер фиксированного размера.Он не может перераспределять и копировать как std :: vector, когда ему нужно расти, потому что другие потоки могут читать его одновременно.

Как описано в руководстве (первый гугл дляboost::lockfree::queue), конструктор explicit queue(size_type) принимает размер.

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

Класс, очевидно, не применяет / не требуетразмером-2, поэтому параметр размера шаблона может быть значительно лучше оптимизирован, если % capacity скомпилировать операции в AND с маской вместо деления.

...