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