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

[m_ax]

Моя задумка была обойтись без счетчика задач, т.е. сохранить весь ф-ционал пула но добавить WaitForDone.

Ну ну.. А mInCalc это тогда что такое? Не счётчик ли

[m_ax]

Кстати а Вашу "футуристическую" реализацию не хотите прогнать на этих задачках?  

Прогнал) Одинаково) (см. аттач)

[Igors]

Ну ну.. А mInCalc это тогда что такое? Не счётчик ли

Счетчик, да не тот - это всего лишь кол-во задач которые сейчас выполняются. Об общем числе задач этот вариант понятия не имеет, он отлавливает ситуацию когда "очередь пуста" + "нет выполняющихся задач". На первый взгляд это совсем просто

Тут я немного попутал

Код

C++ (Qt)
     template <class F>
    auto add_task(F && task)->std::future<decltype(task())>
    {
        std::lock_guard<std::mutex> locker(m_mutex);
        ++m_task_count;
        return m_pool.add_task(std::bind(&wrapper_pool::wrapper_task, this, std::forward<F>(task)));
    }

Это здесь мутекс не нужен. Атомарный счетчик корректно инкременируется/декрементируется любым числом ниток. Из этого часто делается вывод типа "если атомик то мутекс не нужен". Это обычно не так, но в данном случае верно Разумеется полагаем что у m_pool есть свой мутекс для защиты очереди

[m_ax]

Счетчик, да не тот - это всего лишь кол-во задач которые сейчас выполняются. Об общем числе задач этот вариант понятия не имеет, он отлавливает ситуацию когда "очередь пуста" + "нет выполняющихся задач". На первый взгляд это совсем просто

Это я понял  Но это, тем не менее, счётчик)

Это здесь мутекс не нужен.

Это почему это он там не нужен? Он нужен и old выше уже объяснял почему необходимо лочить этот каунтер, дажеесли он атомарен..

[m_ax]

Кстатии, а Вы проверяли тесты с футурами и без них? Я проверял  

update: Там в аттаче небольшая ошибка с wrapper_pool, которая, однако, никак не влияет на результаты.. Правильнее реализовать его так (чтоб любые футуры получать):

Код

C++ (Qt)
class wrapper_pool
{
public:
    wrapper_pool(specmath::thread_pool & pool)
        : m_pool(pool), m_task_count(0)
    {}
 
    template <class F>
    auto add_task(F && task)->std::future<decltype (task())>
    {
        using R = decltype (task());
 
        std::lock_guard<std::mutex> locker(m_mutex);
        ++m_task_count;
        return m_pool.add_task(std::bind(&wrapper_pool::wrapper_task<R>, this, std::forward<F>(task)));
    }
 
    void wait_for_all()
    {
        std::unique_lock<std::mutex> locker(m_mutex);
        m_cond.wait(locker, [&]()
        {
            return !m_task_count;
        });
    }
 
private:
    specmath::thread_pool & m_pool;
    std::atomic_int m_task_count;
    std::mutex m_mutex;
    std::condition_variable m_cond;
 
    template <class R>
    R wrapper_task(const std::function<R()> & task)
    {
        auto res = task();
        {
            std::unique_lock<std::mutex> locker(m_mutex);
            if (--m_task_count == 0)
            {
                locker.unlock();
                m_cond.notify_one();
            }
        }
        return res;
    }
};
 

[Igors]

Это почему это он там не нужен?

Пусть счетчик был 0 и теперь, без всякой защиты, стал 1. В рез-те wait_for_all могло уснуть хотя должно было пробудиться. Но это ничем не грозит, когда-то счетчик обнулится и wait_for_all проснется.

Теперь здесь.

Код

C++ (Qt)
    template <class R>
    R wrapper_task(const std::function<R()> & task)
    {
        auto res = task();
        {
            std::unique_lock<std::mutex> locker(m_mutex);
            if (--m_task_count == 0)
            {
                locker.unlock();
                m_cond.notify_one();
            }
        }
        return res;
    }

О том что лочить все подряд нехорошо мы уже говорили. Но здесь "защищать условие" нет смысла. Как только случился unlock, кто-то может сунуть задачу в очередь, m_task_count станет ненулевым, в рез-те wait_for_all не уснет или продолжит спать. Ситуация та же что и в первом случае - если пул пополняется со стороны, то можно потерять моменты когда он был пуст. Правда это ничем особым не грозит. Лучше так

Код

C++ (Qt)
    template <class R>
    R wrapper_task(const std::function<R()> & task)
    {
        auto res = task();
        if (--m_task_count == 0)
            std::unique_lock<std::mutex> locker(m_mutex);
            if (m_task_count == 0) 
             m_cond.notify_one();
        }
        return res;
    }

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

Да, и опять забыл! Там небольшая помарка: в случае abort нужно отпустить ожидающего

[Igors]

Кстатии, а Вы проверяли тесты с футурами и без них? Я проверял  

Да уж  

1) Контрольная сумма не бьется, где-то насвистели (наверное с рандомом)
2) WEIGHT_TASK = 10000, т.е. опять "только хорошая задача"
3) Проходов - аж один, и все раунды одинаковы, имитировать гранулярность не нужно

Ну ладно, изменил печать и увеличил число проходовю Ставим WEIGHT_TASK = 64

pass 0
with std::futures<double>  result = -1.00525e+06 total time (ms) = 4299
with wait_for_all  result = -1.00526e+06 total time (ms) = 6061

pass 1
with std::futures<double>  result = -1.00517e+06 total time (ms) = 3883
with wait_for_all  result = -1.00547e+06 total time (ms) = 5927

pass 2
with std::futures<double>  result = -1.00534e+06 total time (ms) = 3781
with wait_for_all  result = -1.0053e+06 total time (ms) = 5817

pass 3
with std::futures<double>  result = -1.00513e+06 total time (ms) = 3962
with wait_for_all  result = -1.0052e+06 total time (ms) = 5824

End

Смотрите, как быстрее с футурой! (я же говорил!) и.т.п.   Ладно, теперь уберем бездумно насаженные Вами мутексы (там я поставил define в хедере). Имеем

pass 0
with std::futures<double>  result = -1.00545e+06 total time (ms) = 4353
with wait_for_all  result = -1.00562e+06 total time (ms) = 3767

pass 1
with std::futures<double>  result = -1.00538e+06 total time (ms) = 4097
with wait_for_all  result = -1.00502e+06 total time (ms) = 3749

pass 2
with std::futures<double>  result = -1.0054e+06 total time (ms) = 3925
with wait_for_all  result = -1.00499e+06 total time (ms) = 3758

pass 3
with std::futures<double>  result = -1.00513e+06 total time (ms) = 3911
with wait_for_all  result = -1.00494e+06 total time (ms) = 3776

End

И выясняется что футуры чего-то весят, как оно и должно быть. И что лепить мутексы где ни попадя не так уж хорошо  

Изменения в аттаче

[m_ax]

Пусть счетчик был 0 и теперь, без всякой защиты, стал 1. В рез-те wait_for_all могло уснуть хотя должно было пробудиться. Но это ничем не грозит, когда-то счетчик обнулится и wait_for_all проснется.

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

Чтож.. Убедили, согласен  

Правда у меня Ваши тесты показывают +- одинаковые результаты

Код

C++ (Qt)
const size_t WEIGHT_TASK = 64;
const size_t NUM_TASKS = 1024 * 1000;
const size_t NUM_PASSES = 8;
 
#define M_AX	0
 

Вывод:

Код

Bash
pass 0
with std::futures<double>  result = -52.8008 total time (ms) = 5184
with wait_for_all  result = -156.814 total time (ms) = 5220
 
pass 1
with std::futures<double>  result = -66.3582 total time (ms) = 4875
with wait_for_all  result = -16.7101 total time (ms) = 5256
 
pass 2
with std::futures<double>  result = 209.773 total time (ms) = 4785
with wait_for_all  result = 137.421 total time (ms) = 5227
 
pass 3
with std::futures<double>  result = -109.539 total time (ms) = 4818
with wait_for_all  result = -311.828 total time (ms) = 5272
 
pass 4
with std::futures<double>  result = 64.4322 total time (ms) = 4721
with wait_for_all  result = -55.3244 total time (ms) = 5272
 
pass 5
with std::futures<double>  result = 141.65 total time (ms) = 4828
with wait_for_all  result = 272.018 total time (ms) = 5184
 
pass 6
with std::futures<double>  result = 7.6451 total time (ms) = 5081
with wait_for_all  result = -56.6884 total time (ms) = 5194
 
pass 7
with std::futures<double>  result = -171.047 total time (ms) = 4989
with wait_for_all  result = -130.11 total time (ms) = 5253
 
End
 

1) Контрольная сумма не бьется, где-то насвистели (наверное с рандомом)

Да, насвистел.. Во-первых здесь:

Код

C++ (Qt)
for (size_t i = 0; i < data.size(); ++i)
    {
        double val = dist(gen);
        data[i] = val * 2 - 1.0;
    }
 

нужно заменить на:

Код

C++ (Qt)
 for (size_t i = 0; i < data.size(); ++i)
    {
        data[i] = dist(gen);
    }
 

а во-вторых, при таком WEIGHT_TASK = 64 такие флуктуации совершенно нормальны..

Всё же лучше так:

Код

C++ (Qt)
void wrapper_task(const std::function<void()> & task)
    {
        if (task)
            task();
 
	#if M_AX
        std::unique_lock<std::mutex> locker(m_mutex);
        if (--m_task_count == 0) {
            locker.unlock();
            m_cond.notify_one();
        }
    #else
        if (std::atomic_fetch_sub(&m_task_count, 1) == 1)
        {
            std::unique_lock<std::mutex> locker(m_mutex);
            if (m_task_count == 0)
            {
                locker.unlock(); // <== делаем unlock перед пробудкой..
                m_cond.notify_one();
            }
        }
    #endif
    }
 

[m_ax]

Вообще, в идеале, хотелось бы реализовать идею cpp-taskflow https://github.com/cpp-taskflow/cpp-taskflow
Понравилась мне она очень  

Так, чисто для удовлетворения своих эстетических чувств) К тому же я могу себе это позволить)

[Old]

m_ax, а какой выигрыш принесли последние исправления?

[m_ax]

m_ax, а какой выигрыш принесли последние исправления?

Ну на моём железе, особо, никакой) Я бы вообще бы сказал, что никакой  

Ну там и тест ещё весьма спорный.. Едва ли в реальных задачах так делать  будут..

update: на мой сугубо взгляд, конечно

[Old]

Ну на моём железе, особо, никакой) Я бы вообще бы сказал, что никакой  

Если вы замерите, сколько раз рабочие нитки упирались в заблокированный мьютес, то увидите, на сколько это редкий случай. У меня на машине для 10000 задач блок случался от 10 до 20 раз. И все эти ухищрения с двойными проверками не принесут даже 1% прироста.

Плюс современные мьютексы на основных платформах делают на футексах, которые при захвате свободного мьютекса бесплатны (это изменение атомарной переменной), дальше в зависимости от реализации может покрутиться  spin-lock, а только потом будет syscall в ядро на wait futex'а.

[m_ax]

Если вы замерите, сколько раз рабочие нитки упирались в заблокированный мьютес, то увидите, на сколько это редкий случай. У меня на машине для 10000 задач блок случался от 10 до 20 раз. И все эти ухищрения с двойными проверками не принесут даже 1% прироста.

Вы имеете в виду ситуацию пробки? Когда более одной нитки ждут на мьютексе?

[Old]

да

[m_ax]

да

Ну я, специально, этот случий не расматривал.. Хотя, в общем, это не сложно посмотреть.. (Посмотрю)

Здесь, как я понял, весь затык в том, что когда задачи становятся "дешёвыми" и их много, разница в подходах с футурами и этим методом wait_for_all нивилируется, а то и последний вариант работает быстрей.. Но это, на мой взгляд, проблемы самой реализации распараллеливания  

Не будет нормальный человек так бездумно кидать задачи в пул)