Использование параллелизма в конвейерном исполнении

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

Я делю свой ответ на две части: общее решение без OpenMP и некоторые конкретные изменения для использования OpenMP.

Как уже упоминалось в комментарии, вы сталкиваетесь с проблемой производителя / потребителя, но дважды: один поток заполняет буфер (производит элемент), который затем должен быть прочитан (и изменен) вторым (потребляемым). Особенность вашей проблемы в том, что этот второй поток также является продюсером (изображение, которое нужно нарисовать), а третий поток отвечает за его потребление (визуализатор).

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

Давайте использовать подход очереди с вашей проблемой в следующем примере программы.

• Произведенные предметы будут храниться в очереди. Передняя часть очереди содержит самые старые элементы, те, которые должны быть использованы в первую очередь.
• Существует две очереди: одна для данных, созданных первым манипулятором (и предназначенных для использования вторым манипулятором), и другая для данных, созданных вторым манипулятором (и которые будут визуализироваться другим потоком).
• Этап производства прост: получите эксклюзивный доступ к соответствующей очереди и вставьте элемент в конце.
• Расход аналогичен, но он должен ждать, пока в очереди будет хотя бы один элемент (не быть пустым).
• Я добавил немного спит моделировать другие операции.
• Условие остановки приведено для иллюстрации.

Примечание: я предполагаю, что у вас есть доступ к компилятору C ++ 11 для простоты. Реализации, использующие другие API, относительно похожи.

#include <iostream>
#include <thread>
#include <mutex>
#include <atomic>
#include <chrono>
#include <list>

using namespace std::chrono_literals;

std::mutex g_data_produced_by_m1_mutex;
std::list<int> g_data_produced_by_m1;

std::mutex g_data_produced_by_m2_mutex;
std::list<int> g_data_produced_by_m2;

std::atomic<bool> stop = false;

void manipulator1_kernel()
{
while (!stop) {
// Producer 1: generate data
{
std::lock_guard<std::mutex> lock(g_data_produced_by_m1_mutex);
g_data_produced_by_m1.push_back(rand());
}
std::this_thread::sleep_for(100ms);
}
}

void manipulator2_kernel()
{
int data;

while (!stop) {
// Consumer 1
while (!stop) { // wait until there is an item to be consumed
{
std::lock_guard<std::mutex> lock(g_data_produced_by_m1_mutex);
if (!g_data_produced_by_m1.empty()) { // is there data to be consumed?
data = g_data_produced_by_m1.front(); // consume
g_data_produced_by_m1.pop_front();
break;
}
}
std::this_thread::sleep_for(100ms);
}

// Producer 2: modify and send to the visualizer
{
std::lock_guard<std::mutex> lock(g_data_produced_by_m2_mutex);
g_data_produced_by_m2.push_back(5 * data);
}

std::this_thread::sleep_for(100ms);
}
}

void visualizer_kernel()
{
int data;

while (!stop) {
// Consumer 2
while (!stop) { // wait until there is an item to be visualized
{
std::lock_guard<std::mutex> lock(g_data_produced_by_m2_mutex);
if (!g_data_produced_by_m2.empty()) {
data = g_data_produced_by_m2.front();
g_data_produced_by_m2.pop_front();
break;
}
}
std::this_thread::sleep_for(100ms);
}

std::cout << data << std::endl; // render to display
std::this_thread::sleep_for(100ms);

if (data % 8 == 0) stop = true; // some stop condition for the example
}
}

int main()
{
std::thread manipulator1(manipulator1_kernel);
std::thread manipulator2(manipulator2_kernel);
std::thread visualizer(visualizer_kernel);

visualizer.join();
manipulator2.join();
manipulator1.join();

return 0;
}

Если вы все еще хотите использовать OpenMP, возможно, вы найдете задачи (начиная с OpenMP 3.0 я думаю). Я не очень много их использовал, но вышеприведенную программу можно переписать так:

int main()
{
#pragma omp parallel
{
#pragma omp task
manipulator1_kernel();
#pragma omp task
manipulator2_kernel();
#pragma omp task
visualizer_kernel();

#pragma omp taskwait
}

return 0;
}

Остальную часть кода можно изменить, чтобы использовать функции OpenMP, но я думаю, что это отвечает на ваш вопрос.

Основная проблема с этим подходом заключается в том, что вам нужно создать блок кода для задач, чтобы жить в OpenMP parallel, легко усложняя всю логику и структуру вашего приложения.