К вопросу об организации взаимодействия пула производителей и одного потребителя

[Old]

Вы писали что попробовали бустовский пул. В бусте пул поток есть только в asio и это не то, что пробовали вы. У вас какая то библиотека не буста.

Беру свои слова назад.
Нашел этот пул в экспериментальных расширениях thread.
 

[Igors]

Можно оформить через future+promise

Код

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();
}

Как всегда, std сэкономило пару строк. Но делать футуру на каждую (из многочисленных) задач - явная глупость  

[Old]

Но делать футуру на каждую (из многочисленных) задач - явная глупость  

Ну конечно.
Уже согласились на одну.
Подождем, через пару лет вы будете здесь всем рекомендовать их использовать для всех задач.

Ну хоть что-то полезное случилось, Igors освоил future+promise.
Теперь следующий шажок понять что package_task объединяет future+promise и функтор задачи и основы будут освоены.

[m_ax]

Можно оформить через future+promise

И чем же такое решение лучше? 

[Igors]

И чем же такое решение лучше? 

Лучше предыдущего что я приводил? Да в принципе ничем, ну пару строк сэкономили (правда ценой запоминания глупых классов). То я вижу что для Вас std - пуп Земли, ну и заглянул в справочник 

А вот "ничтоже сумняшеся" создавать потенциально большой контейнер жирных структур - тяжелая ошибка, и никакие познания в std/boost ее не компенсируют

[Old]

А вот "ничтоже сумняшеся" создавать потенциально большой контейнер жирных структур - тяжелая ошибка, и никакие познания в std/boost ее не компенсируют

А почему вы решили, что эти структуры такие жирные для помещения в потенциально большой контейнер?

[m_ax]

Лучше предыдущего что я приводил?

Нет, я имел в виду в сравнении с моей реализацией..
Минусом Вашей реализацией является то, что я должен оборачивать таску в MyTask, заводить там статический каунтер и промис. Уже не безопасно, кто-нибудь да и выстрелит себе в ногу.. Однако, это поправимо и решаемо. Но на этом проблемы не заканчиваются.

Код

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

Здесь мы словим исключение, в попытке более одного раза записать значение в promise.
Но это тоже решаемо..

Давайте посмотрим на примерную реализацию такого пула с методом wait_for_done:

Код

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();
 

Т.е. чтоб не выстрелить себе в ногу, мы как минимум должны подавать в очередь не по одной задачи, а целую группу.
И тем не менее, этот код безопасен?

[Igors]

..заводить там статический каунтер и промис. Уже не безопасно, кто-нибудь да и выстрелит себе в ногу..

Не то чтобы "небезопасно", а просто "коряво" как и любой статик. Конечно сделать эти переменные членами пула - намного лучше, но я ж дустовский пул не мог кромсать  

Здесь мы словим исключение, в попытке более одного раза записать значение в promise.

Не должны, на то он и атомик чтобы установиться лишь раз. Кстати не вижу где у Вас сброс promise перед след раундом? Наверно надо добавить в конец wait_for_done

Код

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();
            for (const auto & task : task_group)
                m_queue.push(task);
        }
 
        m_loop_cv.notify_one();
    }
 

Наверное notify_all (задач-то прибыло не одна). А так все хорошо.  

[Igors]

Код

C++ (Qt)
    void join()
    {
        m_is_stop = true;
        m_loop_cv.notify_all();
 
        for (auto & th : m_threads)
        {
            if (th.joinable())
                th.join();
        }
    }
 

notify_all должно быть под мутексом - на этот момент какие-то нитки могут еще не уснуть.

[Old]

Не должны, на то он и атомик чтобы установиться лишь раз.

Ну давайте посмотрим:

Код

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

Предположим остались 2 задачи в очереди (taskCount = 2).
Завершается первая из них, она успевает уменьшить taskCount на 1, но не успела пройти сравнение с 0 и была вытеснена (taskCount = 1).
Управление получила другая задача, отработала, уменьшила taskCount на единицу и выполнила set_value (taskCount = 0).
Управление получает первая задача, сравнивает taskCount c 0, он действительно 0 и она торжественно вызывает set_value второй раз.

[Old]

на этот момент какие-то нитки могут еще не уснуть.

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

[m_ax]

А так все хорошо.

Нет, нехорошо( Нам в любом случае нужет контейнер task_group. И хз какие по жирности (по размеру) там будут таски.
Уж лучше опционально иметь контейнер футур)

[Igors]

Нет, нехорошо( Нам в любом случае нужет контейнер task_group. И хз какие по жирности (по размеру) там будут таски.

Не понял в чем проблема

[Old]

Не понял в чем проблема

Все хорошо. Сойдет.

[m_ax]

Не понял в чем проблема

Я не вижу чем это решение как минимум лучше.