ghc/rts/Capability.c

   1 /* ---------------------------------------------------------------------------
   2  * (c) The GHC Team, 2003
   3  *
   4  * Capabilities
   5  *
   6  * A Capability represent the token required to execute STG code,
   7  * and all the state an OS thread/task needs to run Haskell code:
   8  * its STG registers, a pointer to its TSO, a nursery etc. During
   9  * STG execution, a pointer to the capabilitity is kept in a
  10  * register (BaseReg).
  11  *
  12  * Only in an SMP build will there be multiple capabilities, for
  13  * the threaded RTS and other non-threaded builds, there is only
  14  * one global capability, namely MainCapability.
  15  *
  16  * --------------------------------------------------------------------------*/
  17
  18 #include "PosixSource.h"
  19 #include "Rts.h"
  20 #include "RtsUtils.h"
  21 #include "RtsFlags.h"
  22 #include "OSThreads.h"
  23 #include "Capability.h"
  24 #include "Schedule.h"  /* to get at EMPTY_RUN_QUEUE() */
  25
  26 #if !defined(SMP)
  27 Capability MainCapability;     /* for non-SMP, we have one global capability */
  28 #endif
  29
  30 nat rts_n_free_capabilities;
  31
  32 #if defined(RTS_SUPPORTS_THREADS)
  33
  34 /* returning_worker_cond: when a worker thread returns from executing an
  35  * external call, it needs to wait for an RTS Capability before passing
  36  * on the result of the call to the Haskell thread that made it.
  37  *
  38  * returning_worker_cond is signalled in Capability.releaseCapability().
  39  *
  40  */
  41 Condition returning_worker_cond = INIT_COND_VAR;
  42
  43 /*
  44  * To avoid starvation of threads blocked on worker_thread_cond,
  45  * the task(s) that enter the Scheduler will check to see whether
  46  * there are one or more worker threads blocked waiting on
  47  * returning_worker_cond.
  48  */
  49 nat rts_n_waiting_workers = 0;
  50
  51 /* thread_ready_cond: when signalled, a thread has become runnable for a
  52  * task to execute.
  53  *
  54  * In the non-SMP case, it also implies that the thread that is woken up has
  55  * exclusive access to the RTS and all its data structures (that are not
  56  * locked by the Scheduler's mutex).
  57  *
  58  * thread_ready_cond is signalled whenever
  59  *      !noCapabilities && !EMPTY_RUN_QUEUE().
  60  */
  61 Condition thread_ready_cond = INIT_COND_VAR;
  62
  63 /*
  64  * To be able to make an informed decision about whether or not
  65  * to create a new task when making an external call, keep track of
  66  * the number of tasks currently blocked waiting on thread_ready_cond.
  67  * (if > 0 => no need for a new task, just unblock an existing one).
  68  *
  69  * waitForWorkCapability() takes care of keeping it up-to-date;
  70  * Task.startTask() uses its current value.
  71  */
  72 nat rts_n_waiting_tasks = 0;
  73
  74 static Condition *passTarget = NULL;
  75 static rtsBool passingCapability = rtsFalse;
  76 #endif
  77
  78 #if defined(SMP)
  79 /*
  80  * Free capability list.
  81  */
  82 Capability *free_capabilities;
  83 #endif
  84
  85 #ifdef SMP
  86 #define UNUSED_IF_NOT_SMP
  87 #else
  88 #define UNUSED_IF_NOT_SMP STG_UNUSED
  89 #endif
  90
  91 #if defined(RTS_USER_SIGNALS)
  92 #define ANY_WORK_TO_DO() (!EMPTY_RUN_QUEUE() || interrupted || blackholes_need_checking || signals_pending())
  93 #else
  94 #define ANY_WORK_TO_DO() (!EMPTY_RUN_QUEUE() || interrupted || blackholes_need_checking)
  95 #endif
  96
  97 /* ----------------------------------------------------------------------------
  98    Initialisation
  99    ------------------------------------------------------------------------- */
 100
 101 static void
 102 initCapability( Capability *cap )
 103 {
 104     cap->f.stgGCEnter1     = (F_)__stg_gc_enter_1;
 105     cap->f.stgGCFun        = (F_)__stg_gc_fun;
 106 }
 107
 108 /* -----------------------------------------------------------------------------
 109  * Function: initCapabilities_(nat)
 110  *
 111  * Purpose:  upon startup, allocate and fill in table
 112  *           holding 'n' Capabilities. Only for SMP, since
 113  *           it is the only build that supports multiple
 114  *           capabilities within the RTS.
 115  * -------------------------------------------------------------------------- */
 116 #if defined(SMP)
 117 static void
 118 initCapabilities_(nat n)
 119 {
 120   nat i;
 121   Capability *cap, *prev;
 122   cap  = NULL;
 123   prev = NULL;
 124   for (i = 0; i < n; i++) {
 125     cap = stgMallocBytes(sizeof(Capability), "initCapabilities");
 126     initCapability(cap);
 127     cap->link = prev;
 128     prev = cap;
 129   }
 130   free_capabilities = cap;
 131   rts_n_free_capabilities = n;
 132   IF_DEBUG(scheduler,
 133            sched_belch("allocated %d capabilities", rts_n_free_capabilities));
 134 }
 135 #endif /* SMP */
 136
 137 /* ---------------------------------------------------------------------------
 138  * Function:  initCapabilities()
 139  *
 140  * Purpose:   set up the Capability handling. For the SMP build,
 141  *            we keep a table of them, the size of which is
 142  *            controlled by the user via the RTS flag RtsFlags.ParFlags.nNodes
 143  *
 144  * ------------------------------------------------------------------------- */
 145 void
 146 initCapabilities( void )
 147 {
 148 #if defined(SMP)
 149   initCapabilities_(RtsFlags.ParFlags.nNodes);
 150 #else
 151   initCapability(&MainCapability);
 152 #endif
 153
 154 #if defined(RTS_SUPPORTS_THREADS)
 155   initCondition(&returning_worker_cond);
 156   initCondition(&thread_ready_cond);
 157 #endif
 158
 159   rts_n_free_capabilities = 1;
 160 }
 161
 162 /* ----------------------------------------------------------------------------
 163    grabCapability( Capability** )
 164
 165    (only externally visible when !RTS_SUPPORTS_THREADS.  In the
 166    threaded RTS, clients must use waitFor*Capability()).
 167    ------------------------------------------------------------------------- */
 168
 169 #if defined(RTS_SUPPORTS_THREADS)
 170 static
 171 #endif
 172 void
 173 grabCapability( Capability** cap )
 174 {
 175 #if defined(SMP)
 176   ASSERT(rts_n_free_capabilities > 0);
 177   *cap = free_capabilities;
 178   free_capabilities = (*cap)->link;
 179   rts_n_free_capabilities--;
 180 #else
 181 # if defined(RTS_SUPPORTS_THREADS)
 182   ASSERT(rts_n_free_capabilities == 1);
 183   rts_n_free_capabilities = 0;
 184 # endif
 185   *cap = &MainCapability;
 186 #endif
 187 #if defined(RTS_SUPPORTS_THREADS)
 188   IF_DEBUG(scheduler, sched_belch("worker: got capability"));
 189 #endif
 190 }
 191
 192 /* ----------------------------------------------------------------------------
 193  * Function:  releaseCapability(Capability*)
 194  *
 195  * Purpose:   Letting go of a capability. Causes a
 196  *            'returning worker' thread or a 'waiting worker'
 197  *            to wake up, in that order.
 198  * ------------------------------------------------------------------------- */
 199
 200 void
 201 releaseCapability( Capability* cap UNUSED_IF_NOT_SMP )
 202 {
 203     // Precondition: sched_mutex is held.
 204 #if defined(RTS_SUPPORTS_THREADS)
 205 #if !defined(SMP)
 206     ASSERT(rts_n_free_capabilities == 0);
 207 #endif
 208 #if defined(SMP)
 209     cap->link = free_capabilities;
 210     free_capabilities = cap;
 211 #endif
 212     // Check to see whether a worker thread can be given
 213     // the go-ahead to return the result of an external call..
 214     if (rts_n_waiting_workers > 0) {
 215         // Decrement the counter here to avoid livelock where the
 216         // thread that is yielding its capability will repeatedly
 217         // signal returning_worker_cond.
 218
 219         rts_n_waiting_workers--;
 220         signalCondition(&returning_worker_cond);
 221         IF_DEBUG(scheduler, sched_belch("worker: released capability to returning worker"));
 222     } else if (passingCapability) {
 223         if (passTarget == NULL) {
 224             signalCondition(&thread_ready_cond);
 225             startSchedulerTaskIfNecessary();
 226         } else {
 227             signalCondition(passTarget);
 228         }
 229 #if defined(SMP)
 230         rts_n_free_capabilities++;
 231 #else
 232         rts_n_free_capabilities = 1;
 233 #endif
 234         IF_DEBUG(scheduler, sched_belch("worker: released capability, passing it"));
 235
 236     } else {
 237 #if defined(SMP)
 238         rts_n_free_capabilities++;
 239 #else
 240         rts_n_free_capabilities = 1;
 241 #endif
 242         // Signal that a capability is available
 243         if (rts_n_waiting_tasks > 0 && ANY_WORK_TO_DO()) {
 244             signalCondition(&thread_ready_cond);
 245         }
 246         startSchedulerTaskIfNecessary();
 247         IF_DEBUG(scheduler, sched_belch("worker: released capability"));
 248     }
 249 #endif
 250     return;
 251 }
 252
 253 #if defined(RTS_SUPPORTS_THREADS)
 254 /*
 255  * When a native thread has completed the execution of an external
 256  * call, it needs to communicate the result back. This is done
 257  * as follows:
 258  *
 259  *  - in resumeThread(), the thread calls waitForReturnCapability().
 260  *  - If no capabilities are readily available, waitForReturnCapability()
 261  *    increments a counter rts_n_waiting_workers, and blocks
 262  *    waiting for the condition returning_worker_cond to become
 263  *    signalled.
 264  *  - upon entry to the Scheduler, a worker thread checks the
 265  *    value of rts_n_waiting_workers. If > 0, the worker thread
 266  *    will yield its capability to let a returning worker thread
 267  *    proceed with returning its result -- this is done via
 268  *    yieldToReturningWorker().
 269  *  - the worker thread that yielded its capability then tries
 270  *    to re-grab a capability and re-enter the Scheduler.
 271  */
 272
 273 /* ----------------------------------------------------------------------------
 274  * waitForReturnCapability( Mutext *pMutex, Capability** )
 275  *
 276  * Purpose:  when an OS thread returns from an external call,
 277  * it calls grabReturnCapability() (via Schedule.resumeThread())
 278  * to wait for permissions to enter the RTS & communicate the
 279  * result of the external call back to the Haskell thread that
 280  * made it.
 281  *
 282  * ------------------------------------------------------------------------- */
 283
 284 void
 285 waitForReturnCapability( Mutex* pMutex, Capability** pCap )
 286 {
 287     // Pre-condition: pMutex is held.
 288
 289     IF_DEBUG(scheduler,
 290              sched_belch("worker: returning; workers waiting: %d",
 291                          rts_n_waiting_workers));
 292
 293     if ( noCapabilities() || passingCapability ) {
 294         rts_n_waiting_workers++;
 295         context_switch = 1;     // make sure it's our turn soon
 296         waitCondition(&returning_worker_cond, pMutex);
 297 #if defined(SMP)
 298         *pCap = free_capabilities;
 299         free_capabilities = (*pCap)->link;
 300         ASSERT(pCap != NULL);
 301 #else
 302         *pCap = &MainCapability;
 303         ASSERT(rts_n_free_capabilities == 0);
 304 #endif
 305     } else {
 306         grabCapability(pCap);
 307     }
 308
 309     // Post-condition: pMutex is held, pCap points to a capability
 310     // which is now held by the current thread.
 311     return;
 312 }
 313
 314
 315 /* ----------------------------------------------------------------------------
 316  * yieldCapability( Mutex* pMutex, Capability** pCap )
 317  * ------------------------------------------------------------------------- */
 318
 319 void
 320 yieldCapability( Capability** pCap )
 321 {
 322     // Pre-condition:  pMutex is assumed held, the current thread
 323     // holds the capability pointed to by pCap.
 324
 325     if ( rts_n_waiting_workers > 0 || passingCapability || !ANY_WORK_TO_DO()) {
 326         IF_DEBUG(scheduler,
 327                  if (rts_n_waiting_workers > 0) {
 328                      sched_belch("worker: giving up capability (returning wkr)");
 329                  } else if (passingCapability) {
 330                      sched_belch("worker: giving up capability (passing capability)");
 331                  } else {
 332                      sched_belch("worker: giving up capability (no threads to run)");
 333                  }
 334             );
 335         releaseCapability(*pCap);
 336         *pCap = NULL;
 337     }
 338
 339     // Post-condition:  either:
 340     //
 341     //  1. *pCap is NULL, in which case the current thread does not
 342     //     hold a capability now, or
 343     //  2. *pCap is not NULL, in which case the current thread still
 344     //     holds the capability.
 345     //
 346     return;
 347 }
 348
 349
 350 /* ----------------------------------------------------------------------------
 351  * waitForCapability( Mutex*, Capability**, Condition* )
 352  *
 353  * Purpose:  wait for a Capability to become available. In
 354  *           the process of doing so, updates the number
 355  *           of tasks currently blocked waiting for a capability/more
 356  *           work. That counter is used when deciding whether or
 357  *           not to create a new worker thread when an external
 358  *           call is made.
 359  *           If pThreadCond is not NULL, a capability can be specifically
 360  *           passed to this thread using passCapability.
 361  * ------------------------------------------------------------------------- */
 362
 363 void
 364 waitForCapability( Mutex* pMutex, Capability** pCap, Condition* pThreadCond )
 365 {
 366     // Pre-condition: pMutex is held.
 367
 368     while ( noCapabilities() ||
 369             (passingCapability && passTarget != pThreadCond) ||
 370             !ANY_WORK_TO_DO()) {
 371         IF_DEBUG(scheduler,
 372                  sched_belch("worker: wait for capability (cond: %p)",
 373                              pThreadCond));
 374
 375         if (pThreadCond != NULL) {
 376             waitCondition(pThreadCond, pMutex);
 377             IF_DEBUG(scheduler, sched_belch("worker: get passed capability"));
 378         } else {
 379             rts_n_waiting_tasks++;
 380             waitCondition(&thread_ready_cond, pMutex);
 381             rts_n_waiting_tasks--;
 382             IF_DEBUG(scheduler, sched_belch("worker: get normal capability"));
 383         }
 384     }
 385     passingCapability = rtsFalse;
 386     grabCapability(pCap);
 387
 388     // Post-condition: pMutex is held and *pCap is held by the current thread
 389     return;
 390 }
 391
 392 /* ----------------------------------------------------------------------------
 393    passCapability, passCapabilityToWorker
 394    ------------------------------------------------------------------------- */
 395
 396 void
 397 passCapability( Condition *pTargetThreadCond )
 398 {
 399     // Pre-condition: pMutex is held and cap is held by the current thread
 400
 401     passTarget = pTargetThreadCond;
 402     passingCapability = rtsTrue;
 403     IF_DEBUG(scheduler, sched_belch("worker: passCapability"));
 404
 405     // Post-condition: pMutex is held; cap is still held, but will be
 406     //                 passed to the target thread when next released.
 407 }
 408
 409 void
 410 passCapabilityToWorker( void )
 411 {
 412     // Pre-condition: pMutex is held and cap is held by the current thread
 413
 414     passTarget = NULL;
 415     passingCapability = rtsTrue;
 416     IF_DEBUG(scheduler, sched_belch("worker: passCapabilityToWorker"));
 417
 418     // Post-condition: pMutex is held; cap is still held, but will be
 419     //                 passed to a worker thread when next released.
 420 }
 421
 422 #endif /* RTS_SUPPORTS_THREADS */
 423
 424 /* ----------------------------------------------------------------------------
 425    threadRunnable()
 426
 427    Signals that a thread has been placed on the run queue, so a worker
 428    might need to be woken up to run it.
 429
 430    ToDo: should check whether the thread at the front of the queue is
 431    bound, and if so wake up the appropriate worker.
 432    -------------------------------------------------------------------------- */
 433 void
 434 threadRunnable ( void )
 435 {
 436 #if defined(RTS_SUPPORTS_THREADS)
 437     if ( !noCapabilities() && ANY_WORK_TO_DO() && rts_n_waiting_tasks > 0 ) {
 438         signalCondition(&thread_ready_cond);
 439     }
 440     startSchedulerTaskIfNecessary();
 441 #endif
 442 }
 443
 444
 445 /* ----------------------------------------------------------------------------
 446    prodWorker()
 447
 448    Wake up... time to die.
 449    -------------------------------------------------------------------------- */
 450 void
 451 prodWorker ( void )
 452 {
 453 #if defined(RTS_SUPPORTS_THREADS)
 454     signalCondition(&thread_ready_cond);
 455 #endif
 456 }