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