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