Russian Qt Forum

C++ (Qt)
do {
     // эта функция может быть распараллелена
     doHardComputing();
 
    // А это легковесная функция. Она использует результаты, полученные doHardComputing 
    doSomething();
 
} while (condition); // Всё это крутится в цикле пока не сработает условие выхода. 
 

C++ (Qt)
do {
    std::vector<std::thread> threads;
    for (size_t i = 0; i < NThreads; ++i)
        threads.emplace_back(worker); 
 
    for (auto & th : threads)
        th.join();
 
    doSomething();
 
} while(condition);
 

C++ (Qt)
#include <iostream>
#include <thread>
#include <mutex>
#include <condition_variable>
#include <atomic>
#include <vector>
#include <algorithm>
 
std::mutex mutex;
 
std::condition_variable cv_worker;
std::condition_variable cv_loop;
 
std::atomic_bool is_done(false);
 
static constexpr int NThreads = 8;
 
void worker(std::vector<bool> & flags, unsigned int id)
{
    while (!is_done)
    {
        // Имитация расчётов..
        std::this_thread::sleep_for(std::chrono::seconds(5));
 
        {
            // Каждый worker имеет свой флажёк, который говорит о том, что он выполнил расчёты и готов спать) 
            std::unique_lock<std::mutex> lk(mutex);
            flags[id] = false;
        }
 
        cv_loop.notify_one(); // Пытаемся разбудить потребителя (main_loop)
        // И засыпаем
        std::unique_lock<std::mutex> lk(mutex); 
        cv_worker.wait(lk, [&]()
        {
            return flags[id] || is_done;
        });
    }
}
 
void main_loop(std::vector<bool> & flags)
{
    int i = 0;
    auto cond = [](int i) { return i > 10; }; // условие выхода, для наглядности
 
    do
    {
        {   // Спим до тех пор, пока все workerы не выполнят вычисления (все флажки = false)
            std::unique_lock<std::mutex> lk(mutex);
            cv_loop.wait(lk, [&]()
            {
                return !std::any_of(flags.begin(), flags.end(), [](bool x){ return x; });
            });
        }
 
        // Имитация вычислений..
        std::this_thread::sleep_for(std::chrono::seconds(1));
 
        {   // После вычислений заряжаем флажки
            std::unique_lock<std::mutex> lk(mutex);
            std::fill(flags.begin(), flags.end(), true);
            if (cond(i++))
                is_done = true;
        }
        // Будим всех воркеров и засыпаем 
        cv_worker.notify_all();
 
    } while (!is_done);
}
 
int main()
{
    std::vector<bool> flags(NThreads, false); // флажки
 
    std::thread mainthread(main_loop, std::ref(flags)); // Это наш потребитель
 
    std::vector<std::thread> threads; // Это наши производители workerы
 
    for (unsigned int i = 0; i < NThreads; ++i)
        threads.emplace_back(worker, std::ref(flags), i);
 
    mainthread.join();
 
    for (auto & th : threads)
        th.join();
 
    return 0;
}
 

C++ (Qt)
 std::thread consumer([&]() {
        while (!abortFlag) {
           std::unique_lock<std::mutex> lock(m);
 
/* каждый из worker'ов, закончив работу, уменьшает атомарный счетчик 
workerInProcess и пытается нас (производителя) будить - 
но мы игнорируем побудку если хотя бы один worker активен. 
Собсно идея в этом */ 
            while (workerInProcess) {
                if (abortFlag) return;
                cond_var.wait(lock);
             }
 
/*  В этот момент все worker'ы отстрелялись и висят на (захваченном нами) мутексе, перезаряжаемся */
            if (CheckDone()) break;
            SetupWorkersData();   // заряжаем новые данные worker'ов 
            workerInProcess = NUM_WORKERS;
            cond_var.notify_all();   // стартуем worker'ов, сами засыпаем
        }
 
/* даем worker'ам выйти */
        abortFlag = true;
        cond_var.notify_all();
 
    });

    std::thread worker([&]() {
        while (true) {
 
// Пресекаем "ложные побудки"
          {
            std::unique_lock<std::mutex> lock(m);
            while (!HasData()) {
               if (abortFlag) return;
               cond_var.wait(lock);    // освобождаем мутекс и спим
             } 
          }

// считаем
           DoWorkerJob();
        
// уменьшаем счетчик и будим всех
           --workerInProcess;
           cond_var.notify_all();
    });

C++ (Qt)
#pragma omp parallel
{
    do {
          #pragma omp task 
          {
               worker();
          }
          #pragma omp taskwait
 
          doSomething();
 
    } while (condition);
}
 

void worker(std::vector<bool> & flags, unsigned int id)
{
    while (!is_done)
    {
        // Имитация расчётов..
        std::this_thread::sleep_for(std::chrono::seconds(5));

C++ (Qt)
        {
            // Каждый worker имеет свой флажёк, который говорит о том, что он выполнил расчёты и готов спать) 
            std::unique_lock<std::mutex> lk(mutex);
            flags[id] = false;
        }
 

C++ (Qt)
#pragma omp parallel for
for (int i = 0; i < data.size(); ++i)
 DoCalc(data[i]);

C++ (Qt)
void worker(bool & flag)
 

C++ (Qt)
bool loop_function() { ... }
 
void do_work() { ... }
 
 
concurrent_loop cloop(num_threads);
 
cloop.start(loop_function, do_work);
 
cloop.join();
 
 

C++ (Qt)
do {
    std::vector<std::thread> threads;
    for (unsigned int i = 0; i < num_threads; ++i)
        threads.emplace_back(do_work);
 
    for (auto & th : threads)
        th.join();
 
} while (loop_function());
 

C++ (Qt)
template <class Done>
void worker(Done done)
{
    // hard computing...
    done();
}
 
void caller()
{
    const std::size_t N = 4;
    std::vector<bool> flags(N, false);
 
    for (int i = 0; i < N; ++i)
        worker([&flags, i]() { flags[i] = true; });
}
 

C++ (Qt)
#include <iostream>
#include "concurrent_loop.h"
#include <list>
 
int main()
{
    static constexpr int num_threads = 4;
    std::mutex print_mutex;
 
    std::list<std::function<void()>> workers;
 
    for (unsigned int i = 0; i < num_threads; ++i)
        workers.push_back([&]()
        {
            std::this_thread::sleep_for(std::chrono::seconds(1));
            std::lock_guard<std::mutex> lk(print_mutex);
            std::cout << std::this_thread::get_id() << std::endl;
        });
 
    int ntests = 3;
 
    auto loop_function = [&]()
    {
        std::cout << --ntests << "---------------------" << std::endl;
        return ntests == 0;
    };
 
    concurrent_loop cloop(loop_function, workers);
 
    cloop.join();
 
    return 0;
}
 

Bash
140269582214912
140269557036800
140269573822208
140269565429504
2---------------------
140269557036800
140269565429504
140269582214912
140269573822208
1---------------------
140269573822208
140269565429504
140269557036800
140269582214912
0---------------------
 

C++ (Qt)
#include <vector>
#include <functional>
#include <thread>
#include <mutex>
#include <iostream>
 
using namespace std;
 
int main()
{
  static constexpr size_t num_threads = 4;
  mutex print_mutex;
 
  vector< function< void() > > workers;
  workers.reserve( num_threads );
 
  for( size_t i = 0; i < num_threads; ++i)
      workers.push_back([&]()
      {
          this_thread::sleep_for( std::chrono::seconds( 1 ) );
          lock_guard<mutex> lk( print_mutex );
          cout << this_thread::get_id() << endl;
      });
 
  const size_t ntests = 3;
  for( size_t n = 0; n < ntests; ++n )
  {
    #pragma omp parallel for
    for( size_t i = 0; i < workers.size(); ++i )
      workers[ i ]();
 
    std::cout << n << "---------------------" << std::endl;
  }
 
  return 0;
}
 

C++ (Qt)
if (--workersInProcess == 0)
 ...

C++ (Qt)
struct Task : public QRunnable {
  Task( Data * data ) : mData(data) {}  
  virtual void run( void ) { DoCalc(*mData); }
 
  Data * mData;
};
 
// main thread
QThreadPool * pool = QThreadPool::globalInstance();
while (!solved) {
 
// заряжаем данные расчетов
  std::vector<Data> data;
  for (size_t i = 0; i < num_tasks; ++i)
    data.push_back(Data(i)); 
 
// запускаем
  for (size_t i = 0; i < num_tasks; ++i)
    pool->start(new Task(&data[i])); 
 
// ждем когда сварится
   pool->waitForDone();
 
   solved = AnalyzeResults(data);
};

C++ (Qt)
#ifndef THREAD_POOL_H
#define THREAD_POOL_H
 
#include <list>
#include <queue>
#include <thread>
#include <mutex>
#include <condition_variable>
#include <atomic>
#include <future>
#include <memory>
#include <functional>
 
class thread_pool
{
public:
    thread_pool(unsigned int nthreads = 1)
        : m_is_stop(false)
    {
        for (unsigned int i = 0; i < nthreads; ++i)
            m_threads.push_back(std::thread(&thread_pool::worker, this));
    }
 
    thread_pool(const thread_pool &) = delete;
    thread_pool(thread_pool &&) = delete;
 
    thread_pool & operator=(const thread_pool &) = delete;
    thread_pool & operator=(thread_pool &&) = delete;
 
    void join()
    {
        m_is_stop = true;
        m_loop_cv.notify_all();
 
        for (auto & th : m_threads)
        {
            if (th.joinable())
                th.join();
        }
    }
 
    template<class F>
    auto add_task(F && work_function)->std::future<decltype(work_function())>
    {
        auto task_ptr = std::make_shared<std::packaged_task<decltype(work_function())()>>(std::forward<F>(work_function));
 
        std::function<void()> wrapper_func = [task_ptr]() { (*task_ptr)(); };
 
        {
            std::lock_guard<std::mutex> lock(m_mutex);
            m_queue.push(wrapper_func);
        }
 
        m_loop_cv.notify_one();
 
        return task_ptr->get_future();
    }
 
private:
    std::atomic_bool m_is_stop;
    std::list<std::thread> m_threads;
    std::queue<std::function<void()>> m_queue;
    std::mutex m_mutex;
    std::condition_variable m_loop_cv;
 
    void worker()
    {
        while (!m_is_stop)
        {
            std::unique_lock<std::mutex> lock(m_mutex);
            m_loop_cv.wait(lock, [&]()
            {
                return !m_queue.empty() || m_is_stop;
            });
 
            if (m_is_stop)
                return;
 
            std::function<void()> work_function(m_queue.front());
            m_queue.pop();
 
            lock.unlock();
 
            work_function();
        }
    }
 
};
 
#endif // THREAD_POOL_H
 

C++ (Qt)
#include <iostream>
 
#include "thread_pool.h"
 
int main()
{
    static constexpr size_t nthreads = 4;
    static constexpr size_t ntests = 3;
    static constexpr size_t ntasks = 12;
    std::mutex mutex;
 
    std::vector<std::future<void>> results(ntasks);
 
    thread_pool pool(nthreads);
 
    for (size_t i = 0; i < ntests; ++i)
    {
        for (size_t j = 0; j < ntasks; ++j)
            results[j] = pool.add_task([&]()
            {
                std::this_thread::sleep_for(std::chrono::seconds(1));
                std::lock_guard<std::mutex> lk(mutex);
                std::cout << std::this_thread::get_id() << std::endl;
            });
 
        for (size_t j = 0; j < results.size(); ++j)
            results[j].wait();
 
        std::cout << i << "---------------------" << std::endl;
    }
 
    pool.join();
 
    return 0;
}
 

C++ (Qt)
...
     {
                std::unique_lock<std::mutex> lock(m_cv_mutex);
                m_loop_cv.wait(lock, [&]()
                {
                    std::lock_guard<std::mutex> lk(m_queue_mutex);
                    return !m_queue.empty() || m_is_stop;
                });
      }
...
 

C++ (Qt)
    void worker()
    {
        while (!m_is_stop)
        {
            {
                std::unique_lock<std::mutex> lock(m_mutex);
                m_loop_cv.wait(lock, [&]()
                {
                    return !m_queue.empty() || m_is_stop;
                });
            }
 
            if (m_is_stop)
                return;
 
            std::function<void()> work_function([](){});
 
            {
                // std::lock_guard<std::mutex> lock(m_mutex);  <<< и так золочен
                // if (!m_queue.empty())  <<< тут точно не пусто - проверено в wait
                {
                    work_function = std::move(m_queue.front());
                    m_queue.pop();
                }
            }
           // Разлочиваем и запускаем работу
           lock.unlock();
 
           work_function();
        }
    }
 

C++ (Qt)
while (!m_is_stop)
        {
            {
                std::unique_lock<std::mutex> lock(m_mutex);
                m_loop_cv.wait(lock, [&]()
                {
                    return !m_queue.empty() || m_is_stop;
                });
            }
            // <<< Здесь lock уже откинулся и освободил m_mutex 
 
            if (m_is_stop)
                return;
 
            std::function<void()> work_function([](){});
 
            {
                // std::lock_guard<std::mutex> lock(m_mutex);  <<< и так золочен 
                // if (!m_queue.empty())  <<< тут точно не пусто - проверено в wait
                {
                    work_function = std::move(m_queue.front());
                    m_queue.pop();
                }
            }
           // Разлочиваем и запускаем работу
           lock.unlock();
 
           work_function();
        }
    }
 

C++ (Qt)
void worker()
    {
        while (!m_is_stop)
        {
            std::unique_lock<std::mutex> lock(m_mutex);
             m_loop_cv.wait(lock, [&]()
             {
                    return !m_queue.empty() || m_is_stop;
             });
 
 
            if (m_is_stop)
                return;
 
            std::function<void()> work_function(m_queue.front());
            m_queue.pop();
            lock.unlock();
 
           work_function();
        }
    }
 

C++ (Qt)
wait( mutex m )
{
  m.unlock()
  yield()
  m.lock()
}
 

C++ (Qt)
auto task_ptr = std::make_shared<std::packaged_task<decltype(work_function())()>>(std::forward<F>(work_function));
 

C++ (Qt)
class thread_pool
{
...
private:
...
    std::list<std::shared_ptr<std::thread>> m_threads;
...
};
 

C++ (Qt)
vector<thread> m_threads;
 

C++ (Qt)
 
for(uint_type steps = 0; !tol(radius, steps); ++steps)
        {
            mean_x = x;
 
            for (uint_type i = 0; i < point_per_step; ++i)
            {
                std::uniform_real_distribution<real_type> dist(mean_x - radius, mean_x + radius);
                vector_type c = dist(rng);
                real_type cur_val = function(cx);
 
                if (cur_val < min)
                {
                    x = std::move(cx);
                    min = cur_val;
                }
            }
 
            dx = x - mean_x;
            real_type sqr_ds = dx*dx; /* ds^2 = (dx, dx) */
 
            if (sqr_ds < radius*radius*mean_sqr_radius)
                radius *= m_params.alpha;
            else
               radius = std::min(m_params.init_radius, radius*m_params.beta);
 
            observer(x, min, steps);
        }
 

C++ (Qt)
#include <iostream>
 
#include "minsearch.h"
#include "model_energy.h"
#include "potential.h"
 
static constexpr unsigned int NAtomsA = 50; // Number of atoms type A
static constexpr unsigned int NAtomsB = 50; // Number of atoms type B
static constexpr unsigned int NAtoms = NAtomsA + NAtomsB; // Total number of atoms
static constexpr unsigned int Dim = 2; // Physical dimension
 
int main()
{
 
    ABModelEnergy<double, NAtomsA, NAtomsB> modelEnergy(std::bind(potential, std::placeholders::_1, 1.0, -10.0, 1.0),
                                                            std::bind(potential, std::placeholders::_1, 1.0, -12.0, 1.0),
                                                            std::bind(potential, std::placeholders::_1, 0.0, -20.0, 4.0));
 
    std::vector<double> x(Dim*NAtoms, 0.0); // The initial phase vector (all atoms are in the zero point now)
 
    double init_radius = 0.1;
    unsigned int point_per_step = 8*10;
    double alpha = 0.9;
    double beta = 2.0;
 
    //thread pool realisation
    specmath::minsearch<double> minsearch(init_radius, point_per_step, alpha, beta);
 
    auto start = std::chrono::high_resolution_clock::now();
 
    const double eps = 1e-6;
    const unsigned long max_steps = 3000000000;
 
    double res = minsearch.find_minimum(modelEnergy, x, specmath::breaker<double>(eps, max_steps));
 
    std::cout << " (multi threads) Energy = " << res << std::endl;
 
    auto stop = std::chrono::high_resolution_clock::now();
 
    auto duration = std::chrono::duration_cast<std::chrono::seconds>(stop-start).count();
 
    std::cout << "duration = " << duration << " sec" << std::endl;
 
    // single thread realisation
    specmath::minsearch_st<double> minsearch_st(init_radius, point_per_step, alpha, beta);
 
    for (auto & xi : x)
        xi = 0.0;
 
    start = std::chrono::high_resolution_clock::now();
 
    res = minsearch_st.find_minimum(modelEnergy, x, specmath::breaker<double>(eps, max_steps));
 
    std::cout << "(single threads) Energy = " << res << std::endl;
 
    stop = std::chrono::high_resolution_clock::now();
 
    duration = std::chrono::duration_cast<std::chrono::seconds>(stop-start).count();
 
    std::cout << "duration = " << duration << " sec" << std::endl;
 
    return 0;
}
 

C++ (Qt)
for(uint_type steps = 0; !tol(radius, steps); ++steps)
        {
            mean_x = x;
            std::atomic_int ntests(m_params.point_per_step);
 
            for (uint_type i = 0; i < m_num_threads; ++i)
                results[i] = pool.add_task(std::bind(task, radius, std::ref(ntests)));
 
            for (auto & res : results)
                res.wait();
            ...
 

C++ (Qt)
auto task = [&](const real_type & r, std::atomic_int & ntests)
        {
            static thread_local std::random_device rd;
            static thread_local rng_type rng(rd());
            static thread_local std::uniform_real_distribution<real_type> dist;
            typedef typename std::uniform_real_distribution<real_type>::param_type range_t;
 
            // ntests - это и есть атомарный счётчик point_per_step
            while (std::atomic_fetch_sub(&ntests, 1) > 0)
            {
                vector_type cx = mean_x;
                std::transform(mean_x.begin(), mean_x.end(), cx.begin(), [&](const real_type & x0){ return dist(rng, range_t(x0-r, x0+r)); });
                real_type cur_val = function(cx);
 
                std::lock_guard<std::mutex> locker(mutex);
                if (cur_val < min)
                {
                    x = std::move(cx);
                    min = cur_val;
                }
            }
        };
 

C++ (Qt)
auto task = [&](const real_type & r, std::atomic_int & ntests)
        {
            static thread_local std::random_device rd;
            static thread_local rng_type rng(rd());
            static thread_local std::uniform_real_distribution<real_type> dist;
            typedef typename std::uniform_real_distribution<real_type>::param_type range_t;
 
            // ntests - это и есть атомарный счётчик point_per_step
            while (std::atomic_fetch_sub(&ntests, 1) > 0)
            {
                vector_type cx = mean_x;
                std::transform(mean_x.begin(), mean_x.end(), cx.begin(), [&](const real_type & x0){ return dist(rng, range_t(x0-r, x0+r)); });
                real_type cur_val = function(cx);
 
                std::lock_guard<std::mutex> locker(mutex);
                if (cur_val < min)
                {
                    x = std::move(cx);
                    min = cur_val;
                }
            }
        };
 

C++ (Qt)
               ...
               real_type cur_val = function(cx);
 
                if (cur_val < min) { 
                  std::lock_guard<std::mutex> locker(mutex);
                  if (cur_val < min)
                  {
                      x = std::move(cx);
                      min = cur_val;
                  }
                }
 

C++ (Qt)
// worker
void MyTask::operator()( void )
{
  ....
  if (--MyTask::taskCount <= 0) {
    std::unique_lock..
    MyTask::cond.notify_one();
  }
}
 
// main thread
MyTask::taskCount = tasks.size();
for (int i = 0; i < tasks.size(); ++i
 pool.submit(tasks[i]);
 
while (MyTask::taskCount) {
 std::unique_lock..
 MyTask::cond.wait(lock);
}

C++ (Qt)
    /**
     * \b Effects: close the \c basic_thread_pool for submissions.
     * The worker threads will work until there is no more closures to run.
     */
    void close()
    {
      work_queue.close();
    }
 

C++ (Qt)
// worker
void MyTask::operator()( void )
{
  ....
 if (--MyTask::taskCount <= 0) 
  MyTask::mPromise.set_value(1);
}
 
// main thread
MyTask::taskCount = tasks.size();
MyTask::mPromise = std::promise<int> ();
std::future<int> future = MyTask::mPromise.get_future();
 
for (int i = 0; i < tasks.size(); ++i)
 pool.submit(tasks[i]);
 
future.wait();
}

C++ (Qt)
// worker
void MyTask::operator()( void )
{
  ....
 if (--MyTask::taskCount <= 0) 
  MyTask::mPromise.set_value(1);
}
 

C++ (Qt)
#ifndef THREAD_POOL_H
#define THREAD_POOL_H
 
#include <list>
#include <queue>
#include <thread>
#include <mutex>
#include <condition_variable>
#include <atomic>
#include <future>
#include <memory>
#include <functional>
 
 
class thread_pool
{
public:
    thread_pool(size_t num_threads = std::max(size_t(1), size_t(std::thread::hardware_concurrency())))
        : m_is_stop(false), m_counter(0)
    {
        for (size_t i = 0; i < num_threads; ++i)
            m_threads.push_back(std::thread(&thread_pool::worker, this));
    }
 
    thread_pool(const thread_pool &) = delete;
    thread_pool(thread_pool &&) = delete;
 
    thread_pool & operator=(const thread_pool &) = delete;
    thread_pool & operator=(thread_pool &&) = delete;
 
    void join()
    {
        m_is_stop = true;
        m_loop_cv.notify_all();
 
        for (auto & th : m_threads)
        {
            if (th.joinable())
                th.join();
        }
    }
 
    template<class G>
    void add_task_group(const G & task_group)
    {
        {
            std::lock_guard<std::mutex> lock(m_mutex);
            m_counter += task_group.size();
            for (const auto & task : task_group)
                m_queue.push(task);
        }
 
        m_loop_cv.notify_one();
    }
 
    void wait_for_done()
    {
        m_promise.get_future().wait();
    }
 
private:
    std::atomic_bool m_is_stop;
    std::atomic_int m_counter;
    std::list<std::thread> m_threads;
    std::queue<std::function<void()>> m_queue;
    std::mutex m_mutex;
    std::condition_variable m_loop_cv;
    std::promise<void> m_promise;
 
    void worker()
    {
        while (!m_is_stop)
        {
            std::unique_lock<std::mutex> lock(m_mutex);
            m_loop_cv.wait(lock, [&]()
            {
                return !m_queue.empty() || m_is_stop;
            });
 
            if (m_is_stop)
                return;
 
            std::function<void()> work_function(m_queue.front());
            m_queue.pop();
 
            lock.unlock();
 
            work_function();
 
            if (!--m_counter)
                m_promise.set_value();
        }
    }
};
 
#endif // THREAD_POOL_H
 

C++ (Qt)
thread_pool pool;
 
    const size_t num_tasks = 100;
 
    std::vector<std::function<void()>> tasks;
    for (size_t i = 0; i < num_tasks; ++i)
        tasks.push_back(task);
 
    pool.add_task_group(tasks);
 
    pool.wait_for_done();
 
    pool.join();
 

C++ (Qt)
// worker
void MyTask::operator()( void )
{
  ....
 if (--MyTask::taskCount <= 0) 
  MyTask::mPromise.set_value(1);
}
 

C++ (Qt)
if (std::atomic_fetch_sub(&MyTask::taskCount, 1) == 1) 
  MyTask::mPromise.set_value(1);
 

C++ (Qt)
std::atomic_int test(5);
if (--test <= 0)
	....;
 

ASM
        lock sub        QWORD PTR taskCount[rip], 1
        je      .L13
        ret
 

C++ (Qt)
template<class G>
    void add_task_group(const G & task_group)
    {
        {
            std::lock_guard<std::mutex> lock(m_mutex);
            m_counter += task_group.size();
            m_promise = std::promise<void>();
            for (const auto & task : task_group)
                m_queue.push(task);
        }
 
        m_loop_cv.notify_one();
    }
 

C++ (Qt)
threap_pool pool;
 
pool.add_tasks_group(tasks_group);
 
...
 
pool.add_tasks_group(tasks_group);
 
pool.wait_for_done();
 

C++ (Qt)
#include <taskflow/taskflow.hpp>  // Cpp-Taskflow is header-only
 
int main(){
 
  tf::Executor executor;
  tf::Taskflow taskflow;
 
  auto [A, B, C, D] = taskflow.emplace(
    [] () { std::cout << "TaskA\n"; },               //  task dependency graph
    [] () { std::cout << "TaskB\n"; },               // 
    [] () { std::cout << "TaskC\n"; },                    //            +---+          
    [] () { std::cout << "TaskD\n"; }                     //    +---->| B |-----+   
  );                                                                    //    |        +---+     |
                                                                        //  +---+           +-v-+ 
  A.precede(B);  // A runs before B                  //  | A  |            | D  | 
  A.precede(C);  // A runs before C                  //  +---+           +-^-+ 
  B.precede(D);  // B runs before D                     //    |     +---+     |    
  C.precede(D);  // C runs before D                  //    +---->| C |-----+    
                                                                            //          +---+          
  executor.run(taskflow).wait();
 
  return 0;
}
 

C++ (Qt)
void worker()
    {
        while (!m_is_stop)
        {
            std::unique_lock<std::mutex> lock(m_mutex);
            m_loop_cv.wait(lock, [&]()
            {
                return !m_queue.empty() || m_is_stop;
            });
 
            if (m_is_stop)
                return;
 
            std::function<void()> work_function(m_queue.front());
            m_queue.pop();
 
            lock.unlock();
 
            work_function();
        }
    }
 

C++ (Qt)
void worker()
    {
        for(;;)
        {
            std::unique_lock<std::mutex> lock(m_mutex);
            m_loop_cv.wait(lock, [&]()
            {
                return !m_queue.empty() || m_is_stop;
            });
 
            if (m_is_stop && m_queue.empty())
                return;
 
            std::function<void()> work_function(m_queue.front());
            m_queue.pop();
 
            lock.unlock();
 
            work_function();
        }
    }
 

C++ (Qt)
  void	thread_run()
  {
    for(;;)
    {
      std::unique_lock<std::mutex> locker( m_mutex );
      for(;;)
      {
        if( m_stop )
          return;
        else if( !m_tasks.empty() )
          break;
        m_cond.wait( locker );
      }
 
      Task task = m_tasks.front();
      m_tasks.pop();
 
      locker.unlock();
 
      if( task )
        task();
    }
  }
 

C++ (Qt)
  ~ThreadPool()
  {
    {
      std::unique_lock<std::mutex> locker( m_mutex );
      m_stop = true;
    }
    m_cond.notify_all();
 
    for( auto &it : m_threads )
      if( it.joinable() )
        it.join();
  }
 

C++ (Qt)
class thread_pool
{
public:
    explicit thread_pool(size_t num_threads = std::max(size_t(1), size_t(std::thread::hardware_concurrency())))
        : m_stop(false)
    {
        for (size_t i = 0; i < num_threads; ++i)
            m_threads.emplace_back(&thread_pool::worker, this);
    }
 
    thread_pool(const thread_pool &) = delete;
    thread_pool(thread_pool &&) = delete;
 
    thread_pool & operator=(const thread_pool &) = delete;
    thread_pool & operator=(thread_pool &&) = delete;
 
    ~thread_pool()
    {
        {
            std::lock_guard<std::mutex> locker(m_mutex);
            m_stop = true;
        }
 
        m_cond.notify_all();
 
        for (auto & th : m_threads)
        {
            if (th.joinable())
                th.join();
        }
    }
 
    void wait_for_all()
    {
        std::unique_lock<std::mutex> locker(m_mutex);
        m_cond_wait.wait(locker, [&]() { return m_queue.empty(); });
    }
 
    template<class F>
    auto add_task(F && f)->std::future<decltype(f())>
    {
        using R = decltype(f());
 
        auto task_ptr = std::make_shared<std::packaged_task<R()>>(std::forward<F>(f));
 
        {
            std::lock_guard<std::mutex> locker(m_mutex);
            m_queue.emplace( [=]() { (*task_ptr)(); } );
        }
 
        m_cond.notify_one();
 
        return task_ptr->get_future();
    }
 
private:
    std::atomic_bool m_stop;
    std::list<std::thread> m_threads;
    std::queue<std::function<void()>> m_queue;
    std::mutex m_mutex;
    std::condition_variable m_cond;
    std::condition_variable m_cond_wait;
 
    void worker()
    {
        for(;;)
        {
            std::unique_lock<std::mutex> locker(m_mutex);
            m_cond.wait(locker, [&]()
            {
                return !m_queue.empty() || m_stop;
            });
 
            if (m_stop)
                return;
 
            std::function<void()> task(m_queue.front());
            m_queue.pop();
 
            locker.unlock();
 
            if (task)
                task();
 
            m_cond_wait.notify_one(); // <== Эта условная переменная для wait_for_all 
        }
    }
};