Распараллеливание программы с процедурами обратного вызова

Question

Набросал программу с "действующими лицами" Producer/Consumer. Программа выполняется в одном потоке. Исполнение примитивное. Producer выполняет определенное действие, периодически "сообщая" Consumer'у о состоянии выполнения, а так же опрашивает необходимость прерывания обработки. В конце обработки сообщает об окончании обработки. Код ниже:

#include <functional>
#include <iostream>
#include <thread>

using namespace std::chrono_literals;

using TypeCallbackRun = std::function<void (int)>;
using TypeCallbackBreak = std::function<bool (int)>;
using TypeCallbackFinish = std::function<void ()>;

// ───────────────────────────────────────────────────────────────────────────────────

class Producer {
    TypeCallbackRun CallbackRun;
    TypeCallbackBreak CallbackBreak;
    TypeCallbackFinish CallbackFinish;
  public:
    void Run() {
      std::cout << "Run started (10 times) ...\n";
      for (int i = 0; i < 10; ++i) {
        if (CallbackBreak && CallbackBreak(i)) break;
        if (CallbackRun) CallbackRun(i);
        std::this_thread::sleep_for(100ms);
      }
      if (CallbackFinish) CallbackFinish();
      std::cout << "\nRun finished\n";
    }
    void SetCallbacks(TypeCallbackRun r, TypeCallbackBreak b, TypeCallbackFinish f) {
      CallbackRun = r;
      CallbackBreak = b;
      CallbackFinish = f;
    }
};

// ───────────────────────────────────────────────────────────────────────────────────

class Consumer {
  public:
    bool Ready = false;
    void ProcCallbackRun(const int i) {
      std::cout << i << " ";
    }
    bool ProcCallbackBreak(const int i) {
      return i > 4;
    }
    void ProcCallbackFinish() {
      Ready = true;
    }
};

// ───────────────────────────────────────────────────────────────────────────────────

int main() {
  Producer Prod;
  Consumer Cons;
  //
  //  так
  //
  Prod.SetCallbacks(
    std::bind(&Consumer::ProcCallbackRun, &Cons, std::placeholders::_1),
    std::bind(&Consumer::ProcCallbackBreak, &Cons, std::placeholders::_1),
    std::bind(&Consumer::ProcCallbackFinish, &Cons)
  );
  //
  //  ну или так
  //
  //  Prod.SetCallbacks(
  //  [&](int i) {
  //    Cons.ProcCallbackRun(i);
  //  },
  //  [&](int i) {
  //    return Cons.ProcCallbackBreak(i);
  //  },
  //  [&]() {
  //    Cons.ProcCallbackFinish();
  //  });
  Prod.Run();
  return 0;
}

Вопрос

Как правильно запустить Producer в отдельном потоке средствами С++11/14/17 так, чтобы функционал "сообщений" о "состоянии обработки", о "необходимости прерывания", об "окончании обработки" остался работоспособным? Ну или как-то реализовать иначе подобное.

Допустим код ожидания обработки пусть будет примерно таким:

for (int i = 0; i < 100000; i++) {
  if (Cons.Ready) break;
  std::this_thread::sleep_for(10ms);
}

Т.e. поток с Producer должен быть отсоединен сразу посредством std::thread::detach

Answer 1

Вроде бы сам решил. Пишите если есть замечания или что-то более лучшее.

#include <functional>
#include <iostream>
#include <thread>
#include <mutex>
#include <atomic>

using namespace std::chrono_literals;

using TypeCallbackRun = std::function<void (int)>;
using TypeCallbackBreak = std::function<bool (int)>;
using TypeCallbackFinish = std::function<void ()>;

std::mutex Mutex;

// ───────────────────────────────────────────────────────────────────────────────────

class Producer {
    TypeCallbackRun CallbackRun;
    TypeCallbackBreak CallbackBreak;
    TypeCallbackFinish CallbackFinish;
  public:
    void Run() {
      Mutex.lock();
      std::cout << "Run started (10 times): " << std::this_thread::get_id() << "\n";
      Mutex.unlock();
      for (int i = 0; i < 1000; ++i) {
        if (CallbackBreak && CallbackBreak(i)) break;
        if (CallbackRun) CallbackRun(i);
        Mutex.lock();
        std::cout << ".";
        Mutex.unlock();
        std::this_thread::sleep_for(10ms);
      }
      if (CallbackFinish) CallbackFinish();
      Mutex.lock();
      std::cout << "\nRun finished: " << std::this_thread::get_id() << "\n";
      Mutex.unlock();
    }
    void SetCallbacks(TypeCallbackRun r, TypeCallbackBreak b, TypeCallbackFinish f) {
      CallbackRun = r;
      CallbackBreak = b;
      CallbackFinish = f;
    }
};

// ───────────────────────────────────────────────────────────────────────────────────

class Consumer {
  public:
    std::atomic<int> Counter;
    std::atomic<bool> Ready;
    Consumer () {
      Counter.store(0);
      Ready.store(false);
    }
    void ProcCallbackRun(const int i) {
      std::lock_guard<std::mutex> Lock(Mutex);
      std::cout << "(" << i << ") ";
    }
    bool ProcCallbackBreak([[maybe_unused]] const int i) {
      return Counter.load() > 75;
    }
    void ProcCallbackFinish() {
      std::lock_guard<std::mutex> Lock(Mutex);
      std::cout << "\nthread finish: " << std::this_thread::get_id() << std::endl;
      Ready = true;
    }
};

// ───────────────────────────────────────────────────────────────────────────────────

int main() {
  Consumer Cons;
  Producer Prod;    
  Prod.SetCallbacks(
      std::bind(&Consumer::ProcCallbackRun, &Cons, std::placeholders::_1),
      std::bind(&Consumer::ProcCallbackBreak, &Cons, std::placeholders::_1),
      std::bind(&Consumer::ProcCallbackFinish, &Cons)
    );  
  // запуск
  std::cout << "main thread: " << std::this_thread::get_id() << std::endl;
  std::thread([&]() {
    Prod.Run();
  }).detach();
  while (!Cons.Ready) {
    std::this_thread::sleep_for(200ms);
    Mutex.lock();
    std::cout << "[" << Cons.Counter.load() << "] ";
    Cons.Counter++;
    Mutex.unlock();
  }
  std::cout << "exit program: " << std::this_thread::get_id() << std::endl;
  return 0;
}

EDIT: Поправил код, чтобы компилировался и в C++11, а не только >= С++17. Увеличил задержку в основном потоке, чтобы компилятор не оптимизировал, и не исполнял все в одном потоке. Пример тут.

БЛОГ НА HUSL

Распараллеливание программы с процедурами обратного вызова

Ответы (1 шт):