i want to apply boost group_thread in my program with a maximal number of Threads. For example
int maxNumberOfThreads
boost::thread_group group;
for (int i = 0; i < N; ++i)
//create new if group.size() is smaller then maximal number of threads
group.create_thread(Worker);
group.join_all();
Someone has an idea how i can realize this ?
Because it will be very inefficient when i start N numbers of thread.
Thank you for your help
What you seem to want is a thread pool.
You can use boost::thread::hardware_concurrency() to determine the number of (logical) cores available on your particular system.
Here's one I rolled for an answer last week:
#include <boost/thread.hpp>
#include <boost/phoenix.hpp>
#include <boost/optional.hpp>
using namespace boost;
using namespace boost::phoenix::arg_names;
boost::atomic_size_t counter(0ul);
class thread_pool
{
private:
mutex mx;
condition_variable cv;
typedef function<void()> job_t;
std::deque<job_t> _queue;
thread_group pool;
boost::atomic_bool shutdown;
static void worker_thread(thread_pool& q)
{
while (optional<job_t> job = q.dequeue())
(*job)();
}
public:
thread_pool() : shutdown(false) {
for (unsigned i = 0; i < boost::thread::hardware_concurrency(); ++i)
pool.create_thread(bind(worker_thread, ref(*this)));
}
void enqueue(job_t job)
{
lock_guard<mutex> lk(mx);
_queue.push_back(job);
cv.notify_one();
}
optional<job_t> dequeue()
{
unique_lock<mutex> lk(mx);
namespace phx = boost::phoenix;
cv.wait(lk, phx::ref(shutdown) || !phx::empty(phx::ref(_queue)));
if (_queue.empty())
return none;
job_t job = _queue.front();
_queue.pop_front();
return job;
}
~thread_pool()
{
shutdown = true;
{
lock_guard<mutex> lk(mx);
cv.notify_all();
}
pool.join_all();
}
};
A typical way to use that is also in that answer:
static const size_t bignumber = 1 << 20;
class myClass
{
thread_pool pool; // uses 1 thread per core
public:
void launch_jobs()
{
std::cout << "enqueuing jobs... " << std::flush;
for(size_t i=0; i<bignumber; ++i)
{
for(int j=0; j<2; ++j) {
pool.enqueue(bind(&myClass::myFunction, this, j, i));
}
}
std::cout << "done\n";
}
private:
void myFunction(int i, int j)
{
boost::this_thread::sleep_for(boost::chrono::milliseconds(1));
counter += 1;
}
};
int main()
{
myClass instance;
instance.launch_jobs();
size_t last = 0;
while (counter < (2*bignumber))
{
boost::this_thread::sleep_for(boost::chrono::milliseconds(100));
if ((counter >> 4u) > last)
{
std::cout << "Progress: " << counter << "/" << (bignumber*2) << "\n";
last = counter >> 4u;
}
}
}
For bonus, at that question, in the comments to another answer, I also posted an equivalent solution based on a lock-free job queue implementation:
This is my (imperfect) implementation :
/**
* \author Christophe Dumeunier
* \brief Extension of boost::thread_group managing a maximum number of threads running in parallel
*/
class thread_group_max : public boost::thread_group
{
public:
/**
* \brief Instanciate a group for threads
* \param max_running_threads Maximum number of threads running in parallel, if 0 use the number of cores
* \param max_sleeping_time Maximum sleeping time (seconds) between two checks for finished threads (must be > sleeping_time_start)
* \param sleeping_time_grow Coefficient increasing sleeping time while waiting for finished threads (must be > 1)
* \param sleeping_time_start Initial sleeping time (must be > 0)
*/
explicit thread_group_max(std::size_t max_running_threads = 0, float max_sleeping_time = 1.0f,
float sleeping_time_grow = 1.1f, float sleeping_time_start = 0.001f);
/**
* \brief Destroy the group
* \note Doesn't join the unterminated threads
*/
~thread_group_max();
/** \brief Wait for an available slot and then create a new thread and launch it */
template<typename F>
boost::thread* create_thread(F f);
private:
std::size_t maxRunningThreads; //!< Maximum number of running threads
float maxSleepingTime; //!< Maximum sleeping time between two checks for finished threads
float sleepingTimeStart; //!< Initial sleeping time
float sleepingTimeGrow; //!< Coefficient increasing sleeping time while waiting for finished threads
std::set<boost::thread*> runningThreads; //!< Pointers to running or finished-but-not-removed-yet threads
};
thread_group_max::thread_group_max(std::size_t max_running_threads, float max_sleeping_time, float sleeping_time_grow, float sleeping_time_start) :
boost::thread_group(),
maxRunningThreads(max_running_threads == 0 ? std::max(boost::thread::hardware_concurrency(), 1u) : max_running_threads),
maxSleepingTime(max_sleeping_time),
sleepingTimeStart(sleeping_time_start),
sleepingTimeGrow(sleeping_time_grow),
runningThreads()
{
assert(this->maxRunningThreads > 0);
assert(this->maxSleepingTime >= this->sleepingTimeStart);
assert(this->sleepingTimeStart > 0.0f);
assert(this->sleepingTimeGrow > 1.0f);
}
thread_group_max::~thread_group_max()
{}
template<typename F>
boost::thread* thread_group_max::create_thread(F f)
{
// First, try to clean already finished threads
if(this->runningThreads.size() >= this->maxRunningThreads)
{
for(std::set<boost::thread*>::iterator it = this->runningThreads.begin(); it != this->runningThreads.end();)
{
const std::set<boost::thread*>::iterator jt = it++;
if((*jt)->timed_join(boost::posix_time::milliseconds(0))) /// @todo timed_join is deprecated
this->runningThreads.erase(jt);
}
}
// If no finished thread found, wait for it
if(this->runningThreads.size() >= this->maxRunningThreads)
{
float sleeping_time = this->sleepingTimeStart;
do
{
boost::this_thread::sleep(boost::posix_time::milliseconds((long int)(1000.0f * sleeping_time)));
for(std::set<boost::thread*>::iterator it = this->runningThreads.begin(); it != this->runningThreads.end();)
{
const std::set<boost::thread*>::iterator jt = it++;
if((*jt)->timed_join(boost::posix_time::milliseconds(0))) /// @todo timed_join is deprecated
this->runningThreads.erase(jt);
}
if(sleeping_time < this->maxSleepingTime)
{
sleeping_time *= this->sleepingTimeGrow;
if(sleeping_time > this->maxSleepingTime)
sleeping_time = this->maxSleepingTime;
}
} while(this->runningThreads.size() >= this->maxRunningThreads);
}
// Now, at least 1 slot is available, use it
return *this->runningThreads.insert(this->boost::thread_group::create_thread(f)).first;
}
Example of use:
thread_group_max group(num_threads);
for(std::size_t i = 0; i < jobs.size(); ++i)
group.create_thread(boost::bind(&my_run_job_function, boost::ref(job[i])));
group.join_all();
