X-Git-Url: http://git.megacz.com/?a=blobdiff_plain;f=rts%2FCapability.c;h=345bd39c0fb9560a555d3bb32caf20241b4e1f21;hb=8625c675de45bdb8bcfa795572ce7c47687c147c;hp=ce6ecebd7214fdf7f4884b14208d0d5cb1a2ed72;hpb=7effbbbbdfe7eb05c6402fa9337e358e7e9fadde;p=ghc-hetmet.git diff --git a/rts/Capability.c b/rts/Capability.c index ce6eceb..345bd39 100644 --- a/rts/Capability.c +++ b/rts/Capability.c @@ -62,9 +62,8 @@ Capability * rts_unsafeGetMyCapability (void) STATIC_INLINE rtsBool globalWorkToDo (void) { - return blackholes_need_checking - || sched_state >= SCHED_INTERRUPTING - ; + return sched_state >= SCHED_INTERRUPTING + || recent_activity == ACTIVITY_INACTIVE; // need to check for deadlock } #endif @@ -220,11 +219,11 @@ initCapability( Capability *cap, nat i ) initMutex(&cap->lock); cap->running_task = NULL; // indicates cap is free cap->spare_workers = NULL; + cap->n_spare_workers = 0; cap->suspended_ccalls = NULL; cap->returning_tasks_hd = NULL; cap->returning_tasks_tl = NULL; - cap->wakeup_queue_hd = END_TSO_QUEUE; - cap->wakeup_queue_tl = END_TSO_QUEUE; + cap->inbox = (Message*)END_TSO_QUEUE; cap->sparks_created = 0; cap->sparks_converted = 0; cap->sparks_pruned = 0; @@ -419,7 +418,7 @@ releaseCapability_ (Capability* cap, // If we have an unbound thread on the run queue, or if there's // anything else to do, give the Capability to a worker thread. if (always_wakeup || - !emptyRunQueue(cap) || !emptyWakeupQueue(cap) || + !emptyRunQueue(cap) || !emptyInbox(cap) || !emptySparkPoolCap(cap) || globalWorkToDo()) { if (cap->spare_workers) { giveCapabilityToTask(cap,cap->spare_workers); @@ -457,14 +456,33 @@ releaseCapabilityAndQueueWorker (Capability* cap USED_IF_THREADS) task = cap->running_task; + // If the Task is stopped, we shouldn't be yielding, we should + // be just exiting. + ASSERT(!task->stopped); + // If the current task is a worker, save it on the spare_workers // list of this Capability. A worker can mark itself as stopped, // in which case it is not replaced on the spare_worker queue. // This happens when the system is shutting down (see // Schedule.c:workerStart()). - if (!isBoundTask(task) && !task->stopped) { - task->next = cap->spare_workers; - cap->spare_workers = task; + if (!isBoundTask(task)) + { + if (cap->n_spare_workers < MAX_SPARE_WORKERS) + { + task->next = cap->spare_workers; + cap->spare_workers = task; + cap->n_spare_workers++; + } + else + { + debugTrace(DEBUG_sched, "%d spare workers already, exiting", + cap->n_spare_workers); + releaseCapability_(cap,rtsFalse); + // hold the lock until after workerTaskStop; c.f. scheduleWorker() + workerTaskStop(task); + RELEASE_LOCK(&cap->lock); + shutdownThread(); + } } // Bound tasks just float around attached to their TSOs. @@ -621,7 +639,8 @@ yieldCapability (Capability** pCap, Task *task) } cap->spare_workers = task->next; task->next = NULL; - } + cap->n_spare_workers--; + } cap->running_task = task; RELEASE_LOCK(&cap->lock); break; @@ -638,50 +657,6 @@ yieldCapability (Capability** pCap, Task *task) } /* ---------------------------------------------------------------------------- - * Wake up a thread on a Capability. - * - * This is used when the current Task is running on a Capability and - * wishes to wake up a thread on a different Capability. - * ------------------------------------------------------------------------- */ - -void -wakeupThreadOnCapability (Capability *my_cap, - Capability *other_cap, - StgTSO *tso) -{ - ACQUIRE_LOCK(&other_cap->lock); - - // ASSUMES: cap->lock is held (asserted in wakeupThreadOnCapability) - if (tso->bound) { - ASSERT(tso->bound->task->cap == tso->cap); - tso->bound->task->cap = other_cap; - } - tso->cap = other_cap; - - ASSERT(tso->bound ? tso->bound->task->cap == other_cap : 1); - - if (other_cap->running_task == NULL) { - // nobody is running this Capability, we can add our thread - // directly onto the run queue and start up a Task to run it. - - other_cap->running_task = myTask(); - // precond for releaseCapability_() and appendToRunQueue() - - appendToRunQueue(other_cap,tso); - - releaseCapability_(other_cap,rtsFalse); - } else { - appendToWakeupQueue(my_cap,other_cap,tso); - other_cap->context_switch = 1; - // someone is running on this Capability, so it cannot be - // freed without first checking the wakeup queue (see - // releaseCapability_). - } - - RELEASE_LOCK(&other_cap->lock); -} - -/* ---------------------------------------------------------------------------- * prodCapability * * If a Capability is currently idle, wake up a Task on it. Used to @@ -754,12 +729,13 @@ shutdownCapability (Capability *cap, Task *task, rtsBool safe) if (!osThreadIsAlive(t->id)) { debugTrace(DEBUG_sched, "worker thread %p has died unexpectedly", (void *)t->id); - if (!prev) { - cap->spare_workers = t->next; - } else { - prev->next = t->next; - } - prev = t; + cap->n_spare_workers--; + if (!prev) { + cap->spare_workers = t->next; + } else { + prev->next = t->next; + } + prev = t; } } } @@ -865,7 +841,7 @@ freeCapabilities (void) void markSomeCapabilities (evac_fn evac, void *user, nat i0, nat delta, - rtsBool prune_sparks USED_IF_THREADS) + rtsBool no_mark_sparks USED_IF_THREADS) { nat i; Capability *cap; @@ -881,8 +857,7 @@ markSomeCapabilities (evac_fn evac, void *user, nat i0, nat delta, evac(user, (StgClosure **)(void *)&cap->run_queue_hd); evac(user, (StgClosure **)(void *)&cap->run_queue_tl); #if defined(THREADED_RTS) - evac(user, (StgClosure **)(void *)&cap->wakeup_queue_hd); - evac(user, (StgClosure **)(void *)&cap->wakeup_queue_tl); + evac(user, (StgClosure **)(void *)&cap->inbox); #endif for (incall = cap->suspended_ccalls; incall != NULL; incall=incall->next) { @@ -890,9 +865,7 @@ markSomeCapabilities (evac_fn evac, void *user, nat i0, nat delta, } #if defined(THREADED_RTS) - if (prune_sparks) { - pruneSparkQueue (evac, user, cap); - } else { + if (!no_mark_sparks) { traverseSparkQueue (evac, user, cap); } #endif @@ -910,3 +883,8 @@ markCapabilities (evac_fn evac, void *user) { markSomeCapabilities(evac, user, 0, 1, rtsFalse); } + +/* ----------------------------------------------------------------------------- + Messages + -------------------------------------------------------------------------- */ +