winsup/cygserver/process.cc

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