Program Listing for File ThreadPool.hxx¶
↰ Return to documentation for file (Utils/ThreadPool.hxx)
/*--------------------------------------------------------------------------*\
| |
| Copyright (C) 2020 |
| |
| , __ , __ |
| /|/ \ /|/ \ |
| | __/ _ ,_ | __/ _ ,_ |
| | \|/ / | | | | \|/ / | | | |
| |(__/|__/ |_/ \_/|/|(__/|__/ |_/ \_/|/ |
| /| /| |
| \| \| |
| |
| Enrico Bertolazzi |
| Dipartimento di Ingegneria Industriale |
| Universita` degli Studi di Trento |
| email: enrico.bertolazzi@unitn.it |
| |
\*--------------------------------------------------------------------------*/
#pragma once
#ifndef DOXYGEN_SHOULD_SKIP_THIS
#ifndef UTILS_THREADPOOL_dot_HH
#define UTILS_THREADPOOL_dot_HH
#endif
#include <algorithm>
#include <utility>
#include <vector>
//#include <type_traits>
#include <thread>
#include <condition_variable>
#include <mutex>
#include <functional>
#ifdef UTILS_OS_LINUX
#include <pthread.h>
#endif
#ifdef max
#undef max
#endif
#ifdef min
#undef min
#endif
namespace Utils {
#ifndef DOXYGEN_SHOULD_SKIP_THIS
using std::pair;
using std::vector;
using std::atomic;
using std::mutex;
using std::condition_variable;
using std::unique_lock;
using std::thread;
using std::function;
using std::move;
using std::bind;
using std::forward;
using std::memory_order_relaxed;
using std::memory_order_acquire;
using std::memory_order_release;
using std::max;
using std::min;
#endif
/*\
| _____ _ _
| |_ _| |__ _ __ ___ __ _ __| |___
| | | | '_ \| '__/ _ \/ _` |/ _` / __|
| | | | | | | | | __/ (_| | (_| \__ \
| |_| |_| |_|_| \___|\__,_|\__,_|___/
\*/
class SpinLock {
// see https://geidav.wordpress.com/2016/03/23/test-and-set-spinlocks/
private:
atomic<bool> m_locked = {false};
public:
SpinLock() {}
void
wait() {
while (m_locked.load(memory_order_relaxed) == true);
}
void
lock() {
do { wait(); } while (m_locked.exchange(true, memory_order_acquire) == true);
}
void
unlock() {
m_locked.store(false, memory_order_release);
}
};
// - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
class WaitWorker {
private:
#ifdef UTILS_OS_WINDOWS
atomic<int> n_worker;
#else
atomic<int> n_worker = {0};
#endif
public:
#ifdef UTILS_OS_WINDOWS
WaitWorker() { n_worker = 0; }
#else
WaitWorker() {}
#endif
void
wait() {
while (n_worker.load(memory_order_relaxed) != 0 );
}
void enter() { ++n_worker; }
void leave() { --n_worker; }
};
// - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
template <typename DATA>
class BinarySearch {
private:
typedef pair<thread::id,DATA*> DATA_TYPE;
mutable vector<DATA_TYPE> m_data;
mutable SpinLock m_spin_write;
mutable WaitWorker m_worker_read;
public:
BinarySearch() {
m_data.clear();
m_data.reserve(64);
}
~BinarySearch() {
m_spin_write.wait();
for ( auto & a : m_data ) delete a.second;
m_data.clear();
m_spin_write.wait();
}
void
clear() {
m_spin_write.wait();
for ( auto & a : m_data ) delete a.second;
m_data.clear(); m_data.reserve(64);
m_spin_write.wait();
}
DATA *
search( thread::id const & id, bool & ok ) const {
m_spin_write.wait(); // wait writing finished
m_worker_read.enter();
ok = true;
size_t U = m_data.size();
if ( U == 0 ) {
m_worker_read.leave();
m_spin_write.lock();
m_worker_read.wait(); // wait all read finished
ok = false;
U = m_data.size();
m_data.resize(1);
DATA_TYPE & dL = m_data[0];
dL.first = id;
DATA * res = dL.second = new DATA();
m_spin_write.unlock();
return res;
}
size_t L = 0;
while ( U-L > 1 ) {
size_t pos = (L+U)>>1;
thread::id const & id_pos = m_data[pos].first;
if ( id_pos < id ) L = pos; else U = pos;
}
DATA_TYPE & dL = m_data[L];
if ( dL.first == id ) { m_worker_read.leave(); return dL.second; }
DATA_TYPE & dU = m_data[U];
if ( dU.first == id ) { m_worker_read.leave(); return dU.second; }
m_worker_read.leave();
// not found must insert
m_spin_write.lock();
m_worker_read.wait(); // wait all read finished
ok = false;
if ( dL.first < id ) ++L;
U = m_data.size();
m_data.resize(U+1);
while ( U > L ) {
--U;
m_data[U+1].first = m_data[U].first;
m_data[U+1].second = m_data[U].second;
}
DATA_TYPE & dL1 = m_data[L];
dL1.first = id;
DATA * res = dL1.second = new DATA();
m_spin_write.unlock();
return res;
}
};
// - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
class SpinLock_barrier {
private:
atomic<unsigned> m_count;
atomic<unsigned> m_generation;
unsigned m_count_reset_value;
public:
SpinLock_barrier( SpinLock_barrier const & ) = delete;
SpinLock_barrier& operator=( SpinLock_barrier const & ) = delete;
explicit
SpinLock_barrier()
: m_generation(0)
{}
void
setup( unsigned count ) {
m_count_reset_value = m_count = count ;
}
void
count_down() {
unsigned gen = m_generation.load();
if ( --m_count == 0 ) {
if ( m_generation.compare_exchange_weak(gen, gen + 1) )
m_count = m_count_reset_value;
return;
}
}
void
wait() {
unsigned gen = m_generation.load();
while ((gen == m_generation) && (m_count != 0))
std::this_thread::yield();
}
void
count_down_and_wait() {
unsigned gen = m_generation.load();
if ( --m_count == 0 ) {
if ( m_generation.compare_exchange_weak(gen, gen + 1) )
m_count = m_count_reset_value;
return;
}
while ((gen == m_generation) && (m_count != 0))
std::this_thread::yield();
}
};
// - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
class Barrier {
int m_to_be_done;
int m_usedThread;
mutex m_mtx;
condition_variable m_cond;
public:
Barrier() : m_to_be_done(0) {}
void
setup( int nthreads )
{ m_usedThread = m_to_be_done = nthreads ; }
void
count_down() {
unique_lock<mutex> lck(m_mtx);
if ( --m_to_be_done <= 0 ) m_cond.notify_all() ; // wake up all tread
}
void
wait() {
unique_lock<mutex> lck(m_mtx);
m_cond.wait(lck);
}
void
count_down_and_wait() {
unique_lock<mutex> lck(m_mtx);
if ( --m_to_be_done <= 0 ) {
m_cond.notify_all() ; // wake up all tread
m_to_be_done = m_usedThread ;
} else {
m_cond.wait(lck);
}
}
};
// - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
class SimpleSemaphore {
private:
bool m_go;
mutex m_mutex;
condition_variable m_cv;
public:
SimpleSemaphore() noexcept : m_go(true) {}
void
green() noexcept {
{ unique_lock<mutex> lock(m_mutex); m_go = true; }
m_cv.notify_one();
}
void
red() noexcept {
{ unique_lock<mutex> lock(m_mutex); m_go = false; }
m_cv.notify_one();
}
void
wait() noexcept {
unique_lock<mutex> lock(m_mutex);
m_cv.wait(lock, [this]()->bool { return this->m_go; });
}
};
/*\
| __ __ _
| \ \ / /__ _ __| | _____ _ __
| \ \ /\ / / _ \| '__| |/ / _ \ '__|
| \ V V / (_) | | | < __/ |
| \_/\_/ \___/|_| |_|\_\___|_|
\*/
class Worker {
friend class ThreadPool;
bool m_active;
SimpleSemaphore m_is_running;
SimpleSemaphore m_job_done;
thread m_running_thread;
function<void()> m_job;
//disable copy
Worker( Worker const & ) = delete;
Worker& operator = ( Worker const & ) = delete;
void
loop() {
while ( m_active ) {
m_is_running.wait();
if ( m_active ) m_job();
m_is_running.red();
m_job_done.green();
}
}
public:
Worker() : m_active(false) { start(); }
~Worker() { stop(); }
Worker( Worker && rhs ) {
m_active = rhs.m_active;
m_job = rhs.m_job;
m_running_thread = move(rhs.m_running_thread);
}
void
start() {
if ( !m_active ) {
m_active = true;
m_is_running.red();
m_job_done.green();
m_running_thread = thread( [this] () -> void { this->loop(); } );
}
}
void
stop() {
if ( m_active ) {
m_active = false; // deactivate computation
m_is_running.green(); // for exiting from the loop
m_running_thread.join(); // wait thread for exiting
}
}
void wait() { m_job_done.wait(); }
template < class Func, class... Args >
void
run( Func && func, Args && ... args ) {
//launch( bind(forward<Func>(func), forward<Args>(args)...) );
m_job_done.wait(); // se gia occupato in task aspetta
m_job = bind(forward<Func>(func),forward<Args>(args)...);
m_job_done.red();
m_is_running.green(); // activate computation
}
};
/*\
| _____ _ _ ____ _
| |_ _| |__ _ __ ___ __ _ __| | _ \ ___ ___ | |
| | | | '_ \| '__/ _ \/ _` |/ _` | |_) / _ \ / _ \| |
| | | | | | | | | __/ (_| | (_| | __/ (_) | (_) | |
| |_| |_| |_|_| \___|\__,_|\__,_|_| \___/ \___/|_|
\*/
class ThreadPool {
// need to keep track of threads so we can join them
vector<Worker> m_workers;
//disable copy
ThreadPool() = delete;
ThreadPool( ThreadPool const & ) = delete;
ThreadPool & operator = ( ThreadPool const & ) = delete;
#if defined(UTILS_OS_WINDOWS)
void
setup() {
for ( auto & w: m_workers ) w.start();
}
#elif defined(UTILS_OS_LINUX)
void
setup() {
sched_param sch;
int policy;
for ( auto & w: m_workers ) {
w.start();
thread & t = w.m_running_thread;
pthread_getschedparam( t.native_handle(), &policy, &sch );
sch.sched_priority = sched_get_priority_max( SCHED_RR );
pthread_setschedparam( t.native_handle(), SCHED_RR, &sch );
}
}
#else
void
setup() {
for ( auto & w: m_workers ) w.start();
}
#endif
public:
ThreadPool(
unsigned nthread = max(
unsigned(1),
unsigned(thread::hardware_concurrency()-1)
)
) {
m_workers.resize( size_t( nthread ) );
setup();
}
template <typename Func, typename... Args>
void
run( unsigned nt, Func && func, Args && ... args ) {
m_workers[size_t(nt)].run( func, args...);
}
void wait_all() { for ( auto && w : m_workers ) w.wait(); }
void start_all() { for ( auto && w : m_workers ) w.start(); }
void stop_all() { for ( auto && w : m_workers ) w.stop(); }
unsigned size() const { return unsigned(m_workers.size()); }
thread::id
get_id( unsigned i ) const
{ return m_workers[size_t(i)].m_running_thread.get_id(); }
thread const &
get_thread( unsigned i ) const
{ return m_workers[size_t(i)].m_running_thread; }
thread &
get_thread( unsigned i )
{ return m_workers[size_t(i)].m_running_thread; }
void
resize( unsigned numThreads ) {
wait_all();
stop_all();
m_workers.resize( size_t(numThreads) );
setup();
}
};
}
#endif
