bmthreadpool.h
Go to the documentation of this file.
#ifndef BMTPOOL__H__INCLUDED__
#define BMTPOOL__H__INCLUDED__
/*
Copyright(c) 2002-2021 Anatoliy Kuznetsov(anatoliy_kuznetsov at yahoo.com)
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
For more information please visit: http://bitmagic.io
*/
#include <type_traits>
#include <queue>
#include <thread>
#include <mutex>
#include <atomic>
#include <condition_variable>
#include "bmbuffer.h"
#include "bmtask.h"
namespace bm
{
/// Pad 60 bytes so that the final ocupiles 64 bytes (1 cache line)
/// @internal
struct pad60_struct { char c[60]; };
/// Empty padding
/// @internal
struct pad0_struct { };
/**
Spin-lock with two-phase acquire (read + cas)
padding parameter optionally adds a buffer to avoid CPU cache
line contention.
TODO: test if padding realy helps in our case
Generally spin_lock does not have advantage over std::mutex
but in some specific cases like WebAssembly it may be better
due no "noexcept" property
@ingroup bmtasks
*/
template<class Pad = bm::pad0_struct>
class spin_lock
{
public:
spin_lock() noexcept : locked_(0) {}
/// Lock the lock
void lock() noexcept
{
while(1) // spin loop
{
unsigned locked = locked_.load(std::memory_order_relaxed);
if (!locked &&
locked_.compare_exchange_weak(locked, true,
std::memory_order_acquire))
break;
#if defined(BMSSE2OPT) || defined(BMSSE42OPT) || defined(BMAVX2OPT) || defined(BMAVX512OPT)
_mm_pause();
#endif
} // while
}
/// Try to acquire the lock, return true if successfull
///
bool try_lock() noexcept
{
unsigned locked = locked_.load(std::memory_order_relaxed);
if (!locked &&
locked_.compare_exchange_weak(locked, true,
std::memory_order_acquire))
return true;
return false;
}
/// Unlock the lock
void unlock() noexcept
{
locked_.store(false, std::memory_order_release);
}
private:
spin_lock(const spin_lock&)=delete;
spin_lock& operator=(const spin_lock&)=delete;
private:
std::atomic<unsigned> locked_;
Pad p_;
};
/// Wait for multiple threads to exit
///
/// @internal
/// @ingroup bmtasks
///
template<typename TCont>
void join_multiple_threads(TCont& tcont)
{
typename TCont::iterator it_end = tcont.end();
for (typename TCont::iterator it = tcont.begin(); it != it_end; ++it)
{
if (it->joinable())
it->join();
} // for it
}
template<typename QValue, typename Lock> class thread_pool;
/**
Thread-sync queue with MT access protecion
@ingroup bmtasks
@internal
*/
template<typename Value, typename Lock>
class queue_sync
{
public:
typedef Value value_type;
typedef Lock lock_type;
/// constructor
///
queue_sync() noexcept {}
/// Push value to the back of the queue
/// @param v - value to put in the queue
///
/// @sa push_no_lock
///
void push(const value_type& v) //noexcept(bm::is_lock_noexcept<lock_type>::value)
{
{
std::lock_guard<lock_type> lg(dq_lock_);
data_queue_.push(v);
}
queue_push_cond_.notify_one(); // noexcept
}
/// Push value to the back of the queue without lock protection
/// It is assumed that processing did not start and we are just staging
/// the batch
///
/// @param v - value to put in the queue
/// @sa push
///
void push_no_lock(const value_type& v)
{
data_queue_.push(v);
}
/// Extract value
/// @param v - [out] value returned
/// @return true if extracted
///
bool try_pop(value_type& v)
{
std::lock_guard<lock_type> guard(dq_lock_);
if (data_queue_.empty())
return false;
v = data_queue_.front();
data_queue_.pop();
return true;
}
/// @return true if empty
bool empty() const //noexcept(bm::is_lock_noexcept<lock_type>::value)
{
std::lock_guard<lock_type> guard(dq_lock_);
return data_queue_.empty();
}
/// lock the queue access
/// @sa push_no_lock, unlock
void lock() noexcept(bm::is_lock_noexcept<lock_type>::value)
{ dq_lock_.lock(); }
/// Try to lock the queue exclusively
///
bool try_lock() noexcept(bm::is_lock_noexcept<lock_type>::value)
{ return dq_lock_.try_lock(); }
/// unlock the queue access
/// @sa push_no_lock, lock
void unlock() noexcept(bm::is_lock_noexcept<lock_type>::value)
{
dq_lock_.unlock();
// lock-unlock is done to protect bulk push, need to wake up
// all waiting workers
queue_push_cond_.notify_all(); // noexcept
}
template<typename QV, typename L> friend class bm::thread_pool;
protected:
typedef std::queue<value_type> queue_type;
private:
queue_sync(const queue_sync&) = delete;
queue_sync& operator=(const queue_sync&) = delete;
private:
queue_type data_queue_; ///< queue object
mutable lock_type dq_lock_; ///< lock for queue
// signal structure for wait on empty queue
protected:
mutable std::mutex signal_mut_; ///< signal mutex for q submissions
std::condition_variable queue_push_cond_; ///< mutex paired conditional
};
/**
Thread pool with custom (thread safe) queue
Thread pool implements a busy-wait task stealing
design pattern
QValue - task queue value parameter
Lock - locking protection type (like std::mutex or spinlock)
@ingroup bmtasks
*/
template<typename QValue, typename Lock>
class thread_pool
{
public:
typedef QValue value_type;
typedef Lock lock_type;
typedef bm::queue_sync<QValue, lock_type> queue_type;
/**
Stop modes for threads:
0 - keep running/waiting for jobs
1 - wait for empty task queue then stop threads
2 - stop threads now even if there are pending tasks
*/
enum stop_mode
{
no_stop = 0, ///< keep spinning on busy-wait
stop_when_done = 1, ///< stop if tsak queue is empty
stop_now = 2 ///< stop right now
};
public:
thread_pool(stop_mode sm = no_stop) noexcept
: stop_flag_(sm)
{}
~thread_pool();
/** Setup the criteria for threads shutdown
Also notifies all threads on a new directive
@param sm - stop mode
*/
void set_stop_mode(stop_mode sm) noexcept;
/**
Request an immediate stop of all threads in the pool
*/
void stop() noexcept { set_stop_mode(stop_now); }
/**
Start thread pool worker threads.
@param tcount - number of threads to start
*/
void start(unsigned tcount);
/**
Wait for threads to finish (or stop if stop was requested)
*/
void join();
/**
Conditional spin-wait for the queue to empty
(Important note: tasks may still be running, but the queue is empty)
*/
void wait_empty_queue();
/// Get access to the job submission queue
///
queue_type& get_job_queue() noexcept { return job_queue_; }
/// Return if thread pool is stopped by a request
int is_stopped() const noexcept
{ return stop_flag_.load(std::memory_order_relaxed); }
protected:
/// Internal worker wrapper with busy-wait spin loop
/// making pthread-like call for tasks
///
void worker_func();
private:
thread_pool(const thread_pool&)=delete;
thread_pool& operator=(const thread_pool&)=delete;
private:
queue_type job_queue_; ///< queue (thread sync)
std::vector<std::thread> thread_vect_; ///< threads servicing queue
std::atomic_int stop_flag_{0}; ///< stop flag to all threads
// notification channel for results wait
mutable std::mutex task_done_mut_; ///< signal mutex for task done
std::condition_variable task_done_cond_;///< mutex paired conditional
};
/**
Utility class to submit task batch to the running thread pool
and optionally wait for it getting done
@ingroup bmtasks
*/
template<typename TPool>
class thread_pool_executor
{
public:
typedef TPool thread_pool_type;
typedef task_batch_base::size_type size_type;
public:
thread_pool_executor() {}
static
void run(thread_pool_type& tpool,
bm::task_batch_base& tasks,
bool wait_for_batch);
/**
Check if all batch jobs in the specified interval are done
Spin wait if not.
*/
static
void wait_for_batch_done(thread_pool_type& tpool,
bm::task_batch_base& tasks,
task_batch_base::size_type from_idx,
task_batch_base::size_type to_idx);
private:
thread_pool_executor(const thread_pool_executor&) = delete;
thread_pool_executor& operator=(const thread_pool_executor&) = delete;
};
// ========================================================================
// thread_pool<> implementations
// ========================================================================
// -----------------------------------------------------------------------
template<typename QValue, typename Lock>
thread_pool<QValue, Lock>::~thread_pool()
{
int is_stop = stop_flag_;
if (!is_stop) // finish the outstanding jobs and close threads
set_stop_mode(stop_when_done);
join();
}
// -----------------------------------------------------------------------
template<typename QValue, typename Lock>
void thread_pool<QValue, Lock>::set_stop_mode(stop_mode sm) noexcept
{
stop_flag_ = sm;
job_queue_.queue_push_cond_.notify_all(); // this is noexcept
}
// -----------------------------------------------------------------------
template<typename QValue, typename Lock>
void thread_pool<QValue, Lock>::start(unsigned tcount)
{
int is_stop = stop_flag_.load(std::memory_order_relaxed);
if (is_stop == stop_now) // immediate stop requested
return;
// TODO: consider lock protect of thread_vect_ member
for(unsigned i = 0;i < tcount; ++i)
{
thread_vect_.emplace_back(
std::thread(&thread_pool::worker_func,this));
} // for
}
// -----------------------------------------------------------------------
template<typename QValue, typename Lock>
void thread_pool<QValue, Lock>::join()
{
bm::join_multiple_threads(thread_vect_);
thread_vect_.resize(0);
}
// -----------------------------------------------------------------------
template<typename QValue, typename Lock>
void thread_pool<QValue, Lock>::wait_empty_queue()
{
const std::chrono::duration<int, std::milli> wait_duration(20);
while(1)
{
if (job_queue_.empty())
break;
std::cv_status wait_res;
{
std::unique_lock<std::mutex> lk(task_done_mut_);
wait_res = task_done_cond_.wait_for(lk, wait_duration);
}
if (wait_res == std::cv_status::timeout)
{
std::this_thread::yield();
int is_stop = is_stopped();
if (is_stop == stop_now) // immediate stop requested
return;
}
} // while
}
// -----------------------------------------------------------------------
template<typename QValue, typename Lock>
void thread_pool<QValue, Lock>::worker_func()
{
const std::chrono::duration<int, std::milli> wait_duration(10);
while(1)
{
int is_stop = is_stopped();
if (is_stop == stop_now) // immediate stop requested
break;
bm::task_descr* task_descr;
if (job_queue_.try_pop(task_descr))
{
BM_ASSERT(task_descr->done == 0);
try
{
task_descr->err_code = task_descr->func(task_descr->argp);
}
catch (...)
{
task_descr->err_code = -1;
}
task_descr->done.store(1, std::memory_order_release);
task_done_cond_.notify_one();
continue;
}
// queue appears to be empty, check if requested to stop
//
is_stop = is_stopped();
if (is_stop)
return;
// enter a temporal condition wait
// notifications are treated as unreliable re-verified
// via spin over the poll of the queue
std::cv_status wait_res;
{
std::unique_lock<std::mutex> lk(job_queue_.signal_mut_);
wait_res =
job_queue_.queue_push_cond_.wait_for(lk, wait_duration);
}
if (wait_res == std::cv_status::timeout)
{
is_stop = is_stopped();
if (is_stop == stop_now) // immediate stop requested
return;
std::this_thread::yield();
}
} // while
return;
}
// ========================================================================
// thread_pool_executor<> implementations
// ========================================================================
template<typename TPool>
void thread_pool_executor<TPool>::run(
thread_pool_type& tpool,
bm::task_batch_base& task_batch,
bool wait_for_batch)
{
typename thread_pool_type::queue_type& qu = tpool.get_job_queue();
task_batch_base::size_type batch_size = task_batch.size();
for (task_batch_base::size_type i = 0; i < batch_size; ++i)
{
bm::task_descr* tdescr = task_batch.get_task(i);
tdescr->argp = tdescr; // restore the self referenece
BM_ASSERT(!tdescr->done);
if (tdescr->flags != bm::task_descr::no_flag) // barrier task ?
{
if (i) // wait until all previously scheduled tasks are done
{
tpool.wait_empty_queue();
wait_for_batch_done(tpool, task_batch, 0, batch_size - 1);
}
// run the barrier proc on the curent thread
tdescr->err_code = tdescr->func(tdescr->argp);
tdescr->done.store(1, std::memory_order_release);
// re-read the batch size, if barrier added more tasks
task_batch_base::size_type new_batch_size = task_batch.size();
if (new_batch_size != batch_size)
batch_size = new_batch_size;
continue;
}
qu.push(tdescr); // locked push to the thread queue
auto is_stop = tpool.is_stopped();
if (is_stop == thread_pool_type::stop_now)
break; // thread pool stop requested
} // for
// implicit wait barrier for all tasks
if (wait_for_batch && batch_size)
{
tpool.wait_empty_queue();
wait_for_batch_done(tpool, task_batch, 0, batch_size - 1);
}
}
// -----------------------------------------------------------------------
template<typename TPool>
void thread_pool_executor<TPool>::wait_for_batch_done(
thread_pool_type& tpool,
bm::task_batch_base& tasks,
task_batch_base::size_type from_idx,
task_batch_base::size_type to_idx)
{
BM_ASSERT(from_idx <= to_idx);
BM_ASSERT(to_idx < tasks.size());
for (task_batch_base::size_type i = from_idx; i <= to_idx; ++i)
{
const bm::task_descr* tdescr = tasks.get_task(i);
auto done = tdescr->done.load(std::memory_order_consume);
while (!done)
{
auto is_stop = tpool.is_stopped();
if (is_stop == thread_pool_type::stop_now)
return; // thread pool stopped, jobs will not be done
std::this_thread::yield();
// TODO: subscribe to a conditional wait for job done in tpool
done = tdescr->done.load(std::memory_order_acquire);
} // while
} // for
}
// -----------------------------------------------------------------------
} // bm
#endif
BM_ASSERT
#define BM_ASSERT
Definition bmdef.h:139
bmtask.h
Task definitions for parallel programming with BitMagic.
bm::queue_sync
Thread-sync queue with MT access protecion.
Definition bmthreadpool.h:130
bm::queue_sync::value_type
Value value_type
Definition bmthreadpool.h:132
bm::queue_sync::queue_type
std::queue< value_type > queue_type
Definition bmthreadpool.h:210
bm::queue_sync::empty
bool empty() const
Definition bmthreadpool.h:181
bm::queue_sync::try_pop
bool try_pop(value_type &v)
Extract value.
Definition bmthreadpool.h:170
bm::queue_sync::push
void push(const value_type &v)
Push value to the back of the queue.
Definition bmthreadpool.h:144
bm::queue_sync::lock
void lock() noexcept(bm::is_lock_noexcept< lock_type >::value)
lock the queue access
Definition bmthreadpool.h:189
bm::queue_sync::push_no_lock
void push_no_lock(const value_type &v)
Push value to the back of the queue without lock protection It is assumed that processing did not sta...
Definition bmthreadpool.h:160
bm::queue_sync< QValue, lock_type >::queue_push_cond_
std::condition_variable queue_push_cond_
Definition bmthreadpool.h:222
bm::queue_sync::lock_type
Lock lock_type
Definition bmthreadpool.h:133
bm::queue_sync< QValue, lock_type >::signal_mut_
std::mutex signal_mut_
Definition bmthreadpool.h:221
bm::queue_sync::queue_sync
queue_sync() noexcept
constructor
Definition bmthreadpool.h:137
bm::queue_sync::try_lock
bool try_lock() noexcept(bm::is_lock_noexcept< lock_type >::value)
Try to lock the queue exclusively.
Definition bmthreadpool.h:194
bm::queue_sync::unlock
void unlock() noexcept(bm::is_lock_noexcept< lock_type >::value)
unlock the queue access
Definition bmthreadpool.h:199
bm::spin_lock
Spin-lock with two-phase acquire (read + cas) padding parameter optionally adds a buffer to avoid CPU...
Definition bmthreadpool.h:56
bm::spin_lock::spin_lock
spin_lock() noexcept
Definition bmthreadpool.h:58
bm::spin_lock::try_lock
bool try_lock() noexcept
Try to acquire the lock, return true if successfull.
Definition bmthreadpool.h:78
bm::spin_lock::unlock
void unlock() noexcept
Unlock the lock.
Definition bmthreadpool.h:89
bm::spin_lock::lock
void lock() noexcept
Lock the lock.
Definition bmthreadpool.h:61
bm::task_batch_base
Interface definition (base class) for a group of tasks (batch).
Definition bmtask.h:118
bm::task_batch_base::size
virtual size_type size() const =0
Return size of batch.
bm::task_batch_base::get_task
virtual bm::task_descr * get_task(size_type task_idx)=0
Get task by index in the batch.
bm::task_batch_base::size_type
unsigned size_type
Definition bmtask.h:120
bm::task_batch
Basic implementation for collection of tasks for parallel execution.
Definition bmtask.h:140
bm::task_batch::size
virtual size_type size() const BMNOEXCEPT
task_batch_base intreface implementation
Definition bmtask.h:158
bm::task_batch::get_task
virtual bm::task_descr * get_task(size_type task_idx)
Get task by index in the batch.
Definition bmtask.h:160
bm::thread_pool_executor
Utility class to submit task batch to the running thread pool and optionally wait for it getting done...
Definition bmthreadpool.h:331
bm::thread_pool_executor::wait_for_batch_done
static void wait_for_batch_done(thread_pool_type &tpool, bm::task_batch_base &tasks, task_batch_base::size_type from_idx, task_batch_base::size_type to_idx)
Check if all batch jobs in the specified interval are done Spin wait if not.
Definition bmthreadpool.h:548
bm::thread_pool_executor::thread_pool_type
TPool thread_pool_type
Definition bmthreadpool.h:333
bm::thread_pool_executor::thread_pool_executor
thread_pool_executor()
Definition bmthreadpool.h:337
bm::thread_pool_executor::run
static void run(thread_pool_type &tpool, bm::task_batch_base &tasks, bool wait_for_batch)
Definition bmthreadpool.h:494
bm::thread_pool_executor::size_type
task_batch_base::size_type size_type
Definition bmthreadpool.h:334
bm::thread_pool
Thread pool with custom (thread safe) queue.
Definition bmthreadpool.h:239
bm::thread_pool::stop
void stop() noexcept
Request an immediate stop of all threads in the pool.
Definition bmthreadpool.h:274
bm::thread_pool::queue_type
bm::queue_sync< QValue, lock_type > queue_type
Definition bmthreadpool.h:243
bm::thread_pool::lock_type
Lock lock_type
Definition bmthreadpool.h:242
bm::thread_pool::join
void join()
Wait for threads to finish (or stop if stop was requested).
Definition bmthreadpool.h:403
bm::thread_pool::thread_pool
thread_pool(stop_mode sm=no_stop) noexcept
Definition bmthreadpool.h:259
bm::thread_pool::worker_func
void worker_func()
Internal worker wrapper with busy-wait spin loop making pthread-like call for tasks.
Definition bmthreadpool.h:437
bm::thread_pool::is_stopped
int is_stopped() const noexcept
Return if thread pool is stopped by a request.
Definition bmthreadpool.h:298
bm::thread_pool::start
void start(unsigned tcount)
Start thread pool worker threads.
Definition bmthreadpool.h:387
bm::thread_pool::wait_empty_queue
void wait_empty_queue()
Conditional spin-wait for the queue to empty (Important note: tasks may still be running,...
Definition bmthreadpool.h:412
bm::thread_pool::stop_mode
stop_mode
Stop modes for threads: 0 - keep running/waiting for jobs 1 - wait for empty task queue then stop thr...
Definition bmthreadpool.h:252
bm::thread_pool::no_stop
@ no_stop
keep spinning on busy-wait
Definition bmthreadpool.h:253
bm::thread_pool::stop_when_done
@ stop_when_done
stop if tsak queue is empty
Definition bmthreadpool.h:254
bm::thread_pool::stop_now
@ stop_now
stop right now
Definition bmthreadpool.h:255
bm::thread_pool::get_job_queue
queue_type & get_job_queue() noexcept
Get access to the job submission queue.
Definition bmthreadpool.h:295
bm::thread_pool::~thread_pool
~thread_pool()
Definition bmthreadpool.h:367
bm::thread_pool::set_stop_mode
void set_stop_mode(stop_mode sm) noexcept
Setup the criteria for threads shutdown Also notifies all threads on a new directive.
Definition bmthreadpool.h:378
bm::thread_pool::value_type
QValue value_type
Definition bmthreadpool.h:241
bm
Definition bm.h:78
bm::join_multiple_threads
void join_multiple_threads(TCont &tcont)
Wait for multiple threads to exit.
Definition bmthreadpool.h:108
bm::is_lock_noexcept
"noexcept" traits detection for T::lock()
Definition bmtask.h:213
bm::pad0_struct
Empty padding.
Definition bmthreadpool.h:40
bm::pad60_struct
Pad 60 bytes so that the final ocupiles 64 bytes (1 cache line).
Definition bmthreadpool.h:37
bm::pad60_struct::c
char c[60]
Definition bmthreadpool.h:37
bm::task_descr
BitMagic task with a captured function.
Definition bmtask.h:62
bm::task_descr::argp
void * argp
arg pointer
Definition bmtask.h:72
bm::task_descr::done
std::atomic_bool done
0 - pending
Definition bmtask.h:77
bm::task_descr::no_flag
@ no_flag
no flag specified
Definition bmtask.h:65
bm::task_descr::flags
bm::id64_t flags
task flags to designate barriers
Definition bmtask.h:75
bm::task_descr::func
task_function_t func
captured function callback
Definition bmtask.h:71
bm::task_descr::err_code
int err_code
error code
Definition bmtask.h:76