winsup/cygserver/process.cc

   1 /* process.cc
   2
   3    Copyright 2001, 2002, 2003, 2004, 2005 Red Hat Inc.
   4
   5    Written by Robert Collins <rbtcollins@hotmail.com>
   6
   7 This file is part of Cygwin.
   8
   9 This software is a copyrighted work licensed under the terms of the
  10 Cygwin license.  Please consult the file "CYGWIN_LICENSE" for
  11 details. */
  12
  13 #ifdef __OUTSIDE_CYGWIN__
  14 #include "woutsup.h"
  15
  16 #include <sys/types.h>
  17
  18 #include <assert.h>
  19 #include <stdlib.h>
  20
  21 #include "process.h"
  22
  23 /*****************************************************************************/
  24
  25 #define elements(ARRAY) (sizeof (ARRAY) / sizeof (*ARRAY))
  26
  27 /*****************************************************************************/
  28
  29 process_cleanup::~process_cleanup ()
  30 {
  31   delete _process;
  32 }
  33
  34 void
  35 process_cleanup::process ()
  36 {
  37   _process->cleanup ();
  38 }
  39
  40 /*****************************************************************************/
  41
  42 process::process (const pid_t cygpid, const DWORD winpid, HANDLE signal_arrived)
  43   : _cygpid (cygpid),
  44     _winpid (winpid),
  45     _hProcess (NULL),
  46     _signal_arrived (INVALID_HANDLE_VALUE),
  47     _cleaning_up (false),
  48     _exit_status (STILL_ACTIVE),
  49     _routines_head (NULL),
  50     _next (NULL)
  51 {
  52   _hProcess = OpenProcess (PROCESS_ALL_ACCESS, FALSE, winpid);
  53   if (!_hProcess)
  54     {
  55       system_printf ("unable to obtain handle for new cache process %d(%lu)",
  56                      _cygpid, _winpid);
  57       _hProcess = INVALID_HANDLE_VALUE;
  58       _exit_status = 0;
  59     }
  60   else
  61     debug_printf ("got handle %p for new cache process %d(%lu)",
  62                   _hProcess, _cygpid, _winpid);
  63   if (!signal_arrived)
  64     system_printf ("signal_arrived NULL for process %d(%lu)",
  65                    _cygpid, _winpid);
  66   else if (signal_arrived != INVALID_HANDLE_VALUE)
  67     {
  68       if (!DuplicateHandle (_hProcess, signal_arrived,
  69                             GetCurrentProcess (), &_signal_arrived,
  70                             0, FALSE, DUPLICATE_SAME_ACCESS))
  71         {
  72           system_printf ("error getting signal_arrived to server (%lu)",
  73                          GetLastError ());
  74           _signal_arrived = INVALID_HANDLE_VALUE;
  75         }
  76     }
  77   InitializeCriticalSection (&_access);
  78   debug ("initialized (%lu)", _cygpid);
  79 }
  80
  81 process::~process ()
  82 {
  83   debug ("deleting (%lu)", _cygpid);
  84   DeleteCriticalSection (&_access);
  85   if (_signal_arrived && _signal_arrived != INVALID_HANDLE_VALUE)
  86     CloseHandle (_signal_arrived);
  87   CloseHandle (_hProcess);
  88 }
  89
  90 /* No need to be thread-safe as this is only ever called by
  91  * process_cache::check_and_remove_process ().  If it has to be made
  92  * thread-safe later on, it should not use the `access' critical section as
  93  * that is held by the client request handlers for an arbitrary length of time,
  94  * i.e. while they do whatever processing is required for a client request.
  95  */
  96 DWORD
  97 process::check_exit_code ()
  98 {
  99   if (_hProcess && _hProcess != INVALID_HANDLE_VALUE
 100       && _exit_status == STILL_ACTIVE
 101       && !GetExitCodeProcess (_hProcess, &_exit_status))
 102     {
 103       system_printf ("failed to retrieve exit code for %d(%lu), error = %lu",
 104                      _cygpid, _winpid, GetLastError ());
 105       _hProcess = INVALID_HANDLE_VALUE;
 106     }
 107   return _exit_status;
 108 }
 109
 110 bool
 111 process::add (cleanup_routine *const entry)
 112 {
 113   assert (entry);
 114
 115   bool res = false;
 116   hold ();
 117
 118   if (!_cleaning_up)
 119     {
 120       entry->_next = _routines_head;
 121       _routines_head = entry;
 122       res = true;
 123     }
 124
 125   release ();
 126   return res;
 127 }
 128
 129 bool
 130 process::remove (const cleanup_routine *const entry)
 131 {
 132   assert (entry);
 133
 134   bool res = false;
 135   hold ();
 136
 137   if (!_cleaning_up)
 138     {
 139       cleanup_routine *previous = NULL;
 140
 141       for (cleanup_routine *ptr = _routines_head;
 142            ptr;
 143            previous = ptr, ptr = ptr->_next)
 144         {
 145           if (*ptr == *entry)
 146             {
 147               if (previous)
 148                 previous->_next = ptr->_next;
 149               else
 150                 _routines_head = ptr->_next;
 151
 152               delete ptr;
 153               res = true;
 154               break;
 155             }
 156         }
 157     }
 158
 159   release ();
 160   return res;
 161 }
 162
 163 /* This is single threaded. It's called after the process is removed
 164  * from the cache, but inserts may be attemped by worker threads that
 165  * have a pointer to it.
 166  */
 167 void
 168 process::cleanup ()
 169 {
 170   hold ();
 171   assert (!is_active ());
 172   assert (!_cleaning_up);
 173   InterlockedExchange (&_cleaning_up, true);
 174   cleanup_routine *entry = _routines_head;
 175   _routines_head = NULL;
 176   release ();
 177
 178   while (entry)
 179     {
 180       cleanup_routine *const ptr = entry;
 181       entry = entry->_next;
 182       ptr->cleanup (this);
 183       delete ptr;
 184     }
 185 }
 186
 187 /*****************************************************************************/
 188
 189 void
 190 process_cache::submission_loop::request_loop ()
 191 {
 192   assert (this);
 193   assert (_cache);
 194   assert (_interrupt_event);
 195
 196   while (_running)
 197     _cache->wait_for_processes (_interrupt_event);
 198 }
 199
 200 /*****************************************************************************/
 201
 202 process_cache::process_cache (const size_t max_procs,
 203                               const unsigned int initial_workers)
 204   : _queue (initial_workers),
 205     _submitter (this, &_queue), // true == interruptible
 206     _processes_count (0),
 207     _max_process_count (max_procs),
 208     _processes_head (NULL),
 209     _cache_add_trigger (NULL)
 210 {
 211   /* there can only be one */
 212   InitializeCriticalSection (&_cache_write_access);
 213
 214   _cache_add_trigger = CreateEvent (NULL,  // SECURITY_ATTRIBUTES
 215                                     TRUE,  // Manual-reset
 216                                     FALSE, // Initially non-signalled
 217                                     NULL); // Anonymous
 218
 219   if (!_cache_add_trigger)
 220     {
 221       system_printf ("failed to create cache add trigger, error = %lu",
 222                      GetLastError ());
 223       abort ();
 224     }
 225
 226   _queue.add_submission_loop (&_submitter);
 227 }
 228
 229 process_cache::~process_cache ()
 230 {
 231   (void) CloseHandle (_cache_add_trigger);
 232   DeleteCriticalSection (&_cache_write_access);
 233 }
 234
 235 /* This returns the process object to the caller already locked, that
 236  * is, with the object's `access' critical region entered.  Thus the
 237  * caller must unlock the object when it's finished with it (via
 238  * process::release ()).  It must then not try to access the object
 239  * afterwards, except by going through this routine again, as it may
 240  * have been deleted once it has been unlocked.
 241  */
 242 class process *
 243 process_cache::process (const pid_t cygpid, const DWORD winpid,
 244                         HANDLE signal_arrived)
 245 {
 246   /* TODO: make this more granular, so a search doesn't involve the
 247    * write lock.
 248    */
 249   EnterCriticalSection (&_cache_write_access);
 250   class process *previous = NULL;
 251   class process *entry = find (winpid, &previous);
 252
 253   if (!entry)
 254     {
 255       if (_processes_count >= _max_process_count)
 256         {
 257           LeaveCriticalSection (&_cache_write_access);
 258           system_printf (("process limit (%d processes) reached; "
 259                           "new connection refused for %d(%lu)"),
 260                          _max_process_count, cygpid, winpid);
 261           return NULL;
 262         }
 263
 264       entry = new class process (cygpid, winpid, signal_arrived);
 265       if (!entry->is_active ())
 266         {
 267           LeaveCriticalSection (&_cache_write_access);
 268           delete entry;
 269           return NULL;
 270         }
 271
 272       if (previous)
 273         {
 274           entry->_next = previous->_next;
 275           previous->_next = entry;
 276         }
 277       else
 278         {
 279           entry->_next = _processes_head;
 280           _processes_head = entry;
 281         }
 282
 283       _processes_count += 1;
 284       SetEvent (_cache_add_trigger);
 285     }
 286
 287   entry->hold (); // To be released by the caller.
 288   LeaveCriticalSection (&_cache_write_access);
 289   assert (entry);
 290   assert (entry->_winpid == winpid);
 291   return entry;
 292 }
 293
 294 struct pcache_wait_t
 295 {
 296   size_t index;
 297   size_t count;
 298   HANDLE *hdls;
 299 };
 300
 301 static DWORD WINAPI
 302 pcache_wait_thread (const LPVOID param)
 303 {
 304   pcache_wait_t *p = (pcache_wait_t *) param;
 305
 306   DWORD rc = WaitForMultipleObjects (p->count, p->hdls, FALSE, INFINITE);
 307   ExitThread (rc == WAIT_FAILED ? rc : rc + p->index);
 308 }
 309
 310 void
 311 process_cache::wait_for_processes (const HANDLE interrupt_event)
 312 {
 313   // Update `_wait_array' with handles of all current processes.
 314   size_t idx;
 315   const size_t count = sync_wait_array (interrupt_event);
 316
 317   debug_printf ("waiting on %u objects in total (%u processes)",
 318                 count, _processes_count);
 319
 320   DWORD rc = WAIT_FAILED;
 321
 322   if (count <= 64)
 323     {
 324       /* If count <= 64, a single WaitForMultipleObjects is sufficient and
 325          we can simply wait in the main thread. */
 326       rc = WaitForMultipleObjects (count, _wait_array, FALSE, INFINITE);
 327       if (rc == WAIT_FAILED)
 328         {
 329           system_printf ("could not wait on the process handles, error = %lu",
 330                          GetLastError ());
 331           abort ();
 332         }
 333     }
 334   else
 335     {
 336       /* If count > 64 we have to create sub-threads which wait for the
 337          actual wait objects and the main thread waits for the termination
 338          of one of the threads. */
 339       HANDLE main_wait_array[5] = { NULL };
 340       DWORD mcount = 0;
 341
 342       for (idx = 0; idx < count; idx += 64)
 343         {
 344           pcache_wait_t p = { idx, min (count - idx, 64), _wait_array + idx };
 345           main_wait_array[mcount++] = CreateThread (NULL, 0, pcache_wait_thread,
 346                                                     &p, 0, NULL);
 347         }
 348
 349       rc = WaitForMultipleObjects (mcount, main_wait_array, FALSE, INFINITE);
 350       if (rc == WAIT_FAILED)
 351         {
 352           system_printf ("could not wait on the process handles, error = %lu",
 353                          GetLastError ());
 354           abort ();
 355         }
 356
 357       /* Check for error condition on signalled sub-thread. */
 358       GetExitCodeThread (main_wait_array[rc], &rc);
 359       if (rc == WAIT_FAILED)
 360         {
 361           system_printf ("could not wait on the process handles, error = %lu",
 362                          GetLastError ());
 363           abort ();
 364         }
 365
 366       /* Wake up all waiting threads.  _cache_add_trigger gets reset
 367          in sync_wait_array again. */
 368       SetEvent (_cache_add_trigger);
 369       WaitForMultipleObjects (mcount, main_wait_array, TRUE, INFINITE);
 370       for (idx = 0; idx < mcount; idx++)
 371         CloseHandle (main_wait_array[idx]);
 372     }
 373
 374   /* Tell all processes the bad news.  This one formerly only checked
 375      processes beginning with the index of the signalled process, but
 376      this can result in processes which are signalled but never removed
 377      under heavy load conditions. */
 378   for (idx = 0; idx < count; idx++)
 379     if (_process_array[idx])
 380       check_and_remove_process (idx);
 381 }
 382
 383 /*
 384  * process_cache::sync_wait_array ()
 385  *
 386  * Fill-in the wait array with the handles that the cache needs to wait on.
 387  * These handles are:
 388  *  - the process_process_param's interrupt event
 389  *  - the process_cache's cache_add_trigger event
 390  *  - the handle for each live process in the cache.
 391  *
 392  * Return value: the number of live handles in the array.
 393  */
 394
 395 size_t
 396 process_cache::sync_wait_array (const HANDLE interrupt_event)
 397 {
 398   assert (this);
 399   assert (interrupt_event && interrupt_event != INVALID_HANDLE_VALUE);
 400
 401   /* Always reset _cache_add_trigger before filling up the array again. */
 402   ResetEvent (_cache_add_trigger);
 403
 404   EnterCriticalSection (&_cache_write_access);
 405
 406   size_t index = 0;
 407
 408   for (class process *ptr = _processes_head; ptr; ptr = ptr->_next)
 409     {
 410       assert (ptr->_hProcess && ptr->_hProcess != INVALID_HANDLE_VALUE);
 411       assert (ptr->is_active ());
 412
 413       _wait_array[index] = ptr->handle ();
 414       _process_array[index++] = ptr;
 415
 416       if (!ptr->_next || index % 64 == 62)
 417         {
 418           /* Added at the end of each thread's array part for efficiency. */
 419           _wait_array[index] = interrupt_event;
 420           _process_array[index++] = NULL;
 421           _wait_array[index] = _cache_add_trigger;
 422           _process_array[index++] = NULL;
 423         }
 424     }
 425
 426   if (!index)
 427     {
 428       /* To get at least *something* to wait for. */
 429       _wait_array[index] = interrupt_event;
 430       _process_array[index++] = NULL;
 431       _wait_array[index] = _cache_add_trigger;
 432       _process_array[index++] = NULL;
 433     }
 434
 435   assert (index <= elements (_wait_array));
 436
 437   LeaveCriticalSection (&_cache_write_access);
 438
 439   return index;
 440 }
 441
 442 void
 443 process_cache::check_and_remove_process (const size_t index)
 444 {
 445   assert (this);
 446   assert (index < elements (_wait_array) - SPECIALS_COUNT);
 447
 448   class process *const process = _process_array[index];
 449
 450   assert (process);
 451   assert (process->handle () == _wait_array[index]);
 452
 453   if (process->check_exit_code () == STILL_ACTIVE)
 454     return;
 455
 456   debug_printf ("process %d(%lu) has left the building ($? = %lu)",
 457                 process->_cygpid, process->_winpid, process->_exit_status);
 458
 459   /* Unlink the process object from the process list. */
 460
 461   EnterCriticalSection (&_cache_write_access);
 462
 463   class process *previous = NULL;
 464
 465   const class process *const tmp = find (process->_winpid, &previous);
 466
 467   assert (tmp == process);
 468   assert (previous ? previous->_next == process : _processes_head == process);
 469
 470   if (previous)
 471     previous->_next = process->_next;
 472   else
 473     _processes_head = process->_next;
 474
 475   _processes_count -= 1;
 476   LeaveCriticalSection (&_cache_write_access);
 477
 478   /* Schedule any cleanup tasks for this process. */
 479   _queue.add (new process_cleanup (process));
 480 }
 481
 482 class process *
 483 process_cache::find (const DWORD winpid, class process **previous)
 484 {
 485   if (previous)
 486     *previous = NULL;
 487
 488   for (class process *ptr = _processes_head; ptr; ptr = ptr->_next)
 489     if (ptr->_winpid == winpid)
 490       return ptr;
 491     else if (ptr->_winpid > winpid) // The list is sorted by winpid.
 492       return NULL;
 493     else if (previous)
 494       *previous = ptr;
 495
 496   return NULL;
 497 }
 498
 499 /*****************************************************************************/
 500 #endif /* __OUTSIDE_CYGWIN__ */