29352231943

Committed 14 Jul 2026 05:04PM UTC coverage: 61.419% (+0.07%) from 61.349%

Build # 29352231943

Build Type

push

github

Committed by

web-flow

Commit Message

Misc modernizations and improvments (#896)

* fixed_allocator cleanup
template argument is now the type and not the size
general cleanup, rewrote refill loop to better understand whats
happening
possibility to use different alignments (not needed for the used types)

* adapted fixed_allocator usage

* simplified message_queue conditial wait, removed unused header

* use moveonlyfunction for logging queue

* cleanup and performance improvment for clienttransmit

* use moveonlyfunction for sql

* fixed use after move

* TaskThreadPool uses moveonlyfunction

* catch exception locally without stopping the complete workerthread

* message_queue pop with timeout

* DynTaskThreadPool changes

- using move_only_function
- general simplification
- minimum of 4 threads
- added idle timeout to get rid of workers

* simplified auxpoolthread usage

* fixed threadmap initialization

* to avoid static initialization problem dont prefill worker threada in
the constructor
update macos to latest which hopefully supports jthreads

* removed debug logs

* seems macos-latest is not always 26, explicitly use it

* added jitter for timeout
secured the logic when to spawn a new thread more
busyguard is now used to track a message compare value against
live_threads to decide if a new thread should be spawned
unique name per poolworker

Coverage Stats

199 of 248 new or added lines in 15 files covered. (80.24%)

1 existing line in 1 file now uncovered.

45056 of 73359 relevant lines covered (61.42%)

634055.61 hits per line

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

86.61

/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()	96✔
45	{
46	static ThreadMap threadmap;	96✔
47	return threadmap;	96✔
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 )	64✔
113	{
114	usleep( millis * 1000L );	64✔
115	}	64✔
116	size_t thread_pid()	99,475,756✔
117	{
118	#ifdef __APPLE__
119	return reinterpret_cast<size_t>( pthread_self() );
120	#else
121	return pthread_self();	99,475,756✔
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 )	49✔
293	{
294	Clib::SpinLockGuard guard( _spinlock );	49✔
295	_contents.insert( std::make_pair( pid, name ) );	49✔
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	}	49✔
307	void ThreadMap::Unregister( size_t pid )	47✔
308	{
309	Clib::SpinLockGuard guard( _spinlock );	47✔
310	_contents.erase( pid );	47✔
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	}	47✔
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 )	23✔
325	{
326	threadmap_instance().Register( thread_pid(), name );	23✔
327	}	23✔
328	ThreadRegister::~ThreadRegister()	23✔
329	{
330	threadmap_instance().Unregister( thread_pid() );	23✔
331	}	23✔
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	{
NEW 377	ERROR_PRINTLN( "Thread exception: {}", ex.what() );	×
NEW 378	Clib::force_backtrace( true );	×
NEW 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	{
NEW 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 )	13✔
496	: _name( name ),	13✔
497	_thread(),	13✔
498	_parent( parent ),	13✔
499	_retired( false ),	13✔
500	_timeout( ( parent->_worker_count % JITTER_STEPS ) * TIMEOUT_JITTER )	13✔
501	{
502	run();	13✔
503	}	13✔
504
505	bool DynTaskThreadPool::PoolWorker::isretired() const	1,235✔
506	{
507	return _retired.load( std::memory_order_relaxed );	1,235✔
508	}
509
510	void DynTaskThreadPool::PoolWorker::run()	13✔
511	{
512	_thread = std::jthread(	13✔
513	[&]()	13✔
514	{
515	ThreadRegister register_thread( _name );	13✔
516	ERROR_PRINTLN( "created pool worker {}", _name );	13✔
517	auto f = msg();	13✔
518	try
519	{
520	while ( !_parent->_done && !Clib::exit_signalled )	289✔
521	{
522	if ( !_parent->_msg_queue.pop_wait_for( &f, _timeout ) )	285✔
523	{
524	std::lock_guard<std::mutex> guard( _parent->_pool_mutex );	1✔
525	// we timed out: can we retire?
526	if ( _parent->_live_threads > MIN_WORKER )	1✔
527	{
NEW 528	--_parent->_live_threads;	×
NEW 529	_retired.store( true, std::memory_order_relaxed );	×
NEW 530	ERROR_PRINTLN( "removed pool worker {}", _name );	×
NEW 531	return;	×
532	}
533	continue; // keep alive	1✔
534	}	1✔
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& )	9✔
549	{
550	}	9✔
551	} );	26✔
552	}	13✔
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,511✔
589	// still atleast one idle worker left?
590	if ( _busy_count.load( std::memory_order_relaxed ) <= _live_threads )	552✔
591	return;	271✔
592
593	_threads.emplace_back( new PoolWorker( this, fmt::format( "{} {}", _name, _worker_count++ ) ) );	10✔
594	++_live_threads;	5✔
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 / 29352231943

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