winsup/cygserver/threaded_queue.cc

   1 /* threaded_queue.cc
   2
   3    Written by Robert Collins <rbtcollins@hotmail.com>
   4
   5 This file is part of Cygwin.
   6
   7 This software is a copyrighted work licensed under the terms of the
   8 Cygwin license.  Please consult the file "CYGWIN_LICENSE" for
   9 details. */
  10
  11 #ifdef __OUTSIDE_CYGWIN__
  12 #include "woutsup.h"
  13
  14 #include <assert.h>
  15 #include <errno.h>
  16 #include <stdio.h>
  17 #include <unistd.h>
  18 #include <sys/types.h>
  19 #include <stdlib.h>
  20 #include "threaded_queue.h"
  21
  22 /*****************************************************************************/
  23
  24 /* queue_request */
  25
  26 queue_request::~queue_request ()
  27 {}
  28
  29 /*****************************************************************************/
  30
  31 /* threaded_queue */
  32
  33 threaded_queue::threaded_queue (const size_t initial_workers)
  34   : _workers_count (0),
  35     _workers_busy (0),
  36     _running (false),
  37     _submitters_head (NULL),
  38     _requests_count (0),
  39     _requests_head (NULL),
  40     _requests_sem (NULL)
  41 {
  42   InitializeCriticalSection (&_queue_lock);
  43
  44   // This semaphore's count is the number of requests on the queue.
  45   // The maximum count (129792) is calculated as MAXIMUM_WAIT_OBJECTS
  46   // multiplied by max. threads per process (2028?), which is (a few)
  47   // more requests than could ever be pending with the current design.
  48
  49   _requests_sem = CreateSemaphore (NULL,   // SECURITY_ATTRIBUTES
  50                                    0,      // Initial count
  51                                    129792, // Maximum count
  52                                    NULL);  // Anonymous
  53
  54   if (!_requests_sem)
  55     {
  56       system_printf (("failed to create the request queue semaphore, "
  57                       "error = %u"),
  58                      GetLastError ());
  59       abort ();
  60     }
  61
  62   create_workers (initial_workers);
  63 }
  64
  65 threaded_queue::~threaded_queue ()
  66 {
  67   if (_running)
  68     stop ();
  69
  70   debug_printf ("deleting all pending queue requests");
  71   queue_request *reqptr = _requests_head;
  72   while (reqptr)
  73     {
  74       queue_request *const ptr = reqptr;
  75       reqptr = reqptr->_next;
  76       delete ptr;
  77     }
  78
  79   DeleteCriticalSection (&_queue_lock);
  80   if (_requests_sem)
  81     (void) CloseHandle (_requests_sem);
  82 }
  83
  84 /* FIXME: return success or failure rather than quitting */
  85 void
  86 threaded_queue::add_submission_loop (queue_submission_loop *const submitter)
  87 {
  88   assert (submitter);
  89   assert (submitter->_queue == this);
  90   assert (!submitter->_next);
  91
  92   submitter->_next =
  93     TInterlockedExchangePointer (&_submitters_head, submitter);
  94
  95   if (_running)
  96     submitter->start ();
  97 }
  98
  99 bool
 100 threaded_queue::start ()
 101 {
 102   EnterCriticalSection (&_queue_lock);
 103   const bool was_running = _running;
 104   _running = true;
 105   queue_submission_loop *loopptr = _submitters_head;
 106   LeaveCriticalSection (&_queue_lock);
 107
 108   if (!was_running)
 109     {
 110       debug_printf ("starting all queue submission loops");
 111
 112       while (loopptr)
 113         {
 114           queue_submission_loop *const ptr = loopptr;
 115           loopptr = loopptr->_next;
 116           ptr->start ();
 117         }
 118     }
 119
 120   return was_running;
 121 }
 122
 123 bool
 124 threaded_queue::stop ()
 125 {
 126   EnterCriticalSection (&_queue_lock);
 127   const bool was_running = _running;
 128   _running = false;
 129   queue_submission_loop *loopptr = _submitters_head;
 130   LeaveCriticalSection (&_queue_lock);
 131
 132   if (was_running)
 133     {
 134       debug_printf ("stopping all queue submission loops");
 135       while (loopptr)
 136         {
 137           queue_submission_loop *const ptr = loopptr;
 138           loopptr = loopptr->_next;
 139           ptr->stop ();
 140         }
 141
 142       ReleaseSemaphore (_requests_sem, _workers_count, NULL);
 143       while (_workers_count)
 144         {
 145           debug_printf (("waiting for worker threads to terminate: "
 146                          "%u still running"),
 147                         _workers_count);
 148           Sleep (1000);
 149         }
 150       debug_printf ("all worker threads have terminated");
 151     }
 152
 153   return was_running;
 154 }
 155
 156 /* FIXME: return success or failure */
 157 void
 158 threaded_queue::add (queue_request *const therequest)
 159 {
 160   assert (therequest);
 161   assert (!therequest->_next);
 162
 163   EnterCriticalSection (&_queue_lock);
 164   if (!_requests_head)
 165     _requests_head = therequest;
 166   else
 167     {
 168       /* Add to the queue end. */
 169       queue_request *reqptr = _requests_head;
 170       for (; reqptr->_next; reqptr = reqptr->_next)
 171         {}
 172       assert (reqptr);
 173       assert (!reqptr->_next);
 174       reqptr->_next = therequest;
 175     }
 176
 177   _requests_count += 1;
 178   assert (_requests_count > 0);
 179   LeaveCriticalSection (&_queue_lock);
 180
 181   (void) ReleaseSemaphore (_requests_sem, 1, NULL);
 182
 183   if (_workers_busy >= _workers_count)
 184     {
 185       create_workers (1);
 186       system_printf ("All threads busy, added one (now %u)", _workers_count);
 187     }
 188 }
 189
 190 /*static*/ DWORD WINAPI
 191 threaded_queue::start_routine (const LPVOID lpParam)
 192 {
 193   class threaded_queue *const queue = (class threaded_queue *) lpParam;
 194   assert (queue);
 195
 196   queue->worker_loop ();
 197
 198   const long count = InterlockedDecrement (&queue->_workers_count);
 199   assert (count >= 0);
 200
 201   if (queue->_running)
 202     debug_printf ("worker loop has exited; thread about to terminate");
 203
 204   return 0;
 205 }
 206
 207 void
 208 threaded_queue::create_workers (const size_t initial_workers)
 209 {
 210   assert (initial_workers > 0);
 211
 212   for (unsigned int i = 0; i < initial_workers; i++)
 213     {
 214       const long count = InterlockedIncrement (&_workers_count);
 215       assert (count > 0);
 216
 217       DWORD tid;
 218       const HANDLE hThread =
 219         CreateThread (NULL, 0, start_routine, this, 0, &tid);
 220
 221       if (!hThread)
 222         {
 223           system_printf ("failed to create thread, error = %u",
 224                          GetLastError ());
 225           abort ();
 226         }
 227
 228       (void) CloseHandle (hThread);
 229     }
 230 }
 231
 232 void
 233 threaded_queue::worker_loop ()
 234 {
 235   while (true)
 236     {
 237       const DWORD rc = WaitForSingleObject (_requests_sem, INFINITE);
 238       if (rc == WAIT_FAILED)
 239         {
 240           system_printf ("wait for request semaphore failed, error = %u",
 241                          GetLastError ());
 242           return;
 243         }
 244       assert (rc == WAIT_OBJECT_0);
 245
 246       EnterCriticalSection (&_queue_lock);
 247       if (!_running)
 248         {
 249           LeaveCriticalSection (&_queue_lock);
 250           return;
 251         }
 252
 253       assert (_requests_head);
 254       queue_request *const reqptr = _requests_head;
 255       _requests_head = reqptr->_next;
 256
 257       _requests_count -= 1;
 258       assert (_requests_count >= 0);
 259       LeaveCriticalSection (&_queue_lock);
 260
 261       assert (reqptr);
 262       InterlockedIncrement (&_workers_busy);
 263       reqptr->process ();
 264       InterlockedDecrement (&_workers_busy);
 265       delete reqptr;
 266     }
 267 }
 268
 269 /*****************************************************************************/
 270
 271 /* queue_submission_loop */
 272
 273 queue_submission_loop::queue_submission_loop (threaded_queue *const queue,
 274                                               const bool ninterruptible)
 275   : _running (false),
 276     _interrupt_event (NULL),
 277     _queue (queue),
 278     _interruptible (ninterruptible),
 279     _hThread (NULL),
 280     _tid (0),
 281     _next (NULL)
 282 {
 283   if (_interruptible)
 284     {
 285       // verbose: debug_printf ("creating an interruptible processing thread");
 286
 287       _interrupt_event = CreateEvent (NULL,  // SECURITY_ATTRIBUTES
 288                                       FALSE, // Auto-reset
 289                                       FALSE, // Initially non-signalled
 290                                       NULL); // Anonymous
 291
 292       if (!_interrupt_event)
 293         {
 294           system_printf ("failed to create interrupt event, error = %u",
 295                          GetLastError ());
 296           abort ();
 297         }
 298     }
 299 }
 300
 301 queue_submission_loop::~queue_submission_loop ()
 302 {
 303   if (_running)
 304     stop ();
 305   if (_interrupt_event)
 306     (void) CloseHandle (_interrupt_event);
 307   if (_hThread)
 308     (void) CloseHandle (_hThread);
 309 }
 310
 311 bool
 312 queue_submission_loop::start ()
 313 {
 314   assert (!_hThread);
 315
 316   const bool was_running = _running;
 317
 318   if (!was_running)
 319     {
 320       _running = true;
 321
 322       _hThread = CreateThread (NULL, 0, start_routine, this, 0, &_tid);
 323       if (!_hThread)
 324         {
 325           system_printf ("failed to create thread, error = %u",
 326                          GetLastError ());
 327           abort ();
 328         }
 329     }
 330
 331   return was_running;
 332 }
 333
 334 bool
 335 queue_submission_loop::stop ()
 336 {
 337   assert (_hThread && _hThread != INVALID_HANDLE_VALUE);
 338
 339   const bool was_running = _running;
 340
 341   if (_running)
 342     {
 343       _running = false;
 344
 345       if (_interruptible)
 346         {
 347           assert (_interrupt_event
 348                   && _interrupt_event != INVALID_HANDLE_VALUE);
 349
 350           SetEvent (_interrupt_event);
 351
 352           if (WaitForSingleObject (_hThread, 1000) == WAIT_TIMEOUT)
 353             {
 354               system_printf (("request loop thread %u failed to shutdown "
 355                               "when asked politely: about to get heavy"),
 356                              _tid);
 357
 358               if (!TerminateThread (_hThread, 0))
 359                 {
 360                   system_printf (("failed to kill request loop thread %u"
 361                                   ", error = %u"),
 362                                  _tid, GetLastError ());
 363                   abort ();
 364                 }
 365             }
 366         }
 367       else
 368         {
 369           // FIXME: could wait to see if the request loop notices that
 370           // the submission loop is no longer running and shuts down
 371           // voluntarily.
 372
 373           debug_printf ("killing request loop thread %u", _tid);
 374
 375           if (!TerminateThread (_hThread, 0))
 376             system_printf (("failed to kill request loop thread %u"
 377                             ", error = %u"),
 378                            _tid, GetLastError ());
 379         }
 380     }
 381
 382   return was_running;
 383 }
 384
 385 /*static*/ DWORD WINAPI
 386 queue_submission_loop::start_routine (const LPVOID lpParam)
 387 {
 388   class queue_submission_loop *const submission_loop =
 389     (class queue_submission_loop *) lpParam;
 390   assert (submission_loop);
 391
 392   submission_loop->request_loop ();
 393
 394   debug_printf ("submission loop has exited; thread about to terminate");
 395
 396   submission_loop->stop ();
 397
 398   return 0;
 399 }
 400
 401 /*****************************************************************************/
 402 #endif /* __OUTSIDE_CYGWIN__ */