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