rts/Capability.c

   1 /* ---------------------------------------------------------------------------
   2  *
   3  * (c) The GHC Team, 2003-2006
   4  *
   5  * Capabilities
   6  *
   7  * A Capability represent the token required to execute STG code,
   8  * and all the state an OS thread/task needs to run Haskell code:
   9  * its STG registers, a pointer to its TSO, a nursery etc. During
  10  * STG execution, a pointer to the capabilitity is kept in a
  11  * register (BaseReg; actually it is a pointer to cap->r).
  12  *
  13  * Only in an THREADED_RTS build will there be multiple capabilities,
  14  * for non-threaded builds there is only one global capability, namely
  15  * MainCapability.
  16  *
  17  * --------------------------------------------------------------------------*/
  18
  19 #include "PosixSource.h"
  20 #include "Rts.h"
  21 #include "RtsUtils.h"
  22 #include "RtsFlags.h"
  23 #include "STM.h"
  24 #include "OSThreads.h"
  25 #include "Capability.h"
  26 #include "Schedule.h"
  27 #include "Sparks.h"
  28 #include "Trace.h"
  29
  30 // one global capability, this is the Capability for non-threaded
  31 // builds, and for +RTS -N1
  32 Capability MainCapability;
  33
  34 nat n_capabilities;
  35 Capability *capabilities = NULL;
  36
  37 // Holds the Capability which last became free.  This is used so that
  38 // an in-call has a chance of quickly finding a free Capability.
  39 // Maintaining a global free list of Capabilities would require global
  40 // locking, so we don't do that.
  41 Capability *last_free_capability;
  42
  43 /* GC indicator, in scope for the scheduler, init'ed to false */
  44 volatile StgWord waiting_for_gc = 0;
  45
  46 #if defined(THREADED_RTS)
  47 STATIC_INLINE rtsBool
  48 globalWorkToDo (void)
  49 {
  50     return blackholes_need_checking
  51         || sched_state >= SCHED_INTERRUPTING
  52         ;
  53 }
  54 #endif
  55
  56 #if defined(THREADED_RTS)
  57 STATIC_INLINE rtsBool
  58 anyWorkForMe( Capability *cap, Task *task )
  59 {
  60     if (task->tso != NULL) {
  61         // A bound task only runs if its thread is on the run queue of
  62         // the capability on which it was woken up.  Otherwise, we
  63         // can't be sure that we have the right capability: the thread
  64         // might be woken up on some other capability, and task->cap
  65         // could change under our feet.
  66         return !emptyRunQueue(cap) && cap->run_queue_hd->bound == task;
  67     } else {
  68         // A vanilla worker task runs if either there is a lightweight
  69         // thread at the head of the run queue, or the run queue is
  70         // empty and (there are sparks to execute, or there is some
  71         // other global condition to check, such as threads blocked on
  72         // blackholes).
  73         if (emptyRunQueue(cap)) {
  74             return !emptySparkPoolCap(cap)
  75                 || !emptyWakeupQueue(cap)
  76                 || globalWorkToDo();
  77         } else
  78             return cap->run_queue_hd->bound == NULL;
  79     }
  80 }
  81 #endif
  82
  83 /* -----------------------------------------------------------------------------
  84  * Manage the returning_tasks lists.
  85  *
  86  * These functions require cap->lock
  87  * -------------------------------------------------------------------------- */
  88
  89 #if defined(THREADED_RTS)
  90 STATIC_INLINE void
  91 newReturningTask (Capability *cap, Task *task)
  92 {
  93     ASSERT_LOCK_HELD(&cap->lock);
  94     ASSERT(task->return_link == NULL);
  95     if (cap->returning_tasks_hd) {
  96         ASSERT(cap->returning_tasks_tl->return_link == NULL);
  97         cap->returning_tasks_tl->return_link = task;
  98     } else {
  99         cap->returning_tasks_hd = task;
 100     }
 101     cap->returning_tasks_tl = task;
 102 }
 103
 104 STATIC_INLINE Task *
 105 popReturningTask (Capability *cap)
 106 {
 107     ASSERT_LOCK_HELD(&cap->lock);
 108     Task *task;
 109     task = cap->returning_tasks_hd;
 110     ASSERT(task);
 111     cap->returning_tasks_hd = task->return_link;
 112     if (!cap->returning_tasks_hd) {
 113         cap->returning_tasks_tl = NULL;
 114     }
 115     task->return_link = NULL;
 116     return task;
 117 }
 118 #endif
 119
 120 /* ----------------------------------------------------------------------------
 121  * Initialisation
 122  *
 123  * The Capability is initially marked not free.
 124  * ------------------------------------------------------------------------- */
 125
 126 static void
 127 initCapability( Capability *cap, nat i )
 128 {
 129     nat g;
 130
 131     cap->no = i;
 132     cap->in_haskell        = rtsFalse;
 133
 134     cap->run_queue_hd      = END_TSO_QUEUE;
 135     cap->run_queue_tl      = END_TSO_QUEUE;
 136
 137 #if defined(THREADED_RTS)
 138     initMutex(&cap->lock);
 139     cap->running_task      = NULL; // indicates cap is free
 140     cap->spare_workers     = NULL;
 141     cap->suspended_ccalling_tasks = NULL;
 142     cap->returning_tasks_hd = NULL;
 143     cap->returning_tasks_tl = NULL;
 144     cap->wakeup_queue_hd    = END_TSO_QUEUE;
 145     cap->wakeup_queue_tl    = END_TSO_QUEUE;
 146 #endif
 147
 148     cap->f.stgGCEnter1     = (F_)__stg_gc_enter_1;
 149     cap->f.stgGCFun        = (F_)__stg_gc_fun;
 150
 151     cap->mut_lists  = stgMallocBytes(sizeof(bdescr *) *
 152                                      RtsFlags.GcFlags.generations,
 153                                      "initCapability");
 154
 155     for (g = 0; g < RtsFlags.GcFlags.generations; g++) {
 156         cap->mut_lists[g] = NULL;
 157     }
 158
 159     cap->free_tvar_watch_queues = END_STM_WATCH_QUEUE;
 160     cap->free_invariant_check_queues = END_INVARIANT_CHECK_QUEUE;
 161     cap->free_trec_chunks = END_STM_CHUNK_LIST;
 162     cap->free_trec_headers = NO_TREC;
 163     cap->transaction_tokens = 0;
 164 }
 165
 166 /* ---------------------------------------------------------------------------
 167  * Function:  initCapabilities()
 168  *
 169  * Purpose:   set up the Capability handling. For the THREADED_RTS build,
 170  *            we keep a table of them, the size of which is
 171  *            controlled by the user via the RTS flag -N.
 172  *
 173  * ------------------------------------------------------------------------- */
 174 void
 175 initCapabilities( void )
 176 {
 177 #if defined(THREADED_RTS)
 178     nat i;
 179
 180 #ifndef REG_Base
 181     // We can't support multiple CPUs if BaseReg is not a register
 182     if (RtsFlags.ParFlags.nNodes > 1) {
 183         errorBelch("warning: multiple CPUs not supported in this build, reverting to 1");
 184         RtsFlags.ParFlags.nNodes = 1;
 185     }
 186 #endif
 187
 188     n_capabilities = RtsFlags.ParFlags.nNodes;
 189
 190     if (n_capabilities == 1) {
 191         capabilities = &MainCapability;
 192         // THREADED_RTS must work on builds that don't have a mutable
 193         // BaseReg (eg. unregisterised), so in this case
 194         // capabilities[0] must coincide with &MainCapability.
 195     } else {
 196         capabilities = stgMallocBytes(n_capabilities * sizeof(Capability),
 197                                       "initCapabilities");
 198     }
 199
 200     for (i = 0; i < n_capabilities; i++) {
 201         initCapability(&capabilities[i], i);
 202     }
 203
 204     debugTrace(DEBUG_sched, "allocated %d capabilities", n_capabilities);
 205
 206 #else /* !THREADED_RTS */
 207
 208     n_capabilities = 1;
 209     capabilities = &MainCapability;
 210     initCapability(&MainCapability, 0);
 211
 212 #endif
 213
 214     // There are no free capabilities to begin with.  We will start
 215     // a worker Task to each Capability, which will quickly put the
 216     // Capability on the free list when it finds nothing to do.
 217     last_free_capability = &capabilities[0];
 218 }
 219
 220 /* ----------------------------------------------------------------------------
 221  * Give a Capability to a Task.  The task must currently be sleeping
 222  * on its condition variable.
 223  *
 224  * Requires cap->lock (modifies cap->running_task).
 225  *
 226  * When migrating a Task, the migrater must take task->lock before
 227  * modifying task->cap, to synchronise with the waking up Task.
 228  * Additionally, the migrater should own the Capability (when
 229  * migrating the run queue), or cap->lock (when migrating
 230  * returning_workers).
 231  *
 232  * ------------------------------------------------------------------------- */
 233
 234 #if defined(THREADED_RTS)
 235 STATIC_INLINE void
 236 giveCapabilityToTask (Capability *cap USED_IF_DEBUG, Task *task)
 237 {
 238     ASSERT_LOCK_HELD(&cap->lock);
 239     ASSERT(task->cap == cap);
 240     trace(TRACE_sched | DEBUG_sched,
 241           "passing capability %d to %s %p",
 242           cap->no, task->tso ? "bound task" : "worker",
 243           (void *)task->id);
 244     ACQUIRE_LOCK(&task->lock);
 245     task->wakeup = rtsTrue;
 246     // the wakeup flag is needed because signalCondition() doesn't
 247     // flag the condition if the thread is already runniing, but we want
 248     // it to be sticky.
 249     signalCondition(&task->cond);
 250     RELEASE_LOCK(&task->lock);
 251 }
 252 #endif
 253
 254 /* ----------------------------------------------------------------------------
 255  * Function:  releaseCapability(Capability*)
 256  *
 257  * Purpose:   Letting go of a capability. Causes a
 258  *            'returning worker' thread or a 'waiting worker'
 259  *            to wake up, in that order.
 260  * ------------------------------------------------------------------------- */
 261
 262 #if defined(THREADED_RTS)
 263 void
 264 releaseCapability_ (Capability* cap)
 265 {
 266     Task *task;
 267
 268     task = cap->running_task;
 269
 270     ASSERT_PARTIAL_CAPABILITY_INVARIANTS(cap,task);
 271
 272     cap->running_task = NULL;
 273
 274     // Check to see whether a worker thread can be given
 275     // the go-ahead to return the result of an external call..
 276     if (cap->returning_tasks_hd != NULL) {
 277         giveCapabilityToTask(cap,cap->returning_tasks_hd);
 278         // The Task pops itself from the queue (see waitForReturnCapability())
 279         return;
 280     }
 281
 282     /* if waiting_for_gc was the reason to release the cap: thread
 283        comes from yieldCap->releaseAndQueueWorker. Unconditionally set
 284        cap. free and return (see default after the if-protected other
 285        special cases). Thread will wait on cond.var and re-acquire the
 286        same cap after GC (GC-triggering cap. calls releaseCap and
 287        enters the spare_workers case)
 288     */
 289     if (waiting_for_gc) {
 290       last_free_capability = cap; // needed?
 291       trace(TRACE_sched | DEBUG_sched,
 292             "GC pending, set capability %d free", cap->no);
 293       return;
 294     }
 295
 296
 297     // If the next thread on the run queue is a bound thread,
 298     // give this Capability to the appropriate Task.
 299     if (!emptyRunQueue(cap) && cap->run_queue_hd->bound) {
 300         // Make sure we're not about to try to wake ourselves up
 301         ASSERT(task != cap->run_queue_hd->bound);
 302         task = cap->run_queue_hd->bound;
 303         giveCapabilityToTask(cap,task);
 304         return;
 305     }
 306
 307     if (!cap->spare_workers) {
 308         // Create a worker thread if we don't have one.  If the system
 309         // is interrupted, we only create a worker task if there
 310         // are threads that need to be completed.  If the system is
 311         // shutting down, we never create a new worker.
 312         if (sched_state < SCHED_SHUTTING_DOWN || !emptyRunQueue(cap)) {
 313             debugTrace(DEBUG_sched,
 314                        "starting new worker on capability %d", cap->no);
 315             startWorkerTask(cap, workerStart);
 316             return;
 317         }
 318     }
 319
 320     // If we have an unbound thread on the run queue, or if there's
 321     // anything else to do, give the Capability to a worker thread.
 322     if (!emptyRunQueue(cap) || !emptyWakeupQueue(cap)
 323               || !emptySparkPoolCap(cap) || globalWorkToDo()) {
 324         if (cap->spare_workers) {
 325             giveCapabilityToTask(cap,cap->spare_workers);
 326             // The worker Task pops itself from the queue;
 327             return;
 328         }
 329     }
 330
 331     last_free_capability = cap;
 332     trace(TRACE_sched | DEBUG_sched, "freeing capability %d", cap->no);
 333 }
 334
 335 void
 336 releaseCapability (Capability* cap USED_IF_THREADS)
 337 {
 338     ACQUIRE_LOCK(&cap->lock);
 339     releaseCapability_(cap);
 340     RELEASE_LOCK(&cap->lock);
 341 }
 342
 343 static void
 344 releaseCapabilityAndQueueWorker (Capability* cap USED_IF_THREADS)
 345 {
 346     Task *task;
 347
 348     ACQUIRE_LOCK(&cap->lock);
 349
 350     task = cap->running_task;
 351
 352     // If the current task is a worker, save it on the spare_workers
 353     // list of this Capability.  A worker can mark itself as stopped,
 354     // in which case it is not replaced on the spare_worker queue.
 355     // This happens when the system is shutting down (see
 356     // Schedule.c:workerStart()).
 357     // Also, be careful to check that this task hasn't just exited
 358     // Haskell to do a foreign call (task->suspended_tso).
 359     if (!isBoundTask(task) && !task->stopped && !task->suspended_tso) {
 360         task->next = cap->spare_workers;
 361         cap->spare_workers = task;
 362     }
 363     // Bound tasks just float around attached to their TSOs.
 364
 365     releaseCapability_(cap);
 366
 367     RELEASE_LOCK(&cap->lock);
 368 }
 369 #endif
 370
 371 /* ----------------------------------------------------------------------------
 372  * waitForReturnCapability( Task *task )
 373  *
 374  * Purpose:  when an OS thread returns from an external call,
 375  * it calls waitForReturnCapability() (via Schedule.resumeThread())
 376  * to wait for permission to enter the RTS & communicate the
 377  * result of the external call back to the Haskell thread that
 378  * made it.
 379  *
 380  * ------------------------------------------------------------------------- */
 381 void
 382 waitForReturnCapability (Capability **pCap, Task *task)
 383 {
 384 #if !defined(THREADED_RTS)
 385
 386     MainCapability.running_task = task;
 387     task->cap = &MainCapability;
 388     *pCap = &MainCapability;
 389
 390 #else
 391     Capability *cap = *pCap;
 392
 393     if (cap == NULL) {
 394         // Try last_free_capability first
 395         cap = last_free_capability;
 396         if (!cap->running_task) {
 397             nat i;
 398             // otherwise, search for a free capability
 399             for (i = 0; i < n_capabilities; i++) {
 400                 cap = &capabilities[i];
 401                 if (!cap->running_task) {
 402                     break;
 403                 }
 404             }
 405             // Can't find a free one, use last_free_capability.
 406             cap = last_free_capability;
 407         }
 408
 409         // record the Capability as the one this Task is now assocated with.
 410         task->cap = cap;
 411
 412     } else {
 413         ASSERT(task->cap == cap);
 414     }
 415
 416     ACQUIRE_LOCK(&cap->lock);
 417
 418     debugTrace(DEBUG_sched, "returning; I want capability %d", cap->no);
 419
 420     if (!cap->running_task) {
 421         // It's free; just grab it
 422         cap->running_task = task;
 423         RELEASE_LOCK(&cap->lock);
 424     } else {
 425         newReturningTask(cap,task);
 426         RELEASE_LOCK(&cap->lock);
 427
 428         for (;;) {
 429             ACQUIRE_LOCK(&task->lock);
 430             // task->lock held, cap->lock not held
 431             if (!task->wakeup) waitCondition(&task->cond, &task->lock);
 432             cap = task->cap;
 433             task->wakeup = rtsFalse;
 434             RELEASE_LOCK(&task->lock);
 435
 436             // now check whether we should wake up...
 437             ACQUIRE_LOCK(&cap->lock);
 438             if (cap->running_task == NULL) {
 439                 if (cap->returning_tasks_hd != task) {
 440                     giveCapabilityToTask(cap,cap->returning_tasks_hd);
 441                     RELEASE_LOCK(&cap->lock);
 442                     continue;
 443                 }
 444                 cap->running_task = task;
 445                 popReturningTask(cap);
 446                 RELEASE_LOCK(&cap->lock);
 447                 break;
 448             }
 449             RELEASE_LOCK(&cap->lock);
 450         }
 451
 452     }
 453
 454     ASSERT_FULL_CAPABILITY_INVARIANTS(cap,task);
 455
 456     trace(TRACE_sched | DEBUG_sched, "resuming capability %d", cap->no);
 457
 458     *pCap = cap;
 459 #endif
 460 }
 461
 462 #if defined(THREADED_RTS)
 463 /* ----------------------------------------------------------------------------
 464  * yieldCapability
 465  * ------------------------------------------------------------------------- */
 466
 467 void
 468 yieldCapability (Capability** pCap, Task *task)
 469 {
 470     Capability *cap = *pCap;
 471
 472     // The fast path has no locking, if we don't enter this while loop
 473
 474     while ( waiting_for_gc
 475             /* i.e. another capability triggered HeapOverflow, is busy
 476                getting capabilities (stopping their owning tasks) */
 477             || cap->returning_tasks_hd != NULL
 478                 /* cap reserved for another task */
 479             || !anyWorkForMe(cap,task)
 480                 /* cap/task have no work */
 481             ) {
 482         debugTrace(DEBUG_sched, "giving up capability %d", cap->no);
 483
 484         // We must now release the capability and wait to be woken up
 485         // again.
 486         task->wakeup = rtsFalse;
 487         releaseCapabilityAndQueueWorker(cap);
 488
 489         for (;;) {
 490             ACQUIRE_LOCK(&task->lock);
 491             // task->lock held, cap->lock not held
 492             if (!task->wakeup) waitCondition(&task->cond, &task->lock);
 493             cap = task->cap;
 494             task->wakeup = rtsFalse;
 495             RELEASE_LOCK(&task->lock);
 496
 497             debugTrace(DEBUG_sched, "woken up on capability %d", cap->no);
 498
 499             ACQUIRE_LOCK(&cap->lock);
 500             if (cap->running_task != NULL) {
 501                 debugTrace(DEBUG_sched,
 502                            "capability %d is owned by another task", cap->no);
 503                 RELEASE_LOCK(&cap->lock);
 504                 continue;
 505             }
 506
 507             if (task->tso == NULL) {
 508                 ASSERT(cap->spare_workers != NULL);
 509                 // if we're not at the front of the queue, release it
 510                 // again.  This is unlikely to happen.
 511                 if (cap->spare_workers != task) {
 512                     giveCapabilityToTask(cap,cap->spare_workers);
 513                     RELEASE_LOCK(&cap->lock);
 514                     continue;
 515                 }
 516                 cap->spare_workers = task->next;
 517                 task->next = NULL;
 518             }
 519             cap->running_task = task;
 520             RELEASE_LOCK(&cap->lock);
 521             break;
 522         }
 523
 524         trace(TRACE_sched | DEBUG_sched, "resuming capability %d", cap->no);
 525         ASSERT(cap->running_task == task);
 526     }
 527
 528     *pCap = cap;
 529
 530     ASSERT_FULL_CAPABILITY_INVARIANTS(cap,task);
 531
 532     return;
 533 }
 534
 535 /* ----------------------------------------------------------------------------
 536  * Wake up a thread on a Capability.
 537  *
 538  * This is used when the current Task is running on a Capability and
 539  * wishes to wake up a thread on a different Capability.
 540  * ------------------------------------------------------------------------- */
 541
 542 void
 543 wakeupThreadOnCapability (Capability *cap, StgTSO *tso)
 544 {
 545     ASSERT(tso->cap == cap);
 546     ASSERT(tso->bound ? tso->bound->cap == cap : 1);
 547     ASSERT_LOCK_HELD(&cap->lock);
 548
 549     tso->cap = cap;
 550
 551     if (cap->running_task == NULL) {
 552         // nobody is running this Capability, we can add our thread
 553         // directly onto the run queue and start up a Task to run it.
 554         appendToRunQueue(cap,tso);
 555
 556         // start it up
 557         cap->running_task = myTask(); // precond for releaseCapability_()
 558         trace(TRACE_sched, "resuming capability %d", cap->no);
 559         releaseCapability_(cap);
 560     } else {
 561         appendToWakeupQueue(cap,tso);
 562         // someone is running on this Capability, so it cannot be
 563         // freed without first checking the wakeup queue (see
 564         // releaseCapability_).
 565     }
 566 }
 567
 568 void
 569 wakeupThreadOnCapability_lock (Capability *cap, StgTSO *tso)
 570 {
 571     ACQUIRE_LOCK(&cap->lock);
 572     migrateThreadToCapability (cap, tso);
 573     RELEASE_LOCK(&cap->lock);
 574 }
 575
 576 void
 577 migrateThreadToCapability (Capability *cap, StgTSO *tso)
 578 {
 579     // ASSUMES: cap->lock is held (asserted in wakeupThreadOnCapability)
 580     if (tso->bound) {
 581         ASSERT(tso->bound->cap == tso->cap);
 582         tso->bound->cap = cap;
 583     }
 584     tso->cap = cap;
 585     wakeupThreadOnCapability(cap,tso);
 586 }
 587
 588 void
 589 migrateThreadToCapability_lock (Capability *cap, StgTSO *tso)
 590 {
 591     ACQUIRE_LOCK(&cap->lock);
 592     migrateThreadToCapability (cap, tso);
 593     RELEASE_LOCK(&cap->lock);
 594 }
 595
 596 /* ----------------------------------------------------------------------------
 597  * prodCapabilities
 598  *
 599  * Used to indicate that the interrupted flag is now set, or some
 600  * other global condition that might require waking up a Task on each
 601  * Capability.
 602  * ------------------------------------------------------------------------- */
 603
 604 static void
 605 prodCapabilities(rtsBool all)
 606 {
 607     nat i;
 608     Capability *cap;
 609     Task *task;
 610
 611     for (i=0; i < n_capabilities; i++) {
 612         cap = &capabilities[i];
 613         ACQUIRE_LOCK(&cap->lock);
 614         if (!cap->running_task) {
 615             if (cap->spare_workers) {
 616                 trace(TRACE_sched, "resuming capability %d", cap->no);
 617                 task = cap->spare_workers;
 618                 ASSERT(!task->stopped);
 619                 giveCapabilityToTask(cap,task);
 620                 if (!all) {
 621                     RELEASE_LOCK(&cap->lock);
 622                     return;
 623                 }
 624             }
 625         }
 626         RELEASE_LOCK(&cap->lock);
 627     }
 628     return;
 629 }
 630
 631 void
 632 prodAllCapabilities (void)
 633 {
 634     prodCapabilities(rtsTrue);
 635 }
 636
 637 /* ----------------------------------------------------------------------------
 638  * prodOneCapability
 639  *
 640  * Like prodAllCapabilities, but we only require a single Task to wake
 641  * up in order to service some global event, such as checking for
 642  * deadlock after some idle time has passed.
 643  * ------------------------------------------------------------------------- */
 644
 645 void
 646 prodOneCapability (void)
 647 {
 648     prodCapabilities(rtsFalse);
 649 }
 650
 651 /* ----------------------------------------------------------------------------
 652  * shutdownCapability
 653  *
 654  * At shutdown time, we want to let everything exit as cleanly as
 655  * possible.  For each capability, we let its run queue drain, and
 656  * allow the workers to stop.
 657  *
 658  * This function should be called when interrupted and
 659  * shutting_down_scheduler = rtsTrue, thus any worker that wakes up
 660  * will exit the scheduler and call taskStop(), and any bound thread
 661  * that wakes up will return to its caller.  Runnable threads are
 662  * killed.
 663  *
 664  * ------------------------------------------------------------------------- */
 665
 666 void
 667 shutdownCapability (Capability *cap, Task *task, rtsBool safe)
 668 {
 669     nat i;
 670
 671     ASSERT(sched_state == SCHED_SHUTTING_DOWN);
 672
 673     task->cap = cap;
 674
 675     // Loop indefinitely until all the workers have exited and there
 676     // are no Haskell threads left.  We used to bail out after 50
 677     // iterations of this loop, but that occasionally left a worker
 678     // running which caused problems later (the closeMutex() below
 679     // isn't safe, for one thing).
 680
 681     for (i = 0; /* i < 50 */; i++) {
 682         debugTrace(DEBUG_sched,
 683                    "shutting down capability %d, attempt %d", cap->no, i);
 684         ACQUIRE_LOCK(&cap->lock);
 685         if (cap->running_task) {
 686             RELEASE_LOCK(&cap->lock);
 687             debugTrace(DEBUG_sched, "not owner, yielding");
 688             yieldThread();
 689             continue;
 690         }
 691         cap->running_task = task;
 692
 693         if (cap->spare_workers) {
 694             // Look for workers that have died without removing
 695             // themselves from the list; this could happen if the OS
 696             // summarily killed the thread, for example.  This
 697             // actually happens on Windows when the system is
 698             // terminating the program, and the RTS is running in a
 699             // DLL.
 700             Task *t, *prev;
 701             prev = NULL;
 702             for (t = cap->spare_workers; t != NULL; t = t->next) {
 703                 if (!osThreadIsAlive(t->id)) {
 704                     debugTrace(DEBUG_sched,
 705                                "worker thread %p has died unexpectedly", (void *)t->id);
 706                         if (!prev) {
 707                             cap->spare_workers = t->next;
 708                         } else {
 709                             prev->next = t->next;
 710                         }
 711                         prev = t;
 712                 }
 713             }
 714         }
 715
 716         if (!emptyRunQueue(cap) || cap->spare_workers) {
 717             debugTrace(DEBUG_sched,
 718                        "runnable threads or workers still alive, yielding");
 719             releaseCapability_(cap); // this will wake up a worker
 720             RELEASE_LOCK(&cap->lock);
 721             yieldThread();
 722             continue;
 723         }
 724
 725         // If "safe", then busy-wait for any threads currently doing
 726         // foreign calls.  If we're about to unload this DLL, for
 727         // example, we need to be sure that there are no OS threads
 728         // that will try to return to code that has been unloaded.
 729         // We can be a bit more relaxed when this is a standalone
 730         // program that is about to terminate, and let safe=false.
 731         if (cap->suspended_ccalling_tasks && safe) {
 732             debugTrace(DEBUG_sched,
 733                        "thread(s) are involved in foreign calls, yielding");
 734             cap->running_task = NULL;
 735             RELEASE_LOCK(&cap->lock);
 736             yieldThread();
 737             continue;
 738         }
 739
 740         debugTrace(DEBUG_sched, "capability %d is stopped.", cap->no);
 741         freeCapability(cap);
 742         RELEASE_LOCK(&cap->lock);
 743         break;
 744     }
 745     // we now have the Capability, its run queue and spare workers
 746     // list are both empty.
 747
 748     // ToDo: we can't drop this mutex, because there might still be
 749     // threads performing foreign calls that will eventually try to
 750     // return via resumeThread() and attempt to grab cap->lock.
 751     // closeMutex(&cap->lock);
 752 }
 753
 754 /* ----------------------------------------------------------------------------
 755  * tryGrabCapability
 756  *
 757  * Attempt to gain control of a Capability if it is free.
 758  *
 759  * ------------------------------------------------------------------------- */
 760
 761 rtsBool
 762 tryGrabCapability (Capability *cap, Task *task)
 763 {
 764     if (cap->running_task != NULL) return rtsFalse;
 765     ACQUIRE_LOCK(&cap->lock);
 766     if (cap->running_task != NULL) {
 767         RELEASE_LOCK(&cap->lock);
 768         return rtsFalse;
 769     }
 770     task->cap = cap;
 771     cap->running_task = task;
 772     RELEASE_LOCK(&cap->lock);
 773     return rtsTrue;
 774 }
 775
 776
 777 #endif /* THREADED_RTS */
 778
 779 void
 780 freeCapability (Capability *cap) {
 781     stgFree(cap->mut_lists);
 782 #if defined(THREADED_RTS) || defined(PARALLEL_HASKELL)
 783     freeSparkPool(&cap->r.rSparks);
 784 #endif
 785 }
 786
 787 /* ---------------------------------------------------------------------------
 788    Mark everything directly reachable from the Capabilities.  When
 789    using multiple GC threads, each GC thread marks all Capabilities
 790    for which (c `mod` n == 0), for Capability c and thread n.
 791    ------------------------------------------------------------------------ */
 792
 793 void
 794 markSomeCapabilities (evac_fn evac, void *user, nat i0, nat delta)
 795 {
 796     nat i;
 797     Capability *cap;
 798     Task *task;
 799
 800     // Each GC thread is responsible for following roots from the
 801     // Capability of the same number.  There will usually be the same
 802     // or fewer Capabilities as GC threads, but just in case there
 803     // are more, we mark every Capability whose number is the GC
 804     // thread's index plus a multiple of the number of GC threads.
 805     for (i = i0; i < n_capabilities; i += delta) {
 806         cap = &capabilities[i];
 807         evac(user, (StgClosure **)(void *)&cap->run_queue_hd);
 808         evac(user, (StgClosure **)(void *)&cap->run_queue_tl);
 809 #if defined(THREADED_RTS)
 810         evac(user, (StgClosure **)(void *)&cap->wakeup_queue_hd);
 811         evac(user, (StgClosure **)(void *)&cap->wakeup_queue_tl);
 812 #endif
 813         for (task = cap->suspended_ccalling_tasks; task != NULL;
 814              task=task->next) {
 815             debugTrace(DEBUG_sched,
 816                        "evac'ing suspended TSO %lu", (unsigned long)task->suspended_tso->id);
 817             evac(user, (StgClosure **)(void *)&task->suspended_tso);
 818         }
 819
 820 #if defined(THREADED_RTS)
 821         markSparkQueue (evac, user, cap);
 822 #endif
 823     }
 824
 825 #if !defined(THREADED_RTS)
 826     evac(user, (StgClosure **)(void *)&blocked_queue_hd);
 827     evac(user, (StgClosure **)(void *)&blocked_queue_tl);
 828     evac(user, (StgClosure **)(void *)&sleeping_queue);
 829 #endif
 830 }
 831
 832 // This function is used by the compacting GC to thread all the
 833 // pointers from spark queues.
 834 void
 835 traverseSparkQueues (evac_fn evac USED_IF_THREADS, void *user USED_IF_THREADS)
 836 {
 837 #if defined(THREADED_RTS)
 838     nat i;
 839     for (i = 0; i < n_capabilities; i++) {
 840         traverseSparkQueue (evac, user, &capabilities[i]);
 841     }
 842 #endif // THREADED_RTS
 843
 844 }
 845
 846 void
 847 markCapabilities (evac_fn evac, void *user)
 848 {
 849     markSomeCapabilities(evac, user, 0, 1);
 850 }