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