#include <UserTaskQueue.hh>

Inheritance diagram for PTL::UserTaskQueue:

Public Types
typedef VTask *	task_pointer

typedef std::vector< TaskSubQueue * >	TaskSubQueueContainer

typedef std::default_random_engine	random_engine_t

typedef std::uniform_int_distribution< int >	int_dist_t

Public Types inherited from PTL::VUserTaskQueue
typedef VTask *	task_pointer

typedef std::atomic< intmax_t >	AtomicInt

typedef uintmax_t	size_type

typedef std::function< void()>	function_type

typedef std::set< ThreadId >	ThreadIdSet

Public Member Functions
	UserTaskQueue (intmax_t nworkers=-1, UserTaskQueue *=nullptr)

virtual	~UserTaskQueue () override

virtual task_pointer	GetTask (intmax_t subq=-1, intmax_t nitr=-1) override

virtual intmax_t	InsertTask (task_pointer, ThreadData *=nullptr, intmax_t subq=-1) override

task_pointer	GetThreadBinTask ()

virtual void	Wait () override

virtual void	resize (intmax_t) override

virtual bool	empty () const override

virtual size_type	size () const override

virtual size_type	bin_size (size_type bin) const override

virtual bool	bin_empty (size_type bin) const override

bool	true_empty () const override

size_type	true_size () const override

virtual void	ExecuteOnAllThreads (ThreadPool *tp, function_type f) override

virtual void	ExecuteOnSpecificThreads (ThreadIdSet tid_set, ThreadPool *tp, function_type f) override

virtual VUserTaskQueue *	clone () override

virtual intmax_t	GetThreadBin () const override

Public Member Functions inherited from PTL::VUserTaskQueue
	VUserTaskQueue (intmax_t nworkers=-1)

virtual	~VUserTaskQueue ()

virtual task_pointer	GetTask (intmax_t subq=-1, intmax_t nitr=-1)=0

virtual intmax_t	InsertTask (task_pointer, ThreadData *=nullptr, intmax_t subq=-1)=0

virtual void	Wait ()=0

virtual intmax_t	GetThreadBin () const =0

virtual void	resize (intmax_t)=0

virtual size_type	size () const =0

virtual bool	empty () const =0

virtual size_type	bin_size (size_type bin) const =0

virtual bool	bin_empty (size_type bin) const =0

virtual size_type	true_size () const

virtual bool	true_empty () const

virtual void	ExecuteOnAllThreads (ThreadPool *tp, function_type f)=0

virtual void	ExecuteOnSpecificThreads (ThreadIdSet tid_set, ThreadPool *tp, function_type f)=0

intmax_t	workers () const

virtual VUserTaskQueue *	clone ()=0

Protected Member Functions
intmax_t	GetInsertBin () const

Additional Inherited Members
Static Public Member Functions inherited from PTL::VUserTaskQueue
template<typename ContainerT , size_t... Idx>
static auto	ContainerToTupleImpl (ContainerT &&container, mpl::index_sequence< Idx... >) -> decltype(std::make_tuple(std::forward< ContainerT >(container)[Idx]...))

template<std::size_t N, typename ContainerT >
static auto	ContainerToTuple (ContainerT &&container) -> decltype(ContainerToTupleImpl(std::forward< ContainerT >(container), mpl::make_index_sequence< N >{}))

template<std::size_t N, std::size_t Nt, typename TupleT , enable_if_t<(N==Nt), int > = 0>
static void	TExecutor (TupleT &&_t)

template<std::size_t N, std::size_t Nt, typename TupleT , enable_if_t<(N< Nt), int > = 0>
static void	TExecutor (TupleT &&_t)

template<typename TupleT , std::size_t N = std::tuple_size<decay_t<TupleT>>::value>
static void	Executor (TupleT &&__t)

template<typename Container , typename std::enable_if< std::is_same< Container, task_pointer >::value, int >::type = 0>
static void	Execute (Container &obj)

template<typename Container , typename std::enable_if<!std::is_same< Container, task_pointer >::value, int >::type = 0>
static void	Execute (Container &tasks)

Protected Attributes inherited from PTL::VUserTaskQueue
intmax_t	m_workers = 0

Detailed Description

Definition at line 47 of file UserTaskQueue.hh.

Member Typedef Documentation

◆ int_dist_t

typedef std::uniform_int_distribution<int> PTL::UserTaskQueue::int_dist_t

Definition at line 53 of file UserTaskQueue.hh.

◆ random_engine_t

typedef std::default_random_engine PTL::UserTaskQueue::random_engine_t

Definition at line 52 of file UserTaskQueue.hh.

◆ task_pointer

typedef VTask* PTL::UserTaskQueue::task_pointer

Definition at line 50 of file UserTaskQueue.hh.

◆ TaskSubQueueContainer

typedef std::vector<TaskSubQueue*> PTL::UserTaskQueue::TaskSubQueueContainer

Definition at line 51 of file UserTaskQueue.hh.

Constructor & Destructor Documentation

◆ UserTaskQueue()

UserTaskQueue::UserTaskQueue	(	intmax_t	nworkers = `-1`,
		UserTaskQueue *	parent = `nullptr`
	)

Definition at line 37 of file UserTaskQueue.cc.

: VUserTaskQueue(nworkers)
, m_is_clone((parent) ? true : false)
, m_thread_bin((parent) ? (ThreadPool::get_this_thread_id() % (nworkers + 1)) : 0)
, m_insert_bin((parent) ? (ThreadPool::get_this_thread_id() % (nworkers + 1)) : 0)
, m_hold((parent) ? parent->m_hold : new std::atomic_bool(false))
, m_ntasks((parent) ? parent->m_ntasks : new std::atomic_uintmax_t(0))
, m_subqueues((parent) ? parent->m_subqueues : new TaskSubQueueContainer())
{
    // create nthreads + 1 subqueues so there is always a subqueue available
    if(!parent)
    {
        for(intmax_t i = 0; i < nworkers + 1; ++i)
            m_subqueues->push_back(new TaskSubQueue(m_ntasks));
    }
 
#if defined(DEBUG)
    if(GetEnv<int>("PTL_VERBOSE", 0) > 3)
    {
        RecursiveAutoLock l(TypeRecursiveMutex<decltype(std::cout)>());
        std::stringstream ss;
        ss << ThreadPool::get_this_thread_id() << "> " << ThisThread::get_id() << " ["
           << __FUNCTION__ << ":" << __LINE__ << "] "
           << "this = " << this << ", "
           << "clone = " << std::boolalpha << m_is_clone << ", "
           << "thread = " << m_thread_bin << ", "
           << "insert = " << m_insert_bin << ", "
           << "hold = " << m_hold->load() << " @ " << m_hold << ", "
           << "tasks = " << m_ntasks->load() << " @ " << m_ntasks << ", "
           << "subqueue = " << m_subqueues << ", "
           << "size = " << true_size() << ", "
           << "empty = " << true_empty();
        std::cout << ss.str() << std::endl;
    }
#endif
}

◆ ~UserTaskQueue()

UserTaskQueue::~UserTaskQueue ( )

overridevirtual

Definition at line 76 of file UserTaskQueue.cc.

{
    if(!m_is_clone)
    {
        for(auto& itr : *m_subqueues)
        {
            assert(itr->size() == 0);
            delete itr;
        }
        m_subqueues->clear();
        delete m_hold;
        delete m_ntasks;
        delete m_subqueues;
    }
}

Member Function Documentation

◆ bin_empty()

bool PTL::UserTaskQueue::bin_empty ( size_type bin ) const

inlineoverridevirtual

Implements PTL::VUserTaskQueue.

Definition at line 146 of file UserTaskQueue.hh.

{
    return (*m_subqueues)[bin]->empty();
}

◆ bin_size()

PTL::UserTaskQueue::size_type PTL::UserTaskQueue::bin_size ( size_type bin ) const

inlineoverridevirtual

Implements PTL::VUserTaskQueue.

Definition at line 138 of file UserTaskQueue.hh.

{
    return (*m_subqueues)[bin]->size();
}

◆ clone()

VUserTaskQueue * UserTaskQueue::clone ( )

overridevirtual

Implements PTL::VUserTaskQueue.

Definition at line 120 of file UserTaskQueue.cc.

{
    return new UserTaskQueue(workers(), this);
}

◆ empty()

bool PTL::UserTaskQueue::empty ( ) const

inlineoverridevirtual

Implements PTL::VUserTaskQueue.

Definition at line 122 of file UserTaskQueue.hh.

{
    return (m_ntasks->load(std::memory_order_relaxed) == 0);
}

◆ ExecuteOnAllThreads()

void UserTaskQueue::ExecuteOnAllThreads	(	ThreadPool *	tp,
		function_type	f
	)

overridevirtual

Implements PTL::VUserTaskQueue.

Definition at line 305 of file UserTaskQueue.cc.

{
    typedef TaskGroup<int, int>     task_group_type;
    typedef std::map<int64_t, bool> thread_execute_map_t;
 
    if(!tp->is_alive())
    {
        func();
        return;
    }
 
    auto join_func = [=](int& ref, int i) {
        ref += i;
        return ref;
    };
    task_group_type* tg = new task_group_type(join_func, tp);
 
    // wait for all threads to finish any work
    // NOTE: will cause deadlock if called from a task
    while(tp->get_active_threads_count() > 0)
        ThisThread::sleep_for(std::chrono::milliseconds(10));
 
    thread_execute_map_t* thread_execute_map = new thread_execute_map_t();
 
    AcquireHold();
    for(int i = 0; i < (m_workers + 1); ++i)
    {
        if(i == GetThreadBin())
            continue;
 
        //--------------------------------------------------------------------//
        auto thread_specific_func = [&]() {
            static Mutex _mtx;
            _mtx.lock();
            bool& _executed = (*thread_execute_map)[GetThreadBin()];
            _mtx.unlock();
            if(!_executed)
            {
                func();
                _executed = true;
                return 1;
            }
            return 0;
        };
        //--------------------------------------------------------------------//
 
        auto _task = tg->wrap(thread_specific_func);
        //++(*m_ntasks);
        // TaskSubQueue* task_subq = (*m_subqueues)[i];
        // task_subq->PushTask(_task);
        InsertTask(_task, ThreadData::GetInstance(), i);
    }
 
    tp->notify_all();
    int nexecuted = tg->join();
    if(nexecuted != m_workers)
    {
        std::stringstream msg;
        msg << "Failure executing routine on all threads! Only " << nexecuted
            << " threads executed function out of " << m_workers;
        std::cerr << msg.str() << std::endl;
        // Exception("UserTaskQueue::ExecuteOnAllThreads", "TaskQueue0000",
        //            JustWarning, msg);
    }
    delete thread_execute_map;
    ReleaseHold();
}

◆ ExecuteOnSpecificThreads()

void UserTaskQueue::ExecuteOnSpecificThreads	(	ThreadIdSet	tid_set,
		ThreadPool *	tp,
		function_type	f
	)

overridevirtual

Implements PTL::VUserTaskQueue.

Definition at line 376 of file UserTaskQueue.cc.

{
    typedef TaskGroup<int, int>     task_group_type;
    typedef std::map<int64_t, bool> thread_execute_map_t;
 
    auto join_func = [=](int& ref, int i) {
        ref += i;
        return ref;
    };
    task_group_type* tg = new task_group_type(join_func, tp);
 
    // wait for all threads to finish any work
    // NOTE: will cause deadlock if called from a task
    while(tp->get_active_threads_count() > 0)
        ThisThread::sleep_for(std::chrono::milliseconds(10));
 
    if(!tp->is_alive())
    {
        func();
        return;
    }
 
    thread_execute_map_t* thread_execute_map = new thread_execute_map_t();
 
    //========================================================================//
    // wrap the function so that it will only be executed if the thread
    // has an ID in the set
    auto thread_specific_func = [=]() {
        static Mutex _mtx;
        _mtx.lock();
        bool& _executed = (*thread_execute_map)[GetThreadBin()];
        _mtx.unlock();
        if(!_executed && tid_set.count(ThisThread::get_id()) > 0)
        {
            func();
            _executed = true;
            return 1;
        }
        return 0;
    };
    //========================================================================//
 
    if(tid_set.count(ThisThread::get_id()) > 0)
        func();
 
    AcquireHold();
    for(int i = 0; i < (m_workers + 1); ++i)
    {
        if(i == GetThreadBin())
            continue;
 
        auto _task = tg->wrap(thread_specific_func);
        InsertTask(_task, ThreadData::GetInstance(), i);
    }
    tp->notify_all();
    int nexecuted = tg->join();
    if(nexecuted != m_workers)
    {
        std::stringstream msg;
        msg << "Failure executing routine on all threads! Only " << nexecuted
            << " threads executed function out of " << tid_set.size();
        std::cerr << msg.str() << std::endl;
        // Exception("UserTaskQueue::ExecuteOnSpecificThreads", "TaskQueue0001",
        //            JustWarning, msg);
    }
    delete thread_execute_map;
    ReleaseHold();
}

◆ GetInsertBin()

intmax_t UserTaskQueue::GetInsertBin ( ) const

protected

Definition at line 138 of file UserTaskQueue.cc.

{
    return (++m_insert_bin % (m_workers + 1));
}

Referenced by InsertTask().

◆ GetTask()

UserTaskQueue::task_pointer UserTaskQueue::GetTask	(	intmax_t	subq = `-1`,
		intmax_t	nitr = `-1`
	)

overridevirtual

Implements PTL::VUserTaskQueue.

Definition at line 180 of file UserTaskQueue.cc.

{
    // exit if empty
    if(this->true_empty())
        return nullptr;
 
    // ensure the thread has a bin assignment
    intmax_t tbin = GetThreadBin();
    intmax_t n    = (subq < 0) ? tbin : subq;
    if(nitr < 1)
        nitr = (m_workers + 1);  // * m_ntasks->load(std::memory_order_relaxed);
 
    if(m_hold->load(std::memory_order_relaxed))
    {
        return GetThreadBinTask();
    }
 
    task_pointer _task = nullptr;
    //------------------------------------------------------------------------//
    auto get_task = [&](intmax_t _n) {
        TaskSubQueue* task_subq = (*m_subqueues)[_n % (m_workers + 1)];
        // try to acquire a claim for the bin
        // if acquired, no other threads will access bin until claim is released
        if(!task_subq->empty() && task_subq->AcquireClaim())
        {
            // pop task out of bin
            _task = task_subq->PopTask(n == tbin);
            // release the claim on the bin
            task_subq->ReleaseClaim();
        }
        if(_task)
            --(*m_ntasks);
        // return success if valid pointer
        return (_task != nullptr);
    };
    //------------------------------------------------------------------------//
 
    // there are num_workers+1 bins so there is always a bin that is open
    // execute num_workers+2 iterations so the thread checks its bin twice
    // while(!empty())
    {
        for(intmax_t i = 0; i < nitr; ++i, ++n)
        {
            if(get_task(n % (m_workers + 1)))
                return _task;
        }
    }
 
    // only reached if looped over all bins (and looked in own bin twice)
    // and found no work so return an empty task and the thread will be put to
    // sleep if there is still no work by the time it reaches its
    // condition variable
    return _task;
}

◆ GetThreadBin()

intmax_t UserTaskQueue::GetThreadBin ( ) const

overridevirtual

Implements PTL::VUserTaskQueue.

Definition at line 127 of file UserTaskQueue.cc.

{
    // get a thread id number
    static thread_local intmax_t tl_bin =
        (m_thread_bin + ThreadPool::get_this_thread_id()) % (m_workers + 1);
    return tl_bin;
}

Referenced by ExecuteOnAllThreads(), ExecuteOnSpecificThreads(), GetTask(), GetThreadBinTask(), and InsertTask().

◆ GetThreadBinTask()

UserTaskQueue::task_pointer UserTaskQueue::GetThreadBinTask ( )

Definition at line 146 of file UserTaskQueue.cc.

{
    intmax_t      tbin      = GetThreadBin();
    TaskSubQueue* task_subq = (*m_subqueues)[tbin % (m_workers + 1)];
    task_pointer  _task     = nullptr;
 
    //------------------------------------------------------------------------//
    auto get_task = [&]() {
        if(task_subq->AcquireClaim())
        {
            // run task
            _task = task_subq->PopTask(true);
            // release the claim on the bin
            task_subq->ReleaseClaim();
        }
        if(_task)
            --(*m_ntasks);
        // return success if valid pointer
        return (_task != nullptr);
    };
    //------------------------------------------------------------------------//
 
    // while not empty
    while(!task_subq->empty())
    {
        if(get_task())
            break;
    }
    return _task;
}

Referenced by GetTask().

◆ InsertTask()

intmax_t UserTaskQueue::InsertTask	(	task_pointer	task,
		ThreadData *	data = `nullptr`,
		intmax_t	subq = `-1`
	)

overridevirtual

Implements PTL::VUserTaskQueue.

Definition at line 238 of file UserTaskQueue.cc.

{
    // skip increment here (handled externally)
    ++(*m_ntasks);
 
    bool     spin = m_hold->load(std::memory_order_relaxed);
    intmax_t tbin = GetThreadBin();
 
    if(data && data->within_task)
    {
        subq = tbin;
        // spin = true;
    }
 
    // subq is -1 unless specified so unless specified
    // GetInsertBin() call increments a counter and returns
    // counter % (num_workers + 1) so that tasks are distributed evenly
    // among the bins
    intmax_t n = (subq < 0) ? GetInsertBin() : subq;
 
    //------------------------------------------------------------------------//
    auto insert_task = [&](intmax_t _n) {
        TaskSubQueue* task_subq = (*m_subqueues)[_n];
        // TaskSubQueue* next_subq = (*m_subqueues)[(_n + 1) % (m_workers + 1)];
        // if not threads bin and size difference, insert into smaller
        // if(n != tbin && next_subq->size() < task_subq->size())
        //    task_subq = next_subq;
        // try to acquire a claim for the bin
        // if acquired, no other threads will access bin until claim is released
        if(task_subq->AcquireClaim())
        {
            // push the task into the bin
            task_subq->PushTask(task);
            // release the claim on the bin
            task_subq->ReleaseClaim();
            // return success
            return true;
        }
        return false;
    };
    //------------------------------------------------------------------------//
 
    // if not in "hold/spin mode", where thread only inserts tasks into
    // specified bin, then move onto next bin
    //
    if(spin)
    {
        n = n % (m_workers + 1);
        while(!insert_task(n))
            ;
        return n;
    }
 
    // there are num_workers+1 bins so there is always a bin that is open
    // execute num_workers+2 iterations so the thread checks its bin twice
    while(true)
    {
        auto _n = (n++) % (m_workers + 1);
        if(insert_task(_n))
            return _n;
    }
    return GetThreadBin();
}

Referenced by ExecuteOnAllThreads(), and ExecuteOnSpecificThreads().

◆ resize()

void UserTaskQueue::resize ( intmax_t n )

overridevirtual

Implements PTL::VUserTaskQueue.

Definition at line 95 of file UserTaskQueue.cc.

{
    AutoLock l(m_mutex);
    if(m_workers < n)
    {
        while(m_workers < n)
        {
            m_subqueues->push_back(new TaskSubQueue(m_ntasks));
            ++m_workers;
        }
    }
    else if(m_workers > n)
    {
        while(m_workers > n)
        {
            delete m_subqueues->back();
            m_subqueues->pop_back();
            --m_workers;
        }
    }
}

◆ size()

PTL::UserTaskQueue::size_type PTL::UserTaskQueue::size ( ) const

inlineoverridevirtual

Implements PTL::VUserTaskQueue.

Definition at line 130 of file UserTaskQueue.hh.

{
    return m_ntasks->load(std::memory_order_relaxed);
}

◆ true_empty()

bool PTL::UserTaskQueue::true_empty ( ) const

inlineoverridevirtual

Reimplemented from PTL::VUserTaskQueue.

Definition at line 154 of file UserTaskQueue.hh.

{
    for(const auto& itr : *m_subqueues)
        if(!itr->empty())
            return false;
    return true;
}

Referenced by GetTask(), and UserTaskQueue().

◆ true_size()

PTL::UserTaskQueue::size_type PTL::UserTaskQueue::true_size ( ) const

inlineoverridevirtual

Reimplemented from PTL::VUserTaskQueue.

Definition at line 165 of file UserTaskQueue.hh.

{
    size_type _n = 0;
    for(const auto& itr : *m_subqueues)
        _n += itr->size();
    return _n;
}

Referenced by UserTaskQueue().

◆ Wait()

virtual void PTL::UserTaskQueue::Wait ( )

inlineoverridevirtual

Implements PTL::VUserTaskQueue.

Definition at line 73 of file UserTaskQueue.hh.

73{}

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

Public Types

Public Member Functions

Protected Member Functions

Additional Inherited Members

Detailed Description

Member Typedef Documentation

◆ int_dist_t

◆ random_engine_t

◆ task_pointer

◆ TaskSubQueueContainer

Constructor & Destructor Documentation

◆ UserTaskQueue()

◆ ~UserTaskQueue()

Member Function Documentation

◆ bin_empty()

◆ bin_size()

◆ clone()

◆ empty()

◆ ExecuteOnAllThreads()

◆ ExecuteOnSpecificThreads()

◆ GetInsertBin()

◆ GetTask()

◆ GetThreadBin()

◆ GetThreadBinTask()

◆ InsertTask()

◆ resize()

◆ size()

◆ true_empty()

◆ true_size()

◆ Wait()