ode/src/threading_pool_win.cpp

   1 /*************************************************************************
   2  *                                                                       *
   3  * Open Dynamics Engine, Copyright (C) 2001-2003 Russell L. Smith.       *
   4  * All rights reserved.  Email: russ@q12.org   Web: www.q12.org          *
   5  *                                                                       *
   6  * Threading Windows thread pool implementation file.                    *
   7  * Copyright (C) 2011-2024 Oleh Derevenko. All rights reserved.          *
   8  * e-mail: odar@eleks.com (change all "a" to "e")                        *
   9  *                                                                       *
  10  * This library is free software; you can redistribute it and/or         *
  11  * modify it under the terms of EITHER:                                  *
  12  *   (1) The GNU Lesser General Public License as published by the Free  *
  13  *       Software Foundation; either version 2.1 of the License, or (at  *
  14  *       your option) any later version. The text of the GNU Lesser      *
  15  *       General Public License is included with this library in the     *
  16  *       file LICENSE.TXT.                                               *
  17  *   (2) The BSD-style license that is included with this library in     *
  18  *       the file LICENSE-BSD.TXT.                                       *
  19  *                                                                       *
  20  * This library is distributed in the hope that it will be useful,       *
  21  * but WITHOUT ANY WARRANTY; without even the implied warranty of        *
  22  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the files    *
  23  * LICENSE.TXT and LICENSE-BSD.TXT for more details.                     *
  24  *                                                                       *
  25  *************************************************************************/
  26
  27 /*
  28  *  Windows thread pool implementation for built-in threading support provider.
  29  */
  30
  31
  32 #if defined(_WIN32)
  33
  34 #include <ode/odeconfig.h>
  35 #include <ode/error.h>
  36 #include <ode/threading_impl.h>
  37 #include <ode/odeinit.h>
  38 #include "config.h"
  39 #include "objects.h"
  40 #include "threading_impl_templates.h"
  41
  42
  43 #if dBUILTIN_THREADING_IMPL_ENABLED
  44
  45 #include <Windows.h>
  46 #include <process.h>
  47 #include <new>
  48
  49
  50 #endif // #if dBUILTIN_THREADING_IMPL_ENABLED
  51
  52
  53 #if dBUILTIN_THREADING_IMPL_ENABLED
  54
  55 #define THREAD_STACK_MAX  ((sizeint)(UINT_MAX - 1)) // The absolute maximum would be UINT_MAX but let it be a little bit less to avoid "Comparison is always false" warnings. ;)
  56
  57
  58 struct dxEventObject
  59 {
  60 public:
  61     dxEventObject(): m_event_handle(NULL) {}
  62     ~dxEventObject() { FinalizeObject(); }
  63
  64     bool InitializeObject(bool manual_reset, bool initial_state);
  65     void FinalizeObject();
  66
  67     bool WaitInfinitely() { return ::WaitForSingleObject(m_event_handle, INFINITE) == WAIT_OBJECT_0; }
  68     void SetEvent();
  69     void ResetEvent();
  70
  71 private:
  72     HANDLE        m_event_handle;
  73 };
  74
  75 bool dxEventObject::InitializeObject(bool manual_reset, bool initial_state)
  76 {
  77     dIASSERT(m_event_handle == NULL);
  78
  79     bool result = false;
  80
  81     do
  82     {
  83         HANDLE event_handle = ::CreateEvent(NULL, manual_reset, initial_state, NULL);
  84         if (event_handle == NULL)
  85         {
  86             break;
  87         }
  88
  89         m_event_handle = event_handle;
  90         result = true;
  91     }
  92     while (false);
  93
  94     return result;
  95 }
  96
  97 void dxEventObject::FinalizeObject()
  98 {
  99     HANDLE event_handle = m_event_handle;
 100     if (event_handle != NULL)
 101     {
 102         BOOL close_result = ::CloseHandle(event_handle);
 103         dICHECK(close_result); // Object destruction should always succeed
 104
 105         m_event_handle = NULL;
 106     }
 107 }
 108
 109 void dxEventObject::SetEvent()
 110 {
 111     BOOL set_result = ::SetEvent(m_event_handle);
 112     dICHECK(set_result);
 113 }
 114
 115 void dxEventObject::ResetEvent()
 116 {
 117     BOOL reset_result = ::ResetEvent(m_event_handle);
 118     dICHECK(reset_result);
 119 }
 120
 121
 122
 123 struct dxThreadPoolThreadInfo
 124 {
 125 public:
 126     dxThreadPoolThreadInfo();
 127     ~dxThreadPoolThreadInfo();
 128
 129     bool Initialize(sizeint stack_size, unsigned int ode_data_allocate_flags);
 130
 131 private:
 132     bool WaitInitStatus();
 133
 134 private:
 135     void Finalize();
 136     void WaitAndCloseThreadHandle(HANDLE thread_handle);
 137
 138 public:
 139     enum dxTHREADCOMMAND
 140     {
 141         dxTHREAD_COMMAND_EXIT,
 142         dxTHREAD_COMMAND_NOOP,
 143         dxTHREAD_COMMAND_SERVE_IMPLEMENTATION,
 144     };
 145
 146     struct dxServeImplementationParams
 147     {
 148         dxServeImplementationParams(dThreadingImplementationID impl, dxEventObject *ready_wait_event):
 149     m_impl(impl), m_ready_wait_event(ready_wait_event)
 150     {
 151     }
 152
 153     dThreadingImplementationID m_impl;
 154     dxEventObject *m_ready_wait_event;
 155     };
 156
 157     void ExecuteThreadCommand(dxTHREADCOMMAND command, void *param, bool wait_response);
 158
 159 private:
 160     static unsigned CALLBACK ThreadProcedure_Callback(void *thread_param);
 161     void ThreadProcedure();
 162     void ReportInitStatus(bool init_result);
 163     void RunCommandHandlingLoop();
 164
 165     void ThreadedServeImplementation(dThreadingImplementationID impl, dxEventObject *ready_wait_event);
 166     static void ProcessThreadServeReadiness_Callback(void *context);
 167
 168 private:
 169     HANDLE      m_thread_handle;
 170
 171     unsigned int m_ode_data_allocate_flags;
 172     dxTHREADCOMMAND m_command_code;
 173     dxEventObject m_command_event;
 174     dxEventObject m_acknowledgement_event;
 175     void        *m_command_param;
 176 };
 177
 178
 179 dxThreadPoolThreadInfo::dxThreadPoolThreadInfo():
 180 m_thread_handle(NULL),
 181 m_ode_data_allocate_flags(0),
 182 m_command_code(dxTHREAD_COMMAND_EXIT),
 183 m_command_event(),
 184 m_acknowledgement_event(),
 185 m_command_param(NULL)
 186 {
 187 }
 188
 189 dxThreadPoolThreadInfo::~dxThreadPoolThreadInfo()
 190 {
 191     Finalize();
 192 }
 193
 194
 195 bool dxThreadPoolThreadInfo::Initialize(sizeint stack_size, unsigned int ode_data_allocate_flags)
 196 {
 197     bool result = false;
 198
 199     bool command_event_allocated = false, acknowledgement_event_allocated = false;
 200
 201     HANDLE thread_handle = NULL;
 202
 203     do
 204     {
 205         if (stack_size > THREAD_STACK_MAX)
 206         {
 207             SetLastError(ERROR_INVALID_PARAMETER);
 208             break;
 209         }
 210
 211         if (!m_command_event.InitializeObject(false, false))
 212         {
 213             break;
 214         }
 215
 216         command_event_allocated = true;
 217
 218         if (!m_acknowledgement_event.InitializeObject(true, false))
 219         {
 220             break;
 221         }
 222
 223         acknowledgement_event_allocated = true;
 224
 225         m_ode_data_allocate_flags = ode_data_allocate_flags;
 226
 227         thread_handle = (HANDLE)_beginthreadex(NULL, (unsigned)stack_size, &ThreadProcedure_Callback, (void *)this, 0, NULL);
 228         if (thread_handle == NULL) // Not a bug!!! _beginthreadex() returns NULL on failure
 229         {
 230             break;
 231         }
 232
 233         // It is OK to alter priority for thread without creating it in suspended state as
 234         // it is anyway going to be waited for (waited for its init result) and
 235         // will not be issues commands until after that.
 236         int own_priority = GetThreadPriority(GetCurrentThread());
 237         if (own_priority != THREAD_PRIORITY_ERROR_RETURN)
 238         {
 239             if (!SetThreadPriority(thread_handle, own_priority))
 240             {
 241                 // own_priority = THREAD_PRIORITY_ERROR_RETURN; -- Well, if priority inheritance fails - just ignore it :-/
 242             }
 243         }
 244
 245         bool thread_init_result = WaitInitStatus();
 246         if (!thread_init_result)
 247         {
 248             DWORD error_save = GetLastError();
 249             WaitAndCloseThreadHandle(thread_handle);
 250             SetLastError(error_save);
 251             break;
 252         }
 253
 254         m_thread_handle = thread_handle;
 255         result = true;
 256     }
 257     while (false);
 258
 259     if (!result)
 260     {
 261         if (command_event_allocated)
 262         {
 263             if (acknowledgement_event_allocated)
 264             {
 265                 m_acknowledgement_event.FinalizeObject();
 266             }
 267
 268             m_command_event.FinalizeObject();
 269         }
 270     }
 271
 272     return result;
 273 }
 274
 275 bool dxThreadPoolThreadInfo::WaitInitStatus()
 276 {
 277     bool acknowledgement_wait_result = m_acknowledgement_event.WaitInfinitely();
 278     dICHECK(acknowledgement_wait_result);
 279
 280     DWORD error_code = (DWORD)(sizeint)m_command_param;
 281
 282     bool init_status = error_code == ERROR_SUCCESS ? true : (SetLastError(error_code), false);
 283     return init_status;
 284 }
 285
 286 void dxThreadPoolThreadInfo::Finalize()
 287 {
 288     HANDLE thread_handle = m_thread_handle;
 289     if (thread_handle != NULL)
 290     {
 291         ExecuteThreadCommand(dxTHREAD_COMMAND_EXIT, NULL, false);
 292
 293         WaitAndCloseThreadHandle(thread_handle);
 294         m_thread_handle = NULL;
 295
 296         m_command_event.FinalizeObject();
 297         m_acknowledgement_event.FinalizeObject();
 298     }
 299 }
 300
 301 void dxThreadPoolThreadInfo::WaitAndCloseThreadHandle(HANDLE thread_handle)
 302 {
 303     DWORD thread_wait_result = WaitForSingleObject(thread_handle, INFINITE);
 304     dICHECK(thread_wait_result == WAIT_OBJECT_0);
 305
 306     BOOL thread_close_result = CloseHandle(thread_handle);
 307     dIVERIFY(thread_close_result);
 308
 309 }
 310
 311 void dxThreadPoolThreadInfo::ExecuteThreadCommand(dxTHREADCOMMAND command, void *param, bool wait_response)
 312 {
 313     bool acknowledgement_wait_result = m_acknowledgement_event.WaitInfinitely();
 314     dICHECK(acknowledgement_wait_result);
 315
 316     m_acknowledgement_event.ResetEvent();
 317
 318     m_command_code = command;
 319     m_command_param = param;
 320
 321     m_command_event.SetEvent();
 322
 323     if (wait_response)
 324     {
 325         bool new_acknowledgement_wait_result = m_acknowledgement_event.WaitInfinitely();
 326         dICHECK(new_acknowledgement_wait_result);
 327     }
 328 }
 329
 330 unsigned CALLBACK dxThreadPoolThreadInfo::ThreadProcedure_Callback(void *thread_param)
 331 {
 332     dxThreadPoolThreadInfo *thread_info = (dxThreadPoolThreadInfo *)thread_param;
 333     thread_info->ThreadProcedure();
 334
 335     return 0;
 336 }
 337
 338 void dxThreadPoolThreadInfo::ThreadProcedure()
 339 {
 340     bool init_result = dAllocateODEDataForThread(m_ode_data_allocate_flags) != 0;
 341
 342     ReportInitStatus(init_result);
 343
 344     if (init_result)
 345     {
 346         RunCommandHandlingLoop();
 347
 348         // dCleanupODEAllDataForThread(); -- this function can only be called if ODE was initialized for manual cleanup. And that is unknown here...
 349     }
 350 }
 351
 352 void dxThreadPoolThreadInfo::ReportInitStatus(bool init_result)
 353 {
 354     DWORD error_code;
 355     m_command_param = (void *)(sizeint)(init_result ? ERROR_SUCCESS : ((error_code = GetLastError()) != ERROR_SUCCESS ? error_code : ERROR_INTERNAL_ERROR));
 356
 357     m_acknowledgement_event.SetEvent();
 358 }
 359
 360 void dxThreadPoolThreadInfo::RunCommandHandlingLoop()
 361 {
 362     bool exit_requested = false;
 363
 364     while (!exit_requested)
 365     {
 366         bool command_wait_result = m_command_event.WaitInfinitely();
 367         dICHECK(command_wait_result);
 368
 369         const dxTHREADCOMMAND command_code = m_command_code;
 370         switch (command_code)
 371         {
 372             case dxTHREAD_COMMAND_EXIT:
 373             {
 374                 m_acknowledgement_event.SetEvent();
 375
 376                 exit_requested = true;
 377                 break;
 378             }
 379
 380             default:
 381             {
 382                 dIASSERT(false);
 383                 // break; -- proceed to case dxTHREAD_COMMAND_NOOP
 384             }
 385
 386             case dxTHREAD_COMMAND_NOOP:
 387             {
 388                 m_acknowledgement_event.SetEvent();
 389
 390                 // Do nothing
 391                 break;
 392             }
 393
 394             case dxTHREAD_COMMAND_SERVE_IMPLEMENTATION:
 395             {
 396                 const dxServeImplementationParams *serve_params = (const dxServeImplementationParams *)m_command_param;
 397                 dThreadingImplementationID impl = serve_params->m_impl;
 398                 dxEventObject *ready_wait_event = serve_params->m_ready_wait_event;
 399
 400                 m_acknowledgement_event.SetEvent();
 401
 402                 ThreadedServeImplementation(impl, ready_wait_event);
 403                 break;
 404             }
 405         }
 406     }
 407 }
 408
 409 void dxThreadPoolThreadInfo::ThreadedServeImplementation(dThreadingImplementationID impl, dxEventObject *ready_wait_event)
 410 {
 411     ((dxIThreadingImplementation *)impl)->StickToJobsProcessing(&ProcessThreadServeReadiness_Callback, (void *)ready_wait_event);
 412 }
 413
 414 void dxThreadPoolThreadInfo::ProcessThreadServeReadiness_Callback(void *context)
 415 {
 416     dxEventObject *ready_wait_event = (dxEventObject *)context;
 417
 418     ready_wait_event->SetEvent();
 419 }
 420
 421
 422
 423 struct dxThreadingThreadPool:
 424     public dBase
 425 {
 426 public:
 427     dxThreadingThreadPool();
 428     ~dxThreadingThreadPool();
 429
 430     bool InitializeThreads(sizeint thread_count, sizeint stack_size, unsigned int ode_data_allocate_flags);
 431
 432 private:
 433     void FinalizeThreads();
 434
 435     bool InitializeIndividualThreadInfos(dxThreadPoolThreadInfo *thread_infos, sizeint thread_count, sizeint stack_size, unsigned int ode_data_allocate_flags);
 436     void FinalizeIndividualThreadInfos(dxThreadPoolThreadInfo *thread_infos, sizeint thread_count);
 437
 438     bool InitializeSingleThreadInfo(dxThreadPoolThreadInfo *thread_info, sizeint stack_size, unsigned int ode_data_allocate_flags);
 439     void FinalizeSingleThreadInfo(dxThreadPoolThreadInfo *thread_info);
 440
 441 public:
 442     void ServeThreadingImplementation(dThreadingImplementationID impl);
 443     void WaitIdleState();
 444
 445 private:
 446     dxThreadPoolThreadInfo  *m_thread_infos;
 447     sizeint                  m_thread_count;
 448     dxEventObject           m_ready_wait_event;
 449 };
 450
 451
 452 dxThreadingThreadPool::dxThreadingThreadPool():
 453 m_thread_infos(NULL),
 454 m_thread_count(0),
 455 m_ready_wait_event()
 456 {
 457 }
 458
 459 dxThreadingThreadPool::~dxThreadingThreadPool()
 460 {
 461     FinalizeThreads();
 462 }
 463
 464
 465 bool dxThreadingThreadPool::InitializeThreads(sizeint thread_count, sizeint stack_size, unsigned int ode_data_allocate_flags)
 466 {
 467     dIASSERT(m_thread_infos == NULL);
 468
 469     bool result = false;
 470
 471     bool wait_event_allocated = false;
 472
 473     dxThreadPoolThreadInfo *thread_infos = NULL;
 474     bool thread_infos_allocated = false;
 475
 476     do
 477     {
 478         if (!m_ready_wait_event.InitializeObject(false, false))
 479         {
 480             break;
 481         }
 482
 483         wait_event_allocated = true;
 484
 485         thread_infos = (dxThreadPoolThreadInfo *)dAlloc(thread_count * sizeof(dxThreadPoolThreadInfo));
 486         if (thread_infos == NULL)
 487         {
 488             break;
 489         }
 490
 491         thread_infos_allocated = true;
 492
 493         if (!InitializeIndividualThreadInfos(thread_infos, thread_count, stack_size, ode_data_allocate_flags))
 494         {
 495             break;
 496         }
 497
 498         m_thread_infos = thread_infos;
 499         m_thread_count = thread_count;
 500         result = true;
 501     }
 502     while (false);
 503
 504     if (!result)
 505     {
 506         if (wait_event_allocated)
 507         {
 508             if (thread_infos_allocated)
 509             {
 510                 dFree(thread_infos, thread_count * sizeof(dxThreadPoolThreadInfo));
 511             }
 512
 513             m_ready_wait_event.FinalizeObject();
 514         }
 515     }
 516
 517     return result;
 518 }
 519
 520 void dxThreadingThreadPool::FinalizeThreads()
 521 {
 522     dxThreadPoolThreadInfo *thread_infos = m_thread_infos;
 523     if (thread_infos != NULL)
 524     {
 525         sizeint thread_count = m_thread_count;
 526
 527         FinalizeIndividualThreadInfos(thread_infos, thread_count);
 528         dFree(thread_infos, thread_count * sizeof(dxThreadPoolThreadInfo));
 529
 530         m_ready_wait_event.FinalizeObject();
 531     }
 532 }
 533
 534
 535 bool dxThreadingThreadPool::InitializeIndividualThreadInfos(dxThreadPoolThreadInfo *thread_infos, sizeint thread_count, sizeint stack_size, unsigned int ode_data_allocate_flags)
 536 {
 537     bool any_fault = false;
 538
 539     dxThreadPoolThreadInfo *const infos_end = thread_infos + thread_count;
 540     for (dxThreadPoolThreadInfo *current_info = thread_infos; current_info != infos_end; ++current_info)
 541     {
 542         if (!InitializeSingleThreadInfo(current_info, stack_size, ode_data_allocate_flags))
 543         {
 544             FinalizeIndividualThreadInfos(thread_infos, current_info - thread_infos);
 545
 546             any_fault = true;
 547             break;
 548         }
 549     }
 550
 551     bool result = !any_fault;
 552     return result;
 553 }
 554
 555 void dxThreadingThreadPool::FinalizeIndividualThreadInfos(dxThreadPoolThreadInfo *thread_infos, sizeint thread_count)
 556 {
 557     dxThreadPoolThreadInfo *const infos_end = thread_infos + thread_count;
 558     for (dxThreadPoolThreadInfo *current_info = thread_infos; current_info != infos_end; ++current_info)
 559     {
 560         FinalizeSingleThreadInfo(current_info);
 561     }
 562 }
 563
 564
 565 bool dxThreadingThreadPool::InitializeSingleThreadInfo(dxThreadPoolThreadInfo *thread_info, sizeint stack_size, unsigned int ode_data_allocate_flags)
 566 {
 567     bool result = false;
 568
 569     new(thread_info) dxThreadPoolThreadInfo();
 570
 571     if (thread_info->Initialize(stack_size, ode_data_allocate_flags))
 572     {
 573         result = true;
 574     }
 575     else
 576     {
 577         thread_info->dxThreadPoolThreadInfo::~dxThreadPoolThreadInfo();
 578     }
 579
 580     return result;
 581 }
 582
 583 void dxThreadingThreadPool::FinalizeSingleThreadInfo(dxThreadPoolThreadInfo *thread_info)
 584 {
 585     if (thread_info != NULL)
 586     {
 587         thread_info->dxThreadPoolThreadInfo::~dxThreadPoolThreadInfo();
 588     }
 589 }
 590
 591
 592 void dxThreadingThreadPool::ServeThreadingImplementation(dThreadingImplementationID impl)
 593 {
 594     dxThreadPoolThreadInfo::dxServeImplementationParams params(impl, &m_ready_wait_event);
 595
 596     dxThreadPoolThreadInfo *const infos_end = m_thread_infos + m_thread_count;
 597     for (dxThreadPoolThreadInfo *current_info = m_thread_infos; current_info != infos_end; ++current_info)
 598     {
 599         current_info->ExecuteThreadCommand(dxThreadPoolThreadInfo::dxTHREAD_COMMAND_SERVE_IMPLEMENTATION, &params, true);
 600
 601         bool ready_wait_result = m_ready_wait_event.WaitInfinitely();
 602         dICHECK(ready_wait_result);
 603     }
 604 }
 605
 606 void dxThreadingThreadPool::WaitIdleState()
 607 {
 608     dxThreadPoolThreadInfo *const infos_end = m_thread_infos + m_thread_count;
 609     for (dxThreadPoolThreadInfo *current_info = m_thread_infos; current_info != infos_end; ++current_info)
 610     {
 611         current_info->ExecuteThreadCommand(dxThreadPoolThreadInfo::dxTHREAD_COMMAND_NOOP, NULL, true);
 612     }
 613 }
 614
 615
 616 #endif // #if dBUILTIN_THREADING_IMPL_ENABLED
 617
 618
 619 /*extern */dThreadingThreadPoolID dThreadingAllocateThreadPool(unsigned thread_count,
 620                                                                sizeint stack_size, unsigned int ode_data_allocate_flags, void *reserved/*=NULL*/)
 621 {
 622     dAASSERT(thread_count != 0);
 623
 624 #if dBUILTIN_THREADING_IMPL_ENABLED
 625     dxThreadingThreadPool *thread_pool = new dxThreadingThreadPool();
 626     if (thread_pool != NULL)
 627     {
 628         if (thread_pool->InitializeThreads(thread_count, stack_size, ode_data_allocate_flags))
 629         {
 630             // do nothing
 631         }
 632         else
 633         {
 634             delete thread_pool;
 635             thread_pool = NULL;
 636         }
 637     }
 638 #else
 639     dThreadingThreadPoolID thread_pool = NULL;
 640 #endif // #if dBUILTIN_THREADING_IMPL_ENABLED
 641
 642     return (dThreadingThreadPoolID)thread_pool;
 643 }
 644
 645 /*extern */void dThreadingThreadPoolServeMultiThreadedImplementation(dThreadingThreadPoolID pool, dThreadingImplementationID impl)
 646 {
 647 #if dBUILTIN_THREADING_IMPL_ENABLED
 648     dxThreadingThreadPool *thread_pool = (dxThreadingThreadPool *)pool;
 649     thread_pool->ServeThreadingImplementation(impl);
 650 #endif // #if dBUILTIN_THREADING_IMPL_ENABLED
 651 }
 652
 653 /*extern */void dThreadingThreadPoolWaitIdleState(dThreadingThreadPoolID pool)
 654 {
 655 #if dBUILTIN_THREADING_IMPL_ENABLED
 656     dxThreadingThreadPool *thread_pool = (dxThreadingThreadPool *)pool;
 657     thread_pool->WaitIdleState();
 658 #endif // #if dBUILTIN_THREADING_IMPL_ENABLED
 659 }
 660
 661 /*extern */void dThreadingFreeThreadPool(dThreadingThreadPoolID pool)
 662 {
 663 #if dBUILTIN_THREADING_IMPL_ENABLED
 664     dxThreadingThreadPool *thread_pool = (dxThreadingThreadPool *)pool;
 665     delete thread_pool;
 666 #endif // #if dBUILTIN_THREADING_IMPL_ENABLED
 667 }
 668
 669
 670 #endif // #if defined(_WIN32)