Import Geant4 10.7.0.beta source tree
This commit is contained in:
Gabriele Cosmo
+40
@@ -0,0 +1,40 @@
|
||||
//
|
||||
// MIT License
|
||||
// Copyright (c) 2020 Jonathan R. Madsen
|
||||
// Permission is hereby granted, free of charge, to any person obtaining a copy
|
||||
// of this software and associated documentation files (the "Software"), to deal
|
||||
// in the Software without restriction, including without limitation the rights
|
||||
// to use, copy, modify, merge, publish, distribute, sublicense, and
|
||||
// copies of the Software, and to permit persons to whom the Software is
|
||||
// furnished to do so, subject to the following conditions:
|
||||
// The above copyright notice and this permission notice shall be included in
|
||||
// all copies or substantial portions of the Software. THE SOFTWARE IS PROVIDED
|
||||
// "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT
|
||||
// LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR
|
||||
// PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT
|
||||
// HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN
|
||||
// ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
|
||||
// WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
|
||||
//
|
||||
// ---------------------------------------------------------------
|
||||
// Tasking class implementation
|
||||
//
|
||||
//
|
||||
|
||||
#include "PTL/TaskAllocator.hh"
|
||||
#include "PTL/TaskAllocatorList.hh"
|
||||
|
||||
using namespace PTL;
|
||||
|
||||
//======================================================================================//
|
||||
|
||||
TaskAllocatorBase::TaskAllocatorBase()
|
||||
{
|
||||
TaskAllocatorList::GetAllocatorList()->Register(this);
|
||||
}
|
||||
|
||||
//======================================================================================//
|
||||
|
||||
TaskAllocatorBase::~TaskAllocatorBase() {}
|
||||
|
||||
//======================================================================================//
|
||||
+129
@@ -0,0 +1,129 @@
|
||||
//
|
||||
// MIT License
|
||||
// Copyright (c) 2020 Jonathan R. Madsen
|
||||
// Permission is hereby granted, free of charge, to any person obtaining a copy
|
||||
// of this software and associated documentation files (the "Software"), to deal
|
||||
// in the Software without restriction, including without limitation the rights
|
||||
// to use, copy, modify, merge, publish, distribute, sublicense, and
|
||||
// copies of the Software, and to permit persons to whom the Software is
|
||||
// furnished to do so, subject to the following conditions:
|
||||
// The above copyright notice and this permission notice shall be included in
|
||||
// all copies or substantial portions of the Software. THE SOFTWARE IS PROVIDED
|
||||
// "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT
|
||||
// LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR
|
||||
// PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT
|
||||
// HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN
|
||||
// ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
|
||||
// WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
|
||||
//
|
||||
// ---------------------------------------------------------------
|
||||
// Tasking class implementation
|
||||
//
|
||||
|
||||
#include <iomanip>
|
||||
|
||||
#include "PTL/TaskAllocator.hh"
|
||||
#include "PTL/TaskAllocatorList.hh"
|
||||
|
||||
using namespace PTL;
|
||||
|
||||
//======================================================================================//
|
||||
|
||||
TaskAllocatorList*&
|
||||
TaskAllocatorList::fAllocatorList()
|
||||
{
|
||||
static thread_local TaskAllocatorList* _instance = nullptr;
|
||||
return _instance;
|
||||
}
|
||||
|
||||
//======================================================================================//
|
||||
|
||||
TaskAllocatorList*
|
||||
TaskAllocatorList::GetAllocatorList()
|
||||
{
|
||||
if(!fAllocatorList())
|
||||
{
|
||||
fAllocatorList() = new TaskAllocatorList;
|
||||
}
|
||||
return fAllocatorList();
|
||||
}
|
||||
|
||||
//======================================================================================//
|
||||
|
||||
TaskAllocatorList*
|
||||
TaskAllocatorList::GetAllocatorListIfExist()
|
||||
{
|
||||
return fAllocatorList();
|
||||
}
|
||||
|
||||
//======================================================================================//
|
||||
|
||||
TaskAllocatorList::TaskAllocatorList() {}
|
||||
|
||||
//======================================================================================//
|
||||
|
||||
TaskAllocatorList::~TaskAllocatorList() { fAllocatorList() = nullptr; }
|
||||
|
||||
//======================================================================================//
|
||||
|
||||
void
|
||||
TaskAllocatorList::Register(TaskAllocatorBase* alloc)
|
||||
{
|
||||
fList.push_back(alloc);
|
||||
}
|
||||
|
||||
//======================================================================================//
|
||||
|
||||
void
|
||||
TaskAllocatorList::Destroy(int nStat, int verboseLevel)
|
||||
{
|
||||
int i = 0, j = 0;
|
||||
double tmem = 0;
|
||||
if(verboseLevel > 0)
|
||||
{
|
||||
std::cout << "================== Deleting memory pools ==================="
|
||||
<< std::endl;
|
||||
}
|
||||
for(auto& itr : fList)
|
||||
{
|
||||
double mem = itr->GetAllocatedSize();
|
||||
if(i < nStat)
|
||||
{
|
||||
i++;
|
||||
tmem += mem;
|
||||
itr->ResetStorage();
|
||||
continue;
|
||||
}
|
||||
j++;
|
||||
tmem += mem;
|
||||
if(verboseLevel > 1)
|
||||
{
|
||||
std::cout << "Pool ID '" << itr->GetPoolType()
|
||||
<< "', size : " << std::setprecision(3) << mem / 1048576
|
||||
<< std::setprecision(6) << " MB" << std::endl;
|
||||
}
|
||||
itr->ResetStorage();
|
||||
// delete *itr;
|
||||
}
|
||||
if(verboseLevel > 0)
|
||||
{
|
||||
std::cout << "Number of memory pools allocated: " << Size()
|
||||
<< "; of which, static: " << i << std::endl;
|
||||
std::cout << "Dynamic pools deleted: " << j
|
||||
<< " / Total memory freed: " << std::setprecision(2) << tmem / 1048576
|
||||
<< std::setprecision(6) << " MB" << std::endl;
|
||||
std::cout << "============================================================"
|
||||
<< std::endl;
|
||||
}
|
||||
fList.clear();
|
||||
}
|
||||
|
||||
//======================================================================================//
|
||||
|
||||
int
|
||||
TaskAllocatorList::Size() const
|
||||
{
|
||||
return fList.size();
|
||||
}
|
||||
|
||||
//======================================================================================//
|
||||
+124
@@ -0,0 +1,124 @@
|
||||
//
|
||||
// MIT License
|
||||
// Copyright (c) 2020 Jonathan R. Madsen
|
||||
// Permission is hereby granted, free of charge, to any person obtaining a copy
|
||||
// of this software and associated documentation files (the "Software"), to deal
|
||||
// in the Software without restriction, including without limitation the rights
|
||||
// to use, copy, modify, merge, publish, distribute, sublicense, and
|
||||
// copies of the Software, and to permit persons to whom the Software is
|
||||
// furnished to do so, subject to the following conditions:
|
||||
// The above copyright notice and this permission notice shall be included in
|
||||
// all copies or substantial portions of the Software. THE SOFTWARE IS PROVIDED
|
||||
// "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT
|
||||
// LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR
|
||||
// PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT
|
||||
// HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN
|
||||
// ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
|
||||
// WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
|
||||
//
|
||||
// ---------------------------------------------------------------
|
||||
// Tasking class implementation
|
||||
//
|
||||
// TaskAllocatorPool
|
||||
//
|
||||
// Implementation file
|
||||
//
|
||||
// Author: G.Cosmo, November 2000
|
||||
//
|
||||
|
||||
#include "PTL/TaskAllocatorPool.hh"
|
||||
|
||||
using namespace PTL;
|
||||
|
||||
// ************************************************************
|
||||
// TaskAllocatorPool constructor
|
||||
// ************************************************************
|
||||
//
|
||||
TaskAllocatorPool::TaskAllocatorPool(unsigned int sz)
|
||||
: esize((sz < sizeof(PoolLink)) ? sizeof(PoolLink) : sz)
|
||||
, csize((sz < 1024 / 2 - 16) ? (1024 - 16) : (sz * 10 - 16))
|
||||
, chunks(nullptr)
|
||||
, head(nullptr)
|
||||
, nchunks(0)
|
||||
{}
|
||||
|
||||
// ************************************************************
|
||||
// TaskAllocatorPool copy constructor
|
||||
// ************************************************************
|
||||
//
|
||||
TaskAllocatorPool::TaskAllocatorPool(const TaskAllocatorPool& right)
|
||||
: esize(right.esize)
|
||||
, csize(right.csize)
|
||||
, chunks(right.chunks)
|
||||
, head(right.head)
|
||||
, nchunks(right.nchunks)
|
||||
{}
|
||||
|
||||
// ************************************************************
|
||||
// TaskAllocatorPool operator=
|
||||
// ************************************************************
|
||||
//
|
||||
TaskAllocatorPool&
|
||||
TaskAllocatorPool::operator=(const TaskAllocatorPool& right)
|
||||
{
|
||||
if(&right == this)
|
||||
return *this;
|
||||
chunks = right.chunks;
|
||||
head = right.head;
|
||||
nchunks = right.nchunks;
|
||||
return *this;
|
||||
}
|
||||
|
||||
// ************************************************************
|
||||
// TaskAllocatorPool destructor
|
||||
// ************************************************************
|
||||
//
|
||||
TaskAllocatorPool::~TaskAllocatorPool() { Reset(); }
|
||||
|
||||
// ************************************************************
|
||||
// Reset
|
||||
// ************************************************************
|
||||
//
|
||||
void
|
||||
TaskAllocatorPool::Reset()
|
||||
{
|
||||
// Free all chunks
|
||||
//
|
||||
PoolChunk* n = chunks;
|
||||
PoolChunk* p = nullptr;
|
||||
while(n)
|
||||
{
|
||||
p = n;
|
||||
n = n->next;
|
||||
delete p;
|
||||
}
|
||||
head = nullptr;
|
||||
chunks = nullptr;
|
||||
nchunks = 0;
|
||||
}
|
||||
|
||||
// ************************************************************
|
||||
// Grow
|
||||
// ************************************************************
|
||||
//
|
||||
void
|
||||
TaskAllocatorPool::Grow()
|
||||
{
|
||||
// Allocate new chunk, organize it as a linked list of
|
||||
// elements of size 'esize'
|
||||
//
|
||||
PoolChunk* n = new PoolChunk(csize);
|
||||
n->next = chunks;
|
||||
chunks = n;
|
||||
nchunks++;
|
||||
|
||||
const int nelem = csize / esize;
|
||||
char* start = n->mem;
|
||||
char* last = &start[(nelem - 1) * esize];
|
||||
for(char* p = start; p < last; p += esize)
|
||||
{
|
||||
reinterpret_cast<PoolLink*>(p)->next = reinterpret_cast<PoolLink*>(p + esize);
|
||||
}
|
||||
reinterpret_cast<PoolLink*>(last)->next = nullptr;
|
||||
head = reinterpret_cast<PoolLink*>(start);
|
||||
}
|
||||
+152
@@ -0,0 +1,152 @@
|
||||
//
|
||||
// MIT License
|
||||
// Copyright (c) 2020 Jonathan R. Madsen
|
||||
// Permission is hereby granted, free of charge, to any person obtaining a copy
|
||||
// of this software and associated documentation files (the "Software"), to deal
|
||||
// in the Software without restriction, including without limitation the rights
|
||||
// to use, copy, modify, merge, publish, distribute, sublicense, and
|
||||
// copies of the Software, and to permit persons to whom the Software is
|
||||
// furnished to do so, subject to the following conditions:
|
||||
// The above copyright notice and this permission notice shall be included in
|
||||
// all copies or substantial portions of the Software. THE SOFTWARE IS PROVIDED
|
||||
// "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT
|
||||
// LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR
|
||||
// PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT
|
||||
// HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN
|
||||
// ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
|
||||
// WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
|
||||
//
|
||||
// ---------------------------------------------------------------
|
||||
// Tasking class implementation
|
||||
|
||||
#include "PTL/TaskRunManager.hh"
|
||||
#include "PTL/AutoLock.hh"
|
||||
#include "PTL/Task.hh"
|
||||
#include "PTL/TaskGroup.hh"
|
||||
#include "PTL/TaskManager.hh"
|
||||
#include "PTL/ThreadPool.hh"
|
||||
#include "PTL/Threading.hh"
|
||||
#include "PTL/Utility.hh"
|
||||
|
||||
#include <cstdlib>
|
||||
#include <cstring>
|
||||
#include <iterator>
|
||||
|
||||
using namespace PTL;
|
||||
|
||||
//======================================================================================//
|
||||
|
||||
TaskRunManager::pointer&
|
||||
TaskRunManager::GetPrivateMasterRunManager(bool init, bool useTBB)
|
||||
{
|
||||
static pointer _instance = (init) ? new TaskRunManager(useTBB) : nullptr;
|
||||
return _instance;
|
||||
}
|
||||
|
||||
//======================================================================================//
|
||||
|
||||
TaskRunManager*
|
||||
TaskRunManager::GetMasterRunManager(bool useTBB)
|
||||
{
|
||||
static pointer& _instance = GetPrivateMasterRunManager(true, useTBB);
|
||||
return _instance;
|
||||
}
|
||||
|
||||
//======================================================================================//
|
||||
|
||||
TaskRunManager*
|
||||
TaskRunManager::GetInstance(bool useTBB)
|
||||
{
|
||||
return GetMasterRunManager(useTBB);
|
||||
}
|
||||
|
||||
//======================================================================================//
|
||||
|
||||
TaskRunManager::TaskRunManager(bool useTBB)
|
||||
: m_is_initialized(false)
|
||||
, m_verbose(0)
|
||||
, m_workers(std::thread::hardware_concurrency())
|
||||
, m_task_queue(nullptr)
|
||||
, m_thread_pool(nullptr)
|
||||
, m_task_manager(nullptr)
|
||||
{
|
||||
if(!GetPrivateMasterRunManager(false))
|
||||
{
|
||||
GetPrivateMasterRunManager(false) = this;
|
||||
}
|
||||
|
||||
#ifdef PTL_USE_TBB
|
||||
auto _useTBB = GetEnv<bool>("FORCE_TBB", useTBB);
|
||||
if(_useTBB)
|
||||
useTBB = true;
|
||||
#endif
|
||||
|
||||
// handle TBB
|
||||
ThreadPool::set_use_tbb(useTBB);
|
||||
m_workers = GetEnv<uint64_t>("PTL_NUM_THREADS", m_workers);
|
||||
}
|
||||
|
||||
//======================================================================================//
|
||||
|
||||
TaskRunManager::~TaskRunManager() {}
|
||||
|
||||
//======================================================================================//
|
||||
|
||||
void
|
||||
TaskRunManager::Initialize(uint64_t n)
|
||||
{
|
||||
m_workers = n;
|
||||
|
||||
// create threadpool if needed + task manager
|
||||
if(!m_thread_pool)
|
||||
{
|
||||
if(m_verbose > 0)
|
||||
std::cout << "TaskRunManager :: Creating thread pool..." << std::endl;
|
||||
m_thread_pool = new ThreadPool(m_workers, m_task_queue);
|
||||
if(m_verbose > 0)
|
||||
std::cout << "TaskRunManager :: Creating task manager..." << std::endl;
|
||||
m_task_manager = new TaskManager(m_thread_pool);
|
||||
}
|
||||
// or resize
|
||||
else if(m_workers != m_thread_pool->size())
|
||||
{
|
||||
if(m_verbose > 0)
|
||||
{
|
||||
std::cout << "TaskRunManager :: Resizing thread pool from "
|
||||
<< m_thread_pool->size() << " to " << m_workers << " threads ..."
|
||||
<< std::endl;
|
||||
}
|
||||
m_thread_pool->resize(m_workers);
|
||||
}
|
||||
|
||||
// create the joiners
|
||||
if(ThreadPool::using_tbb())
|
||||
{
|
||||
if(m_verbose > 0)
|
||||
std::cout << "TaskRunManager :: Using TBB..." << std::endl;
|
||||
}
|
||||
else
|
||||
{
|
||||
if(m_verbose > 0)
|
||||
std::cout << "TaskRunManager :: Using ThreadPool..." << std::endl;
|
||||
}
|
||||
|
||||
m_is_initialized = true;
|
||||
if(m_verbose > 0)
|
||||
std::cout << "TaskRunManager :: initialized..." << std::endl;
|
||||
}
|
||||
|
||||
//======================================================================================//
|
||||
|
||||
void
|
||||
TaskRunManager::Terminate()
|
||||
{
|
||||
m_is_initialized = false;
|
||||
m_thread_pool->destroy_threadpool();
|
||||
delete m_task_manager;
|
||||
delete m_thread_pool;
|
||||
m_task_manager = nullptr;
|
||||
m_thread_pool = nullptr;
|
||||
}
|
||||
|
||||
//======================================================================================//
|
||||
+65
@@ -0,0 +1,65 @@
|
||||
//
|
||||
// MIT License
|
||||
// Copyright (c) 2020 Jonathan R. Madsen
|
||||
// Permission is hereby granted, free of charge, to any person obtaining a copy
|
||||
// of this software and associated documentation files (the "Software"), to deal
|
||||
// in the Software without restriction, including without limitation the rights
|
||||
// to use, copy, modify, merge, publish, distribute, sublicense, and
|
||||
// copies of the Software, and to permit persons to whom the Software is
|
||||
// furnished to do so, subject to the following conditions:
|
||||
// The above copyright notice and this permission notice shall be included in
|
||||
// all copies or substantial portions of the Software. THE SOFTWARE IS PROVIDED
|
||||
// "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT
|
||||
// LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR
|
||||
// PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT
|
||||
// HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN
|
||||
// ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
|
||||
// WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
|
||||
//
|
||||
// ---------------------------------------------------------------
|
||||
// Tasking class implementation
|
||||
//
|
||||
|
||||
#include "PTL/ThreadData.hh"
|
||||
#include "PTL/ThreadPool.hh"
|
||||
#include "PTL/Threading.hh"
|
||||
#include "PTL/VUserTaskQueue.hh"
|
||||
|
||||
#include <iostream>
|
||||
|
||||
using namespace PTL;
|
||||
|
||||
//======================================================================================//
|
||||
|
||||
ThreadData*&
|
||||
ThreadData::GetInstance()
|
||||
{
|
||||
static thread_local ThreadData* _instance = nullptr;
|
||||
return _instance;
|
||||
}
|
||||
|
||||
//======================================================================================//
|
||||
|
||||
ThreadData::ThreadData(ThreadPool* tp)
|
||||
: is_master(tp->is_master())
|
||||
, within_task(false)
|
||||
, task_depth(0)
|
||||
, thread_pool(tp)
|
||||
, current_queue(tp->get_queue())
|
||||
, queue_stack({ current_queue })
|
||||
{}
|
||||
|
||||
//======================================================================================//
|
||||
|
||||
void
|
||||
ThreadData::update()
|
||||
{
|
||||
current_queue = thread_pool->get_queue();
|
||||
queue_stack.push_back(current_queue);
|
||||
}
|
||||
|
||||
//======================================================================================//
|
||||
|
||||
ThreadData::~ThreadData() {}
|
||||
|
||||
//======================================================================================//
|
||||
+759
@@ -0,0 +1,759 @@
|
||||
//
|
||||
// MIT License
|
||||
// Copyright (c) 2020 Jonathan R. Madsen
|
||||
// Permission is hereby granted, free of charge, to any person obtaining a copy
|
||||
// of this software and associated documentation files (the "Software"), to deal
|
||||
// in the Software without restriction, including without limitation the rights
|
||||
// to use, copy, modify, merge, publish, distribute, sublicense, and
|
||||
// copies of the Software, and to permit persons to whom the Software is
|
||||
// furnished to do so, subject to the following conditions:
|
||||
// The above copyright notice and this permission notice shall be included in
|
||||
// all copies or substantial portions of the Software. THE SOFTWARE IS PROVIDED
|
||||
// "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT
|
||||
// LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR
|
||||
// PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT
|
||||
// HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN
|
||||
// ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
|
||||
// WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
|
||||
//
|
||||
// ---------------------------------------------------------------
|
||||
// Tasking class implementation
|
||||
//
|
||||
// Class Description:
|
||||
//
|
||||
// This file creates a class for an efficient thread-pool that
|
||||
// accepts work in the form of tasks.
|
||||
//
|
||||
// ---------------------------------------------------------------
|
||||
// Author: Jonathan Madsen (Feb 13th 2018)
|
||||
// ---------------------------------------------------------------
|
||||
|
||||
#include "PTL/ThreadPool.hh"
|
||||
#include "PTL/Globals.hh"
|
||||
#include "PTL/ThreadData.hh"
|
||||
#include "PTL/UserTaskQueue.hh"
|
||||
#include "PTL/VUserTaskQueue.hh"
|
||||
|
||||
#include <cstdlib>
|
||||
|
||||
using namespace PTL;
|
||||
|
||||
//======================================================================================//
|
||||
|
||||
inline intmax_t
|
||||
ncores()
|
||||
{
|
||||
return static_cast<intmax_t>(Thread::hardware_concurrency());
|
||||
}
|
||||
|
||||
//======================================================================================//
|
||||
|
||||
ThreadPool::thread_id_map_t ThreadPool::f_thread_ids;
|
||||
|
||||
//======================================================================================//
|
||||
|
||||
namespace
|
||||
{
|
||||
ThreadData*&
|
||||
thread_data()
|
||||
{
|
||||
return ThreadData::GetInstance();
|
||||
}
|
||||
} // namespace
|
||||
|
||||
//======================================================================================//
|
||||
|
||||
bool ThreadPool::f_use_tbb = false;
|
||||
|
||||
//======================================================================================//
|
||||
// static member function that calls the member function we want the thread to
|
||||
// run
|
||||
void
|
||||
ThreadPool::start_thread(ThreadPool* tp, intmax_t _idx)
|
||||
{
|
||||
{
|
||||
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;
|
||||
}
|
||||
static thread_local std::unique_ptr<ThreadData> _unique_data(new ThreadData(tp));
|
||||
thread_data() = _unique_data.get();
|
||||
tp->record_entry();
|
||||
tp->execute_thread(thread_data()->current_queue);
|
||||
tp->record_exit();
|
||||
}
|
||||
|
||||
//======================================================================================//
|
||||
// static member function that checks enabling of tbb library
|
||||
bool
|
||||
ThreadPool::using_tbb()
|
||||
{
|
||||
return f_use_tbb;
|
||||
}
|
||||
|
||||
//======================================================================================//
|
||||
// static member function that initialized tbb library
|
||||
void
|
||||
ThreadPool::set_use_tbb(bool enable)
|
||||
{
|
||||
#if defined(PTL_USE_TBB)
|
||||
f_use_tbb = enable;
|
||||
#else
|
||||
ConsumeParameters<bool>(enable);
|
||||
#endif
|
||||
}
|
||||
|
||||
//======================================================================================//
|
||||
|
||||
const ThreadPool::thread_id_map_t&
|
||||
ThreadPool::GetThreadIDs()
|
||||
{
|
||||
return f_thread_ids;
|
||||
}
|
||||
|
||||
//======================================================================================//
|
||||
|
||||
uintmax_t
|
||||
ThreadPool::GetThisThreadID()
|
||||
{
|
||||
auto _tid = ThisThread::get_id();
|
||||
{
|
||||
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;
|
||||
}
|
||||
}
|
||||
return f_thread_ids[_tid];
|
||||
}
|
||||
|
||||
//======================================================================================//
|
||||
|
||||
ThreadPool::ThreadPool(const size_type& pool_size, VUserTaskQueue* task_queue,
|
||||
bool _use_affinity, const affinity_func_t& _affinity_func)
|
||||
: m_use_affinity(_use_affinity)
|
||||
, m_tbb_tp(false)
|
||||
, m_verbose(0)
|
||||
, m_pool_size(0)
|
||||
, m_master_tid(ThisThread::get_id())
|
||||
, m_alive_flag(std::make_shared<std::atomic_bool>(false))
|
||||
, m_pool_state(std::make_shared<std::atomic_short>(thread_pool::state::NONINIT))
|
||||
, m_thread_awake(std::make_shared<std::atomic_uintmax_t>(0))
|
||||
, m_task_lock(std::make_shared<Mutex>())
|
||||
, m_task_cond(std::make_shared<Condition>())
|
||||
, m_task_queue(task_queue)
|
||||
, m_tbb_task_group(nullptr)
|
||||
, m_init_func([]() { return; })
|
||||
, m_affinity_func(_affinity_func)
|
||||
{
|
||||
m_verbose = GetEnv<int>("PTL_VERBOSE", m_verbose);
|
||||
|
||||
auto master_id = GetThisThreadID();
|
||||
if(master_id != 0 && m_verbose > 1)
|
||||
std::cerr << "ThreadPool created on non-master slave" << std::endl;
|
||||
|
||||
thread_data() = new ThreadData(this);
|
||||
|
||||
// initialize after GetThisThreadID so master is zero
|
||||
this->initialize_threadpool(pool_size);
|
||||
|
||||
if(!m_task_queue)
|
||||
m_task_queue = new UserTaskQueue(m_pool_size);
|
||||
}
|
||||
|
||||
//======================================================================================//
|
||||
|
||||
ThreadPool::~ThreadPool()
|
||||
{
|
||||
//------------------------------------------------------------------------//
|
||||
// set state to stopped
|
||||
m_pool_state->store(thread_pool::state::STOPPED);
|
||||
|
||||
//------------------------------------------------------------------------//
|
||||
// notify all threads we are shutting down
|
||||
m_task_lock->lock();
|
||||
CONDITIONBROADCAST(m_task_cond.get());
|
||||
m_task_lock->unlock();
|
||||
|
||||
//--------------------------------------------------------------------//
|
||||
// set to dead
|
||||
m_alive_flag->store(false);
|
||||
|
||||
//--------------------------------------------------------------------//
|
||||
// delete tbb task scheduler
|
||||
#ifdef PTL_USE_TBB
|
||||
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;
|
||||
std::cout << "ThreadPool [TBB] destroyed" << std::endl;
|
||||
}
|
||||
|
||||
//--------------------------------------------------------------------//
|
||||
// delete tbb task-group
|
||||
if(m_tbb_task_group)
|
||||
{
|
||||
m_tbb_task_group->wait();
|
||||
delete m_tbb_task_group;
|
||||
m_tbb_task_group = nullptr;
|
||||
}
|
||||
#endif
|
||||
|
||||
auto _tid = std::this_thread::get_id();
|
||||
if(f_thread_ids.find(_tid) != f_thread_ids.end())
|
||||
f_thread_ids.erase(f_thread_ids.find(_tid));
|
||||
|
||||
for(auto& itr : m_main_threads)
|
||||
if(f_thread_ids.find(itr) != f_thread_ids.end())
|
||||
f_thread_ids.erase(f_thread_ids.find(itr));
|
||||
|
||||
m_thread_awake.reset();
|
||||
|
||||
std::cout << "ThreadPool destroyed" << std::endl;
|
||||
|
||||
/*
|
||||
// Release resources
|
||||
if(m_pool_state.load() != thread_pool::state::STOPPED)
|
||||
{
|
||||
size_type ret = destroy_threadpool();
|
||||
while(ret > 0)
|
||||
ret = stop_thread();
|
||||
}
|
||||
|
||||
// wait until thread pool is fully destroyed
|
||||
while(is_alive())
|
||||
;
|
||||
|
||||
// delete thread-local allocator and erase thread IDS
|
||||
auto _tid = std::this_thread::get_id();
|
||||
|
||||
if(f_thread_ids.find(_tid) != f_thread_ids.end())
|
||||
f_thread_ids.erase(f_thread_ids.find(_tid));
|
||||
|
||||
// deleted by ThreadData
|
||||
// delete m_task_queue;
|
||||
*/
|
||||
}
|
||||
|
||||
//======================================================================================//
|
||||
|
||||
bool
|
||||
ThreadPool::is_initialized() const
|
||||
{
|
||||
return !(m_pool_state->load() == thread_pool::state::NONINIT);
|
||||
}
|
||||
|
||||
//======================================================================================//
|
||||
|
||||
void
|
||||
ThreadPool::set_affinity(intmax_t i, Thread& _thread)
|
||||
{
|
||||
try
|
||||
{
|
||||
NativeThread native_thread = _thread.native_handle();
|
||||
intmax_t _pin = m_affinity_func(i);
|
||||
if(m_verbose > 0)
|
||||
{
|
||||
std::cout << "Setting pin affinity for thread " << _thread.get_id() << " to "
|
||||
<< _pin << std::endl;
|
||||
}
|
||||
Threading::SetPinAffinity(_pin, native_thread);
|
||||
} catch(std::runtime_error& e)
|
||||
{
|
||||
std::cout << "Error setting pin affinity" << std::endl;
|
||||
std::cerr << e.what() << std::endl; // issue assigning affinity
|
||||
}
|
||||
}
|
||||
|
||||
//======================================================================================//
|
||||
|
||||
ThreadPool::size_type
|
||||
ThreadPool::initialize_threadpool(size_type proposed_size)
|
||||
{
|
||||
//--------------------------------------------------------------------//
|
||||
// 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);
|
||||
|
||||
//--------------------------------------------------------------------//
|
||||
// handle tbb task scheduler
|
||||
#ifdef PTL_USE_TBB
|
||||
if(f_use_tbb)
|
||||
{
|
||||
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)
|
||||
{
|
||||
std::cout << "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();
|
||||
return m_pool_size;
|
||||
}
|
||||
|
||||
// NOLINT(readability-else-after-return)
|
||||
if(f_use_tbb && tbb_global_control())
|
||||
{
|
||||
m_tbb_tp = false;
|
||||
tbb_global_control_t*& _global_control = tbb_global_control();
|
||||
if(_global_control)
|
||||
{
|
||||
delete _global_control;
|
||||
_global_control = nullptr;
|
||||
}
|
||||
// delete task group (used for async)
|
||||
if(m_tbb_task_group)
|
||||
{
|
||||
m_tbb_task_group->wait();
|
||||
delete m_tbb_task_group;
|
||||
m_tbb_task_group = nullptr;
|
||||
}
|
||||
}
|
||||
#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)
|
||||
{
|
||||
std::cout << "ThreadPool initialized with " << m_pool_size << " threads."
|
||||
<< std::endl;
|
||||
}
|
||||
if(!m_task_queue)
|
||||
m_task_queue = new UserTaskQueue(m_pool_size);
|
||||
return m_pool_size;
|
||||
}
|
||||
else if(m_pool_size == proposed_size) // NOLINT
|
||||
{
|
||||
if(m_verbose > 0)
|
||||
{
|
||||
std::cout << "ThreadPool initialized with " << m_pool_size << " threads."
|
||||
<< std::endl;
|
||||
}
|
||||
if(!m_task_queue)
|
||||
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);
|
||||
}
|
||||
|
||||
auto this_tid = GetThisThreadID();
|
||||
for(size_type i = m_pool_size; i < proposed_size; ++i)
|
||||
{
|
||||
// add the threads
|
||||
try
|
||||
{
|
||||
Thread tid(ThreadPool::start_thread, this, this_tid + i + 1);
|
||||
// only reaches here if successful creation of thread
|
||||
++m_pool_size;
|
||||
// store thread
|
||||
m_main_threads.push_back(tid.get_id());
|
||||
// list of joined thread booleans
|
||||
m_is_joined.push_back(false);
|
||||
// set the affinity
|
||||
if(m_use_affinity)
|
||||
set_affinity(i, tid);
|
||||
// detach
|
||||
tid.detach();
|
||||
} catch(std::runtime_error& e)
|
||||
{
|
||||
std::cerr << e.what() << std::endl; // issue creating thread
|
||||
continue;
|
||||
} catch(std::bad_alloc& e)
|
||||
{
|
||||
std::cerr << 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)
|
||||
{
|
||||
std::cout << "ThreadPool initialized with " << m_pool_size << " threads."
|
||||
<< std::endl;
|
||||
}
|
||||
|
||||
if(!m_task_queue)
|
||||
m_task_queue = new UserTaskQueue(m_main_threads.size());
|
||||
|
||||
return m_main_threads.size();
|
||||
}
|
||||
|
||||
//======================================================================================//
|
||||
|
||||
ThreadPool::size_type
|
||||
ThreadPool::destroy_threadpool()
|
||||
{
|
||||
// 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
|
||||
#ifdef PTL_USE_TBB
|
||||
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;
|
||||
std::cout << "ThreadPool [TBB] destroyed" << std::endl;
|
||||
}
|
||||
if(m_tbb_task_group)
|
||||
{
|
||||
m_tbb_task_group->wait();
|
||||
delete m_tbb_task_group;
|
||||
m_tbb_task_group = nullptr;
|
||||
}
|
||||
#endif
|
||||
|
||||
if(!m_alive_flag->load())
|
||||
return 0;
|
||||
|
||||
//------------------------------------------------------------------------//
|
||||
// notify all threads we are shutting down
|
||||
m_task_lock->lock();
|
||||
CONDITIONBROADCAST(m_task_cond.get());
|
||||
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 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;
|
||||
|
||||
//--------------------------------------------------------------------//
|
||||
// try waking up a bunch of threads that are still waiting
|
||||
CONDITIONBROADCAST(m_task_cond.get());
|
||||
//--------------------------------------------------------------------//
|
||||
}
|
||||
|
||||
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(_active == 0)
|
||||
std::cout << "ThreadPool destroyed" << std::endl;
|
||||
else
|
||||
std::cout << "ThreadPool destroyed but " << _active
|
||||
<< " threads might still be active (and cause a termination error)"
|
||||
<< std::endl;
|
||||
|
||||
return 0;
|
||||
}
|
||||
|
||||
//======================================================================================//
|
||||
|
||||
ThreadPool::size_type
|
||||
ThreadPool::stop_thread()
|
||||
{
|
||||
if(!m_alive_flag->load() || m_pool_size == 0)
|
||||
return 0;
|
||||
|
||||
//------------------------------------------------------------------------//
|
||||
// notify all threads we are shutting down
|
||||
m_task_lock->lock();
|
||||
m_is_stopped.push_back(true);
|
||||
CONDITIONNOTIFY(m_task_cond.get());
|
||||
m_task_lock->unlock();
|
||||
//------------------------------------------------------------------------//
|
||||
|
||||
// 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_size = m_main_threads.size();
|
||||
return m_main_threads.size();
|
||||
}
|
||||
|
||||
//======================================================================================//
|
||||
|
||||
void
|
||||
ThreadPool::execute_thread(VUserTaskQueue* _task_queue)
|
||||
{
|
||||
// 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();
|
||||
|
||||
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)();
|
||||
if(!_task->group())
|
||||
delete _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 && 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 && 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)();
|
||||
if(!_task->group())
|
||||
delete _task;
|
||||
}
|
||||
}
|
||||
//----------------------------------------------------------------//
|
||||
|
||||
// disable guard against recursive deadlock
|
||||
data->within_task = false;
|
||||
//----------------------------------------------------------------//
|
||||
}
|
||||
}
|
||||
|
||||
//======================================================================================//
|
||||
+127
@@ -0,0 +1,127 @@
|
||||
//
|
||||
// MIT License
|
||||
// Copyright (c) 2020 Jonathan R. Madsen
|
||||
// Permission is hereby granted, free of charge, to any person obtaining a copy
|
||||
// of this software and associated documentation files (the "Software"), to deal
|
||||
// in the Software without restriction, including without limitation the rights
|
||||
// to use, copy, modify, merge, publish, distribute, sublicense, and
|
||||
// copies of the Software, and to permit persons to whom the Software is
|
||||
// furnished to do so, subject to the following conditions:
|
||||
// The above copyright notice and this permission notice shall be included in
|
||||
// all copies or substantial portions of the Software. THE SOFTWARE IS PROVIDED
|
||||
// "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT
|
||||
// LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR
|
||||
// PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT
|
||||
// HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN
|
||||
// ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
|
||||
// WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
|
||||
//
|
||||
// ---------------------------------------------------------------
|
||||
// Tasking class implementation
|
||||
//
|
||||
// Threading.cc
|
||||
//
|
||||
// ---------------------------------------------------------------
|
||||
// Author: Andrea Dotti (15 Feb 2013): First Implementation
|
||||
// ---------------------------------------------------------------
|
||||
|
||||
#include "PTL/Threading.hh"
|
||||
#include "PTL/AutoLock.hh"
|
||||
#include "PTL/Globals.hh"
|
||||
|
||||
#if defined(WIN32) || defined(_WIN32) || defined(WIN64) || defined(_WIN64)
|
||||
# include <Windows.h>
|
||||
#else
|
||||
# include <sys/syscall.h>
|
||||
# include <sys/types.h>
|
||||
# include <unistd.h>
|
||||
#endif
|
||||
|
||||
#include <atomic>
|
||||
|
||||
using namespace PTL;
|
||||
|
||||
//======================================================================================//
|
||||
|
||||
namespace
|
||||
{
|
||||
thread_local int ThreadID = Threading::MASTER_ID;
|
||||
std::atomic_int numActThreads(0);
|
||||
} // namespace
|
||||
|
||||
//======================================================================================//
|
||||
|
||||
Pid_t
|
||||
Threading::GetPidId()
|
||||
{
|
||||
// In multithreaded mode return Thread ID
|
||||
return std::this_thread::get_id();
|
||||
}
|
||||
|
||||
//======================================================================================//
|
||||
|
||||
unsigned
|
||||
Threading::GetNumberOfCores()
|
||||
{
|
||||
return std::thread::hardware_concurrency();
|
||||
}
|
||||
|
||||
//======================================================================================//
|
||||
|
||||
void
|
||||
Threading::SetThreadId(int value)
|
||||
{
|
||||
ThreadID = value;
|
||||
}
|
||||
int
|
||||
Threading::GetThreadId()
|
||||
{
|
||||
return ThreadID;
|
||||
}
|
||||
bool
|
||||
Threading::IsWorkerThread()
|
||||
{
|
||||
return (ThreadID >= 0);
|
||||
}
|
||||
bool
|
||||
Threading::IsMasterThread()
|
||||
{
|
||||
return (ThreadID == MASTER_ID);
|
||||
}
|
||||
|
||||
//======================================================================================//
|
||||
|
||||
bool
|
||||
Threading::SetPinAffinity(int cpu, NativeThread& aT)
|
||||
{
|
||||
#if defined(__linux__) || defined(_AIX)
|
||||
cpu_set_t* aset = new cpu_set_t;
|
||||
CPU_ZERO(aset);
|
||||
CPU_SET(cpu, aset);
|
||||
pthread_t& _aT = static_cast<pthread_t&>(aT);
|
||||
return (pthread_setaffinity_np(_aT, sizeof(cpu_set_t), aset) == 0);
|
||||
#else // Not available for Mac, WIN,...
|
||||
ConsumeParameters(cpu, aT);
|
||||
return true;
|
||||
#endif
|
||||
}
|
||||
|
||||
//======================================================================================//
|
||||
|
||||
int
|
||||
Threading::WorkerThreadLeavesPool()
|
||||
{
|
||||
return numActThreads--;
|
||||
}
|
||||
int
|
||||
Threading::WorkerThreadJoinsPool()
|
||||
{
|
||||
return numActThreads++;
|
||||
}
|
||||
int
|
||||
Threading::GetNumberOfRunningWorkerThreads()
|
||||
{
|
||||
return numActThreads.load();
|
||||
}
|
||||
|
||||
//======================================================================================//
|
||||
+472
@@ -0,0 +1,472 @@
|
||||
//
|
||||
// MIT License
|
||||
// Copyright (c) 2020 Jonathan R. Madsen
|
||||
// Permission is hereby granted, free of charge, to any person obtaining a copy
|
||||
// of this software and associated documentation files (the "Software"), to deal
|
||||
// in the Software without restriction, including without limitation the rights
|
||||
// to use, copy, modify, merge, publish, distribute, sublicense, and
|
||||
// copies of the Software, and to permit persons to whom the Software is
|
||||
// furnished to do so, subject to the following conditions:
|
||||
// The above copyright notice and this permission notice shall be included in
|
||||
// all copies or substantial portions of the Software. THE SOFTWARE IS PROVIDED
|
||||
// "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT
|
||||
// LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR
|
||||
// PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT
|
||||
// HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN
|
||||
// ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
|
||||
// WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
|
||||
//
|
||||
// ---------------------------------------------------------------
|
||||
// Tasking class implementation
|
||||
// Class Description:
|
||||
// ---------------------------------------------------------------
|
||||
// Author: Jonathan Madsen
|
||||
// ---------------------------------------------------------------
|
||||
|
||||
#include "PTL/UserTaskQueue.hh"
|
||||
#include "PTL/Task.hh"
|
||||
#include "PTL/TaskGroup.hh"
|
||||
#include "PTL/ThreadPool.hh"
|
||||
|
||||
#include <cassert>
|
||||
|
||||
using namespace PTL;
|
||||
|
||||
//======================================================================================//
|
||||
|
||||
UserTaskQueue::UserTaskQueue(intmax_t nworkers, UserTaskQueue* parent)
|
||||
: VUserTaskQueue(nworkers)
|
||||
, m_is_clone((parent) ? true : false)
|
||||
, m_thread_bin((parent) ? (ThreadPool::GetThisThreadID() % (nworkers + 1)) : 0)
|
||||
, m_insert_bin((parent) ? (ThreadPool::GetThisThreadID() % (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::GetThisThreadID() << "> " << 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()
|
||||
{
|
||||
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;
|
||||
}
|
||||
}
|
||||
|
||||
//======================================================================================//
|
||||
|
||||
void
|
||||
UserTaskQueue::resize(intmax_t n)
|
||||
{
|
||||
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;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
//======================================================================================//
|
||||
|
||||
VUserTaskQueue*
|
||||
UserTaskQueue::clone()
|
||||
{
|
||||
return new UserTaskQueue(workers(), this);
|
||||
}
|
||||
//======================================================================================//
|
||||
|
||||
intmax_t
|
||||
UserTaskQueue::GetThreadBin() const
|
||||
{
|
||||
// get a thread id number
|
||||
static thread_local intmax_t tl_bin =
|
||||
(m_thread_bin + ThreadPool::GetThisThreadID()) % (m_workers + 1);
|
||||
return tl_bin;
|
||||
}
|
||||
|
||||
//======================================================================================//
|
||||
|
||||
intmax_t
|
||||
UserTaskQueue::GetInsertBin() const
|
||||
{
|
||||
return (++m_insert_bin % (m_workers + 1));
|
||||
}
|
||||
|
||||
//======================================================================================//
|
||||
|
||||
UserTaskQueue::task_pointer
|
||||
UserTaskQueue::GetThreadBinTask()
|
||||
{
|
||||
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;
|
||||
}
|
||||
|
||||
//======================================================================================//
|
||||
|
||||
UserTaskQueue::task_pointer
|
||||
UserTaskQueue::GetTask(intmax_t subq, intmax_t nitr)
|
||||
{
|
||||
// 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;
|
||||
}
|
||||
|
||||
//======================================================================================//
|
||||
|
||||
intmax_t
|
||||
UserTaskQueue::InsertTask(task_pointer task, ThreadData* data, intmax_t subq)
|
||||
{
|
||||
// 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();
|
||||
}
|
||||
|
||||
//======================================================================================//
|
||||
|
||||
void
|
||||
UserTaskQueue::ExecuteOnAllThreads(ThreadPool* tp, function_type func)
|
||||
{
|
||||
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();
|
||||
}
|
||||
|
||||
//======================================================================================//
|
||||
|
||||
void
|
||||
UserTaskQueue::ExecuteOnSpecificThreads(ThreadIdSet tid_set, ThreadPool* tp,
|
||||
function_type func)
|
||||
{
|
||||
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();
|
||||
}
|
||||
|
||||
//======================================================================================//
|
||||
|
||||
void
|
||||
UserTaskQueue::AcquireHold()
|
||||
{
|
||||
bool _hold;
|
||||
while(!(_hold = m_hold->load(std::memory_order_relaxed)))
|
||||
{
|
||||
m_hold->compare_exchange_strong(_hold, true, std::memory_order_release,
|
||||
std::memory_order_relaxed);
|
||||
}
|
||||
}
|
||||
|
||||
//======================================================================================//
|
||||
|
||||
void
|
||||
UserTaskQueue::ReleaseHold()
|
||||
{
|
||||
bool _hold;
|
||||
while((_hold = m_hold->load(std::memory_order_relaxed)))
|
||||
{
|
||||
m_hold->compare_exchange_strong(_hold, false, std::memory_order_release,
|
||||
std::memory_order_relaxed);
|
||||
}
|
||||
}
|
||||
|
||||
//======================================================================================//
|
||||
Vendored
+108
@@ -0,0 +1,108 @@
|
||||
//
|
||||
// MIT License
|
||||
// Copyright (c) 2020 Jonathan R. Madsen
|
||||
// Permission is hereby granted, free of charge, to any person obtaining a copy
|
||||
// of this software and associated documentation files (the "Software"), to deal
|
||||
// in the Software without restriction, including without limitation the rights
|
||||
// to use, copy, modify, merge, publish, distribute, sublicense, and
|
||||
// copies of the Software, and to permit persons to whom the Software is
|
||||
// furnished to do so, subject to the following conditions:
|
||||
// The above copyright notice and this permission notice shall be included in
|
||||
// all copies or substantial portions of the Software. THE SOFTWARE IS PROVIDED
|
||||
// "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT
|
||||
// LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR
|
||||
// PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT
|
||||
// HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN
|
||||
// ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
|
||||
// WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
|
||||
//
|
||||
// ---------------------------------------------------------------
|
||||
// Tasking class implementation
|
||||
//
|
||||
// Class Description:
|
||||
//
|
||||
// This file creates an abstract base class for the thread-pool tasking
|
||||
// system
|
||||
//
|
||||
// ---------------------------------------------------------------
|
||||
// Author: Jonathan Madsen (Feb 13th 2018)
|
||||
// ---------------------------------------------------------------
|
||||
|
||||
#include "PTL/VTask.hh"
|
||||
#include "PTL/ThreadData.hh"
|
||||
#include "PTL/ThreadPool.hh"
|
||||
#include "PTL/VTaskGroup.hh"
|
||||
|
||||
using namespace PTL;
|
||||
|
||||
//======================================================================================//
|
||||
|
||||
VTask::VTask()
|
||||
: m_depth(0)
|
||||
, m_group(nullptr)
|
||||
, m_pool(nullptr)
|
||||
{}
|
||||
|
||||
//======================================================================================//
|
||||
|
||||
VTask::VTask(VTaskGroup* task_group)
|
||||
: m_depth(0)
|
||||
, m_group(task_group)
|
||||
, m_pool((m_group) ? task_group->pool() : nullptr)
|
||||
{}
|
||||
|
||||
//======================================================================================//
|
||||
|
||||
VTask::VTask(ThreadPool* tp)
|
||||
: m_depth(0)
|
||||
, m_group(nullptr)
|
||||
, m_pool(tp)
|
||||
{}
|
||||
|
||||
//======================================================================================//
|
||||
|
||||
VTask::~VTask() {}
|
||||
|
||||
//======================================================================================//
|
||||
|
||||
void
|
||||
VTask::operator--()
|
||||
{
|
||||
if(m_group)
|
||||
{
|
||||
intmax_t _count = --(*m_group);
|
||||
if(_count < 2)
|
||||
{
|
||||
try
|
||||
{
|
||||
m_group->task_cond()->notify_all();
|
||||
} catch(std::system_error& e)
|
||||
{
|
||||
auto tid = ThreadPool::GetThisThreadID();
|
||||
AutoLock l(TypeMutex<decltype(std::cerr)>(), std::defer_lock);
|
||||
if(!l.owns_lock())
|
||||
l.lock();
|
||||
std::cerr << "[" << tid << "] Caught system error: " << e.what()
|
||||
<< std::endl;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
//======================================================================================//
|
||||
|
||||
bool
|
||||
VTask::is_native_task() const
|
||||
{
|
||||
return (m_group) ? m_group->is_native_task_group() : false;
|
||||
}
|
||||
|
||||
//======================================================================================//
|
||||
|
||||
ThreadPool*
|
||||
VTask::pool() const
|
||||
{
|
||||
return (!m_pool && m_group) ? m_group->pool() : m_pool;
|
||||
}
|
||||
|
||||
//======================================================================================//
|
||||
+220
@@ -0,0 +1,220 @@
|
||||
//
|
||||
// MIT License
|
||||
// Copyright (c) 2020 Jonathan R. Madsen
|
||||
// Permission is hereby granted, free of charge, to any person obtaining a copy
|
||||
// of this software and associated documentation files (the "Software"), to deal
|
||||
// in the Software without restriction, including without limitation the rights
|
||||
// to use, copy, modify, merge, publish, distribute, sublicense, and
|
||||
// copies of the Software, and to permit persons to whom the Software is
|
||||
// furnished to do so, subject to the following conditions:
|
||||
// The above copyright notice and this permission notice shall be included in
|
||||
// all copies or substantial portions of the Software. THE SOFTWARE IS PROVIDED
|
||||
// "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT
|
||||
// LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR
|
||||
// PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT
|
||||
// HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN
|
||||
// ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
|
||||
// WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
|
||||
//
|
||||
// ---------------------------------------------------------------
|
||||
// Tasking class implementation
|
||||
//
|
||||
// Class Description:
|
||||
//
|
||||
// This file creates an abstract base class for the grouping the thread-pool
|
||||
// tasking system into independently joinable units
|
||||
//
|
||||
// ---------------------------------------------------------------
|
||||
// Author: Jonathan Madsen (Feb 13th 2018)
|
||||
// ---------------------------------------------------------------
|
||||
|
||||
#include "PTL/VTaskGroup.hh"
|
||||
#include "PTL/Globals.hh"
|
||||
#include "PTL/Task.hh"
|
||||
#include "PTL/TaskRunManager.hh"
|
||||
#include "PTL/ThreadData.hh"
|
||||
#include "PTL/ThreadPool.hh"
|
||||
#include "PTL/VTask.hh"
|
||||
|
||||
using namespace PTL;
|
||||
|
||||
//======================================================================================//
|
||||
|
||||
std::atomic_uintmax_t&
|
||||
vtask_group_counter()
|
||||
{
|
||||
static std::atomic_uintmax_t _instance(0);
|
||||
return _instance;
|
||||
}
|
||||
|
||||
//======================================================================================//
|
||||
|
||||
int VTaskGroup::f_verbose = GetEnv<int>("PTL_VERBOSE", 0);
|
||||
|
||||
//======================================================================================//
|
||||
|
||||
VTaskGroup::VTaskGroup(ThreadPool* tp)
|
||||
: m_id(vtask_group_counter()++)
|
||||
, m_pool(tp)
|
||||
, m_tot_task_count(std::make_shared<atomic_int>(0))
|
||||
, m_task_cond(std::make_shared<condition_t>())
|
||||
, m_task_lock(std::make_shared<lock_t>())
|
||||
, m_main_tid(std::this_thread::get_id())
|
||||
{
|
||||
if(!m_pool && TaskRunManager::GetMasterRunManager())
|
||||
m_pool = TaskRunManager::GetMasterRunManager()->GetThreadPool();
|
||||
|
||||
if(!m_pool)
|
||||
{
|
||||
std::cerr << __FUNCTION__ << "@" << __LINE__ << " :: Warning! "
|
||||
<< "nullptr to thread pool!" << std::endl;
|
||||
}
|
||||
}
|
||||
|
||||
//======================================================================================//
|
||||
|
||||
VTaskGroup::~VTaskGroup() {}
|
||||
|
||||
//======================================================================================//
|
||||
|
||||
void
|
||||
VTaskGroup::wait()
|
||||
{
|
||||
// if no pool was initially present at creation
|
||||
if(!m_pool)
|
||||
{
|
||||
// check for master MT run-manager
|
||||
if(TaskRunManager::GetMasterRunManager())
|
||||
m_pool = TaskRunManager::GetMasterRunManager()->GetThreadPool();
|
||||
|
||||
// if MTRunManager does not exist or no thread pool created
|
||||
if(!m_pool)
|
||||
{
|
||||
if(f_verbose > 0)
|
||||
{
|
||||
fprintf(stderr, "%s @ %i :: Warning! nullptr to thread-pool (%p)\n",
|
||||
__FUNCTION__, __LINE__, static_cast<void*>(m_pool));
|
||||
std::cerr << __FUNCTION__ << "@" << __LINE__ << " :: Warning! "
|
||||
<< "nullptr to thread pool!" << std::endl;
|
||||
}
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
||||
ThreadData* data = ThreadData::GetInstance();
|
||||
if(!data)
|
||||
return;
|
||||
|
||||
ThreadPool* tpool = (m_pool) ? m_pool : data->thread_pool;
|
||||
VUserTaskQueue* taskq = (tpool) ? tpool->get_queue() : data->current_queue;
|
||||
|
||||
bool _is_master = data->is_master;
|
||||
bool _within_task = data->within_task;
|
||||
|
||||
auto is_active_state = [&]() {
|
||||
return (tpool->state()->load(std::memory_order_relaxed) !=
|
||||
thread_pool::state::STOPPED);
|
||||
};
|
||||
|
||||
auto execute_this_threads_tasks = [&]() {
|
||||
if(!taskq)
|
||||
return;
|
||||
|
||||
// only want to process if within a task
|
||||
if((!_is_master || tpool->size() < 2) && _within_task)
|
||||
{
|
||||
int bin = static_cast<int>(taskq->GetThreadBin());
|
||||
// const auto nitr = (tpool) ? tpool->size() : Thread::hardware_concurrency();
|
||||
while(this->pending() > 0)
|
||||
{
|
||||
task_pointer _task = taskq->GetTask(bin);
|
||||
if(_task)
|
||||
(*_task)();
|
||||
}
|
||||
}
|
||||
};
|
||||
|
||||
// checks for validity
|
||||
if(!is_native_task_group())
|
||||
{
|
||||
// for external threads
|
||||
if(!_is_master || tpool->size() < 2)
|
||||
return;
|
||||
}
|
||||
else if(f_verbose > 0)
|
||||
{
|
||||
if(!tpool || !taskq)
|
||||
{
|
||||
// something is wrong, didn't create thread-pool?
|
||||
fprintf(
|
||||
stderr,
|
||||
"%s @ %i :: Warning! nullptr to thread data (%p) or task-queue (%p)\n",
|
||||
__FUNCTION__, __LINE__, static_cast<void*>(tpool),
|
||||
static_cast<void*>(taskq));
|
||||
}
|
||||
// return if thread pool isn't built
|
||||
else if(is_native_task_group() && !tpool->is_alive())
|
||||
{
|
||||
fprintf(stderr, "%s @ %i :: Warning! thread-pool is not alive!\n",
|
||||
__FUNCTION__, __LINE__);
|
||||
}
|
||||
else if(!is_active_state())
|
||||
{
|
||||
fprintf(stderr, "%s @ %i :: Warning! thread-pool is not active!\n",
|
||||
__FUNCTION__, __LINE__);
|
||||
}
|
||||
}
|
||||
|
||||
intmax_t wake_size = 2;
|
||||
AutoLock _lock(*m_task_lock, std::defer_lock);
|
||||
|
||||
while(is_active_state())
|
||||
{
|
||||
execute_this_threads_tasks();
|
||||
|
||||
// while loop protects against spurious wake-ups
|
||||
while(_is_master && pending() > 0 && is_active_state())
|
||||
{
|
||||
// auto _wake = [&]() { return (wake_size > pending() || !is_active_state());
|
||||
// };
|
||||
|
||||
// lock before sleeping on condition
|
||||
if(!_lock.owns_lock())
|
||||
_lock.lock();
|
||||
|
||||
// Wait until signaled that a task has been competed
|
||||
// Unlock mutex while wait, then lock it back when signaled
|
||||
// when true, this wakes the thread
|
||||
if(pending() >= wake_size)
|
||||
{
|
||||
m_task_cond->wait(_lock);
|
||||
}
|
||||
else
|
||||
{
|
||||
m_task_cond->wait_for(_lock, std::chrono::microseconds(100));
|
||||
}
|
||||
// unlock
|
||||
if(_lock.owns_lock())
|
||||
_lock.unlock();
|
||||
}
|
||||
|
||||
// if pending is not greater than zero, we are joined
|
||||
if(pending() <= 0)
|
||||
break;
|
||||
}
|
||||
|
||||
if(_lock.owns_lock())
|
||||
_lock.unlock();
|
||||
|
||||
intmax_t ntask = this->task_count().load();
|
||||
if(ntask > 0)
|
||||
{
|
||||
std::stringstream ss;
|
||||
ss << "\nWarning! Join operation issue! " << ntask << " tasks still "
|
||||
<< "are running!" << std::endl;
|
||||
std::cerr << ss.str();
|
||||
this->wait();
|
||||
}
|
||||
}
|
||||
|
||||
//======================================================================================//
|
||||
+60
@@ -0,0 +1,60 @@
|
||||
//
|
||||
// MIT License
|
||||
// Copyright (c) 2020 Jonathan R. Madsen
|
||||
// Permission is hereby granted, free of charge, to any person obtaining a copy
|
||||
// of this software and associated documentation files (the "Software"), to deal
|
||||
// in the Software without restriction, including without limitation the rights
|
||||
// to use, copy, modify, merge, publish, distribute, sublicense, and
|
||||
// copies of the Software, and to permit persons to whom the Software is
|
||||
// furnished to do so, subject to the following conditions:
|
||||
// The above copyright notice and this permission notice shall be included in
|
||||
// all copies or substantial portions of the Software. THE SOFTWARE IS PROVIDED
|
||||
// "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT
|
||||
// LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR
|
||||
// PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT
|
||||
// HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN
|
||||
// ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
|
||||
// WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
|
||||
//
|
||||
// ---------------------------------------------------------------
|
||||
// Tasking class implementation
|
||||
// Class Description:
|
||||
// Abstract base class for creating a task queue used by
|
||||
// ThreadPool
|
||||
// ---------------------------------------------------------------
|
||||
// Author: Jonathan Madsen
|
||||
// ---------------------------------------------------------------
|
||||
|
||||
#include "PTL/VUserTaskQueue.hh"
|
||||
#include "PTL/TaskRunManager.hh"
|
||||
|
||||
using namespace PTL;
|
||||
|
||||
//======================================================================================//
|
||||
|
||||
VUserTaskQueue::VUserTaskQueue(intmax_t nworkers)
|
||||
: m_workers(nworkers)
|
||||
{
|
||||
if(m_workers < 0)
|
||||
{
|
||||
TaskRunManager* rm = TaskRunManager::GetMasterRunManager();
|
||||
m_workers = (rm) ? rm->GetNumberOfThreads() + 1 // number of threads + 1
|
||||
: (2 * std::thread::hardware_concurrency()) + 1;
|
||||
// hyperthreads + 1
|
||||
}
|
||||
}
|
||||
|
||||
//======================================================================================//
|
||||
|
||||
VUserTaskQueue::~VUserTaskQueue() {}
|
||||
|
||||
//======================================================================================//
|
||||
/*
|
||||
intmax_t& VUserTaskQueue::ThisThreadNumber() const
|
||||
{
|
||||
// get a thread id number
|
||||
static thread_local intmax_t _tid;
|
||||
return _tid;
|
||||
}
|
||||
*/
|
||||
//======================================================================================//
|
||||
Reference in New Issue
Block a user