PTL::UserTaskQueue Class Reference

#include <UserTaskQueue.hh>

Inheritance diagram for PTL::UserTaskQueue:

Public Types
using	task_pointer = std::shared_ptr< VTask >
using	TaskSubQueueContainer = std::vector< TaskSubQueue * >
using	random_engine_t = std::default_random_engine
using	int_dist_t = std::uniform_int_distribution< int >
Public Types inherited from PTL::VUserTaskQueue
using	task_pointer = std::shared_ptr< VTask >
using	AtomicInt = std::atomic< intmax_t >
using	size_type = uintmax_t
using	function_type = std::function< void()>
using	ThreadIdSet = std::set< ThreadId >

Public Member Functions
	UserTaskQueue (intmax_t nworkers=-1, UserTaskQueue *=nullptr)
	~UserTaskQueue () override
task_pointer	GetTask (intmax_t subq=-1, intmax_t nitr=-1) override
intmax_t	InsertTask (task_pointer &&, ThreadData *=nullptr, intmax_t subq=-1) override PTL_NO_SANITIZE_THREAD
task_pointer	GetThreadBinTask ()
void	Wait () override
void	resize (intmax_t) override
bool	empty () const override
size_type	size () const override
size_type	bin_size (size_type bin) const override
bool	bin_empty (size_type bin) const override
bool	true_empty () const override
size_type	true_size () const override
void	ExecuteOnAllThreads (ThreadPool *tp, function_type f) override
void	ExecuteOnSpecificThreads (ThreadIdSet tid_set, ThreadPool *tp, function_type f) override
VUserTaskQueue *	clone () override
intmax_t	GetThreadBin () const override
Public Member Functions inherited from PTL::VUserTaskQueue
	VUserTaskQueue (intmax_t nworkers=-1)
virtual	~VUserTaskQueue ()=default
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) PTL_NO_SANITIZE_THREAD=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
Protected Attributes inherited from PTL::VUserTaskQueue
intmax_t	m_workers = 0

Detailed Description

Definition at line 42 of file UserTaskQueue.hh.

Member Typedef Documentation

int_dist_t

using PTL::UserTaskQueue::int_dist_t = std::uniform_int_distribution<int>

Definition at line 48 of file UserTaskQueue.hh.

random_engine_t

using PTL::UserTaskQueue::random_engine_t = std::default_random_engine

Definition at line 47 of file UserTaskQueue.hh.

task_pointer

using PTL::UserTaskQueue::task_pointer = std::shared_ptr<VTask>

Definition at line 45 of file UserTaskQueue.hh.

TaskSubQueueContainer

using PTL::UserTaskQueue::TaskSubQueueContainer = std::vector<TaskSubQueue*>

Definition at line 46 of file UserTaskQueue.hh.

Constructor & Destructor Documentation

UserTaskQueue()

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

Definition at line 48 of file UserTaskQueue.cc.

: VUserTaskQueue(nworkers)
, m_is_clone((parent) != nullptr)
, 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_mutex((parent) ? parent->m_mutex : new Mutex{})
, 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->emplace_back(new TaskSubQueue(m_ntasks));
    }
 
#if defined(DEBUG)
    if(GetEnv<int>("PTL_VERBOSE", 0) > 3)
    {
        RecursiveAutoLock l(TypeMutex<decltype(std::cout), RecursiveMutex>());
        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 ( )

override

Definition at line 88 of file UserTaskQueue.cc.

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

Member Function Documentation

bin_empty()

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

inlineoverridevirtual

Implements PTL::VUserTaskQueue.

Definition at line 135 of file UserTaskQueue.hh.

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

bin_size()

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

inlineoverridevirtual

Implements PTL::VUserTaskQueue.

Definition at line 127 of file UserTaskQueue.hh.

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

clone()

VUserTaskQueue * UserTaskQueue::clone ( )

overridevirtual

Implements PTL::VUserTaskQueue.

Definition at line 135 of file UserTaskQueue.cc.

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

empty()

bool PTL::UserTaskQueue::empty ( ) const

inlineoverridevirtual

Implements PTL::VUserTaskQueue.

Definition at line 111 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 320 of file UserTaskQueue.cc.

{
    using task_group_type      = TaskGroup<int, int>;
    using thread_execute_map_t = std::map<int64_t, bool>;
 
    if(!tp->is_alive())
    {
        func();
        return;
    }
 
    task_group_type tg{ [](int& ref, int i) { return (ref += i); }, 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{};
    std::vector<std::shared_ptr<VTask>> _tasks{};
    _tasks.reserve(m_workers + 1);
 
    AcquireHold();
    for(int i = 0; i < (m_workers + 1); ++i)
    {
        if(i == GetThreadBin())
            continue;
 
        //--------------------------------------------------------------------//
        auto thread_specific_func = [&]() {
            ScopeDestructor _dtor = tg.get_scope_destructor();
            static Mutex    _mtx;
            _mtx.lock();
            bool& _executed = thread_execute_map[GetThreadBin()];
            _mtx.unlock();
            if(!_executed)
            {
                func();
                _executed = true;
                return 1;
            }
            return 0;
        };
        //--------------------------------------------------------------------//
 
        InsertTask(tg.wrap(thread_specific_func), 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 << " workers";
        std::cerr << msg.str() << std::endl;
    }
    ReleaseHold();
}

ExecuteOnSpecificThreads()

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

overridevirtual

Implements PTL::VUserTaskQueue.

Definition at line 383 of file UserTaskQueue.cc.

{
    using task_group_type      = TaskGroup<int, int>;
    using thread_execute_map_t = std::map<int64_t, bool>;
 
    task_group_type tg{ [](int& ref, int i) { return (ref += i); }, 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{};
 
    //========================================================================//
    // wrap the function so that it will only be executed if the thread
    // has an ID in the set
    auto thread_specific_func = [&]() {
        ScopeDestructor _dtor = tg.get_scope_destructor();
        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;
 
        InsertTask(tg.wrap(thread_specific_func), ThreadData::GetInstance(), i);
    }
    tp->notify_all();
    decltype(tid_set.size()) nexecuted = tg.join();
    if(nexecuted != tid_set.size())
    {
        std::stringstream msg;
        msg << "Failure executing routine on specific threads! Only " << nexecuted
            << " threads executed function out of " << tid_set.size() << " workers";
        std::cerr << msg.str() << std::endl;
    }
    ReleaseHold();
}

GetInsertBin()

intmax_t UserTaskQueue::GetInsertBin ( ) const

protected

Definition at line 153 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 195 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 142 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 161 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 253 of file UserTaskQueue.cc.

{
    // increment number of tasks
    ++(*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(std::move(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 108 of file UserTaskQueue.cc.

{
    if(!m_mutex)
        throw std::runtime_error("nullptr to mutex");
    AutoLock lk(m_mutex);
    if(m_workers < n)
    {
        while(m_workers < n)
        {
            m_subqueues->emplace_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()

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

inlineoverridevirtual

Implements PTL::VUserTaskQueue.

Definition at line 119 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 143 of file UserTaskQueue.hh.

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

Referenced by GetTask(), and UserTaskQueue().

true_size()

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

inlineoverridevirtual

Reimplemented from PTL::VUserTaskQueue.

Definition at line 154 of file UserTaskQueue.hh.

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

Referenced by UserTaskQueue().

Wait()

void PTL::UserTaskQueue::Wait ( )

inlineoverridevirtual

Implements PTL::VUserTaskQueue.

Definition at line 68 of file UserTaskQueue.hh.

{}

---

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

• UserTaskQueue.hh
• UserTaskQueue.cc