Потокобезопасность с общими очередями

Question

У меня нет большого опыта работы с многопоточными приложениями на C ++, и в настоящее время я пытаюсь разработать приложение с четырьмя потоками и тремя общими очередями.Это должно быть C ++-версия приложения, которое я написал на Python год назад, но оно не работает достаточно быстро, чтобы соответствовать новым требованиям (я всегда хотел научиться многопоточности на C ++, так что это казалось хорошей возможностью).

Запланированные потоки:

1) Поток для очистки строк от конца файла (эквивалент выполнения tail -f в linux CL)

2) Поток кстроки процесса из файла

3) Поток для отправки обработанных строк в другое место

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

Настройка очереди, которую я планирую использовать:

Одна общая очередь между потоками 1 и 2;нить 1 помещает элементы, нить 2 выводит их.Никогда наоборот.Одна общая очередь для потоков 2 и 3, которая ведет себя так же, как и первая.Третья необязательная очередь, которая записывается всеми первыми тремя потоками и читается четвертым потоком.

Проблема, с которой я столкнулся, состоит в том, что в отличие от объектов multiprocessing.Queue () в Python, очереди C ++ не являются 'По умолчанию он безопасен для потоков, и я еще не нашел стандартную реализацию общей очереди.

Вопросы:

1) Можно ли восстановить базовую схему, которую я использовал в Python (см. выше)используется в C ++, не создавая никаких проблем?

2) Существует ли стандартная реализация потоковой очереди, которую я еще не нашел, или мне придется прибегнуть к экзотическим библиотекам (например Boost) или сделать свой собственный?

Answer 1

Если вы не хотите использовать boost, вот более или менее полная реализация:

#include <mutex>
#include <condition_variable>
#include <queue>

class Queue
{
public:
  Queue() = default;
  ~Queue() = default;

  int pop()
  {
    std::unique_lock<std::mutex> lock_guard(mutex_);

    while (internal_queue_.empty())
    {
      cond_.wait(lock_guard);
    }

    auto item = i_queue_.front();
    i_queue_.pop();
    return item;
  }

  void push(int item)
  {
    std::unique_lock<std::mutex> lock_guard(mutex_);
    i_queue_.push(item);
    lock_guard.unlock();
    cond_.notify_one();
  }

private:
  std::queue<int> i_queue_;
  std::mutex mutex_;
  std::condition_variable cond_;
};

int main()
{
  Queue queue;
  queue.push(1);
  auto i = queue.pop();
  return 0;
}

Обратите внимание, что pop заблокирует ваши потоки, пока очередь не опустеет.Push уведомит ожидающие темы.То есть вам нужен только цикл в ваших потоках, чтобы выталкивать элементы один-на-один.Цикл будет ожидать, если больше нет элементов, готовых к употреблению.