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

[Old]

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

Как минимум?
Да оно дырявое. Я выше все расписал.
Мы взяли надежные future с promise и сделали на их основе дырявый механизм. Для чего? Что бы счетчик атомарный задействовать?
А потом все время надеяться, что бы компилятор хорошо код сгенерил без команды сравнения. В отладке соберешь - будет дыра, в релизе соберешь - будет зависеть от компилятора и уровня оптимизации.

[m_ax]

Мы взяли надежные future с promise и сделали на их основе дырявый механизм. Для чего? Что бы счетчик атомарный задействовать?

Я так понял, что это всё для того, чтоб сделать подобие waitForDone 

А потом все время надеяться, что бы компилятор хорошо код сгенерил без команды сравнения. В отладке соберешь - будет дыра, в релизе соберешь - будет зависеть от компилятора и уровня оптимизации.

Не, ну там, справедливости ради, можно сравнение заменить на

Код

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

Но всё равно дизайн кривой получается. Да и потом повторный вызов wait_for_done, до того, как мы положим в очередь очередную группу задач выкинет исключение.

[Old]

Я так понял, что это всё для того, чтоб сделать подобие waitForDone  

Ну так взяли уже футуры, используем - там уже все есть.
Нет, добавим счетчик, для чего? Просто потому что обещал? Так вроде обещано было другое - атомики без усыпления. Где?

Не, ну там, справедливости ради, можно сравнение заменить на

Вот. Сделали решение, отдали заказчику. А у него 32 ядра и нихрена не работает. Не знаю почему.
А потом уже заменим.
Хотя есть проверенные решения в стандартной библиотеке, даже буст брать не надо.

[Old]

Интересную библиотеку нашел на github:
https://github.com/cpp-taskflow/cpp-taskflow

[m_ax]

Интересную библиотеку нашел на github:
https://github.com/cpp-taskflow/cpp-taskflow

Крутой проект, судя по описанию и отзывам) Спасибо, буду иметь в виду)

[Igors]

Не, ну там, справедливости ради, можно сравнение заменить на

Ладно, пожуем азы

Код

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

Этот код выливается по существу в команду xaddl (аттач) которая выполняется с префиксом lock. Команда или выполняется до вытеснения нитки или нет. Если вытеснение произошло после, то при возврате управления в еax торчит рез-т, т.е последующее сравнение имеет абсолютно тот же эффект что и без вытеснения. Да, если бы этот код выполнялся неск нитками, то возвращаемое значение могло быть напр 3 или 2 (а не 4), т.е. кто-то "успел до нас". И на момент сравнения атомик мог уже стать другим, это нормально. НО возвращаемое значение обязательно УНИКАЛЬНО. Это широко известный и популярный прием.

"А если вот компилятор сгенерит др код.." НЕТ, не имеет права, такое поведение атомика регламентировано. Вот если бы мы понадеялись что, мол, "int и так атомарный" - тогда да.

Я так понял, что это всё для того, чтоб сделать подобие waitForDone 

А разве Вам не это нужно?

Но всё равно дизайн кривой получается.

Да в чем же его кривизна?

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

Ну очевидно в wait_for_done надо провериться на "нет задач" и сбросить промису. Опять Вы в 2 соснах заблудились   

[Old]

Этот код выливается по существу в команду xaddl (аттач) которая выполняется с префиксом lock.

Этот код также может вылиться (gcc и clang) в

Код

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

и выполнится корректно.
Только, насколько я помню (хотя я и не знаток стандарта), это нигде не гарантируется в стандарте, то что современные компиляторы умные и генерируют корректный код, не значит что так будет всегда и везде.
Это скорее побочный эффект: в случае с xadd в регистре останется предыдущее значение счетчика, в случае с sub - будут установлены флаги, и компилятор использует их в дальнейшем.
Но компилятору никто не запрещает генерировать отдельно декремент и load для сравнения.

[m_ax]

А разве Вам не это нужно?

Но не в такой реализации.

Код

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

Т.е. при повторном вызове add_tasks_group  после ожидания на wait_for_done  мы можем получить, что не все задачи отработали.. И это косяк..
Примером изящного дизайна, решающего, в частности, и эту проблему есть Cpp-Taskflow, обозначенная по ссылке выше)  

Да, и потом, здесь мы создаём контейнер тасков. А чем это лучше контейнера футур?

[Igors]

Примером изящного дизайна, решающего, в частности, и эту проблему есть Cpp-Taskflow, обозначенная по ссылке выше) 

Плиз "ткните носиком" в это место, интересно что же там такого хорошего.

И вот здесь можно подорваться на мине (возможно ошибаюсь):
...
Т.е. при повторном вызове add_tasks_group  после ожидания на wait_for_done  мы можем получить, что не все задачи отработали.. И это косяк..

А это уже та самая "архитектура". Вы/мы механически следуем идеям/методам пула, что давно неверно. Пул предназначен для "подбрасывния" задач по мере их поступления, и это действительно очень полезный ф-ционал. Но нужен ли он здесь? Вам виднее, но из того что я слышал - думаю что нет. У Вас четко виден "раунд/сессия" вычислений - известно число задач и данные для каждой. Поэтому логичнее делать по образцу OpenMP или QtConcurrent, т.е.

- заряжаем контейнер задач в главной нитке
- запускаем наш класс и ждем завершения

Да, при этом мы полагаем что "зарядка" достаточно дешева, она выполняется в главной нитке. Если зарядка обнаруживает что (предыдущие) вычисления еще в процессе - сразу assert.

Владение (о котором так любят говорить). Очень неплохо сделано в QThreadPool - пул может и сам удалять задачи и нет. Почему бы не позаимствовать эту удачную идею? Также мне нравится что в пул добавляется указатель на базовый класс QRunnable (кажется он из жабы но могу путать). Ну можно и дусто-любимый оператор () call. Но копировать данные в "вычислялку" не вижу никакого смысла - у нее контейнер указателей. Также добавлять контейнер не очень - "по одному" проще и лучше, ведь вычислялка не должна немедленно толкать выполнение.

Ну и наверно метод Run (запуск вычислений) должен иметь флажок async. Ну и геттеров там подрисовать (типа GetProgress и.т.п)

Да, и потом, здесь мы создаём контейнер тасков. А чем это лучше контейнера футур?

Тем что первый вызывается необходимостью, а второй нет.

[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]

Пул предназначен для "подбрасывния" задач по мере их поступления, и это действительно очень полезный ф-ционал. Но нужен ли он здесь? Вам виднее, но из того что я слышал - думаю что нет. У Вас четко виден "раунд/сессия" вычислений - известно число задач и данные для каждой. Поэтому логичнее делать по образцу OpenMP или QtConcurrent

Под капотом и у OpenMP и у QtConcurrent все тот же пул. Эти инструменты берут на себя "нарезку на задачи" и загрузку пула, но пул никуда не девается. Здесь можно не пытаться уйти от пула, а написать алгоритм по аналогии с каким нибудь QtConcurrent::mapped и все средства для этого в пуле m_ax уже есть.

[Old]

Не хочу быть навязчивым, но это классическая ошибка которую делают ВСЕ (что бы потом ни свистели  ). Видимо Вам это показалось не очень важным - ну подождем пока Вы ее опять сделаете 

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

[m_ax]

Плиз "ткните носиком" в это место, интересно что же там такого хорошего.

Ну хотя бы вот как можно легко создавать различные графы исполнения тасков, что частенько требуется во всяких вычислениях (пример с сайта: https://github.com/cpp-taskflow/cpp-taskflow)

Код

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

У меня, фактически, тоже граф.. циклический)

А это уже та самая "архитектура". Вы/мы механически следуем идеям/методам пула, что давно неверно.

Вы же вначале говорили, что всё хорошо:

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

Если такое решение попадёт в руки, какому-нибудь пользователю, то наверняка он себе что-нить да отстрелит) Это пример плохой, непродуманной архитектуры.

Вы/мы механически следуем идеям/методам пула, что давно неверно.

Что Вы предлагаете? Можно это формализовать в виде кода, чтоб понять о чём идея?

Если зарядка обнаруживает что (предыдущие) вычисления еще в процессе - сразу assert.

Нет уж.. мне такие неожиданности не нужны.. Мне нужно все расчёты безопасно довести до конца. Без всяких возможных асёртов)

Почему бы не позаимствовать эту удачную идею?

Я считаю, что вариант, когда пул имеет возможность предоставлять футуру для любой задачи более гибкая.
  

Не хочу быть навязчивым, но это классическая ошибка которую делают ВСЕ

Ошибка в другом месте.. Покажу позже..

[m_ax]

Не хочу быть навязчивым, но это классическая ошибка которую делают ВСЕ

Да, накасячил, конечно  Но проблема не в condition_variable а здесь:

Код

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

Если сразу после заполнения очереди вызвать join(), m_is_stop == true и в очереди могут остаться невыполненные задачи.
Стыдно на такие грабли было наступить)
Выходить из цикла мы должны только при совместном выполнении условий: когда и очередь пуста и флаг m_is_stop == true.
Т.е. правильно так:

Код

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

   

[Old]

 m_ax, это спорно.
У вас join нужно переименовать в stop, потому что он не ждет завершения работы задач, а завершает работу пула вообще. Если нам нужно завершить программу, уже все равно остались задачи в очереди или нет.
А завершения всех задач вы ждете по футурам.
Кстати, чтобы не плодить проверок, я делаю проверку в нитках чуть по другому. Когда доберусь до компьютера покажу как.