Один поток производителей, несколько потребителей

Question

У меня есть одна нить производителя и несколько потребителей, каждый потребитель имеет: собственную очередь с данными и уникальный идентификатор. Я использую std :: map, чтобы идентифицировать каждую очередь для потока.

Каждый потребительский поток работает с данными в своей очереди, если очередь пуста, поток должен ждать данных. Если я хочу сделать это только с одним потоком, я могу использовать std :: condition_variable с std :: unique_lock, но у меня есть несколько пользователей, поэтому мне нужно несколько std :: condition_variable, но я не могу их сохранить в контейнерах (копирование/назначение удаляются). Так я использовать код, как этот

while(q.empty()) { 
    std::cout << "waiting...\n"; 
    std::this_thread::sleep_for(std::chrono::milliseconds(100)); 
} 

где д ссылка на очереди. Но как я могу синхронизировать его с лучшим способом? Спасибо заранее. P.S. Очередь всегда будет иметь данные, последние данные должны указывать «exit».

Answer 1

вызова ООП для решения этой задачи:

class ConcurentTaskQueue{ 
    std::mutex lock; 
    std::condition_variable m_ConditionVariable; 
    std::queue<Task> m_TaskQueue; 


public: 
    Task getTask(){ 
     std::unique_lock<std::mutex> synchLock (lock); //NOTE: consider doing this with a while(programIsRunning){} loop 
     while(m_TaskQueue.empty()){ 
      m_ConditionVariable.wait(synchLock); 
     } 
     Task task(std::move(m_TaskQueue.front())); 
     m_TaskQueue.pop(); 
     return task; 
    } 

    void addTask (Task task){ 
     std::unique_lock<std::mutex> synchLock (lock); 
     m_TaskQueue.push(std::move(task)); 
     m_ConditionVariable.notify_one(); 
    } 
}; 

и теперь просто:

std::map<size_t,ConcurentTaskQueue> inputQueue; 
std::thread producer ([&]{ 
     Task task = produceTask() 
     inputQueue[ID].addTask(task); 
}); 

std::thread consumer1([&]{ 
     Task task = inputQueue[ID].getTask(); 
}); 


std::thread consumer2([&]{ 
     Task task = inputQueue[ID].getTask(); 
}); 

EDIT2: использовать ThreadPool

Answer 2

Одного std::conditional_variable с одним std::mutex должно быть достаточно.

Task t; 
{ 
    std::unique_lock<std::mutex> lock(mtx); 
    while (q.empty()) 
    cond_var.wait(lock); 
    t = std::move(q.front()); 
    q.pop_front(); 
} 

и основной поток будет делать

{ 
    std::lock_guard<std::mutex> lock(mtx); 
    q.emplace_front(/*...*/); 
    cond_var.notify_all(); 
} 

Главной нити проснется все темы, но большинство из них будет идти спать, так как их очередь еще пуста.

Answer 3

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

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

Answer 4

Если вы не делаете это для изучения/обучения, но и для производственный код, я бы рекомендовал использовать одну из многих реализаций. Реализация этого правильно, эффективно и масштабируемо не так просто, но проблемы были решены другими людьми. Существует много вариантов, например.

• Intel Threading Build Blocks
• Microsoft PPL
• Apple Grand Central Dispatch.