Это код для создания thread_group и выполнения всех потоков параллельно:
boost::thread_group group;
for (int i = 0; i < 15; ++i)
group.create_thread(aFunctionToExecute);
group.join_all();
Этот код будет выполнять все потоки сразу. Я хочу, чтобы они выполняли все, кроме 4 максимум параллельно. Когда on завершено, другой выполняется до тех пор, пока больше не будет выполнено.
Другим, более эффективным решением было бы, чтобы каждый поток обратился к основному потоку, когда они были закончены, и обработчик в основном потоке мог запускать новый поток каждый раз. Это предотвращает повторные вызовы timed_join, поскольку основной поток ничего не сделает, пока не будет вызван обратный вызов.
У меня есть что-то вроде этого:
boost::mutex mutex_;
boost::condition_variable condition_;
const size_t throttle_;
size_t size_;
bool wait_;
template <typename Env, class F>
void eval_(const Env &env, const F &f) {
{
boost::unique_lock<boost::mutex> lock(mutex_);
size_ = std::min(size_+1, throttle_);
while (throttle_ <= size_) condition_.wait(lock);
}
f.eval(env);
{
boost::lock_guard<boost::mutex> lock(mutex_);
--size_;
}
condition_.notify_one();
}
Я думаю, что вы ищете thread_pool, которая доступна .
Кроме того, я заметил, что если вы создадите вектор std:: future и сохраните фьючерсы многих std:: async_tasks в нем, и у вас нет кода блокировки в функции, переданной потоку, VS2013 (по крайней мере от того, что Я могу подтвердить) запустит точно соответствующее количество потоков, которые может обрабатывать ваша машина. Он повторно использует потоки, созданные после создания.
Я создал свой собственный упрощенный интерфейс boost::thread_group для выполнения этой задачи:
class ThreadGroup : public boost::noncopyable
{
private:
boost::thread_group group;
std::size_t maxSize;
float sleepStart;
float sleepCoef;
float sleepMax;
std::set<boost::thread*> running;
public:
ThreadGroup(std::size_t max_size = 0,
float max_sleeping_time = 1.0f,
float sleeping_time_coef = 1.5f,
float sleeping_time_start = 0.001f) :
boost::noncopyable(),
group(),
maxSize(max_size),
sleepStart(sleeping_time_start),
sleepCoef(sleeping_time_coef),
sleepMax(max_sleeping_time),
running()
{
if(max_size == 0)
this->maxSize = (std::size_t)std::max(boost::thread::hardware_concurrency(), 1u);
assert(max_sleeping_time >= sleeping_time_start);
assert(sleeping_time_start > 0.0f);
assert(sleeping_time_coef > 1.0f);
}
~ThreadGroup()
{
this->joinAll();
}
template<typename F> boost::thread* createThread(F f)
{
float sleeping_time = this->sleepStart;
while(this->running.size() >= this->maxSize)
{
for(std::set<boost::thread*>::iterator it = running.begin(); it != running.end();)
{
const std::set<boost::thread*>::iterator jt = it++;
if((*jt)->timed_join(boost::posix_time::milliseconds((long int)(1000.0f * sleeping_time))))
running.erase(jt);
}
if(sleeping_time < this->sleepMax)
{
sleeping_time *= this->sleepCoef;
if(sleeping_time > this->sleepMax)
sleeping_time = this->sleepMax;
}
}
return *this->running.insert(this->group.create_thread(f)).first;
}
void joinAll()
{
this->group.join_all();
}
void interruptAll()
{
#ifdef BOOST_THREAD_PROVIDES_INTERRUPTIONS
this->group.interrupt_all();
#endif
}
std::size_t size() const
{
return this->group.size();
}
};
Вот пример использования, очень похожий на boost::thread_group с основным отличием в том, что создание потока является точкой ожидания:
{
ThreadGroup group(4);
for(int i = 0; i < 15; ++i)
group.createThread(aFunctionToExecute);
} // join all at destruction