Work stealing balances a thread pool when workers have uneven workloads: an idle worker takes a task from another worker’s queue. It complements local queues, which reduce contention but can leave one worker overloaded while others are idle. section 9.1.4
A worker looking for a task checks:
- Its own local queue, then the shared pool queue. section 9.1.4
- Another worker’s queue if both are empty. The owner pushes and pops at one end; a stealer takes from the other. section 9.1.5
Listing 9.7. Lock-based queue for work stealing
class work_stealing_queue
{
private:
typedef function_wrapper data_type;
std::deque<data_type> the_queue; #1
mutable std::mutex the_mutex;
public:
work_stealing_queue()
{}
work_stealing_queue(const work_stealing_queue& other)=delete;
work_stealing_queue& operator=(
const work_stealing_queue& other)=delete;
void push(data_type data) #2
{
std::lock_guard<std::mutex> lock(the_mutex);
the_queue.push_front(std::move(data));
}
bool empty() const
{
std::lock_guard<std::mutex> lock(the_mutex);
return the_queue.empty();
}
bool try_pop(data_type& res) #3
{
std::lock_guard<std::mutex> lock(the_mutex);
if(the_queue.empty())
{
return false;
}
res=std::move(the_queue.front());
the_queue.pop_front();
return true;
}
bool try_steal(data_type& res) #4
{
std::lock_guard<std::mutex> lock(the_mutex);
if(the_queue.empty())
{
return false;
}
res=std::move(the_queue.back());
the_queue.pop_back();
return true;
}
};
!@%STYLE%@!
{"css":"{\"css\": \"font-weight: bold;\"}","target":"[[{\"line\":4,\"ch\":55},{\"line\":4,\"ch\":57}],[{\"line\":12,\"ch\":55},{\"line\":12,\"ch\":57}],[{\"line\":22,\"ch\":55},{\"line\":22,\"ch\":57}],[{\"line\":33,\"ch\":55},{\"line\":33,\"ch\":57}]]"}
!@%STYLE%@!
The lock-based queue protects its deque with a mutex and separates owner operations from stealing by using opposite ends. section 9.1.5
Listing 9.8. A thread pool that uses work stealing
class thread_pool
{
typedef function_wrapper task_type;
std::atomic_bool done;
threadsafe_queue<task_type> pool_work_queue;
std::vector<std::unique_ptr<work_stealing_queue> > queues; #1
std::vector<std::thread> threads;
join_threads joiner;
static thread_local work_stealing_queue* local_work_queue; #2
static thread_local unsigned my_index;
void worker_thread(unsigned my_index_)
{
my_index=my_index_;
local_work_queue=queues[my_index].get(); #3
while(!done)
{
run_pending_task();
}
}
bool pop_task_from_local_queue(task_type& task)
{
return local_work_queue && local_work_queue->try_pop(task);
}
bool pop_task_from_pool_queue(task_type& task)
{
return pool_work_queue.try_pop(task);
}
bool pop_task_from_other_thread_queue(task_type& task) #4
{
for(unsigned i=0;i<queues.size();++i)
{
unsigned const index=(my_index+i+1)%queues.size(); #5
if(queues[index]->try_steal(task))
{
return true;
}
}
return false;
}
public:
thread_pool():
done(false),joiner(threads)
{
unsigned const thread_count=std::thread::hardware_concurrency();
try
{
for(unsigned i=0;i<thread_count;++i)
{
queues.push_back(std::unique_ptr<work_stealing_queue>( #6
new work_stealing_queue));
}
for(unsigned i=0;i<thread_count;++i)
{
threads.push_back(
std::thread(&thread_pool::worker_thread,this,i));
}
}
catch(...)
{
done=true;
throw;
}
}
~thread_pool()
{
done=true;
}
template<typename FunctionType>
std::future<typename std::result_of<FunctionType()>::type> submit(
FunctionType f)
{
typedef typename std::result_of<FunctionType()>::type result_type;
std::packaged_task<result_type()> task(f);
std::future<result_type> res(task.get_future());
if(local_work_queue)
{
local_work_queue->push(std::move(task));
}
else
{
pool_work_queue.push(std::move(task));
}
return res;
}
void run_pending_task()
{
task_type task;
if(pop_task_from_local_queue(task) || #7
pop_task_from_pool_queue(task) || #8
pop_task_from_other_thread_queue(task)) #9
{
task();
}
else
{
std::this_thread::yield();
}
}
};
!@%STYLE%@!
{"css":"{\"css\": \"font-weight: bold;\"}","target":"[[{\"line\":5,\"ch\":72},{\"line\":5,\"ch\":74}],[{\"line\":8,\"ch\":72},{\"line\":8,\"ch\":74}],[{\"line\":13,\"ch\":72},{\"line\":13,\"ch\":74}],[{\"line\":27,\"ch\":72},{\"line\":27,\"ch\":74}],[{\"line\":31,\"ch\":72},{\"line\":31,\"ch\":74}],[{\"line\":48,\"ch\":72},{\"line\":48,\"ch\":74}],[{\"line\":87,\"ch\":72},{\"line\":87,\"ch\":74}],[{\"line\":88,\"ch\":72},{\"line\":88,\"ch\":74}],[{\"line\":89,\"ch\":72},{\"line\":89,\"ch\":74}]]"}
!@%STYLE%@!
The thread-pool implementation shows how workers use local queues alongside the shared pool queue. section 9.1.5
Using opposite ends can reduce contention between an owner and a stealer. A mutex-based queue demonstrates the approach; a lock-free implementation is left beyond scope. section 9.1.5
