// locking, so we don't do that.
Capability *last_free_capability;
+/* GC indicator, in scope for the scheduler, init'ed to false */
+volatile StgWord waiting_for_gc = 0;
+
#if defined(THREADED_RTS)
STATIC_INLINE rtsBool
globalWorkToDo (void)
#endif
#if defined(THREADED_RTS)
-STATIC_INLINE rtsBool
-anyWorkForMe( Capability *cap, Task *task )
+StgClosure *
+stealWork (Capability *cap)
{
- if (task->tso != NULL) {
- // A bound task only runs if its thread is on the run queue of
- // the capability on which it was woken up. Otherwise, we
- // can't be sure that we have the right capability: the thread
- // might be woken up on some other capability, and task->cap
- // could change under our feet.
- return !emptyRunQueue(cap) && cap->run_queue_hd->bound == task;
- } else {
- // A vanilla worker task runs if either there is a lightweight
- // thread at the head of the run queue, or the run queue is
- // empty and (there are sparks to execute, or there is some
- // other global condition to check, such as threads blocked on
- // blackholes).
- if (emptyRunQueue(cap)) {
- return !emptySparkPoolCap(cap)
- || !emptyWakeupQueue(cap)
- || globalWorkToDo();
- } else
- return cap->run_queue_hd->bound == NULL;
+ /* use the normal Sparks.h interface (internally modified to enable
+ concurrent stealing)
+ and immediately turn the spark into a thread when successful
+ */
+ Capability *robbed;
+ StgClosurePtr spark;
+ rtsBool retry;
+ nat i = 0;
+
+ debugTrace(DEBUG_sched,
+ "cap %d: Trying to steal work from other capabilities",
+ cap->no);
+
+ if (n_capabilities == 1) { return NULL; } // makes no sense...
+
+ do {
+ retry = rtsFalse;
+
+ /* visit cap.s 0..n-1 in sequence until a theft succeeds. We could
+ start at a random place instead of 0 as well. */
+ for ( i=0 ; i < n_capabilities ; i++ ) {
+ robbed = &capabilities[i];
+ if (cap == robbed) // ourselves...
+ continue;
+
+ if (emptySparkPoolCap(robbed)) // nothing to steal here
+ continue;
+
+ spark = tryStealSpark(robbed);
+ if (spark == NULL && !emptySparkPoolCap(robbed)) {
+ // we conflicted with another thread while trying to steal;
+ // try again later.
+ retry = rtsTrue;
+ }
+
+ if (spark != NULL) {
+ debugTrace(DEBUG_sched,
+ "cap %d: Stole a spark from capability %d",
+ cap->no, robbed->no);
+ return spark;
+ }
+ // otherwise: no success, try next one
+ }
+ } while (retry);
+
+ debugTrace(DEBUG_sched, "No sparks stolen");
+ return NULL;
+}
+
+// Returns True if any spark pool is non-empty at this moment in time
+// The result is only valid for an instant, of course, so in a sense
+// is immediately invalid, and should not be relied upon for
+// correctness.
+rtsBool
+anySparks (void)
+{
+ nat i;
+
+ for (i=0; i < n_capabilities; i++) {
+ if (!emptySparkPoolCap(&capabilities[i])) {
+ return rtsTrue;
+ }
}
+ return rtsFalse;
}
#endif
cap->returning_tasks_tl = NULL;
cap->wakeup_queue_hd = END_TSO_QUEUE;
cap->wakeup_queue_tl = END_TSO_QUEUE;
+ cap->sparks_created = 0;
+ cap->sparks_converted = 0;
+ cap->sparks_pruned = 0;
#endif
cap->f.stgGCEnter1 = (F_)__stg_gc_enter_1;
cap->free_trec_chunks = END_STM_CHUNK_LIST;
cap->free_trec_headers = NO_TREC;
cap->transaction_tokens = 0;
+ cap->context_switch = 0;
}
/* ---------------------------------------------------------------------------
}
/* ----------------------------------------------------------------------------
+ * setContextSwitches: cause all capabilities to context switch as
+ * soon as possible.
+ * ------------------------------------------------------------------------- */
+
+void setContextSwitches(void)
+{
+ nat i;
+ for (i=0; i < n_capabilities; i++) {
+ capabilities[i].context_switch = 1;
+ }
+}
+
+/* ----------------------------------------------------------------------------
* Give a Capability to a Task. The task must currently be sleeping
* on its condition variable.
*
#if defined(THREADED_RTS)
void
-releaseCapability_ (Capability* cap)
+releaseCapability_ (Capability* cap,
+ rtsBool always_wakeup)
{
Task *task;
return;
}
+ /* if waiting_for_gc was the reason to release the cap: thread
+ comes from yieldCap->releaseAndQueueWorker. Unconditionally set
+ cap. free and return (see default after the if-protected other
+ special cases). Thread will wait on cond.var and re-acquire the
+ same cap after GC (GC-triggering cap. calls releaseCap and
+ enters the spare_workers case)
+ */
+ if (waiting_for_gc) {
+ last_free_capability = cap; // needed?
+ trace(TRACE_sched | DEBUG_sched,
+ "GC pending, set capability %d free", cap->no);
+ return;
+ }
+
+
// If the next thread on the run queue is a bound thread,
// give this Capability to the appropriate Task.
if (!emptyRunQueue(cap) && cap->run_queue_hd->bound) {
// 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 (!emptyRunQueue(cap) || !emptyWakeupQueue(cap)
- || !emptySparkPoolCap(cap) || globalWorkToDo()) {
+ if (always_wakeup ||
+ !emptyRunQueue(cap) || !emptyWakeupQueue(cap) ||
+ !emptySparkPoolCap(cap) || globalWorkToDo()) {
if (cap->spare_workers) {
giveCapabilityToTask(cap,cap->spare_workers);
// The worker Task pops itself from the queue;
releaseCapability (Capability* cap USED_IF_THREADS)
{
ACQUIRE_LOCK(&cap->lock);
- releaseCapability_(cap);
+ releaseCapability_(cap, rtsFalse);
+ RELEASE_LOCK(&cap->lock);
+}
+
+void
+releaseAndWakeupCapability (Capability* cap USED_IF_THREADS)
+{
+ ACQUIRE_LOCK(&cap->lock);
+ releaseCapability_(cap, rtsTrue);
RELEASE_LOCK(&cap->lock);
}
}
// Bound tasks just float around attached to their TSOs.
- releaseCapability_(cap);
+ releaseCapability_(cap,rtsFalse);
RELEASE_LOCK(&cap->lock);
}
{
Capability *cap = *pCap;
- // The fast path has no locking, if we don't enter this while loop
-
- while ( cap->returning_tasks_hd != NULL || !anyWorkForMe(cap,task) ) {
debugTrace(DEBUG_sched, "giving up capability %d", cap->no);
// We must now release the capability and wait to be woken up
trace(TRACE_sched | DEBUG_sched, "resuming capability %d", cap->no);
ASSERT(cap->running_task == task);
- }
*pCap = cap;
* ------------------------------------------------------------------------- */
void
-wakeupThreadOnCapability (Capability *cap, StgTSO *tso)
+wakeupThreadOnCapability (Capability *my_cap,
+ Capability *other_cap,
+ StgTSO *tso)
{
- ASSERT(tso->cap == cap);
- ASSERT(tso->bound ? tso->bound->cap == cap : 1);
- ASSERT_LOCK_HELD(&cap->lock);
+ ACQUIRE_LOCK(&other_cap->lock);
+
+ // ASSUMES: cap->lock is held (asserted in wakeupThreadOnCapability)
+ if (tso->bound) {
+ ASSERT(tso->bound->cap == tso->cap);
+ tso->bound->cap = other_cap;
+ }
+ tso->cap = other_cap;
- tso->cap = cap;
+ ASSERT(tso->bound ? tso->bound->cap == other_cap : 1);
- if (cap->running_task == NULL) {
+ 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.
- appendToRunQueue(cap,tso);
- // start it up
- cap->running_task = myTask(); // precond for releaseCapability_()
- trace(TRACE_sched, "resuming capability %d", cap->no);
- releaseCapability_(cap);
+ other_cap->running_task = myTask();
+ // precond for releaseCapability_() and appendToRunQueue()
+
+ appendToRunQueue(other_cap,tso);
+
+ trace(TRACE_sched, "resuming capability %d", other_cap->no);
+ releaseCapability_(other_cap,rtsFalse);
} else {
- appendToWakeupQueue(cap,tso);
+ 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_).
}
-}
-
-void
-wakeupThreadOnCapability_lock (Capability *cap, StgTSO *tso)
-{
- ACQUIRE_LOCK(&cap->lock);
- migrateThreadToCapability (cap, tso);
- RELEASE_LOCK(&cap->lock);
-}
-
-void
-migrateThreadToCapability (Capability *cap, StgTSO *tso)
-{
- // ASSUMES: cap->lock is held (asserted in wakeupThreadOnCapability)
- if (tso->bound) {
- ASSERT(tso->bound->cap == tso->cap);
- tso->bound->cap = cap;
- }
- tso->cap = cap;
- wakeupThreadOnCapability(cap,tso);
-}
-void
-migrateThreadToCapability_lock (Capability *cap, StgTSO *tso)
-{
- ACQUIRE_LOCK(&cap->lock);
- migrateThreadToCapability (cap, tso);
- RELEASE_LOCK(&cap->lock);
+ RELEASE_LOCK(&other_cap->lock);
}
/* ----------------------------------------------------------------------------
* ------------------------------------------------------------------------- */
void
-shutdownCapability (Capability *cap, Task *task)
+shutdownCapability (Capability *cap, Task *task, rtsBool safe)
{
nat i;
for (t = cap->spare_workers; t != NULL; t = t->next) {
if (!osThreadIsAlive(t->id)) {
debugTrace(DEBUG_sched,
- "worker thread %p has died unexpectedly", t->id);
+ "worker thread %p has died unexpectedly", (void *)t->id);
if (!prev) {
cap->spare_workers = t->next;
} else {
if (!emptyRunQueue(cap) || cap->spare_workers) {
debugTrace(DEBUG_sched,
"runnable threads or workers still alive, yielding");
- releaseCapability_(cap); // this will wake up a worker
+ releaseCapability_(cap,rtsFalse); // this will wake up a worker
RELEASE_LOCK(&cap->lock);
yieldThread();
continue;
}
+
+ // If "safe", then busy-wait for any threads currently doing
+ // foreign calls. If we're about to unload this DLL, for
+ // example, we need to be sure that there are no OS threads
+ // that will try to return to code that has been unloaded.
+ // We can be a bit more relaxed when this is a standalone
+ // program that is about to terminate, and let safe=false.
+ if (cap->suspended_ccalling_tasks && safe) {
+ debugTrace(DEBUG_sched,
+ "thread(s) are involved in foreign calls, yielding");
+ cap->running_task = NULL;
+ RELEASE_LOCK(&cap->lock);
+ yieldThread();
+ continue;
+ }
+
debugTrace(DEBUG_sched, "capability %d is stopped.", cap->no);
- freeCapability(cap);
RELEASE_LOCK(&cap->lock);
break;
}
#endif /* THREADED_RTS */
-void
-freeCapability (Capability *cap) {
+static void
+freeCapability (Capability *cap)
+{
stgFree(cap->mut_lists);
#if defined(THREADED_RTS) || defined(PARALLEL_HASKELL)
- freeSparkPool(&cap->r.rSparks);
+ freeSparkPool(cap->sparks);
+#endif
+}
+
+void
+freeCapabilities (void)
+{
+#if defined(THREADED_RTS)
+ nat i;
+ for (i=0; i < n_capabilities; i++) {
+ freeCapability(&capabilities[i]);
+ }
+#else
+ freeCapability(&MainCapability);
+#endif
+}
+
+/* ---------------------------------------------------------------------------
+ Mark everything directly reachable from the Capabilities. When
+ using multiple GC threads, each GC thread marks all Capabilities
+ for which (c `mod` n == 0), for Capability c and thread n.
+ ------------------------------------------------------------------------ */
+
+void
+markSomeCapabilities (evac_fn evac, void *user, nat i0, nat delta,
+ rtsBool prune_sparks USED_IF_THREADS)
+{
+ nat i;
+ Capability *cap;
+ Task *task;
+
+ // Each GC thread is responsible for following roots from the
+ // Capability of the same number. There will usually be the same
+ // or fewer Capabilities as GC threads, but just in case there
+ // are more, we mark every Capability whose number is the GC
+ // thread's index plus a multiple of the number of GC threads.
+ for (i = i0; i < n_capabilities; i += delta) {
+ cap = &capabilities[i];
+ 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);
#endif
+ for (task = cap->suspended_ccalling_tasks; task != NULL;
+ task=task->next) {
+ debugTrace(DEBUG_sched,
+ "evac'ing suspended TSO %lu", (unsigned long)task->suspended_tso->id);
+ evac(user, (StgClosure **)(void *)&task->suspended_tso);
+ }
+
+#if defined(THREADED_RTS)
+ if (prune_sparks) {
+ pruneSparkQueue (evac, user, cap);
+ } else {
+ traverseSparkQueue (evac, user, cap);
+ }
+#endif
+ }
+
+#if !defined(THREADED_RTS)
+ evac(user, (StgClosure **)(void *)&blocked_queue_hd);
+ evac(user, (StgClosure **)(void *)&blocked_queue_tl);
+ evac(user, (StgClosure **)(void *)&sleeping_queue);
+#endif
}
+void
+markCapabilities (evac_fn evac, void *user)
+{
+ markSomeCapabilities(evac, user, 0, 1, rtsFalse);
+}
projects / ghc-hetmet.git / blobdiff