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