#include <ThreadPool.hh>

Classes
struct	Config

Public Types
template<typename KeyT , typename MappedT , typename HashT = KeyT>
using	uomap = std::unordered_map< KeyT, MappedT, std::hash< HashT > >

using	size_type = size_t

using	task_count_type = std::shared_ptr< std::atomic_uintmax_t >

using	atomic_int_type = std::shared_ptr< std::atomic_uintmax_t >

using	pool_state_type = std::shared_ptr< std::atomic_short >

using	atomic_bool_type = std::shared_ptr< std::atomic_bool >

using	task_type = VTask

using	lock_t = std::shared_ptr< Mutex >

using	condition_t = std::shared_ptr< Condition >

using	task_pointer = std::shared_ptr< task_type >

using	task_queue_t = VUserTaskQueue

using	thread_list_t = std::deque< ThreadId >

using	bool_list_t = std::vector< bool >

using	thread_id_map_t = std::map< ThreadId, uintmax_t >

using	thread_index_map_t = std::map< uintmax_t, ThreadId >

using	thread_vec_t = std::vector< Thread >

using	thread_data_t = std::vector< std::shared_ptr< ThreadData > >

using	initialize_func_t = std::function< void()>

using	finalize_func_t = std::function< void()>

using	affinity_func_t = std::function< intmax_t(intmax_t)>

Public Member Functions
	ThreadPool (const Config &)

	ThreadPool (const size_type &pool_size, VUserTaskQueue *task_queue=nullptr, bool _use_affinity=f_use_cpu_affinity(), affinity_func_t=affinity_functor(), initialize_func_t=initialization_functor(), finalize_func_t=finalization_functor())

	ThreadPool (const size_type &pool_size, initialize_func_t, finalize_func_t, bool _use_affinity=f_use_cpu_affinity(), affinity_func_t=affinity_functor(), VUserTaskQueue *task_queue=nullptr)

virtual	~ThreadPool ()

	ThreadPool (const ThreadPool &)=delete

	ThreadPool (ThreadPool &&)=default

ThreadPool &	operator= (const ThreadPool &)=delete

ThreadPool &	operator= (ThreadPool &&)=default

size_type	initialize_threadpool (size_type)

size_type	destroy_threadpool ()

size_type	stop_thread ()

template<typename FuncT >
void	execute_on_all_threads (FuncT &&_func)

template<typename FuncT >
void	execute_on_specific_threads (const std::set< std::thread::id > &_tid, FuncT &&_func)

task_queue_t *	get_queue () const

task_queue_t *&	get_valid_queue (task_queue_t *&) const

bool	is_tbb_threadpool () const

size_type	add_task (task_pointer &&task, int bin=-1)

template<typename ListT >
size_type	add_tasks (ListT &)

Thread *	get_thread (size_type _n) const

Thread *	get_thread (std::thread::id id) const

void	set_initialization (initialize_func_t f)

void	set_finalization (finalize_func_t f)

void	reset_initialization ()

void	reset_finalization ()

const pool_state_type &	state () const

size_type	size () const

void	resize (size_type _n)

bool	using_affinity () const

bool	is_alive ()

void	notify ()

void	notify_all ()

void	notify (size_type)

bool	is_initialized () const

int	get_active_threads_count () const

void	set_affinity (affinity_func_t f)

void	set_affinity (intmax_t i, Thread &) const

void	set_priority (int _prio, Thread &) const

void	set_verbose (int n)

int	get_verbose () const

bool	is_main () const

tbb_task_arena_t *	get_task_arena ()

Static Public Member Functions
static affinity_func_t &	affinity_functor ()

static initialize_func_t &	initialization_functor ()

static finalize_func_t &	finalization_functor ()

static bool	using_tbb ()

static void	set_use_tbb (bool _v)

static void	set_default_use_tbb (bool _v)
	set the default use of tbb

static void	set_default_use_cpu_affinity (bool _v)
	set the default use of cpu affinity

static void	set_default_scheduling_priority (int _v)
	set the default scheduling priority of threads in thread-pool

static void	set_default_verbose (int _v)
	set the default verbosity

static void	set_default_size (size_type _v)
	set the default pool size

static bool	get_default_use_tbb ()
	get the default use of tbb

static bool	get_default_use_cpu_affinity ()
	get the default use of cpu affinity

static int	get_default_scheduling_priority ()
	get the default scheduling priority of threads in thread-pool

static int	get_default_verbose ()
	get the default verbosity

static size_type	get_default_size ()
	get the default pool size

static tbb_global_control_t *&	tbb_global_control ()

static const thread_id_map_t &	get_thread_ids ()

static uintmax_t	get_thread_id (ThreadId)

static uintmax_t	get_this_thread_id ()

static uintmax_t	add_thread_id (ThreadId=ThisThread::get_id())

Protected Member Functions
void	execute_thread (VUserTaskQueue *)

int	insert (task_pointer &&, int=-1)

int	run_on_this (task_pointer &&)

void	record_entry ()

void	record_exit ()

Static Protected Member Functions
static void	start_thread (ThreadPool , thread_data_t , intmax_t=-1)

Detailed Description

Definition at line 87 of file ThreadPool.hh.

Member Typedef Documentation

◆ affinity_func_t

using PTL::ThreadPool::affinity_func_t = std::function<intmax_t(intmax_t)>

Definition at line 115 of file ThreadPool.hh.

◆ atomic_bool_type

using PTL::ThreadPool::atomic_bool_type = std::shared_ptr<std::atomic_bool>

Definition at line 98 of file ThreadPool.hh.

◆ atomic_int_type

using PTL::ThreadPool::atomic_int_type = std::shared_ptr<std::atomic_uintmax_t>

Definition at line 96 of file ThreadPool.hh.

◆ bool_list_t

using PTL::ThreadPool::bool_list_t = std::vector<bool>

Definition at line 107 of file ThreadPool.hh.

◆ condition_t

using PTL::ThreadPool::condition_t = std::shared_ptr<Condition>

Definition at line 102 of file ThreadPool.hh.

◆ finalize_func_t

using PTL::ThreadPool::finalize_func_t = std::function<void()>

Definition at line 114 of file ThreadPool.hh.

◆ initialize_func_t

using PTL::ThreadPool::initialize_func_t = std::function<void()>

Definition at line 113 of file ThreadPool.hh.

◆ lock_t

using PTL::ThreadPool::lock_t = std::shared_ptr<Mutex>

Definition at line 101 of file ThreadPool.hh.

◆ pool_state_type

using PTL::ThreadPool::pool_state_type = std::shared_ptr<std::atomic_short>

Definition at line 97 of file ThreadPool.hh.

◆ size_type

using PTL::ThreadPool::size_type = size_t

Definition at line 94 of file ThreadPool.hh.

◆ task_count_type

using PTL::ThreadPool::task_count_type = std::shared_ptr<std::atomic_uintmax_t>

Definition at line 95 of file ThreadPool.hh.

◆ task_pointer

using PTL::ThreadPool::task_pointer = std::shared_ptr<task_type>

Definition at line 103 of file ThreadPool.hh.

◆ task_queue_t

using PTL::ThreadPool::task_queue_t = VUserTaskQueue

Definition at line 104 of file ThreadPool.hh.

◆ task_type

using PTL::ThreadPool::task_type = VTask

Definition at line 100 of file ThreadPool.hh.

◆ thread_data_t

using PTL::ThreadPool::thread_data_t = std::vector<std::shared_ptr<ThreadData> >

Definition at line 111 of file ThreadPool.hh.

◆ thread_id_map_t

using PTL::ThreadPool::thread_id_map_t = std::map<ThreadId, uintmax_t>

Definition at line 108 of file ThreadPool.hh.

◆ thread_index_map_t

using PTL::ThreadPool::thread_index_map_t = std::map<uintmax_t, ThreadId>

Definition at line 109 of file ThreadPool.hh.

◆ thread_list_t

using PTL::ThreadPool::thread_list_t = std::deque<ThreadId>

Definition at line 106 of file ThreadPool.hh.

◆ thread_vec_t

using PTL::ThreadPool::thread_vec_t = std::vector<Thread>

Definition at line 110 of file ThreadPool.hh.

◆ uomap

template<typename KeyT , typename MappedT , typename HashT = KeyT>

using PTL::ThreadPool::uomap = std::unordered_map<KeyT, MappedT, std::hash<HashT> >

Definition at line 91 of file ThreadPool.hh.

Constructor & Destructor Documentation

◆ ThreadPool() [1/5]

ThreadPool::ThreadPool ( const Config & _cfg )

explicit

Definition at line 257 of file ThreadPool.cc.

: m_use_affinity{ _cfg.use_affinity }
, m_tbb_tp{ _cfg.use_tbb }
, m_verbose{ _cfg.verbose }
, m_priority{ _cfg.priority }
, m_pool_state{ std::make_shared<std::atomic_short>(thread_pool::state::NONINIT) }
, m_task_queue{ _cfg.task_queue }
, m_init_func{ _cfg.initializer }
, m_fini_func{ _cfg.finalizer }
, m_affinity_func{ _cfg.set_affinity }
{
    auto master_id = get_this_thread_id();
    if(master_id != 0 && m_verbose > 1)
    {
        AutoLock lock(TypeMutex<decltype(std::cerr)>());
        std::cerr << "[PTL::ThreadPool] ThreadPool created on worker thread" << std::endl;
    }
 
    thread_data() = new ThreadData(this);
 
    // initialize after get_this_thread_id so master is zero
    if(_cfg.init)
        this->initialize_threadpool(_cfg.pool_size);
}

◆ ThreadPool() [2/5]

ThreadPool::ThreadPool	(	const size_type &	pool_size,
		VUserTaskQueue *	task_queue = `nullptr`,
		bool	_use_affinity = `f_use_cpu_affinity()`,
		affinity_func_t	_affinity_func = `affinity_functor()`,
		initialize_func_t	_init_func = `initialization_functor()`,
		finalize_func_t	_fini_func = `finalization_functor()`
	)

Definition at line 282 of file ThreadPool.cc.

: ThreadPool{ Config{ true, f_use_tbb(), _use_affinity, f_verbose(), f_thread_priority(),
                      _pool_size, _task_queue, std::move(_affinity_func),
                      std::move(_init_func), std::move(_fini_func) } }
{}

◆ ThreadPool() [3/5]

ThreadPool::ThreadPool	(	const size_type &	pool_size,
		initialize_func_t	_init_func,
		finalize_func_t	_fini_func,
		bool	_use_affinity = `f_use_cpu_affinity()`,
		affinity_func_t	_affinity_func = `affinity_functor()`,
		VUserTaskQueue *	task_queue = `nullptr`
	)

Definition at line 290 of file ThreadPool.cc.

: ThreadPool{ pool_size,
              task_queue,
              _use_affinity,
              std::move(_affinity_func),
              std::move(_init_func),
              std::move(_fini_func) }
{}

◆ ~ThreadPool()

ThreadPool::~ThreadPool ( )

virtual

Definition at line 303 of file ThreadPool.cc.

{
    if(m_alive_flag->load())
    {
        std::cerr << "Warning! ThreadPool was not properly destroyed! Call "
                     "destroy_threadpool() before deleting the ThreadPool object to "
                     "eliminate this message."
                  << std::endl;
        m_pool_state->store(thread_pool::state::STOPPED);
        m_task_lock->lock();
        m_task_cond->notify_all();
        m_task_lock->unlock();
        for(auto& itr : m_threads)
            itr.join();
        m_threads.clear();
    }
 
    // delete owned resources
    if(m_delete_task_queue)
        delete m_task_queue;
 
    delete m_tbb_task_arena;
    delete m_tbb_task_group;
}

◆ ThreadPool() [4/5]

PTL::ThreadPool::ThreadPool ( const ThreadPool & )

delete

◆ ThreadPool() [5/5]

PTL::ThreadPool::ThreadPool ( ThreadPool && )

default

Member Function Documentation

◆ add_task()

ThreadPool::size_type PTL::ThreadPool::add_task	(	task_pointer &&	task,
		int	bin = `-1`
	)

inline

Definition at line 450 of file ThreadPool.hh.

{
    // if not native (i.e. TBB) or we haven't built thread-pool, just execute
    if(m_tbb_tp || !task->is_native_task() || !m_alive_flag->load())
        return static_cast<size_type>(run_on_this(std::move(task)));
 
    return static_cast<size_type>(insert(std::move(task), bin));
}

Referenced by PTL::TaskManager::async(), and PTL::TaskManager::exec().

◆ add_tasks()

template<typename ListT >

ThreadPool::size_type PTL::ThreadPool::add_tasks ( ListT & c )

inline

Definition at line 461 of file ThreadPool.hh.

{
    if(!m_alive_flag)  // if we haven't built thread-pool, just execute
    {
        for(auto& itr : c)
            run(itr);
        c.clear();
        return 0;
    }
 
    // TODO: put a limit on how many tasks can be added at most
    auto c_size = c.size();
    for(auto& itr : c)
    {
        if(!itr->is_native_task())
            --c_size;
        else
        {
            //++(m_task_queue);
            get_valid_queue(m_task_queue)->InsertTask(itr);
        }
    }
    c.clear();
 
    // notify sleeping threads
    notify(c_size);
 
    return c_size;
}

◆ add_thread_id()

uintmax_t ThreadPool::add_thread_id ( ThreadId _tid = ThisThread::get_id() )

static

Definition at line 223 of file ThreadPool.cc.

{
    AutoLock lock(TypeMutex<ThreadPool>(), std::defer_lock);
    if(!lock.owns_lock())
        lock.lock();
    if(f_thread_ids().find(_tid) == f_thread_ids().end())
    {
        auto _idx            = f_thread_ids().size();
        f_thread_ids()[_tid] = _idx;
        Threading::SetThreadId((int)_idx);
    }
    return f_thread_ids().at(_tid);
}

Referenced by execute_on_all_threads(), and execute_on_specific_threads().

◆ affinity_functor()

static affinity_func_t & PTL::ThreadPool::affinity_functor ( )

inlinestatic

Definition at line 117 of file ThreadPool.hh.

    {
        static affinity_func_t _v = [](intmax_t) {
            static std::atomic<intmax_t> assigned;
            intmax_t                     _assign = assigned++;
            return _assign % Thread::hardware_concurrency();
        };
        return _v;
    }

◆ destroy_threadpool()

ThreadPool::size_type ThreadPool::destroy_threadpool ( )

Definition at line 572 of file ThreadPool.cc.

{
    // Note: this is not for synchronization, its for thread communication!
    // destroy_threadpool() will only be called from the main thread, yet
    // the modified m_pool_state may not show up to other threads until its
    // modified in a lock!
    //------------------------------------------------------------------------//
    m_pool_state->store(thread_pool::state::STOPPED);
 
    //--------------------------------------------------------------------//
    // handle tbb task scheduler
#if defined(PTL_USE_TBB)
    if(m_tbb_task_group)
    {
        execute_on_all_threads([this]() { m_fini_func(); });
        auto _func = [&]() { m_tbb_task_group->wait(); };
        if(m_tbb_task_arena)
            m_tbb_task_arena->execute(_func);
        else
            _func();
        delete m_tbb_task_group;
        m_tbb_task_group = nullptr;
    }
    if(m_tbb_task_arena)
    {
        delete m_tbb_task_arena;
        m_tbb_task_arena = nullptr;
    }
    if(m_tbb_tp && tbb_global_control())
    {
        tbb_global_control_t*& _global_control = tbb_global_control();
        delete _global_control;
        _global_control = nullptr;
        m_tbb_tp        = false;
        AutoLock lock(TypeMutex<decltype(std::cerr)>());
        if(m_verbose > 0)
        {
            std::cerr << "[PTL::ThreadPool] ThreadPool [TBB] destroyed" << std::endl;
        }
    }
#endif
 
    if(!m_alive_flag->load())
        return 0;
 
    //------------------------------------------------------------------------//
    // notify all threads we are shutting down
    m_task_lock->lock();
    m_task_cond->notify_all();
    m_task_lock->unlock();
    //------------------------------------------------------------------------//
 
    if(m_is_joined.size() != m_main_threads.size())
    {
        std::stringstream ss;
        ss << "   ThreadPool::destroy_thread_pool - boolean is_joined vector "
           << "is a different size than threads vector: " << m_is_joined.size() << " vs. "
           << m_main_threads.size() << " (tid: " << std::this_thread::get_id() << ")";
 
        throw std::runtime_error(ss.str());
    }
 
    for(size_type i = 0; i < m_is_joined.size(); i++)
    {
        //--------------------------------------------------------------------//
        //
        if(i < m_threads.size())
            m_threads.at(i).join();
 
        //--------------------------------------------------------------------//
        // if its joined already, nothing else needs to be done
        if(m_is_joined.at(i))
            continue;
 
        //--------------------------------------------------------------------//
        // join
        if(std::this_thread::get_id() == m_main_threads[i])
            continue;
 
        //--------------------------------------------------------------------//
        // thread id and index
        auto _tid = m_main_threads[i];
 
        //--------------------------------------------------------------------//
        // erase thread from thread ID list
        if(f_thread_ids().find(_tid) != f_thread_ids().end())
            f_thread_ids().erase(f_thread_ids().find(_tid));
 
        //--------------------------------------------------------------------//
        // it's joined
        m_is_joined.at(i) = true;
    }
 
    m_thread_data.clear();
    m_threads.clear();
    m_main_threads.clear();
    m_is_joined.clear();
 
    m_alive_flag->store(false);
 
    auto start   = std::chrono::steady_clock::now();
    auto elapsed = std::chrono::duration<double>{};
    // wait maximum of 30 seconds for threads to exit
    while(m_thread_active->load() > 0 && elapsed.count() < 30)
    {
        std::this_thread::sleep_for(std::chrono::milliseconds(50));
        elapsed = std::chrono::steady_clock::now() - start;
    }
 
    auto _active = m_thread_active->load();
 
    if(get_verbose() > 0)
    {
        if(_active == 0)
        {
            AutoLock lock(TypeMutex<decltype(std::cerr)>());
            std::cerr << "[PTL::ThreadPool] ThreadPool destroyed" << std::endl;
        }
        else
        {
            AutoLock lock(TypeMutex<decltype(std::cerr)>());
            std::cerr << "[PTL::ThreadPool] ThreadPool destroyed but " << _active
                      << " threads might still be active (and cause a termination error)"
                      << std::endl;
        }
    }
 
    if(m_delete_task_queue)
    {
        delete m_task_queue;
        m_task_queue = nullptr;
    }
 
    return 0;
}

Referenced by PTL::TaskManager::finalize(), PTL::TaskRunManager::Terminate(), and G4TaskRunManager::~G4TaskRunManager().

◆ execute_on_all_threads()

template<typename FuncT >

void PTL::ThreadPool::execute_on_all_threads ( FuncT && _func )

inline

Definition at line 493 of file ThreadPool.hh.

{
    if(m_tbb_tp && m_tbb_task_group)
    {
#if defined(PTL_USE_TBB)
        // TBB lazily activates threads to process tasks and the main thread
        // participates in processing the tasks so getting a specific
        // function to execute only on the worker threads requires some trickery
        //
        std::set<std::thread::id> _first{};
        Mutex                     _mutex{};
        // init function which executes function and returns 1 only once
        auto _init = [&]() {
            int _once = 0;
            _mutex.lock();
            if(_first.find(std::this_thread::get_id()) == _first.end())
            {
                // we need to reset this thread-local static for multiple invocations
                // of the same template instantiation
                _once = 1;
                _first.insert(std::this_thread::get_id());
            }
            _mutex.unlock();
            if(_once != 0)
            {
                _func();
                return 1;
            }
            return 0;
        };
        // this will collect the number of threads which have
        // executed the _init function above
        std::atomic<size_t> _total_init{ 0 };
        // max parallelism by TBB
        size_t _maxp = tbb_global_control()->active_value(
            tbb::global_control::max_allowed_parallelism);
        // create a task arean
        auto* _arena = get_task_arena();
        // size of the thread-pool
        size_t _sz = size();
        // number of cores
        size_t _ncore = Threading::GetNumberOfCores();
        // maximum depth for recursion
        size_t _dmax = std::max<size_t>(_ncore, 8);
        // how many threads we need to initialize
        size_t _num = std::min(_maxp, std::min(_sz, _ncore));
        // this is the task passed to the task-group
        std::function<void()> _init_task;
        _init_task = [&]() {
            add_thread_id();
            static thread_local size_type _depth = 0;
            int                           _ret   = 0;
            // don't let the main thread execute the function
            if(!is_main())
            {
                // execute the function
                _ret = _init();
                // add the result
                _total_init += _ret;
            }
            // if the function did not return anything, recursively execute
            // two more tasks
            ++_depth;
            if(_ret == 0 && _depth < _dmax && _total_init.load() < _num)
            {
                tbb::task_group tg{};
                tg.run([&]() { _init_task(); });
                tg.run([&]() { _init_task(); });
                ThisThread::sleep_for(std::chrono::milliseconds{ 1 });
                tg.wait();
            }
            --_depth;
        };
 
        // TBB won't oversubscribe so we need to limit by ncores - 1
        size_t nitr        = 0;
        auto   _fname      = __FUNCTION__;
        auto   _write_info = [&]() {
            std::cout << "[" << _fname << "]> Total initialized: " << _total_init
                      << ", expected: " << _num << ", max-parallel: " << _maxp
                      << ", size: " << _sz << ", ncore: " << _ncore << std::endl;
        };
        while(_total_init < _num)
        {
            auto _n = 2 * _num;
            while(--_n > 0)
            {
                _arena->execute(
                    [&]() { m_tbb_task_group->run([&]() { _init_task(); }); });
            }
            _arena->execute([&]() { m_tbb_task_group->wait(); });
            // don't loop infinitely but use a strict condition
            if(nitr++ > 2 * (_num + 1) && (_total_init - 1) == _num)
            {
                _write_info();
                break;
            }
            // at this point we need to exit
            if(nitr > 4 * (_ncore + 1))
            {
                _write_info();
                break;
            }
        }
        if(get_verbose() > 3)
            _write_info();
#endif
    }
    else if(get_queue())
    {
        get_queue()->ExecuteOnAllThreads(this, std::forward<FuncT>(_func));
    }
}

Referenced by G4TaskRunManager::CreateAndStartWorkers(), destroy_threadpool(), initialize_threadpool(), G4TaskRunManager::RequestWorkersProcessCommandsStack(), G4TaskRunManager::TerminateWorkers(), and G4TaskRunManager::WaitForEndEventLoopWorkers().

◆ execute_on_specific_threads()

template<typename FuncT >

void PTL::ThreadPool::execute_on_specific_threads	(	const std::set< std::thread::id > &	_tid,
		FuncT &&	_func
	)

inline

Definition at line 611 of file ThreadPool.hh.

{
    if(m_tbb_tp && m_tbb_task_group)
    {
#if defined(PTL_USE_TBB)
        // TBB lazily activates threads to process tasks and the main thread
        // participates in processing the tasks so getting a specific
        // function to execute only on the worker threads requires some trickery
        //
        std::set<std::thread::id> _first{};
        Mutex                     _mutex{};
        // init function which executes function and returns 1 only once
        auto _exec = [&]() {
            int _once = 0;
            _mutex.lock();
            if(_first.find(std::this_thread::get_id()) == _first.end())
            {
                // we need to reset this thread-local static for multiple invocations
                // of the same template instantiation
                _once = 1;
                _first.insert(std::this_thread::get_id());
            }
            _mutex.unlock();
            if(_once != 0)
            {
                _func();
                return 1;
            }
            return 0;
        };
        // this will collect the number of threads which have
        // executed the _exec function above
        std::atomic<size_t> _total_exec{ 0 };
        // number of cores
        size_t _ncore = Threading::GetNumberOfCores();
        // maximum depth for recursion
        size_t _dmax = std::max<size_t>(_ncore, 8);
        // how many threads we need to initialize
        size_t _num = _tids.size();
        // create a task arena
        auto* _arena = get_task_arena();
        // this is the task passed to the task-group
        std::function<void()> _exec_task;
        _exec_task = [&]() {
            add_thread_id();
            static thread_local size_type _depth    = 0;
            int                           _ret      = 0;
            auto                          _this_tid = std::this_thread::get_id();
            // don't let the main thread execute the function
            if(_tids.count(_this_tid) > 0)
            {
                // execute the function
                _ret = _exec();
                // add the result
                _total_exec += _ret;
            }
            // if the function did not return anything, recursively execute
            // two more tasks
            ++_depth;
            if(_ret == 0 && _depth < _dmax && _total_exec.load() < _num)
            {
                tbb::task_group tg{};
                tg.run([&]() { _exec_task(); });
                tg.run([&]() { _exec_task(); });
                ThisThread::sleep_for(std::chrono::milliseconds{ 1 });
                tg.wait();
            }
            --_depth;
        };
 
        // TBB won't oversubscribe so we need to limit by ncores - 1
        size_t nitr        = 0;
        auto   _fname      = __FUNCTION__;
        auto   _write_info = [&]() {
            std::cout << "[" << _fname << "]> Total executed: " << _total_exec
                      << ", expected: " << _num << ", size: " << size() << std::endl;
        };
        while(_total_exec < _num)
        {
            auto _n = 2 * _num;
            while(--_n > 0)
            {
                _arena->execute(
                    [&]() { m_tbb_task_group->run([&]() { _exec_task(); }); });
            }
            _arena->execute([&]() { m_tbb_task_group->wait(); });
            // don't loop infinitely but use a strict condition
            if(nitr++ > 2 * (_num + 1) && (_total_exec - 1) == _num)
            {
                _write_info();
                break;
            }
            // at this point we need to exit
            if(nitr > 8 * (_num + 1))
            {
                _write_info();
                break;
            }
        }
        if(get_verbose() > 3)
            _write_info();
#endif
    }
    else if(get_queue())
    {
        get_queue()->ExecuteOnSpecificThreads(_tids, this, std::forward<FuncT>(_func));
    }
}

◆ execute_thread()

void ThreadPool::execute_thread ( VUserTaskQueue * _task_queue )

protected

Definition at line 768 of file ThreadPool.cc.

{
    // how long the thread waits on condition variable
    // static int wait_time = GetEnv<int>("PTL_POOL_WAIT_TIME", 5);
 
    ++(*m_thread_awake);
 
    // initialization function
    m_init_func();
    // finalization function (executed when scope is destroyed)
    ScopeDestructor _fini{ [this]() { m_fini_func(); } };
 
    ThreadId    tid  = ThisThread::get_id();
    ThreadData* data = thread_data();
    // auto        thread_bin = _task_queue->GetThreadBin();
    // auto        workers    = _task_queue->workers();
 
    auto start   = std::chrono::steady_clock::now();
    auto elapsed = std::chrono::duration<double>{};
    // check for updates for 60 seconds max
    while(!_task_queue && elapsed.count() < 60)
    {
        elapsed = std::chrono::steady_clock::now() - start;
        data->update();
        _task_queue = data->current_queue;
    }
 
    if(!_task_queue)
    {
        --(*m_thread_awake);
        throw std::runtime_error("No task queue was found after 60 seconds!");
    }
 
    assert(data->current_queue != nullptr);
    assert(_task_queue == data->current_queue);
 
    // essentially a dummy run
    if(_task_queue)
    {
        data->within_task = true;
        auto _task        = _task_queue->GetTask();
        if(_task)
        {
            (*_task)();
        }
        data->within_task = false;
    }
 
    // threads stay in this loop forever until thread-pool destroyed
    while(true)
    {
        static thread_local auto p_task_lock = m_task_lock;
 
        //--------------------------------------------------------------------//
        // Try to pick a task
        AutoLock _task_lock(*p_task_lock, std::defer_lock);
        //--------------------------------------------------------------------//
 
        auto leave_pool = [&]() {
            auto _state      = [&]() { return static_cast<int>(m_pool_state->load()); };
            auto _pool_state = _state();
            if(_pool_state > 0)
            {
                // stop whole pool
                if(_pool_state == thread_pool::state::STOPPED)
                {
                    if(_task_lock.owns_lock())
                        _task_lock.unlock();
                    return true;
                }
                // single thread stoppage
                else if(_pool_state == thread_pool::state::PARTIAL)  // NOLINT
                {
                    if(!_task_lock.owns_lock())
                        _task_lock.lock();
                    if(!m_is_stopped.empty() && m_is_stopped.back())
                    {
                        m_stop_threads.push_back(tid);
                        m_is_stopped.pop_back();
                        if(_task_lock.owns_lock())
                            _task_lock.unlock();
                        // exit entire function
                        return true;
                    }
                    if(_task_lock.owns_lock())
                        _task_lock.unlock();
                }
            }
            return false;
        };
 
        // We need to put condition.wait() in a loop for two reasons:
        // 1. There can be spurious wake-ups (due to signal/ENITR)
        // 2. When mutex is released for waiting, another thread can be woken up
        //    from a signal/broadcast and that thread can mess up the condition.
        //    So when the current thread wakes up the condition may no longer be
        //    actually true!
        while(_task_queue->empty())
        {
            auto _state = [&]() { return static_cast<int>(m_pool_state->load()); };
            auto _size  = [&]() { return _task_queue->true_size(); };
            auto _empty = [&]() { return _task_queue->empty(); };
            auto _wake  = [&]() { return (!_empty() || _size() > 0 || _state() > 0); };
 
            if(leave_pool())
                return;
 
            if(_task_queue->true_size() == 0)
            {
                if(m_thread_awake->load() > 0)
                    --(*m_thread_awake);
 
                // lock before sleeping on condition
                if(!_task_lock.owns_lock())
                    _task_lock.lock();
 
                // Wait until there is a task in the queue
                // Unlocks mutex while waiting, then locks it back when signaled
                // use lambda to control waking
                m_task_cond->wait(_task_lock, _wake);
 
                if(_state() == thread_pool::state::STOPPED)
                    return;
 
                // unlock if owned
                if(_task_lock.owns_lock())
                    _task_lock.unlock();
 
                // notify that is awake
                if(m_thread_awake->load() < m_pool_size)
                    ++(*m_thread_awake);
            }
            else
                break;
        }
 
        // release the lock
        if(_task_lock.owns_lock())
            _task_lock.unlock();
 
        //----------------------------------------------------------------//
 
        // leave pool if conditions dictate it
        if(leave_pool())
            return;
 
        // activate guard against recursive deadlock
        data->within_task = true;
        //----------------------------------------------------------------//
 
        // execute the task(s)
        while(!_task_queue->empty())
        {
            auto _task = _task_queue->GetTask();
            if(_task)
            {
                (*_task)();
            }
        }
        //----------------------------------------------------------------//
 
        // disable guard against recursive deadlock
        data->within_task = false;
        //----------------------------------------------------------------//
    }
}

◆ finalization_functor()

static finalize_func_t & PTL::ThreadPool::finalization_functor ( )

inlinestatic

Definition at line 133 of file ThreadPool.hh.

    {
        static finalize_func_t _v = []() {};
        return _v;
    }

◆ get_active_threads_count()

int PTL::ThreadPool::get_active_threads_count ( ) const

inline

Definition at line 262 of file ThreadPool.hh.

262{ return (int)m_thread_awake->load(); }

◆ get_default_scheduling_priority()

static int PTL::ThreadPool::get_default_scheduling_priority ( )

inlinestatic

get the default scheduling priority of threads in thread-pool

Definition at line 216 of file ThreadPool.hh.

216{ return f_thread_priority(); }

◆ get_default_size()

static size_type PTL::ThreadPool::get_default_size ( )

inlinestatic

get the default pool size

Definition at line 220 of file ThreadPool.hh.

220{ return f_default_pool_size(); }

◆ get_default_use_cpu_affinity()

static bool PTL::ThreadPool::get_default_use_cpu_affinity ( )

inlinestatic

get the default use of cpu affinity

Definition at line 214 of file ThreadPool.hh.

214{ return f_use_cpu_affinity(); }

◆ get_default_use_tbb()

static bool PTL::ThreadPool::get_default_use_tbb ( )

inlinestatic

get the default use of tbb

Definition at line 212 of file ThreadPool.hh.

212{ return f_use_tbb(); }

◆ get_default_verbose()

static int PTL::ThreadPool::get_default_verbose ( )

inlinestatic

get the default verbosity

Definition at line 218 of file ThreadPool.hh.

218{ return f_verbose(); }

◆ get_queue()

task_queue_t * PTL::ThreadPool::get_queue ( ) const

inline

Definition at line 189 of file ThreadPool.hh.

189{ return m_task_queue; }

Referenced by execute_on_all_threads(), execute_on_specific_threads(), PTL::ThreadData::update(), and PTL::TaskGroup< Tp, Arg, MaxDepth >::wait().

◆ get_task_arena()

tbb_task_arena_t * PTL::ThreadPool::get_task_arena ( )

inline

Definition at line 392 of file ThreadPool.hh.

{
#if defined(PTL_USE_TBB)
    // create a task arena
    if(!m_tbb_task_arena)
    {
        auto _sz = (tbb_global_control())
                       ? tbb_global_control()->active_value(
                             tbb::global_control::max_allowed_parallelism)
                       : size();
        m_tbb_task_arena = new tbb_task_arena_t(::tbb::task_arena::attach{});
        m_tbb_task_arena->initialize(_sz, 1);
    }
#else
    if(!m_tbb_task_arena)
        m_tbb_task_arena = new tbb_task_arena_t{};
#endif
    return m_tbb_task_arena;
}

Referenced by execute_on_all_threads(), execute_on_specific_threads(), and run_on_this().

◆ get_this_thread_id()

uintmax_t ThreadPool::get_this_thread_id ( )

static

Definition at line 215 of file ThreadPool.cc.

{
    return get_thread_id(ThisThread::get_id());
}

Referenced by PTL::UserTaskQueue::GetThreadBin(), initialize_threadpool(), G4TaskRunManagerKernel::InitializeWorker(), ThreadPool(), and PTL::UserTaskQueue::UserTaskQueue().

◆ get_thread() [1/2]

Thread * PTL::ThreadPool::get_thread ( size_type _n ) const

◆ get_thread() [2/2]

Thread * PTL::ThreadPool::get_thread ( std::thread::id id ) const

◆ get_thread_id()

uintmax_t ThreadPool::get_thread_id ( ThreadId _tid )

static

Definition at line 191 of file ThreadPool.cc.

{
    uintmax_t _idx = 0;
    {
        AutoLock lock(TypeMutex<ThreadPool>(), std::defer_lock);
        if(!lock.owns_lock())
            lock.lock();
        auto itr = f_thread_ids().find(_tid);
        if(itr == f_thread_ids().end())
        {
            _idx                 = f_thread_ids().size();
            f_thread_ids()[_tid] = _idx;
        }
        else
        {
            _idx = itr->second;
        }
    }
    return _idx;
}

Referenced by get_this_thread_id(), set_affinity(), and set_priority().

◆ get_thread_ids()

const ThreadPool::thread_id_map_t & ThreadPool::get_thread_ids ( )

static

Definition at line 183 of file ThreadPool.cc.

{
    return f_thread_ids();
}

◆ get_valid_queue()

ThreadPool::task_queue_t *& ThreadPool::get_valid_queue ( task_queue_t *& _queue ) const

Definition at line 759 of file ThreadPool.cc.

{
    if(!_queue)
        _queue = new UserTaskQueue{ static_cast<intmax_t>(m_pool_size) };
    return _queue;
}

Referenced by add_tasks(), and insert().

◆ get_verbose()

int PTL::ThreadPool::get_verbose ( ) const

inline

Definition at line 269 of file ThreadPool.hh.

269{ return m_verbose; }

Referenced by destroy_threadpool(), execute_on_all_threads(), and execute_on_specific_threads().

◆ initialization_functor()

static initialize_func_t & PTL::ThreadPool::initialization_functor ( )

inlinestatic

Definition at line 127 of file ThreadPool.hh.

    {
        static initialize_func_t _v = []() {};
        return _v;
    }

◆ initialize_threadpool()

ThreadPool::size_type ThreadPool::initialize_threadpool ( size_type proposed_size )

Definition at line 402 of file ThreadPool.cc.

{
    //--------------------------------------------------------------------//
    // return before initializing
    if(proposed_size < 1)
        return 0;
 
    //--------------------------------------------------------------------//
    // store that has been started
    if(!m_alive_flag->load())
        m_pool_state->store(thread_pool::state::STARTED);
 
#if defined(PTL_USE_TBB)
    //--------------------------------------------------------------------//
    // handle tbb task scheduler
    if(m_tbb_tp)
    {
        m_tbb_tp                               = true;
        m_pool_size                            = proposed_size;
        tbb_global_control_t*& _global_control = tbb_global_control();
        // delete if wrong size
        if(m_pool_size != proposed_size)
        {
            delete _global_control;
            _global_control = nullptr;
        }
 
        if(!_global_control)
        {
            _global_control = new tbb_global_control_t(
                tbb::global_control::max_allowed_parallelism, proposed_size + 1);
            if(m_verbose > 0)
            {
                AutoLock lock(TypeMutex<decltype(std::cerr)>());
                std::cerr << "[PTL::ThreadPool] ThreadPool [TBB] initialized with "
                          << m_pool_size << " threads." << std::endl;
            }
        }
 
        // create task group (used for async)
        if(!m_tbb_task_group)
        {
            m_tbb_task_group = new tbb_task_group_t{};
            execute_on_all_threads([this]() { m_init_func(); });
        }
 
        return m_pool_size;
    }
#endif
 
    m_alive_flag->store(true);
 
    //--------------------------------------------------------------------//
    // if started, stop some thread if smaller or return if equal
    if(m_pool_state->load() == thread_pool::state::STARTED)
    {
        if(m_pool_size > proposed_size)
        {
            while(stop_thread() > proposed_size)
                ;
            if(m_verbose > 0)
            {
                AutoLock lock(TypeMutex<decltype(std::cerr)>());
                std::cerr << "[PTL::ThreadPool] ThreadPool initialized with "
                          << m_pool_size << " threads." << std::endl;
            }
            if(!m_task_queue)
            {
                m_delete_task_queue = true;
                m_task_queue        = new UserTaskQueue(m_pool_size);
            }
            else
            {
                m_task_queue->resize(m_pool_size);
            }
            return m_pool_size;
        }
        else if(m_pool_size == proposed_size)  // NOLINT
        {
            if(m_verbose > 0)
            {
                AutoLock lock(TypeMutex<decltype(std::cerr)>());
                std::cerr << "ThreadPool initialized with " << m_pool_size << " threads."
                          << std::endl;
            }
            if(!m_task_queue)
            {
                m_delete_task_queue = true;
                m_task_queue        = new UserTaskQueue(m_pool_size);
            }
            return m_pool_size;
        }
    }
 
    //--------------------------------------------------------------------//
    // reserve enough space to prevent realloc later
    {
        AutoLock _task_lock(*m_task_lock);
        m_is_joined.reserve(proposed_size);
    }
 
    if(!m_task_queue)
    {
        m_delete_task_queue = true;
        m_task_queue        = new UserTaskQueue(proposed_size);
    }
 
    auto this_tid = get_this_thread_id();
    for(size_type i = m_pool_size; i < proposed_size; ++i)
    {
        // add the threads
        try
        {
            // create thread
            Thread thr{ ThreadPool::start_thread, this, &m_thread_data,
                        this_tid + i + 1 };
            // only reaches here if successful creation of thread
            ++m_pool_size;
            // store thread
            m_main_threads.push_back(thr.get_id());
            // list of joined thread booleans
            m_is_joined.push_back(false);
            // set the affinity
            if(m_use_affinity)
                set_affinity(i, thr);
            set_priority(m_priority, thr);
            // store
            m_threads.emplace_back(std::move(thr));
        } catch(std::runtime_error& e)
        {
            AutoLock lock(TypeMutex<decltype(std::cerr)>());
            std::cerr << "[PTL::ThreadPool] " << e.what()
                      << std::endl;  // issue creating thread
            continue;
        } catch(std::bad_alloc& e)
        {
            AutoLock lock(TypeMutex<decltype(std::cerr)>());
            std::cerr << "[PTL::ThreadPool] " << e.what() << std::endl;
            continue;
        }
    }
    //------------------------------------------------------------------------//
 
    AutoLock _task_lock(*m_task_lock);
 
    // thread pool size doesn't match with join vector
    // this will screw up joining later
    if(m_is_joined.size() != m_main_threads.size())
    {
        std::stringstream ss;
        ss << "ThreadPool::initialize_threadpool - boolean is_joined vector "
           << "is a different size than threads vector: " << m_is_joined.size() << " vs. "
           << m_main_threads.size() << " (tid: " << std::this_thread::get_id() << ")";
 
        throw std::runtime_error(ss.str());
    }
 
    if(m_verbose > 0)
    {
        AutoLock lock(TypeMutex<decltype(std::cerr)>());
        std::cerr << "[PTL::ThreadPool] ThreadPool initialized with " << m_pool_size
                  << " threads." << std::endl;
    }
 
    return m_main_threads.size();
}

Referenced by resize(), and ThreadPool().

◆ insert()

int PTL::ThreadPool::insert	(	task_pointer &&	task,
		int	bin = `-1`
	)

inlineprotected

Definition at line 439 of file ThreadPool.hh.

{
    static thread_local ThreadData* _data = ThreadData::GetInstance();
 
    // pass the task to the queue
    auto ibin = get_valid_queue(m_task_queue)->InsertTask(std::move(task), _data, bin);
    notify();
    return (int)ibin;
}

Referenced by add_task().

◆ is_alive()

bool PTL::ThreadPool::is_alive ( )

inline

Definition at line 257 of file ThreadPool.hh.

257{ return m_alive_flag->load(); }

Referenced by PTL::TaskGroup< Tp, Arg, MaxDepth >::wait().

◆ is_initialized()

bool ThreadPool::is_initialized ( ) const

Definition at line 331 of file ThreadPool.cc.

{
    return !(m_pool_state->load() == thread_pool::state::NONINIT);
}

◆ is_main()

bool PTL::ThreadPool::is_main ( ) const

inline

Definition at line 270 of file ThreadPool.hh.

270{ return ThisThread::get_id() == m_main_tid; }

Referenced by execute_on_all_threads().

◆ is_tbb_threadpool()

bool PTL::ThreadPool::is_tbb_threadpool ( ) const

inline

Definition at line 192 of file ThreadPool.hh.

192{ return m_tbb_tp; }

Referenced by G4TaskRunManager::InitializeThreadPool().

◆ notify() [1/2]

void PTL::ThreadPool::notify ( )

inline

Definition at line 342 of file ThreadPool.hh.

{
    // wake up one thread that is waiting for a task to be available
    if(m_thread_awake->load() < m_pool_size)
    {
        AutoLock l(*m_task_lock);
        m_task_cond->notify_one();
    }
}

Referenced by add_tasks(), and insert().

◆ notify() [2/2]

void PTL::ThreadPool::notify ( size_type ntasks )

inline

Definition at line 361 of file ThreadPool.hh.

{
    if(ntasks == 0)
        return;
 
    // wake up as many threads that tasks just added
    if(m_thread_awake->load() < m_pool_size)
    {
        AutoLock l(*m_task_lock);
        if(ntasks < this->size())
        {
            for(size_type i = 0; i < ntasks; ++i)
                m_task_cond->notify_one();
        }
        else
        {
            m_task_cond->notify_all();
        }
    }
}

◆ notify_all()

void PTL::ThreadPool::notify_all ( )

inline

Definition at line 353 of file ThreadPool.hh.

{
    // wake all threads
    AutoLock l(*m_task_lock);
    m_task_cond->notify_all();
}

◆ operator=() [1/2]

ThreadPool & PTL::ThreadPool::operator= ( const ThreadPool & )

delete

◆ operator=() [2/2]

ThreadPool & PTL::ThreadPool::operator= ( ThreadPool && )

default

◆ record_entry()

void ThreadPool::record_entry ( )

protected

Definition at line 339 of file ThreadPool.cc.

{
    ++(*m_thread_active);
}

◆ record_exit()

void ThreadPool::record_exit ( )

protected

Definition at line 347 of file ThreadPool.cc.

{
    --(*m_thread_active);
}

◆ reset_finalization()

void PTL::ThreadPool::reset_finalization ( )

inline

Definition at line 243 of file ThreadPool.hh.

    {
        m_fini_func = []() {};
    }

◆ reset_initialization()

void PTL::ThreadPool::reset_initialization ( )

inline

Definition at line 239 of file ThreadPool.hh.

    {
        m_init_func = []() {};
    }

◆ resize()

void PTL::ThreadPool::resize ( size_type _n )

inline

Definition at line 413 of file ThreadPool.hh.

{
    initialize_threadpool(_n);
    if(m_task_queue)
        m_task_queue->resize(static_cast<intmax_t>(_n));
}

Referenced by PTL::TaskRunManager::Initialize(), and G4TaskRunManager::SetNumberOfThreads().

◆ run_on_this()

int PTL::ThreadPool::run_on_this ( task_pointer && _task )

inlineprotected

Definition at line 421 of file ThreadPool.hh.

{
    auto&& _func = [_task]() { (*_task)(); };
 
    if(m_tbb_tp && m_tbb_task_group)
    {
        auto* _arena = get_task_arena();
        _arena->execute([this, _func]() { this->m_tbb_task_group->run(_func); });
    }
    else
    {
        _func();
    }
    // return the number of tasks added to task-list
    return 0;
}

Referenced by add_task().

◆ set_affinity() [1/2]

void PTL::ThreadPool::set_affinity ( affinity_func_t f )

inline

Definition at line 264 of file ThreadPool.hh.

264{ m_affinity_func = std::move(f); }

Referenced by initialize_threadpool().

◆ set_affinity() [2/2]

void ThreadPool::set_affinity	(	intmax_t	i,
		Thread &	_thread
	)		const

Definition at line 355 of file ThreadPool.cc.

{
    try
    {
        NativeThread native_thread = _thread.native_handle();
        intmax_t     _pin          = m_affinity_func(i);
        if(m_verbose > 0)
        {
            AutoLock lock(TypeMutex<decltype(std::cerr)>());
            std::cerr << "[PTL::ThreadPool] Setting pin affinity for thread "
                      << get_thread_id(_thread.get_id()) << " to " << _pin << std::endl;
        }
        Threading::SetPinAffinity((int)_pin, native_thread);
    } catch(std::runtime_error& e)
    {
        std::cerr << "[PTL::ThreadPool] Error setting pin affinity: " << e.what()
                  << std::endl;
    }
}

◆ set_default_scheduling_priority()

static void PTL::ThreadPool::set_default_scheduling_priority ( int _v )

inlinestatic

set the default scheduling priority of threads in thread-pool

Definition at line 205 of file ThreadPool.hh.

205{ f_thread_priority() = _v; }

◆ set_default_size()

static void PTL::ThreadPool::set_default_size ( size_type _v )

inlinestatic

set the default pool size

Definition at line 209 of file ThreadPool.hh.

209{ f_default_pool_size() = _v; }

◆ set_default_use_cpu_affinity()

void ThreadPool::set_default_use_cpu_affinity ( bool _v )

static

set the default use of cpu affinity

Definition at line 171 of file ThreadPool.cc.

{
#if defined(PTL_USE_TBB)
    f_use_cpu_affinity() = enable;
#else
    ConsumeParameters(enable);
#endif
}

◆ set_default_use_tbb()

static void PTL::ThreadPool::set_default_use_tbb ( bool _v )

inlinestatic

set the default use of tbb

Definition at line 201 of file ThreadPool.hh.

201{ set_use_tbb(_v); }

PTL::ThreadPool::set_use_tbb

static void set_use_tbb(bool _v)

Definition: ThreadPool.cc:159

◆ set_default_verbose()

static void PTL::ThreadPool::set_default_verbose ( int _v )

inlinestatic

set the default verbosity

Definition at line 207 of file ThreadPool.hh.

207{ f_verbose() = _v; }

◆ set_finalization()

void PTL::ThreadPool::set_finalization ( finalize_func_t f )

inline

Definition at line 237 of file ThreadPool.hh.

237{ m_fini_func = std::move(f); }

◆ set_initialization()

void PTL::ThreadPool::set_initialization ( initialize_func_t f )

inline

Definition at line 236 of file ThreadPool.hh.

236{ m_init_func = std::move(f); }

◆ set_priority()

void ThreadPool::set_priority	(	int	_prio,
		Thread &	_thread
	)		const

Definition at line 378 of file ThreadPool.cc.

{
    try
    {
        NativeThread native_thread = _thread.native_handle();
        if(m_verbose > 0)
        {
            AutoLock lock(TypeMutex<decltype(std::cerr)>());
            std::cerr << "[PTL::ThreadPool] Setting thread "
                      << get_thread_id(_thread.get_id()) << " priority to " << _prio
                      << std::endl;
        }
        Threading::SetThreadPriority(_prio, native_thread);
    } catch(std::runtime_error& e)
    {
        AutoLock lock(TypeMutex<decltype(std::cerr)>());
        std::cerr << "[PTL::ThreadPool] Error setting thread priority: " << e.what()
                  << std::endl;
    }
}

Referenced by initialize_threadpool().

◆ set_use_tbb()

void ThreadPool::set_use_tbb ( bool _v )

static

Definition at line 159 of file ThreadPool.cc.

{
#if defined(PTL_USE_TBB)
    f_use_tbb() = enable;
#else
    ConsumeParameters(enable);
#endif
}

Referenced by G4TaskRunManager::G4TaskRunManager(), set_default_use_tbb(), and PTL::TaskRunManager::TaskRunManager().

◆ set_verbose()

void PTL::ThreadPool::set_verbose ( int n )

inline

Definition at line 268 of file ThreadPool.hh.

268{ m_verbose = n; }

CLHEP::detail::n

n

Definition: Ranlux64Engine.cc:90

◆ size()

size_type PTL::ThreadPool::size ( ) const

inline

Definition at line 252 of file ThreadPool.hh.

252{ return m_pool_size; }

Referenced by G4TaskRunManager::ComputeNumberOfTasks(), execute_on_all_threads(), execute_on_specific_threads(), get_task_arena(), PTL::TaskRunManager::GetNumberActiveThreads(), PTL::TaskRunManager::GetNumberOfThreads(), PTL::TaskRunManager::Initialize(), G4TaskRunManagerKernel::InitializeWorker(), notify(), PTL::TaskManager::size(), and PTL::TaskGroup< Tp, Arg, MaxDepth >::wait().

◆ start_thread()

void ThreadPool::start_thread	(	ThreadPool *	tp,
		thread_data_t *	_data,
		intmax_t	_idx = `-1`
	)

staticprotected

Definition at line 116 of file ThreadPool.cc.

{
    if(tp->get_verbose() > 0)
    {
        AutoLock lock(TypeMutex<decltype(std::cerr)>());
        std::cerr << "[PTL::ThreadPool] Starting thread " << _idx << "..." << std::endl;
    }
 
    auto _thr_data = std::make_shared<ThreadData>(tp);
    {
        AutoLock lock(TypeMutex<ThreadPool>(), std::defer_lock);
        if(!lock.owns_lock())
            lock.lock();
        if(_idx < 0)
            _idx = f_thread_ids().size();
        f_thread_ids()[std::this_thread::get_id()] = _idx;
        Threading::SetThreadId((int)_idx);
        _data->emplace_back(_thr_data);
    }
    thread_data() = _thr_data.get();
    tp->record_entry();
    tp->execute_thread(thread_data()->current_queue);
    tp->record_exit();
 
    if(tp->get_verbose() > 0)
    {
        AutoLock lock(TypeMutex<decltype(std::cerr)>());
        std::cerr << "[PTL::ThreadPool] Thread " << _idx << " terminating..."
                  << std::endl;
    }
}

Referenced by initialize_threadpool().

◆ state()

const pool_state_type & PTL::ThreadPool::state ( ) const

inline

Definition at line 250 of file ThreadPool.hh.

250{ return m_pool_state; }

Referenced by PTL::TaskGroup< Tp, Arg, MaxDepth >::wait().

◆ stop_thread()

ThreadPool::size_type ThreadPool::stop_thread ( )

Definition at line 711 of file ThreadPool.cc.

{
    if(!m_alive_flag->load() || m_pool_size == 0)
        return 0;
 
    m_pool_state->store(thread_pool::state::PARTIAL);
 
    //------------------------------------------------------------------------//
    // notify all threads we are shutting down
    m_task_lock->lock();
    m_is_stopped.push_back(true);
    m_task_cond->notify_one();
    m_task_lock->unlock();
    //------------------------------------------------------------------------//
 
    while(!m_is_stopped.empty() && m_stop_threads.empty())
        ;
 
    // lock up the task queue
    AutoLock _task_lock(*m_task_lock);
 
    while(!m_stop_threads.empty())
    {
        auto tid = m_stop_threads.front();
        // remove from stopped
        m_stop_threads.pop_front();
        // remove from main
        for(auto itr = m_main_threads.begin(); itr != m_main_threads.end(); ++itr)
        {
            if(*itr == tid)
            {
                m_main_threads.erase(itr);
                break;
            }
        }
        // remove from join list
        m_is_joined.pop_back();
    }
 
    m_pool_state->store(thread_pool::state::STARTED);
 
    m_pool_size = m_main_threads.size();
    return m_main_threads.size();
}

Referenced by initialize_threadpool().

◆ tbb_global_control()

tbb_global_control_t *& PTL::ThreadPool::tbb_global_control ( )

inlinestatic

Definition at line 384 of file ThreadPool.hh.

{
    static thread_local tbb_global_control_t* _instance = nullptr;
    return _instance;
}

Referenced by destroy_threadpool(), execute_on_all_threads(), get_task_arena(), and initialize_threadpool().

◆ using_affinity()

bool PTL::ThreadPool::using_affinity ( ) const

inline

Definition at line 256 of file ThreadPool.hh.

256{ return m_use_affinity; }

◆ using_tbb()

bool ThreadPool::using_tbb ( )

static

Definition at line 151 of file ThreadPool.cc.

{
    return f_use_tbb();
}

Referenced by PTL::TaskRunManager::Initialize().

The documentation for this class was generated from the following files:

Classes

Public Types

Public Member Functions

Static Public Member Functions

Protected Member Functions

Static Protected Member Functions

Detailed Description

Member Typedef Documentation

◆ affinity_func_t

◆ atomic_bool_type

◆ atomic_int_type

◆ bool_list_t

◆ condition_t

◆ finalize_func_t

◆ initialize_func_t

◆ lock_t

◆ pool_state_type

◆ size_type

◆ task_count_type

◆ task_pointer

◆ task_queue_t

◆ task_type

◆ thread_data_t

◆ thread_id_map_t

◆ thread_index_map_t

◆ thread_list_t

◆ thread_vec_t

◆ uomap

Constructor & Destructor Documentation

◆ ThreadPool() [1/5]

◆ ThreadPool() [2/5]

◆ ThreadPool() [3/5]

◆ ~ThreadPool()

◆ ThreadPool() [4/5]

◆ ThreadPool() [5/5]

Member Function Documentation

◆ add_task()

◆ add_tasks()

◆ add_thread_id()

◆ affinity_functor()

◆ destroy_threadpool()

◆ execute_on_all_threads()

◆ execute_on_specific_threads()

◆ execute_thread()

◆ finalization_functor()

◆ get_active_threads_count()

◆ get_default_scheduling_priority()

◆ get_default_size()

◆ get_default_use_cpu_affinity()

◆ get_default_use_tbb()

◆ get_default_verbose()

◆ get_queue()

◆ get_task_arena()

◆ get_this_thread_id()

◆ get_thread() [1/2]

◆ get_thread() [2/2]

◆ get_thread_id()

◆ get_thread_ids()

◆ get_valid_queue()

◆ get_verbose()

◆ initialization_functor()

◆ initialize_threadpool()

◆ insert()

◆ is_alive()

◆ is_initialized()

◆ is_main()

◆ is_tbb_threadpool()

◆ notify() [1/2]

◆ notify() [2/2]

◆ notify_all()

◆ operator=() [1/2]

◆ operator=() [2/2]

◆ record_entry()

◆ record_exit()

◆ reset_finalization()

◆ reset_initialization()

◆ resize()

◆ run_on_this()

◆ set_affinity() [1/2]

◆ set_affinity() [2/2]