winsup/cygserver/process.cc

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