16683711876

Committed 01 Aug 2025 07:40PM UTC coverage: 59.756% (-0.03%) from 59.79%

Build # 16683711876

Build Type

push

github

Committed by

web-flow

Commit Message

use c++20 (#799)

* use c++20
* increased used clang version
* compiler report and logfacility use now compile time formatting, which
means that the formatstring gets checked at compile time. (which found 2 errors)
adapted a few places since report only accepts formatting arguments
adapted a few places with logging since char[] is compile time
formatting and string or chr* is runtime formatting
* use std::ranges instead of boost
* disabled pragma_assume vs specific macro (I guess noone cares)
* needed to fix ancient ms exception code
* modernized SpinLock
* removed unused code in ECompile
* replaced std::filesystem::path::u8string with string. It now returns an actual u8string type
* cleanup layers.h added the defines for other layers which where before
  only defined in the pkt
* osmod::OpenConnection and HTTPRequest cleanup: early outs, dont check for pChild which is
  only needed for startscript and placed suspend at the very last
  position

* fix warning

* rebuild cache with new compiler version

* define c++ standard for external libs where possible

* added fixme

Run Details

156 of 212 new or added lines in 26 files covered. (73.58%)

45 existing lines in 19 files now uncovered.

43621 of 72999 relevant lines covered (59.76%)

409874.73 hits per line

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

76.29

/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 <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
{
namespace threadhelp
{
ThreadMap 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 )( void ) )
{
  // 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.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.Unregister( thread_pid() );
}

class ThreadData
{
public:
  std::string name;
  void ( *entry )( void* );
  void ( *entry_noparam )( void );
  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 )( void ) = td->entry_noparam;
  void* arg = td->arg;

  threadmap.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 )( void ), 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.Register( thread_pid(), name );
}
ThreadRegister::~ThreadRegister()
{
  threadmap.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 )
            {
              _msg_queue.pop_wait( &f );
              f();
            }
          }
          catch ( msg_queue::Canceled& )
          {
          }
          catch ( std::exception& ex )
          {
            ERROR_PRINTLN( "Thread exception: {}", ex.what() );
            Clib::force_backtrace( true );
            return;
          }
          // 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(
      [&]()
      {
        _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( const msg& msg )
{
  _msg_queue.push( msg );
}

/// returns a future which will be set once the msg is processed
std::future<bool> TaskThreadPool::checked_push( const msg& msg )
{
  auto promise = std::make_shared<std::promise<bool>>();
  auto ret = promise->get_future();
  _msg_queue.push(
      [=]()
      {
        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;
  bool isbusy() const;
  void join();
  void run();

private:
  std::string _name;
  std::atomic<bool> _busy;
  std::thread _thread;
  DynTaskThreadPool* _parent;
  struct BusyGuard
  {
    std::atomic<bool>* _busy;
    BusyGuard( std::atomic<bool>* busy ) : _busy( busy ) { ( *_busy ) = true; }
    ~BusyGuard() { ( *_busy ) = false; }
  };
};
DynTaskThreadPool::PoolWorker::PoolWorker( DynTaskThreadPool* parent, const std::string& name )
    : _name( name ), _busy( false ), _thread(), _parent( parent )
{
  run();
}
bool DynTaskThreadPool::PoolWorker::isbusy() const
{
  return _busy;
}

void DynTaskThreadPool::PoolWorker::join()
{
  _thread.join();
}

void DynTaskThreadPool::PoolWorker::run()
{
  _thread = std::thread(
      [&]()
      {
        ThreadRegister register_thread( _name );
        auto f = msg();
        try
        {
          while ( !_parent->_done && !Clib::exit_signalled )
          {
            _parent->_msg_queue.pop_wait( &f );
            {
              BusyGuard busy( &_busy );
              f();
            }
          }
        }
        catch ( msg_queue::Canceled& )
        {
        }
        catch ( std::exception& ex )
        {
          ERROR_PRINTLN( "Thread exception: {}", ex.what() );
          Clib::force_backtrace( true );
          return;
        }
      } );
}

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

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

void DynTaskThreadPool::create_thread()
{
  std::lock_guard<std::mutex> guard( _pool_mutex );
  for ( const auto& worker : _threads )
  {
    if ( !worker->isbusy() )  // check for a idle instance
    {
      return;
    }
  }
  size_t thread_num = _threads.size();
  _threads.emplace_back( new PoolWorker( this, _name + " " + std::to_string( thread_num ) ) );
  ERROR_PRINTLN( "create pool worker {} {}", _name, thread_num );
}

DynTaskThreadPool::~DynTaskThreadPool()
{
  // send both done and cancel to wake up all workers
  _msg_queue.push(
      [&]()
      {
        _done = true;
        _msg_queue.cancel();
      } );
  for ( auto& thread : _threads )
    thread->join();
}

/// simply fire and forget only the deconstructor ensures the msg to be finished
void DynTaskThreadPool::push( const msg& msg )
{
  create_thread();
  _msg_queue.push( msg );
}

/// returns a future which will be set once the msg is processed
std::future<bool> DynTaskThreadPool::checked_push( const msg& msg )
{
  auto promise = std::make_shared<std::promise<bool>>();
  auto ret = promise->get_future();
  create_thread();
  _msg_queue.push(
      [=]()
      {
        try
        {
          msg();
          promise->set_value( true );
        }
        catch ( ... )
        {
          promise->set_exception( std::current_exception() );
        }
      } );
  return ret;
}
}  // namespace threadhelp
}  // namespace Pol

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

polserver / polserver / 16683711876

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