импульсный пул потоков
Мне нужен пул потоков для моего приложения, и я бы хотел как можно больше полагаться на стандартные вещи (C ++ 11 или boost). Я понимаю, что существует неофициальный (!) Класс пула потоков наддува, который в основном решает то, что мне нужно, однако я бы предпочел этого избегать, потому что его нет в самой библиотеке наддува — почему он до сих пор не находится в основной библиотеке много лет?
В некоторых постах на этой странице и в других местах люди предлагали использовать boost :: asio для достижения поведения, похожего на пул потоков. На первый взгляд, это выглядело как то, что я хотел сделать, однако я обнаружил, что все реализации, которые я видел, не имеют средств для присоединения к текущим активным задачам, что делает его бесполезным для моего приложения. Чтобы выполнить соединение, они посылают сигнал остановки всем потокам и затем присоединяются к ним. Однако это полностью сводит на нет преимущество потоковых пулов в моем случае использования, потому что это делает новые задачи требующими создания нового потока.
Что я хочу сделать, это:
ThreadPool pool(4);
for (...)
{
for (int i=0;i<something;i++)
pool.pushTask(...);
pool.join();
// do something with the results
}
Может кто-нибудь предложить решение (за исключением использования существующего неофициального пула потоков на sourceforge)? Есть ли что-нибудь в C ++ 11 или усилении ядра, которые могут мне помочь?
Решение
На первый взгляд, это выглядело как то, что я хотел сделать, однако я обнаружил, что все реализации, которые я видел, не имеют средств для присоединения к текущим активным задачам, что делает его бесполезным для моего приложения. Чтобы выполнить соединение, они посылают сигнал остановки всем потокам и затем присоединяются к ним. Однако это полностью сводит на нет преимущество потоковых пулов в моем случае использования, потому что это делает новые задачи требующими создания нового потока.
Я думаю, вы могли неправильно понять пример asio:
IIRC (и это было давно) каждый поток, запущенный в пуле потоков, вызвал io_service::run Это означает, что фактически каждый поток имеет цикл обработки событий и планировщик. Затем, чтобы получить готовые задания, вы публикуете их на io_service Использование метода io_service :: post и механизма планирования asio позаботится об остальном. Пока ты не звонишь io_service::stop, пул потоков продолжит работу, используя столько потоков, сколько вы запустили (при условии, что каждый поток имеет работу или ему назначен io_service::work объект).
Так что вы не необходимо создавать новые потоки для новых задач, что противоречило бы концепции пула потоков.
Другие решения
Пусть каждый класс задач является производным от задачи, в которой есть метод / событие OnCompletion (task). Потоки потоков могут затем вызывать это после вызова основного метода run () задачи.
Ожидание выполнения одной задачи легко. OnCompletion () может выполнить все, что требуется для сигнализации исходного потока, сигнализации condvar, постановки задачи в очередь в очередь производителя-потребителя, вызова API-интерфейсов SendMessage / PostMessage, Invoke / BeginInvoke и т. Д.
Если потоку-ординатору нужно дождаться завершения нескольких задач, вы можете расширить вышеприведенное и выпустить одну «Задачу ожидания» для пула. Задача ожидания имеет собственный OnCompletion для сообщения о завершении других задач и имеет поточно-ориентированный «счетчик задач» (атомарные операции или блокировка), установленный в число «главных» задач, которые должны быть выполнены. Задача ожидания запускается сначала для пула, и поток, который ее запускает, ожидает частного condvar ‘allDone’ в задаче ожидания. Затем «основные» задачи передаются в пул с установленным OnCompletion для вызова метода задачи ожидания, который уменьшает счетчик задач до нуля. Когда счетчик задач достигает нуля, поток, который достигает этого, сигнализирует condvar allDone. Затем запускается задача ожидания OnCompletion, которая сигнализирует о завершении всех основных задач.
Такой механизм не требует постоянного создания / завершения / объединения / удаления потоков пула потоков, не накладывает никаких ограничений на то, как должна сигнализироваться исходная задача, и вы можете создавать столько групп задач, сколько пожелаете. Однако следует отметить, что каждая задача ожидания блокирует один поток пула потоков, поэтому убедитесь, что вы создали несколько дополнительных потоков в пуле (обычно это не проблема).
Это похоже на работу для повышение :: фьючерсы. Пример в документации, кажется, демонстрирует, что именно вы хотите сделать.
Присоединение к потоку означает остановку до тех пор, пока он не остановится, и если он остановился и вы хотите назначить ему новое задание, вы должны создать новый поток. Так что в вашем случае вы должны подождать условие (например, boost::condition_variable) чтобы указать конец задач. Таким образом, используя эту технику, это очень легко реализовать, используя boost::asio а также boost::condition_variable, Каждый поток вызова boost::asio::io_service::run и задачи будут запланированы и выполнены в разных потоках, и в конце каждая задача установит boost::condition_variable или событие уменьшается std::atomic указать конец работы! это действительно легко, не так ли?