29808310540

Committed 21 Jul 2026 06:50AM UTC coverage: 61.522% (+0.1%) from 61.394%

Build # 29808310540

Build Type

push

github

Committed by

web-flow

Commit Message

removed printOn method these debug strings where only used at a single (#899)

place
use formatter for Token instead of ostream

Coverage Stats

33 of 169 new or added lines in 4 files covered. (19.53%)

16 existing lines in 4 files now uncovered.

45084 of 73281 relevant lines covered (61.52%)

449728.5 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

85.04

/pol-core/clib/threadhelp.cpp

/** @file
 *
 * @par History
 * - 2005/12/13 Shinigami: added error code printing in create_thread for debugging
 * - 2006/02/06 Shinigami: smaller bugfix in logging
 *                         error code printing in create_thread extended
 * - 2007/02/28 Shinigami: error code printing in create_thread added for linux
 * - 2007/03/08 Shinigami: added pthread_exit and _endhreadex to close threads
 * - 2008/03/02 Nando:     Added bool dec_child to create_thread, used to dec_child_thread_count()
 *                         if there is an error on create_thread. Will fix some of the zombies.
 */


#include "threadhelp.h"

#include <algorithm>
#include <atomic>
#include <chrono>
#include <cstring>
#include <exception>
#include <thread>

#include "esignal.h"
#include "logfacility.h"
#include "passert.h"

#ifndef _WIN32
#include <errno.h>
#include <pthread.h>
#include <unistd.h>
#endif

// TODO: fix trunc cast warnings
#ifdef _MSC_VER
#pragma warning( disable : 4311 )  // trunc cast
#pragma warning( disable : 4302 )  // trunc cast
#endif


namespace Pol::threadhelp
{
using namespace std::chrono_literals;

ThreadMap& threadmap_instance()
{
  static ThreadMap threadmap;
  return threadmap;
}

std::atomic<unsigned int> child_threads( 0 );
static int threads = 0;

#ifdef _WIN32
void init_threadhelp() {}

void thread_sleep_ms( unsigned millis )
{
  Sleep( millis );
}
size_t thread_pid()
{
  return GetCurrentThreadId();
}

const DWORD MS_VC_EXCEPTION = 0x406D1388;

#pragma pack( push, 8 )
typedef struct tagTHREADNAME_INFO
{
  DWORD dwType;      // Must be 0x1000.
  LPCSTR szName;     // Pointer to name (in user addr space).
  DWORD dwThreadID;  // Thread ID (-1=caller thread).
  DWORD dwFlags;     // Reserved for future use, must be zero.
} THREADNAME_INFO;
#pragma pack( pop )

void _SetThreadName( DWORD dwThreadID, const char* name )
{
  THREADNAME_INFO info;
  info.dwType = 0x1000;
  info.szName = name;
  info.dwThreadID = dwThreadID;
  info.dwFlags = 0;

  __try
  {  // oh my god i hate ms ...
    RaiseException( MS_VC_EXCEPTION, 0, sizeof( info ) / sizeof( ULONG_PTR ), (ULONG_PTR*)&info );
  }
  __except ( EXCEPTION_EXECUTE_HANDLER )
  {
  }
}
void SetThreadName( int threadid, std::string threadName )
{
  // This redirection is needed because std::string has a destructor
  // which isn't compatible with __try
  _SetThreadName( threadid, threadName.c_str() );
}
#else
static pthread_attr_t create_detached_attr;
static Clib::SpinLock pthread_attr_lock;

void init_threadhelp()
{
  int res;
  res = pthread_attr_init( &create_detached_attr );
  passert_always( res == 0 );
  res = pthread_attr_setdetachstate( &create_detached_attr, PTHREAD_CREATE_DETACHED );
  passert_always( res == 0 );
}

void thread_sleep_ms( unsigned millis )
{
  usleep( millis * 1000L );
}
size_t thread_pid()
{
#ifdef __APPLE__
  return reinterpret_cast<size_t>( pthread_self() );
#else
  return pthread_self();
#endif
}
#endif

void run_thread( void ( *threadf )() )
{
  // thread creator calls inc_child_thread_count before starting thread
  try
  {
    ( *threadf )();
  }
  catch ( std::exception& ex )
  {
    ERROR_PRINTLN( "Thread exception: {}", ex.what() );
  }

  --child_threads;

  threadmap_instance().Unregister( thread_pid() );
}
void run_thread( void ( *threadf )( void* ), void* arg )
{
  // thread creator calls inc_child_thread_count before starting thread
  try
  {
    ( *threadf )( arg );
  }
  catch ( std::exception& ex )
  {
    ERROR_PRINTLN( "Thread exception: {}", ex.what() );
  }

  --child_threads;

  threadmap_instance().Unregister( thread_pid() );
}

class ThreadData
{
public:
  std::string name;
  void ( *entry )( void* );
  void ( *entry_noparam )();
  void* arg;
};

#ifdef _WIN32
unsigned __stdcall thread_stub2( void* v_td )
#else
void* thread_stub2( void* v_td )
#endif
{
  ThreadData* td = reinterpret_cast<ThreadData*>( v_td );

  void ( *entry )( void* ) = td->entry;
  void ( *entry_noparam )() = td->entry_noparam;
  void* arg = td->arg;

  threadmap_instance().Register( thread_pid(), td->name );

  delete td;
  td = nullptr;

  if ( entry != nullptr )
    run_thread( entry, arg );
  else
    run_thread( entry_noparam );

#ifdef _WIN32
  _endthreadex( 0 );
  return 0;
#else
  pthread_exit( nullptr );
  return nullptr;
#endif
}

#ifdef _WIN32
void create_thread( ThreadData* td, bool dec_child = false )
{
  // If the thread starts successfully, td will be deleted by thread_stub2.
  // So we must save the threadName for later.
  std::string threadName = td->name;

  unsigned threadid = 0;
  HANDLE h = (HANDLE)_beginthreadex( nullptr, 0, thread_stub2, td, 0, &threadid );
  if ( h == 0 )  // added for better debugging
  {
    POLLOG( "error in create_thread: {} {} \"{}\" \"{}\" {} {} {} {} {} {}\n", errno, _doserrno,
            strerror( errno ), strerror( _doserrno ), threads++, (unsigned)thread_stub2,
            td->name.c_str(), (unsigned)td->entry, (unsigned)td->entry_noparam, td->arg );

    // dec_child says that we should dec_child_threads when there's an error... :)
    if ( dec_child )
      --child_threads;
  }
  else
  {
    SetThreadName( threadid, threadName );
    CloseHandle( h );
  }
}
#else
void create_thread( ThreadData* td, bool dec_child = false )
{
  Clib::SpinLockGuard guard( pthread_attr_lock );
  pthread_t thread;
  int result = pthread_create( &thread, &create_detached_attr, thread_stub2, td );
  if ( result != 0 )  // added for better debugging
  {
    POLLOG( "error in create_thread: {} {} \"{}\" {} {} {} {} {} {}\n", result, errno,
            strerror( errno ), threads++, reinterpret_cast<const void*>( thread_stub2 ),
            td->name.c_str(), reinterpret_cast<const void*>( td->entry ),
            reinterpret_cast<const void*>( td->entry_noparam ), td->arg );

    // dec_child says that we should dec_child_threads when there's an error... :)
    if ( dec_child )
      --child_threads;
  }
}
#endif

void start_thread( void ( *entry )( void* ), const char* thread_name, void* arg )
{
  auto td = new ThreadData;
  td->name = thread_name;
  td->entry = entry;
  td->entry_noparam = nullptr;
  td->arg = arg;

  ++child_threads;

  create_thread( td, true );
}

void start_thread( void ( *entry )(), const char* thread_name )
{
  auto td = new ThreadData;
  td->name = thread_name;
  td->entry = nullptr;
  td->entry_noparam = entry;
  td->arg = nullptr;

  ++child_threads;

  create_thread( td, true );
}

ThreadMap::ThreadMap()
    : _spinlock(),
      _contents()
#ifdef _WIN32
      ,
      _handles()
#endif
{
}

#ifdef _WIN32
HANDLE ThreadMap::getThreadHandle( size_t pid ) const
{
  Clib::SpinLockGuard guard( _spinlock );
  auto itr = _handles.find( pid );
  if ( itr == _handles.end() )
  {
    return 0;
  }
  return itr->second;
}
#endif
void ThreadMap::Register( size_t pid, const std::string& name )
{
  Clib::SpinLockGuard guard( _spinlock );
  _contents.insert( std::make_pair( pid, name ) );
#ifdef _WIN32
  HANDLE hThread = 0;
  if ( !DuplicateHandle( GetCurrentProcess(), GetCurrentThread(), GetCurrentProcess(), &hThread, 0,
                         FALSE, DUPLICATE_SAME_ACCESS ) )
  {
    ERROR_PRINTLN( "failed to duplicate thread handle" );
    return;
  }
  _handles.insert( std::make_pair( pid, hThread ) );
#endif
}
void ThreadMap::Unregister( size_t pid )
{
  Clib::SpinLockGuard guard( _spinlock );
  _contents.erase( pid );
#ifdef _WIN32
  auto itr = _handles.find( pid );
  if ( itr != _handles.end() )
    CloseHandle( itr->second );
  _handles.erase( pid );
#endif
}
void ThreadMap::CopyContents( Contents& out ) const
{
  Clib::SpinLockGuard guard( _spinlock );
  out = _contents;
}

ThreadRegister::ThreadRegister( const std::string& name )
{
  threadmap_instance().Register( thread_pid(), name );
}
ThreadRegister::~ThreadRegister()
{
  threadmap_instance().Unregister( thread_pid() );
}


/// Creates a threadpool of workers.
/// blocks on deconstruction
/// eg:
/// TaskThreadPool workers;
/// for (....)
///   workers.push([&](){dosomework();});
TaskThreadPool::TaskThreadPool() : _done( false ), _msg_queue() {}

TaskThreadPool::TaskThreadPool( const std::string& name ) : _done( false ), _msg_queue()
{
  // get the count of processors
  unsigned int max_count = std::thread::hardware_concurrency();
  if ( !max_count )  // can fail so at least one
    max_count = 1;
  init( max_count, name );
}

TaskThreadPool::TaskThreadPool( unsigned int max_count, const std::string& name )
    : _done( false ), _msg_queue()
{
  init( max_count, name );
}

void TaskThreadPool::init( unsigned int max_count, const std::string& name )
{
  for ( unsigned int i = 0; i < max_count; ++i )
  {
    _threads.emplace_back(
        [this, name]()
        {
          ThreadRegister register_thread( "TaskPool " + name );
          auto f = msg();
          try
          {
            while ( !_done )
            {
              try
              {
                _msg_queue.pop_wait( &f );
                f();
              }
              catch ( std::exception& ex )
              {
                ERROR_PRINTLN( "Thread exception: {}", ex.what() );
                Clib::force_backtrace( true );
              }
            }
          }
          catch ( msg_queue::Canceled& )
          {
          }
          // purge the queue empty
          std::list<msg> remaining;
          _msg_queue.pop_remaining( &remaining );
          for ( auto& _f : remaining )
            _f();
        } );
  }
}

void TaskThreadPool::init_pool( unsigned int max_count, const std::string& name )
{
  if ( !_threads.empty() )
    return;
  init( max_count, name );
}

void TaskThreadPool::deinit_pool()
{
  if ( _threads.empty() )
    return;
  // send both done and cancel to wake up all workers
  _msg_queue.push_move(
      [&]()
      {
        _done = true;
        _msg_queue.cancel();
      } );
  for ( auto& thread : _threads )
    thread.join();
  _threads.clear();
}
TaskThreadPool::~TaskThreadPool()
{
  deinit_pool();
}

/// simply fire and forget only the deconstructor ensures the msg to be finished
void TaskThreadPool::push( msg&& msg )
{
  _msg_queue.push_move( std::move( msg ) );
}

/// returns a future which will be set once the msg is processed
std::future<bool> TaskThreadPool::checked_push( msg&& msg )
{
  auto promise = std::promise<bool>();
  auto ret = promise.get_future();
  _msg_queue.push_move(
      [promise = std::move( promise ), msg = std::move( msg )]() mutable
      {
        try
        {
          msg();
          promise.set_value( true );
        }
        catch ( ... )
        {
          promise.set_exception( std::current_exception() );
        }
      } );
  return ret;
}

size_t TaskThreadPool::size() const
{
  return _threads.size();
}


class DynTaskThreadPool::PoolWorker
{
public:
  PoolWorker( DynTaskThreadPool* parent, const std::string& name );
  PoolWorker( const PoolWorker& ) = delete;
  PoolWorker& operator=( const PoolWorker& ) = delete;
  void run();
  bool isretired() const;

  constexpr static size_t MIN_WORKER = 4;
  constexpr static size_t JITTER_STEPS = 10;
  constexpr static auto TIMEOUT = 5min;
  constexpr static auto TIMEOUT_JITTER = 15s;

  // guard which is alive as long as the task is in the pool
  struct BusyGuard
  {
    std::atomic<size_t>* _busy;
    BusyGuard( std::atomic<size_t>* busy ) : _busy( busy )
    {
      _busy->fetch_add( 1, std::memory_order_relaxed );
    }
    BusyGuard( BusyGuard&& o ) noexcept : _busy( o._busy ) { o._busy = nullptr; }
    BusyGuard( const BusyGuard& ) = delete;
    BusyGuard& operator=( const BusyGuard& ) = delete;
    BusyGuard& operator=( BusyGuard&& ) = delete;
    ~BusyGuard()
    {
      if ( _busy )
        _busy->fetch_sub( 1, std::memory_order_relaxed );
    }
  };

private:
  std::string _name;
  std::jthread _thread;
  DynTaskThreadPool* _parent;
  std::atomic<bool> _retired;
  std::chrono::seconds _timeout;
};

DynTaskThreadPool::PoolWorker::PoolWorker( DynTaskThreadPool* parent, const std::string& name )
    : _name( name ),
      _thread(),
      _parent( parent ),
      _retired( false ),
      _timeout( ( parent->_worker_count % JITTER_STEPS ) * TIMEOUT_JITTER )
{
  run();
}

bool DynTaskThreadPool::PoolWorker::isretired() const
{
  return _retired.load( std::memory_order_relaxed );
}

void DynTaskThreadPool::PoolWorker::run()
{
  _thread = std::jthread(
      [&]()
      {
        ThreadRegister register_thread( _name );
        ERROR_PRINTLN( "created pool worker {}", _name );
        auto f = msg();
        try
        {
          while ( !_parent->_done && !Clib::exit_signalled )
          {
            if ( !_parent->_msg_queue.pop_wait_for( &f, _timeout ) )
            {
              std::lock_guard<std::mutex> guard( _parent->_pool_mutex );
              // we timed out: can we retire?
              if ( _parent->_live_threads > MIN_WORKER )
              {
                --_parent->_live_threads;
                _retired.store( true, std::memory_order_relaxed );
                ERROR_PRINTLN( "removed pool worker {}", _name );
                return;
              }
              continue;  // keep alive
            }

            try
            {
              f();
            }
            catch ( std::exception& ex )
            {
              ERROR_PRINTLN( "Thread exception: {}", ex.what() );
              Clib::force_backtrace( false );
            }
            f = nullptr;  // reset BusyGuard
          }
        }
        catch ( msg_queue::Canceled& )
        {
        }
      } );
}

/// Creates a dynamic threadpool of workers.
/// if no idle worker is found creates a new worker thread
/// blocks on deconstruction
/// idle worker threads get destroyed after timeout until minimum count is reached
/// eg:
/// DynTaskThreadPool workers;
/// for (....)
///   workers.push([&](){dosomework();});
DynTaskThreadPool::DynTaskThreadPool( const std::string& name )
    : _done( false ), _msg_queue(), _pool_mutex(), _name( "DynTaskPool" + name )
{
}

void DynTaskThreadPool::prefill_workers()
{
  std::lock_guard<std::mutex> guard( _pool_mutex );
  for ( size_t i = 0; i < PoolWorker::MIN_WORKER; ++i )
  {
    _threads.emplace_back( new PoolWorker( this, fmt::format( "{} {}", _name, _worker_count++ ) ) );
    ++_live_threads;
  }
}

size_t DynTaskThreadPool::threadpoolsize() const
{
  std::lock_guard<std::mutex> guard( _pool_mutex );
  return _live_threads;
}

void DynTaskThreadPool::create_thread()
{
  std::lock_guard<std::mutex> guard( _pool_mutex );
  // TODO move to some task and run every X seconds
  // remove timeout threads
  std::erase_if( _threads, []( const auto& w ) { return w->isretired(); } );
  // still atleast one idle worker left?
  if ( _busy_count.load( std::memory_order_relaxed ) <= _live_threads )
    return;

  _threads.emplace_back( new PoolWorker( this, fmt::format( "{} {}", _name, _worker_count++ ) ) );
  ++_live_threads;
}

DynTaskThreadPool::~DynTaskThreadPool()
{
  // send both done and cancel to wake up all workers
  _msg_queue.push_move(
      [&]()
      {
        _done = true;
        _msg_queue.cancel();
      } );
  create_thread();
  _threads.clear();  // dont rely on order
}

/// simply fire and forget only the deconstructor ensures the msg to be finished
void DynTaskThreadPool::push( msg&& msg )
{
  _msg_queue.push_move(
      [msg = std::move( msg ), busy = PoolWorker::BusyGuard( &_busy_count )]() mutable { msg(); } );
  create_thread();  // needs to be the last so that busy_count is updated
}

/// returns a future which will be set once the msg is processed
std::future<bool> DynTaskThreadPool::checked_push( msg&& msg )
{
  auto promise = std::promise<bool>();
  auto ret = promise.get_future();
  _msg_queue.push_move(
      [promise = std::move( promise ), msg = std::move( msg ),
       busy = PoolWorker::BusyGuard( &_busy_count )]() mutable
      {
        try
        {
          msg();
          promise.set_value( true );
        }
        catch ( ... )
        {
          promise.set_exception( std::current_exception() );
        }
      } );
  create_thread();  // needs to be the last so that busy_count is updated
  return ret;
}
}  // namespace Pol::threadhelp

1	/** @file
2	*
3	* @par History
4	* - 2005/12/13 Shinigami: added error code printing in create_thread for debugging
5	* - 2006/02/06 Shinigami: smaller bugfix in logging
6	* error code printing in create_thread extended
7	* - 2007/02/28 Shinigami: error code printing in create_thread added for linux
8	* - 2007/03/08 Shinigami: added pthread_exit and _endhreadex to close threads
9	* - 2008/03/02 Nando: Added bool dec_child to create_thread, used to dec_child_thread_count()
10	* if there is an error on create_thread. Will fix some of the zombies.
11	*/
12
13
14	#include "threadhelp.h"
15
16	#include <algorithm>
17	#include <atomic>
18	#include <chrono>
19	#include <cstring>
20	#include <exception>
21	#include <thread>
22
23	#include "esignal.h"
24	#include "logfacility.h"
25	#include "passert.h"
26
27	#ifndef _WIN32
28	#include <errno.h>
29	#include <pthread.h>
30	#include <unistd.h>
31	#endif
32
33	// TODO: fix trunc cast warnings
34	#ifdef _MSC_VER
35	#pragma warning( disable : 4311 ) // trunc cast
36	#pragma warning( disable : 4302 ) // trunc cast
37	#endif
38
39
40	namespace Pol::threadhelp
41	{
42	using namespace std::chrono_literals;
43
44	ThreadMap& threadmap_instance()	92✔
45	{
46	static ThreadMap threadmap;	92✔
47	return threadmap;	92✔
48	}
49
50	std::atomic<unsigned int> child_threads( 0 );
51	static int threads = 0;
52
53	#ifdef _WIN32
54	void init_threadhelp() {}
55
56	void thread_sleep_ms( unsigned millis )
57	{
58	Sleep( millis );
59	}
60	size_t thread_pid()
61	{
62	return GetCurrentThreadId();
63	}
64
65	const DWORD MS_VC_EXCEPTION = 0x406D1388;
66
67	#pragma pack( push, 8 )
68	typedef struct tagTHREADNAME_INFO
69	{
70	DWORD dwType; // Must be 0x1000.
71	LPCSTR szName; // Pointer to name (in user addr space).
72	DWORD dwThreadID; // Thread ID (-1=caller thread).
73	DWORD dwFlags; // Reserved for future use, must be zero.
74	} THREADNAME_INFO;
75	#pragma pack( pop )
76
77	void _SetThreadName( DWORD dwThreadID, const char* name )
78	{
79	THREADNAME_INFO info;
80	info.dwType = 0x1000;
81	info.szName = name;
82	info.dwThreadID = dwThreadID;
83	info.dwFlags = 0;
84
85	__try
86	{ // oh my god i hate ms ...
87	RaiseException( MS_VC_EXCEPTION, 0, sizeof( info ) / sizeof( ULONG_PTR ), (ULONG_PTR*)&info );
88	}
89	__except ( EXCEPTION_EXECUTE_HANDLER )
90	{
91	}
92	}
93	void SetThreadName( int threadid, std::string threadName )
94	{
95	// This redirection is needed because std::string has a destructor
96	// which isn't compatible with __try
97	_SetThreadName( threadid, threadName.c_str() );
98	}
99	#else
100	static pthread_attr_t create_detached_attr;
101	static Clib::SpinLock pthread_attr_lock;
102
103	void init_threadhelp()	3✔
104	{
105	int res;
106	res = pthread_attr_init( &create_detached_attr );	3✔
107	passert_always( res == 0 );	3✔
108	res = pthread_attr_setdetachstate( &create_detached_attr, PTHREAD_CREATE_DETACHED );	3✔
109	passert_always( res == 0 );	3✔
110	}	3✔
111
112	void thread_sleep_ms( unsigned millis )	67✔
113	{
114	usleep( millis * 1000L );	67✔
115	}	67✔
116	size_t thread_pid()	81,682,441✔
117	{
118	#ifdef __APPLE__
119	return reinterpret_cast<size_t>( pthread_self() );
120	#else
121	return pthread_self();	81,682,441✔
122	#endif
123	}
124	#endif
125
126	void run_thread( void ( *threadf )() )	18✔
127	{
128	// thread creator calls inc_child_thread_count before starting thread
129	try
130	{
131	( *threadf )();	18✔
132	}
133	catch ( std::exception& ex )	×
134	{
135	ERROR_PRINTLN( "Thread exception: {}", ex.what() );	×
136	}	×
137
138	--child_threads;	18✔
139
140	threadmap_instance().Unregister( thread_pid() );	18✔
141	}	18✔
142	void run_thread( void ( threadf )( void ), void* arg )	6✔
143	{
144	// thread creator calls inc_child_thread_count before starting thread
145	try
146	{
147	( *threadf )( arg );	6✔
148	}
149	catch ( std::exception& ex )	×
150	{
151	ERROR_PRINTLN( "Thread exception: {}", ex.what() );	×
152	}	×
153
154	--child_threads;	6✔
155
156	threadmap_instance().Unregister( thread_pid() );	6✔
157	}	6✔
158
159	class ThreadData
160	{
161	public:
162	std::string name;
163	void ( entry )( void );
164	void ( *entry_noparam )();
165	void* arg;
166	};
167
168	#ifdef _WIN32
169	unsigned __stdcall thread_stub2( void* v_td )
170	#else
171	void* thread_stub2( void* v_td )	24✔
172	#endif
173	{
174	ThreadData* td = reinterpret_cast<ThreadData*>( v_td );	24✔
175
176	void ( entry )( void ) = td->entry;	24✔
177	void ( *entry_noparam )() = td->entry_noparam;	24✔
178	void* arg = td->arg;	24✔
179
180	threadmap_instance().Register( thread_pid(), td->name );	24✔
181
182	delete td;	24✔
183	td = nullptr;	24✔
184
185	if ( entry != nullptr )	24✔
186	run_thread( entry, arg );	6✔
187	else
188	run_thread( entry_noparam );	18✔
189
190	#ifdef _WIN32
191	_endthreadex( 0 );
192	return 0;
193	#else
194	pthread_exit( nullptr );	24✔
195	return nullptr;
196	#endif
197	}
198
199	#ifdef _WIN32
200	void create_thread( ThreadData* td, bool dec_child = false )
201	{
202	// If the thread starts successfully, td will be deleted by thread_stub2.
203	// So we must save the threadName for later.
204	std::string threadName = td->name;
205
206	unsigned threadid = 0;
207	HANDLE h = (HANDLE)_beginthreadex( nullptr, 0, thread_stub2, td, 0, &threadid );
208	if ( h == 0 ) // added for better debugging
209	{
210	POLLOG( "error in create_thread: {} {} \"{}\" \"{}\" {} {} {} {} {} {}\n", errno, _doserrno,
211	strerror( errno ), strerror( _doserrno ), threads++, (unsigned)thread_stub2,
212	td->name.c_str(), (unsigned)td->entry, (unsigned)td->entry_noparam, td->arg );
213
214	// dec_child says that we should dec_child_threads when there's an error... :)
215	if ( dec_child )
216	--child_threads;
217	}
218	else
219	{
220	SetThreadName( threadid, threadName );
221	CloseHandle( h );
222	}
223	}
224	#else
225	void create_thread( ThreadData* td, bool dec_child = false )	24✔
226	{
227	Clib::SpinLockGuard guard( pthread_attr_lock );	24✔
228	pthread_t thread;
229	int result = pthread_create( &thread, &create_detached_attr, thread_stub2, td );	24✔
230	if ( result != 0 ) // added for better debugging	24✔
231	{
232	POLLOG( "error in create_thread: {} {} \"{}\" {} {} {} {} {} {}\n", result, errno,	×
233	strerror( errno ), threads++, reinterpret_cast<const void*>( thread_stub2 ),	×
234	td->name.c_str(), reinterpret_cast<const void*>( td->entry ),	×
235	reinterpret_cast<const void*>( td->entry_noparam ), td->arg );	×
236
237	// dec_child says that we should dec_child_threads when there's an error... :)
238	if ( dec_child )	×
239	--child_threads;	×
240	}
241	}	24✔
242	#endif
243
244	void start_thread( void ( entry )( void ), const char* thread_name, void* arg )	6✔
245	{
246	auto td = new ThreadData;	6✔
247	td->name = thread_name;	6✔
248	td->entry = entry;	6✔
249	td->entry_noparam = nullptr;	6✔
250	td->arg = arg;	6✔
251
252	++child_threads;	6✔
253
254	create_thread( td, true );	6✔
255	}	6✔
256
257	void start_thread( void ( entry )(), const char thread_name )	18✔
258	{
259	auto td = new ThreadData;	18✔
260	td->name = thread_name;	18✔
261	td->entry = nullptr;	18✔
262	td->entry_noparam = entry;	18✔
263	td->arg = nullptr;	18✔
264
265	++child_threads;	18✔
266
267	create_thread( td, true );	18✔
268	}	18✔
269
270	ThreadMap::ThreadMap()	3✔
271	: _spinlock(),	3✔
272	_contents()	3✔
273	#ifdef _WIN32
274	,
275	_handles()
276	#endif
277	{
278	}	3✔
279
280	#ifdef _WIN32
281	HANDLE ThreadMap::getThreadHandle( size_t pid ) const
282	{
283	Clib::SpinLockGuard guard( _spinlock );
284	auto itr = _handles.find( pid );
285	if ( itr == _handles.end() )
286	{
287	return 0;
288	}
289	return itr->second;
290	}
291	#endif
292	void ThreadMap::Register( size_t pid, const std::string& name )	47✔
293	{
294	Clib::SpinLockGuard guard( _spinlock );	47✔
295	_contents.insert( std::make_pair( pid, name ) );	47✔
296	#ifdef _WIN32
297	HANDLE hThread = 0;
298	if ( !DuplicateHandle( GetCurrentProcess(), GetCurrentThread(), GetCurrentProcess(), &hThread, 0,
299	FALSE, DUPLICATE_SAME_ACCESS ) )
300	{
301	ERROR_PRINTLN( "failed to duplicate thread handle" );
302	return;
303	}
304	_handles.insert( std::make_pair( pid, hThread ) );
305	#endif
306	}	47✔
307	void ThreadMap::Unregister( size_t pid )	45✔
308	{
309	Clib::SpinLockGuard guard( _spinlock );	45✔
310	_contents.erase( pid );	45✔
311	#ifdef _WIN32
312	auto itr = _handles.find( pid );
313	if ( itr != _handles.end() )
314	CloseHandle( itr->second );
315	_handles.erase( pid );
316	#endif
317	}	45✔
318	void ThreadMap::CopyContents( Contents& out ) const	×
319	{
320	Clib::SpinLockGuard guard( _spinlock );	×
321	out = _contents;	×
322	}	×
323
324	ThreadRegister::ThreadRegister( const std::string& name )	21✔
325	{
326	threadmap_instance().Register( thread_pid(), name );	21✔
327	}	21✔
328	ThreadRegister::~ThreadRegister()	21✔
329	{
330	threadmap_instance().Unregister( thread_pid() );	21✔
331	}	21✔
332
333
334	/// Creates a threadpool of workers.
335	/// blocks on deconstruction
336	/// eg:
337	/// TaskThreadPool workers;
338	/// for (....)
339	/// workers.push([&](){dosomework();});
340	TaskThreadPool::TaskThreadPool() : _done( false ), _msg_queue() {}	3✔
341
342	TaskThreadPool::TaskThreadPool( const std::string& name ) : _done( false ), _msg_queue()	×
343	{
344	// get the count of processors
345	unsigned int max_count = std::thread::hardware_concurrency();	×
346	if ( !max_count ) // can fail so at least one	×
347	max_count = 1;	×
348	init( max_count, name );	×
349	}	×
350
351	TaskThreadPool::TaskThreadPool( unsigned int max_count, const std::string& name )	1✔
352	: _done( false ), _msg_queue()	1✔
353	{
354	init( max_count, name );	1✔
355	}	1✔
356
357	void TaskThreadPool::init( unsigned int max_count, const std::string& name )	4✔
358	{
359	for ( unsigned int i = 0; i < max_count; ++i )	14✔
360	{
361	_threads.emplace_back(	10✔
362	[this, name]()	20✔
363	{
364	ThreadRegister register_thread( "TaskPool " + name );	10✔
365	auto f = msg();	10✔
366	try
367	{
368	while ( !_done )	103✔
369	{
370	try
371	{
372	_msg_queue.pop_wait( &f );	99✔
373	f();	93✔
374	}
375	catch ( std::exception& ex )	6✔
376	{
377	ERROR_PRINTLN( "Thread exception: {}", ex.what() );	×
378	Clib::force_backtrace( true );	×
379	}	×
380	}
381	}
382	catch ( msg_queue::Canceled& )	6✔
383	{
384	}	6✔
385	// purge the queue empty
386	std::list<msg> remaining;	10✔
387	_msg_queue.pop_remaining( &remaining );	10✔
388	for ( auto& _f : remaining )	10✔
389	_f();	×
390	} );	10✔
391	}
392	}	4✔
393
394	void TaskThreadPool::init_pool( unsigned int max_count, const std::string& name )	3✔
395	{
396	if ( !_threads.empty() )	3✔
397	return;	×
398	init( max_count, name );	3✔
399	}
400
401	void TaskThreadPool::deinit_pool()	7✔
402	{
403	if ( _threads.empty() )	7✔
404	return;	3✔
405	// send both done and cancel to wake up all workers
406	_msg_queue.push_move(	4✔
407	[&]()	4✔
408	{
409	_done = true;	4✔
410	_msg_queue.cancel();	4✔
411	} );	4✔
412	for ( auto& thread : _threads )	14✔
413	thread.join();	10✔
414	_threads.clear();	4✔
415	}
416	TaskThreadPool::~TaskThreadPool()	4✔
417	{
418	deinit_pool();	4✔
419	}	4✔
420
421	/// simply fire and forget only the deconstructor ensures the msg to be finished
422	void TaskThreadPool::push( msg&& msg )	45✔
423	{
424	_msg_queue.push_move( std::move( msg ) );	45✔
425	}	45✔
426
427	/// returns a future which will be set once the msg is processed
428	std::future<bool> TaskThreadPool::checked_push( msg&& msg )	44✔
429	{
430	auto promise = std::promise<bool>();	44✔
431	auto ret = promise.get_future();	44✔
432	_msg_queue.push_move(	44✔
433	[promise = std::move( promise ), msg = std::move( msg )]() mutable	88✔
434	{
435	try
436	{
437	msg();	44✔
438	promise.set_value( true );	44✔
439	}
440	catch ( ... )	×
441	{
442	promise.set_exception( std::current_exception() );	×
443	}	×
444	} );	44✔
445	return ret;	88✔
446	}	44✔
447
448	size_t TaskThreadPool::size() const	3✔
449	{
450	return _threads.size();	3✔
451	}
452
453
454	class DynTaskThreadPool::PoolWorker
455	{
456	public:
457	PoolWorker( DynTaskThreadPool* parent, const std::string& name );
458	PoolWorker( const PoolWorker& ) = delete;
459	PoolWorker& operator=( const PoolWorker& ) = delete;
460	void run();
461	bool isretired() const;
462
463	constexpr static size_t MIN_WORKER = 4;
464	constexpr static size_t JITTER_STEPS = 10;
465	constexpr static auto TIMEOUT = 5min;
466	constexpr static auto TIMEOUT_JITTER = 15s;
467
468	// guard which is alive as long as the task is in the pool
469	struct BusyGuard
470	{
471	std::atomic<size_t>* _busy;
472	BusyGuard( std::atomic<size_t>* busy ) : _busy( busy )	272✔
473	{
474	_busy->fetch_add( 1, std::memory_order_relaxed );	272✔
475	}	272✔
476	BusyGuard( BusyGuard&& o ) noexcept : _busy( o._busy ) { o._busy = nullptr; }	272✔
477	BusyGuard( const BusyGuard& ) = delete;
478	BusyGuard& operator=( const BusyGuard& ) = delete;
479	BusyGuard& operator=( BusyGuard&& ) = delete;
480	~BusyGuard()	544✔
481	{
482	if ( _busy )	544✔
483	_busy->fetch_sub( 1, std::memory_order_relaxed );	272✔
484	}	544✔
485	};
486
487	private:
488	std::string _name;
489	std::jthread _thread;
490	DynTaskThreadPool* _parent;
491	std::atomic<bool> _retired;
492	std::chrono::seconds _timeout;
493	};
494
495	DynTaskThreadPool::PoolWorker::PoolWorker( DynTaskThreadPool* parent, const std::string& name )	11✔
496	: _name( name ),	11✔
497	_thread(),	11✔
498	_parent( parent ),	11✔
499	_retired( false ),	11✔
500	_timeout( ( parent->_worker_count % JITTER_STEPS ) * TIMEOUT_JITTER )	11✔
501	{
502	run();	11✔
503	}	11✔
504
505	bool DynTaskThreadPool::PoolWorker::isretired() const	1,184✔
506	{
507	return _retired.load( std::memory_order_relaxed );	1,184✔
508	}
509
510	void DynTaskThreadPool::PoolWorker::run()	11✔
511	{
512	_thread = std::jthread(	11✔
513	[&]()	11✔
514	{
515	ThreadRegister register_thread( _name );	11✔
516	ERROR_PRINTLN( "created pool worker {}", _name );	11✔
517	auto f = msg();	11✔
518	try
519	{
520	while ( !_parent->_done && !Clib::exit_signalled )	286✔
521	{
522	if ( !_parent->_msg_queue.pop_wait_for( &f, _timeout ) )	282✔
523	{
UNCOV 524	std::lock_guard<std::mutex> guard( _parent->_pool_mutex );	×
525	// we timed out: can we retire?
UNCOV 526	if ( _parent->_live_threads > MIN_WORKER )	×
527	{
528	--_parent->_live_threads;	×
529	_retired.store( true, std::memory_order_relaxed );	×
530	ERROR_PRINTLN( "removed pool worker {}", _name );	×
531	return;	×
532	}
UNCOV 533	continue; // keep alive	×
UNCOV 534	}	×
535
536	try
537	{
538	f();	275✔
539	}
540	catch ( std::exception& ex )	1✔
541	{
542	ERROR_PRINTLN( "Thread exception: {}", ex.what() );	1✔
543	Clib::force_backtrace( false );	1✔
544	}	1✔
545	f = nullptr; // reset BusyGuard	275✔
546	}
547	}
548	catch ( msg_queue::Canceled& )	7✔
549	{
550	}	7✔
551	} );	22✔
552	}	11✔
553
554	/// Creates a dynamic threadpool of workers.
555	/// if no idle worker is found creates a new worker thread
556	/// blocks on deconstruction
557	/// idle worker threads get destroyed after timeout until minimum count is reached
558	/// eg:
559	/// DynTaskThreadPool workers;
560	/// for (....)
561	/// workers.push([&](){dosomework();});
562	DynTaskThreadPool::DynTaskThreadPool( const std::string& name )	4✔
563	: _done( false ), _msg_queue(), _pool_mutex(), _name( "DynTaskPool" + name )	4✔
564	{
565	}	4✔
566
567	void DynTaskThreadPool::prefill_workers()	2✔
568	{
569	std::lock_guard<std::mutex> guard( _pool_mutex );	2✔
570	for ( size_t i = 0; i < PoolWorker::MIN_WORKER; ++i )	10✔
571	{
572	_threads.emplace_back( new PoolWorker( this, fmt::format( "{} {}", _name, _worker_count++ ) ) );	16✔
573	++_live_threads;	8✔
574	}
575	}	2✔
576
577	size_t DynTaskThreadPool::threadpoolsize() const	2✔
578	{
579	std::lock_guard<std::mutex> guard( _pool_mutex );	2✔
580	return _live_threads;	2✔
581	}	2✔
582
583	void DynTaskThreadPool::create_thread()	276✔
584	{
585	std::lock_guard<std::mutex> guard( _pool_mutex );	276✔
586	// TODO move to some task and run every X seconds
587	// remove timeout threads
588	std::erase_if( _threads, []( const auto& w ) { return w->isretired(); } );	1,460✔
589	// still atleast one idle worker left?
590	if ( _busy_count.load( std::memory_order_relaxed ) <= _live_threads )	552✔
591	return;	273✔
592
593	_threads.emplace_back( new PoolWorker( this, fmt::format( "{} {}", _name, _worker_count++ ) ) );	6✔
594	++_live_threads;	3✔
595	}	276✔
596
597	DynTaskThreadPool::~DynTaskThreadPool()	4✔
598	{
599	// send both done and cancel to wake up all workers
600	_msg_queue.push_move(	4✔
601	[&]()	4✔
602	{
603	_done = true;	3✔
604	_msg_queue.cancel();	3✔
605	} );	3✔
606	create_thread();	4✔
607	_threads.clear(); // dont rely on order	4✔
608	}	4✔
609
610	/// simply fire and forget only the deconstructor ensures the msg to be finished
611	void DynTaskThreadPool::push( msg&& msg )	269✔
612	{
613	_msg_queue.push_move(	269✔
614	[msg = std::move( msg ), busy = PoolWorker::BusyGuard( &_busy_count )]() mutable { msg(); } );	807✔
615	create_thread(); // needs to be the last so that busy_count is updated	269✔
616	}	269✔
617
618	/// returns a future which will be set once the msg is processed
619	std::future<bool> DynTaskThreadPool::checked_push( msg&& msg )	3✔
620	{
621	auto promise = std::promise<bool>();	3✔
622	auto ret = promise.get_future();	3✔
623	_msg_queue.push_move(	3✔
624	[promise = std::move( promise ), msg = std::move( msg ),	6✔
625	busy = PoolWorker::BusyGuard( &_busy_count )]() mutable
626	{
627	try
628	{
629	msg();	3✔
630	promise.set_value( true );	2✔
631	}
632	catch ( ... )	1✔
633	{
634	promise.set_exception( std::current_exception() );	1✔
635	}	1✔
636	} );	3✔
637	create_thread(); // needs to be the last so that busy_count is updated	3✔
638	return ret;	6✔
639	}	3✔
640	} // namespace Pol::threadhelp

polserver / polserver / 29808310540

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous