Привет желающим присоединиться к исследованию Thread Pool
Насколько я понимаю, Thread Pool это вариация на тему multiple producers/multiple consumers.
Схематически, должно быть так:
0. Нужна конкурентная queue очередь
1. Делается N worker threads — например, N = количеству аппаратных параллельных ядер
2. В каждом worker запускается "выгребатель" работы из очереди
3. Функция dispatch принимает кусочки работы (например std::function) и ложит их в очередь
С точки зрения поставщика "кусочков работы" АПИ будет выглядеть например так:
void doer() {
Work work1 = [](){} /// Work обертка может быть std::function
Pool::dispatch(work1);
}
Эффективно реализовать очередь можно с помощью condition variable
Т.е auto work = queue.pop(); заблокируется до появления новых данных в очереди
Т.е каждый поток может сделать push() и каждый же поток — заблокироваться и заснуть на pop() до поступления работы.
Вопросы, которые не очень ясны.
1) Какой механизм меж-потоковой синхронизации?
Например я хочу гарантировать очередь исполнения моих ворков
допустим, из разных потоков я диспатчу:
dispatch(work1) | dispatch(work2)
я хочу чтобы work1 гарантированно исполнилась перед work2
при этом, чтобы потоки не простаивали
для этого work1 нужно не просто разместить в очереди до work2
а и гарантировать что поток, исполняющий work1 закончится до того, как другой поток возьмется за work2
Возможно тут нужны некие барьеры
2) Как эффективно передается Work между потоками?
Я так понимаю, что все thread используют общую память т.е. нет никакой нужды ничего никуда копировать? Или как?
Будет ли эффективной передача std::function по значению?
3) Как вообще делаются барьеры?
т.е чтобы был некий dispatch_barrier() который заставит все задиспатченные ранее Work выполниться до него гарантированно,
при этом сохранив возможность после него делать новые dispatch() просто добавляя новые Work в очередь
Здравствуйте, barney, Вы писали:
B>для этого work1 нужно не просто разместить в очереди до work2 B>а и гарантировать что поток, исполняющий work1 закончится до того, как другой поток возьмется за work2 B>Возможно тут нужны некие барьеры
Барьеры тут не очень нужны. Можно построить логику на условной переменной и проверке условий при выборе следующей задачи для исполнения.
B>3) Как вообще делаются барьеры? B>т.е чтобы был некий dispatch_barrier() который заставит все задиспатченные ранее Work выполниться до него гарантированно, B>при этом сохранив возможность после него делать новые dispatch() просто добавляя новые Work в очередь
Барьер это отдельный примитив синхронизации, который используется, например, в MPI на кластерах.
B>2) Как эффективно передается Work между потоками? B>Я так понимаю, что все thread используют общую память т.е. нет никакой нужды ничего никуда копировать? Или как?
Память потоков общая, хотя может быть еще свой TLS(Thread Local Storage). B>Будет ли эффективной передача std::function по значению?
Здесь это заметным образом не повлияет не на что.
У сложных вещей обычно есть и хорошие, и плохие аспекты.
Берегите Родину, мать вашу. (ДДТ)
лол да что за троллинг -то опять?
не понимаю, в чем ваша трудность? не хотите добровольно общаться об этом — не общайтесь?
я нож у горла вроде бы не держу)
мне интересно найти единомышленников, равных по уровню, которые горят, и им все еще любопытно
в этом покопаться, подискутировать
а гуру которые уже выгорели, и посылают читать мануалы — нам до вашего уровня далеко, сорян)
тут интересно понять базовые принципы для начала
а не закопаться в промышленной библиотеке на тысячи строк сложного кода
причем здесь выгорели ?
все уже давно придумано (с)
остается лишь изучить то что уже есть, а вы не хотите
если не можете читать исходники какого бы размера они не были, наверное программирование не для вас ... я так думаю
тот конкуренси по ссылкам или тот же asio не самые большие проекты, если вы в них не сможете разобраться ....
заучивание чужого — не научит мыслить, лишь мыслям.
даст готовое решение, а не навык решать.
чтобы по настоящему разобраться в теме — важно пройти путь, который прошел автор "с нуля"
понять, какие проблемы он решал, какие фундаментальные причины лежали в основе принятых решений.
ну а если привычнее потреблять готовое — то в чем проблема? пройдите мимо
зачем провоцировать меня на офтопы? зачем засирать ветку философскими вопросами?
эта ветка для таких же как я — строителей велосипедов)
тех, кто любит учиться, разбирая и изобретая с нуля. а не заучивая чужое
ну и кто здесь троллит?
B>заучивание чужого — не научит мыслить, лишь мыслям. даст готовое решение, а не навык решать.
ошибка раз, заучивают в школе, взрослые дяди изучают чужие решения. B>чтобы по настоящему разобраться в теме — важно пройти путь, который прошел автор "с нуля"
ошибка два, вся информация по теме давно уже описана теми самыми авторами, вместо хождения по граблям достаточно прочитать книгу по многопоточности. там умные люди доходчиво и структурированно описали: B>какие проблемы он решал, какие фундаментальные причины лежали в основе принятых решений.
а также достоинства и недостатки разных подходов к реализации.
B>зачем провоцировать меня на офтопы? зачем засирать ветку философскими вопросами?
потому что велосипедить с нуля тут никому не интересно.
вопрос по теме должен звучать как-то так: "использую тредпул из boost::asio, не устраивает тем-то и тем-то, попробовал исправить то, что не устраивает, но получилось не очень. подскажите что делать?"
XMK>потому что велосипедить с нуля тут никому не интересно.
лол, очередной пояснятель "за всех")
пшел вон ) ветка для тех, кому интересно.
не относишь себя к таковым? в чем проблема?
пройди мимо.
это клуб любителей самодельных велосипедов)
что за синдром "защемленного яичка"?)
надо обязательно прийти и накакать?))
Здравствуйте, barney, Вы писали:
B>1) Какой механизм меж-потоковой синхронизации?
В общем случае код воркера выглядит где-то так:
queue_mutex.lock();
for(;;) {
if (shutdown_flag) {
break;
}
if (!queue.empty()) {
auto work = queue.pop_front(); // C++ - front, но потом pop_front, чтобы удалить
queue_mutex.unlock();
work();
queue_mutex.lock();
} else {
queue_cv.wait(&queue_mutex);
}
}
queue_mutex.unlock();
}
На входе в цикл воркера лочишь мьютекс и проверяешь очередь. Если она пуста, становишься в wait. wait() атомарно становится в очередь на сигнал и разлочивает мьютекс. Как именно ей указать мьютекс — зависит от API; в одних конкретный мьютекс это аргумент метода wait(), в других надо установить проперть у CV.
Дальше оно ждёт сигнала (одного или на всех), и выходит из wait(), залочивая тот же мьютекс.
Проверяешь снова, есть ли работа или приказ на shutdown.
Во время работы мьютекс надо явно отпустить, но после того, как работа вынута из очереди. Потом снова захватить.
Учти, что любой CV может получать stray interrupts, то есть из wait() вышли, но соответствующего signal()/notify() или signal_all()/broadcast() не было. Это норма, тогда надо без жалоб уйти на следующий цикл сна.
В общем всё. Можно обвесить счётчиками и т.п., но не принципиально.
Обычно очередь одна на всех, но можно делать, что диспетчер выбирает, к кому добавлять в очередь, это может быть удобнее в среде с дофига ядер и медленной синхронизацией кэша.
B>Например я хочу гарантировать очередь исполнения моих ворков
B>допустим, из разных потоков я диспатчу: B>dispatch(work1) | dispatch(work2)
B>я хочу чтобы work1 гарантированно исполнилась перед work2 B>при этом, чтобы потоки не простаивали
А вот тут никак, кроме как что диспетчер следит, что work2 требует завершения work1, и до этого завершения держит work2 в отдельной очереди, которую не кидает целевым воркерам.
Как именно — от темы thread pool никак не зависит, это тема структуры данных типа "граф задач".
Назовёшь это барьером или нет — несущественно.
B>2) Как эффективно передается Work между потоками? B>Я так понимаю, что все thread используют общую память т.е. нет никакой нужды ничего никуда копировать? Или как?
Общая память. Но каждая передача между ядрами будет требовать сотню тактов. Поэтому часто их делать нельзя.
B>Будет ли эффективной передача std::function по значению?
Привет, netch80
О, спасибо за содержательный ответ, любопытно!
N>В общем случае код воркера выглядит где-то так:
N>
N> queue_mutex.lock();
N> for(;;) {
N> if (shutdown_flag) {
N> break;
N> }
N> if (!queue.empty()) {
N> auto work = queue.pop_front(); // C++ - front, но потом pop_front, чтобы удалить
N> queue_mutex.unlock();
N> work();
N> queue_mutex.lock();
N> } else {
N> queue_cv.wait(&queue_mutex);
N> }
N> }
N> queue_mutex.unlock();
N>}
N>
Т.е все воркеры сразу блокируются на мьютексе очереди.
Затем один из них выгребает (непустую) очередь, убирает мьютекс и сразу же запускает work().
В это время какой то другой воркер ухватит мьютекс и пойдет запускать свою работу.
Только не понял, зачем мьютекс лочится вконце после work? work(); queue_mutex.lock();
Только для симметрии чтобы компенсировать общий unlock? Я так понимаю можно просто сделать брейк без блокировки мьютекса, это будет чуть эффективнее, так?
Второй момент, возможно ли внутри воркера (вызванного из work()) вложенное обращение к диспетчеру же с какой то новой работой?
ну например вот так:
Dispatcher::dispatch([](){
do_worker1(); // эта лямбда выполняется внутри queue_mutex.unlock(); work(); queue_mutex.lock();
Dispatcher::dispatch([]() // здесь диспетчер попытается захватить мьютекс снова, но если он захвачен другим потоком - это безопасно ли? нет ли deadlock?
{
some_more_work();
});
});
N>На входе в цикл воркера лочишь мьютекс и проверяешь очередь. Если она пуста, становишься в wait. wait() атомарно становится в очередь на сигнал и разлочивает мьютекс. Как именно ей указать мьютекс — зависит от API; в одних конкретный мьютекс это аргумент метода wait(), в других надо установить проперть у CV. N>Дальше оно ждёт сигнала (одного или на всех), и выходит из wait(), залочивая тот же мьютекс. N>Проверяешь снова, есть ли работа или приказ на shutdown. N>Во время работы мьютекс надо явно отпустить, но после того, как работа вынута из очереди. Потом снова захватить. N>Учти, что любой CV может получать stray interrupts, то есть из wait() вышли, но соответствующего signal()/notify() или signal_all()/broadcast() не было. Это норма, тогда надо без жалоб уйти на следующий цикл сна.
N>В общем всё. Можно обвесить счётчиками и т.п., но не принципиально. N>Обычно очередь одна на всех, но можно делать, что диспетчер выбирает, к кому добавлять в очередь, это может быть удобнее в среде с дофига ядер и медленной синхронизацией кэша.
N>Укладка в очередь для такого подхода:
N>
B>>dispatch(work1) | dispatch(work2) B>>я хочу чтобы work1 гарантированно исполнилась перед work2 B>>при этом, чтобы потоки не простаивали
N>А вот тут никак, кроме как что диспетчер следит, что work2 требует завершения work1, и до этого завершения держит work2 в отдельной очереди, которую не кидает целевым воркерам. N>Как именно — от темы thread pool никак не зависит, это тема структуры данных типа "граф задач". N>Назовёшь это барьером или нет — несущественно.
Понятно, тут можно по сути вручную прикрутить любую кастомную логику.
Например помечать воркеры тегами — или ставить как то зависимости.
Можно даже приоритеты навернуть, чтобы при выборке работы — сначала шерстили всю очередь на предмет приоритетов -1, а потом брали 0 и +1 соотв.
Здравствуйте, barney, Вы писали:
B>Т.е все воркеры сразу блокируются на мьютексе очереди. B>Затем один из них выгребает (непустую) очередь, убирает мьютекс и сразу же запускает work(). B>В это время какой то другой воркер ухватит мьютекс и пойдет запускать свою работу.
B>Только не понял, зачем мьютекс лочится вконце после work? work(); queue_mutex.lock();
Потому что
1) проверять очередь можно только под локом — иначе вы будете лезть без сериализации в общие потроха.
2) входить в queue_cv.wait() можно только имея захваченный лок (если это не сделать, скорее всего, оно вылетит).
Поэтому удобнее всего сделать lock() сразу после конца выполнения рабочей функции.
B>Только для симметрии чтобы компенсировать общий unlock? Я так понимаю можно просто сделать брейк без блокировки мьютекса, это будет чуть эффективнее, так?
Можно делать lock в начале итерации цикла и unlock в конце. Но тогда будет лишняя пара unlock/lock на каждую итерацию. Смысла нет.
B>Второй момент, возможно ли внутри воркера (вызванного из work()) вложенное обращение к диспетчеру же с какой то новой работой?
Возможно. Никаких проблем. Лок при этом не держится.
B>ну например вот так:
B>[code] B>Dispatcher::dispatch([](){ B> do_worker1(); // эта лямбда выполняется внутри queue_mutex.unlock(); work(); queue_mutex.lock(); B> Dispatcher::dispatch([]() // здесь диспетчер попытается захватить мьютекс снова, но если он захвачен другим потоком — это безопасно ли? нет ли deadlock?
Нет. Главное, чтобы текущая задача не держала его.
B>Понятно, тут можно по сути вручную прикрутить любую кастомную логику. B>Например помечать воркеры тегами — или ставить как то зависимости. B>Можно даже приоритеты навернуть, чтобы при выборке работы — сначала шерстили всю очередь на предмет приоритетов -1, а потом брали 0 и +1 соотв.
Здравствуйте, barney, Вы писали:
B>Привет желающим присоединиться к исследованию Thread Pool B>Насколько я понимаю, Thread Pool это вариация на тему multiple producers/multiple consumers.
Не совсем. Потому что производители и потребители здесь начинают смешиваться.
B>Вопросы, которые не очень ясны.
B>1) Какой механизм меж-потоковой синхронизации? B>Например я хочу гарантировать очередь исполнения моих ворков
B>допустим, из разных потоков я диспатчу: B>dispatch(work1) | dispatch(work2) B>я хочу чтобы work1 гарантированно исполнилась перед work2
Это значит, что ты по договорённости в этих потоках создаёшь составное задание.
Например, из первого потока отправляешь
// в первом потоке
dispatch( []() {
work1(); // в случайном потоке
dispatch( []() {
work2(); // в другом случайном потоке
} );
} );
Либо
// в первом потокеauto future = dispatch(work1);
send_to_second_thread(future);
.....
// во втором потокеauto future = receive_from_first_thread();
wait(future);
dispatch(work2);
Кстати, речь идёт о потоках из пула или о произвольных потоках?
B>при этом, чтобы потоки не простаивали B>для этого work1 нужно не просто разместить в очереди до work2 B>а и гарантировать что поток, исполняющий work1 закончится до того, как другой поток возьмется за work2 B>Возможно тут нужны некие барьеры
Такие барьеры — это фьючерсы. В 11 стандарте уже есть в стандарте, а до того были в бусте.
Кури std::promise / std::future.
Ожидание фьючерсов, по-хорошему, должно возвращать поток обратно в пул.
Иначе может сложиться такая ситуация, что все работчие потоки что-то там модально запланировали и сидят-кукуют, а диспетчер видит, что, с одной стороны, у него очередь заявок, а с другой — что свободной кассы нет.
Вот именно поэтому я написал не future.wait(), а wait(future) — что диспетчер воспользуется моментом и превратит блокирующий вызов во что-то более человекополезное.
К>// в первом потоке
К>auto future = dispatch(work1);
К>send_to_second_thread(future);
К>.....
К>// во втором потоке
К>auto future = receive_from_first_thread();
К>wait(future);
К>dispatch(work2);
К>
Эта схема мне не нравится — тк забираем ядро у пула — т.к просто в потоке ждем фьючера, а это значит что поток заблокирован
и изьят из пула.
Инетерсно можно ли это сделать как то элегантнее, т.е чтобы воркер просто не вызвался до получения этого фьючера.
Некий колбэк по факту завершения вычисления.
Впрочем, самое простое — делать по первой схеме — вложенный диспатч, т.е просто вызывать сл. воркер по факту завершения работы первого.
Тогда и фьючерсы не нужны
К>Кстати, речь идёт о потоках из пула или о произвольных потоках?
О пуле потоков. Тк издержки на поднятие thread большие, чем усыпить его по condition_variable
Поэтому я делаю N потоков, они в цикле выгребают воркеров из очереди и если работы нет — спят по cond_variable
К>Такие барьеры — это фьючерсы. В 11 стандарте уже есть в стандарте, а до того были в бусте. К>Кури std::promise / std::future.
К>Ожидание фьючерсов, по-хорошему, должно возвращать поток обратно в пул.
Вот, то о чем я и говорю
К>Иначе может сложиться такая ситуация, что все работчие потоки что-то там модально запланировали и сидят-кукуют, а диспетчер видит, что, с одной стороны, у него очередь заявок, а с другой — что свободной кассы нет. К>Вот именно поэтому я написал не future.wait(), а wait(future) — что диспетчер воспользуется моментом и превратит блокирующий вызов во что-то более человекополезное.
Вот как это можно реализовать? Т.е future.get() это блокирующий вызов, а периодически опрашивать фьючер — тоже съедает ресурсы CPU
Т.к уже promise поток делает некие вычисления — то блокироваться на фьючере из другого потока — вообще смысла ни малейшего.
Нужен каокй то механизм уведомления.
Т.е promise.set должен не просто передавать данные в соотв. future — а каким то образом обращаться к диспетчеру, и просить запланировать след. воркер,
который сделает работу связанную с данным фьючером, future.get
не понятно как бы это на уровне апи выразить... т.к future.get() может быть внутри лямбды, и сам future ничего не знает о своей лямбде, да и вообще откуда он вызывается
B>Схематически, должно быть так:
Блокировок и ожиданий будет меньше если сделать не одну очередь на все потоки, а у каждого потока своя очередь. И только если локальная очередь пуста, тогда поток идёт и ищет работу у других.
Это называется work-stealing, и реализовано, полагаю, в каждой библиотеке в исходники которой советовали посмотреть.
Здравствуйте, barney, Вы писали:
B>Вот как это можно реализовать? Т.е future.get() это блокирующий вызов, а периодически опрашивать фьючер — тоже съедает ресурсы CPU B>Т.к уже promise поток делает некие вычисления — то блокироваться на фьючере из другого потока — вообще смысла ни малейшего.
future::then позволяет указать колбэк, который вызвовится когда результат будет готов. 'future::then' есть в boost, facebook folly и будет в следующем стандарте.
"For every complex problem, there is a solution that is simple, neat,
and wrong."
К>>// в первом потоке
К>>auto future = dispatch(work1);
К>>send_to_second_thread(future);
К>>.....
К>>// во втором потоке
К>>auto future = receive_from_first_thread();
К>>wait(future);
К>>dispatch(work2);
К>>
B>Эта схема мне не нравится — тк забираем ядро у пула — т.к просто в потоке ждем фьючера, а это значит что поток заблокирован B>и изьят из пула. B>Инетерсно можно ли это сделать как то элегантнее, т.е чтобы воркер просто не вызвался до получения этого фьючера. B>Некий колбэк по факту завершения вычисления.
Я думаю, тут нужно сделать просто специальные фьючерсы — специфичные для тредпула.
Тогда ожидание как бы отдаёт поток в распоряжение тредпула, а исполнение обещания — планирует вернуть управление в поток, когда он размотает стек.
К сожалению, в стандартной библиотеке нет из коробки блокирующего множественного ожидания, — это можно навелосипедить на условных переменных. Но это частности.
Гораздо бОльшая проблема состоит в том, что в плюсах нет сопрограмм со стеком. Поэтому, когда поток отдаётся в тредпул, он привозит к себе абсолютно неуправляемую латентность.
второй поток исполнитель work1
second_task...
second_task...
wait(THE_FUTURE)
another_task...
another_task...
wait(something_else) work1...
third_task... THE_PROMISE.set_value
third_task...
awaited something_else - и вернёмся в another_task
another_task... а отнюдь не в second_task
another_task...
awaited THE_FUTURE!
second_task...
B>Впрочем, самое простое — делать по первой схеме — вложенный диспатч, т.е просто вызывать сл. воркер по факту завершения работы первого. B>Тогда и фьючерсы не нужны
Тогда нужен канал передачи информации из второго потока к исполнителю этого составного задания. (В общем случае, разумеется).
Либо мы наловчимся разбивать задачи так, чтобы они были бесстековыми.
B>не понятно как бы это на уровне апи выразить... т.к future.get() может быть внутри лямбды, и сам future ничего не знает о своей лямбде, да и вообще откуда он вызывается
Просто сделать my_threadpool_promise и my_threadpool_future.
Которые конструируются конкретным тредпулом, потому что они завязаны не на планировщика ОС, а на диспетчера конкретного тредпула.
Перекуём баги на фичи!
