X-Git-Url: http://git.megacz.com/?p=ghc-hetmet.git;a=blobdiff_plain;f=rts%2FCapability.c;h=fe5dbdca40536b1cfecca1a8579855b2ef416d20;hp=0012c24cc862eecfee242b5c12e081bd3f4262c9;hb=b2524b3960999fffdb3767900f58825903f6560f;hpb=5270423a6afe69f1dc57e5e5a474812182718d40 diff --git a/rts/Capability.c b/rts/Capability.c index 0012c24..fe5dbdc 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 @@ -98,7 +97,7 @@ findSpark (Capability *cap) cap->sparks_converted++; // Post event for running a spark from capability's own pool. - traceSchedEvent(cap, EVENT_RUN_SPARK, cap->r.rCurrentTSO, 0); + traceEventRunSpark(cap, cap->r.rCurrentTSO); return spark; } @@ -132,8 +131,7 @@ findSpark (Capability *cap) if (spark != NULL) { cap->sparks_converted++; - traceSchedEvent(cap, EVENT_STEAL_SPARK, - cap->r.rCurrentTSO, robbed->no); + traceEventStealSpark(cap, cap->r.rCurrentTSO, robbed->no); return spark; } @@ -174,10 +172,10 @@ STATIC_INLINE void newReturningTask (Capability *cap, Task *task) { ASSERT_LOCK_HELD(&cap->lock); - ASSERT(task->return_link == NULL); + ASSERT(task->next == NULL); if (cap->returning_tasks_hd) { - ASSERT(cap->returning_tasks_tl->return_link == NULL); - cap->returning_tasks_tl->return_link = task; + ASSERT(cap->returning_tasks_tl->next == NULL); + cap->returning_tasks_tl->next = task; } else { cap->returning_tasks_hd = task; } @@ -191,11 +189,11 @@ popReturningTask (Capability *cap) Task *task; task = cap->returning_tasks_hd; ASSERT(task); - cap->returning_tasks_hd = task->return_link; + cap->returning_tasks_hd = task->next; if (!cap->returning_tasks_hd) { cap->returning_tasks_tl = NULL; } - task->return_link = NULL; + task->next = NULL; return task; } #endif @@ -213,7 +211,6 @@ initCapability( Capability *cap, nat i ) cap->no = i; cap->in_haskell = rtsFalse; - cap->in_gc = rtsFalse; cap->run_queue_hd = END_TSO_QUEUE; cap->run_queue_tl = END_TSO_QUEUE; @@ -222,14 +219,16 @@ initCapability( Capability *cap, nat i ) initMutex(&cap->lock); cap->running_task = NULL; // indicates cap is free cap->spare_workers = NULL; - cap->suspended_ccalling_tasks = 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_dud = 0; cap->sparks_converted = 0; - cap->sparks_pruned = 0; + cap->sparks_gcd = 0; + cap->sparks_fizzled = 0; #endif cap->f.stgEagerBlackholeInfo = (W_)&__stg_EAGER_BLACKHOLE_info; @@ -254,6 +253,8 @@ initCapability( Capability *cap, nat i ) cap->transaction_tokens = 0; cap->context_switch = 0; cap->pinned_object_block = NULL; + + traceCapsetAssignCap(CAPSET_OSPROCESS_DEFAULT, i); } /* --------------------------------------------------------------------------- @@ -267,6 +268,10 @@ initCapability( Capability *cap, nat i ) void initCapabilities( void ) { + /* Declare a single capability set representing the process. + Each capability will get added to this capset. */ + traceCapsetCreate(CAPSET_OSPROCESS_DEFAULT, CapsetTypeOsProcess); + #if defined(THREADED_RTS) nat i; @@ -344,7 +349,7 @@ giveCapabilityToTask (Capability *cap USED_IF_DEBUG, Task *task) ASSERT_LOCK_HELD(&cap->lock); ASSERT(task->cap == cap); debugTrace(DEBUG_sched, "passing capability %d to %s %p", - cap->no, task->tso ? "bound task" : "worker", + cap->no, task->incall->tso ? "bound task" : "worker", (void *)task->id); ACQUIRE_LOCK(&task->lock); task->wakeup = rtsTrue; @@ -396,8 +401,11 @@ releaseCapability_ (Capability* cap, // give this Capability to the appropriate Task. if (!emptyRunQueue(cap) && cap->run_queue_hd->bound) { // Make sure we're not about to try to wake ourselves up - ASSERT(task != cap->run_queue_hd->bound); - task = cap->run_queue_hd->bound; + // ASSERT(task != cap->run_queue_hd->bound); + // assertion is false: in schedule() we force a yield after + // ThreadBlocked, but the thread may be back on the run queue + // by now. + task = cap->run_queue_hd->bound->task; giveCapabilityToTask(cap,task); return; } @@ -410,7 +418,7 @@ releaseCapability_ (Capability* cap, if (sched_state < SCHED_SHUTTING_DOWN || !emptyRunQueue(cap)) { debugTrace(DEBUG_sched, "starting new worker on capability %d", cap->no); - startWorkerTask(cap, workerStart); + startWorkerTask(cap); return; } } @@ -418,7 +426,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); @@ -456,16 +464,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()). - // Also, be careful to check that this task hasn't just exited - // Haskell to do a foreign call (task->suspended_tso). - if (!isBoundTask(task) && !task->stopped && !task->suspended_tso) { - 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. @@ -580,9 +605,9 @@ yieldCapability (Capability** pCap, Task *task) Capability *cap = *pCap; if (waiting_for_gc == PENDING_GC_PAR) { - traceSchedEvent(cap, EVENT_GC_START, 0, 0); + traceEventGcStart(cap); gcWorkerThread(cap); - traceSchedEvent(cap, EVENT_GC_END, 0, 0); + traceEventGcEnd(cap); return; } @@ -611,7 +636,7 @@ yieldCapability (Capability** pCap, Task *task) continue; } - if (task->tso == NULL) { + if (task->incall->tso == NULL) { ASSERT(cap->spare_workers != NULL); // if we're not at the front of the queue, release it // again. This is unlikely to happen. @@ -622,7 +647,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; @@ -639,67 +665,48 @@ yieldCapability (Capability** pCap, Task *task) } /* ---------------------------------------------------------------------------- - * Wake up a thread on a Capability. + * prodCapability * - * This is used when the current Task is running on a Capability and - * wishes to wake up a thread on a different Capability. + * If a Capability is currently idle, wake up a Task on it. Used to + * get every Capability into the GC. * ------------------------------------------------------------------------- */ void -wakeupThreadOnCapability (Capability *my_cap, - Capability *other_cap, - StgTSO *tso) +prodCapability (Capability *cap, Task *task) { - 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; - - ASSERT(tso->bound ? tso->bound->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_). + ACQUIRE_LOCK(&cap->lock); + if (!cap->running_task) { + cap->running_task = task; + releaseCapability_(cap,rtsTrue); } - - RELEASE_LOCK(&other_cap->lock); + RELEASE_LOCK(&cap->lock); } /* ---------------------------------------------------------------------------- - * prodCapability + * tryGrabCapability + * + * Attempt to gain control of a Capability if it is free. * - * If a Capability is currently idle, wake up a Task on it. Used to - * get every Capability into the GC. * ------------------------------------------------------------------------- */ -void -prodCapability (Capability *cap, Task *task) +rtsBool +tryGrabCapability (Capability *cap, Task *task) { + if (cap->running_task != NULL) return rtsFalse; ACQUIRE_LOCK(&cap->lock); - if (!cap->running_task) { - cap->running_task = task; - releaseCapability_(cap,rtsTrue); + if (cap->running_task != NULL) { + RELEASE_LOCK(&cap->lock); + return rtsFalse; } + task->cap = cap; + cap->running_task = task; RELEASE_LOCK(&cap->lock); + return rtsTrue; } + +#endif /* THREADED_RTS */ + /* ---------------------------------------------------------------------------- * shutdownCapability * @@ -716,8 +723,11 @@ prodCapability (Capability *cap, Task *task) * ------------------------------------------------------------------------- */ void -shutdownCapability (Capability *cap, Task *task, rtsBool safe) +shutdownCapability (Capability *cap, + Task *task USED_IF_THREADS, + rtsBool safe USED_IF_THREADS) { +#if defined(THREADED_RTS) nat i; task->cap = cap; @@ -755,12 +765,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; } } } @@ -780,16 +791,24 @@ shutdownCapability (Capability *cap, Task *task, rtsBool safe) // 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) { + if (cap->suspended_ccalls && safe) { debugTrace(DEBUG_sched, "thread(s) are involved in foreign calls, yielding"); cap->running_task = NULL; RELEASE_LOCK(&cap->lock); + // The IO manager thread might have been slow to start up, + // so the first attempt to kill it might not have + // succeeded. Just in case, try again - the kill message + // will only be sent once. + // + // To reproduce this deadlock: run ffi002(threaded1) + // repeatedly on a loaded machine. + ioManagerDie(); yieldThread(); continue; } - - traceSchedEvent(cap, EVENT_SHUTDOWN, 0, 0); + + traceEventShutdown(cap); RELEASE_LOCK(&cap->lock); break; } @@ -800,33 +819,23 @@ shutdownCapability (Capability *cap, Task *task, rtsBool safe) // threads performing foreign calls that will eventually try to // return via resumeThread() and attempt to grab cap->lock. // closeMutex(&cap->lock); -} + +#endif /* THREADED_RTS */ -/* ---------------------------------------------------------------------------- - * tryGrabCapability - * - * Attempt to gain control of a Capability if it is free. - * - * ------------------------------------------------------------------------- */ + traceCapsetRemoveCap(CAPSET_OSPROCESS_DEFAULT, cap->no); +} -rtsBool -tryGrabCapability (Capability *cap, Task *task) +void +shutdownCapabilities(Task *task, rtsBool safe) { - if (cap->running_task != NULL) return rtsFalse; - ACQUIRE_LOCK(&cap->lock); - if (cap->running_task != NULL) { - RELEASE_LOCK(&cap->lock); - return rtsFalse; + nat i; + for (i=0; i < n_capabilities; i++) { + ASSERT(task->incall->tso == NULL); + shutdownCapability(&capabilities[i], task, safe); } - task->cap = cap; - cap->running_task = task; - RELEASE_LOCK(&cap->lock); - return rtsTrue; + traceCapsetDelete(CAPSET_OSPROCESS_DEFAULT); } - -#endif /* THREADED_RTS */ - static void freeCapability (Capability *cap) { @@ -857,49 +866,41 @@ freeCapabilities (void) ------------------------------------------------------------------------ */ void -markSomeCapabilities (evac_fn evac, void *user, nat i0, nat delta, - rtsBool prune_sparks USED_IF_THREADS) +markCapability (evac_fn evac, void *user, Capability *cap, + rtsBool no_mark_sparks USED_IF_THREADS) { - nat i; - Capability *cap; - Task *task; + InCall *incall; // 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); + 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 (task = cap->suspended_ccalling_tasks; task != NULL; - task=task->next) { - evac(user, (StgClosure **)(void *)&task->suspended_tso); - } + for (incall = cap->suspended_ccalls; incall != NULL; + incall=incall->next) { + evac(user, (StgClosure **)(void *)&incall->suspended_tso); + } #if defined(THREADED_RTS) - if (prune_sparks) { - pruneSparkQueue (evac, user, cap); - } else { - traverseSparkQueue (evac, user, cap); - } -#endif + if (!no_mark_sparks) { + 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 + // Free STM structures for this Capability + stmPreGCHook(cap); } void markCapabilities (evac_fn evac, void *user) { - markSomeCapabilities(evac, user, 0, 1, rtsFalse); + nat n; + for (n = 0; n < n_capabilities; n++) { + markCapability(evac, user, &capabilities[n], rtsFalse); + } }