X-Git-Url: http://git.megacz.com/?a=blobdiff_plain;f=rts%2FSchedule.c;h=915530f5d2d220fec7691b88e702a04bc10e58c8;hb=200c73fdfea734765c48309cc8dcbcf44b69c8c5;hp=5ebb685a027d750690dacbabb435efb0cc3c5411;hpb=45202530612593a0ba7a6c559a38dc1ff26670a4;p=ghc-hetmet.git diff --git a/rts/Schedule.c b/rts/Schedule.c index 5ebb685..915530f 100644 --- a/rts/Schedule.c +++ b/rts/Schedule.c @@ -118,12 +118,6 @@ StgTSO *blackhole_queue = NULL; */ rtsBool blackholes_need_checking = rtsFalse; -/* Linked list of all threads. - * Used for detecting garbage collected threads. - * LOCK: sched_mutex+capability, or all capabilities - */ -StgTSO *all_threads = NULL; - /* flag set by signal handler to precipitate a context switch * LOCK: none (just an advisory flag) */ @@ -222,6 +216,7 @@ static Capability *scheduleDoGC(Capability *cap, Task *task, static rtsBool checkBlackHoles(Capability *cap); static StgTSO *threadStackOverflow(Capability *cap, StgTSO *tso); +static StgTSO *threadStackUnderflow(Task *task, StgTSO *tso); static void deleteThread (Capability *cap, StgTSO *tso); static void deleteAllThreads (Capability *cap); @@ -591,14 +586,14 @@ run_thread: cap->in_haskell = rtsTrue; - dirtyTSO(t); + dirty_TSO(cap,t); #if defined(THREADED_RTS) if (recent_activity == ACTIVITY_DONE_GC) { // ACTIVITY_DONE_GC means we turned off the timer signal to // conserve power (see #1623). Re-enable it here. nat prev; - prev = xchg(&recent_activity, ACTIVITY_YES); + prev = xchg((P_)&recent_activity, ACTIVITY_YES); if (prev == ACTIVITY_DONE_GC) { startTimer(); } @@ -683,6 +678,8 @@ run_thread: schedulePostRunThread(); + t = threadStackUnderflow(task,t); + ready_to_gc = rtsFalse; switch (ret) { @@ -718,9 +715,6 @@ run_thread: cap = scheduleDoGC(cap,task,rtsFalse); } } /* end of while() */ - - debugTrace(PAR_DEBUG_verbose, - "== Leaving schedule() after having received Finish"); } /* ---------------------------------------------------------------------------- @@ -768,7 +762,7 @@ schedulePushWork(Capability *cap USED_IF_THREADS, // Check whether we have more threads on our run queue, or sparks // in our pool, that we could hand to another Capability. - if ((emptyRunQueue(cap) || cap->run_queue_hd->link == END_TSO_QUEUE) + if ((emptyRunQueue(cap) || cap->run_queue_hd->_link == END_TSO_QUEUE) && sparkPoolSizeCap(cap) < 2) { return; } @@ -809,21 +803,21 @@ schedulePushWork(Capability *cap USED_IF_THREADS, if (cap->run_queue_hd != END_TSO_QUEUE) { prev = cap->run_queue_hd; - t = prev->link; - prev->link = END_TSO_QUEUE; + t = prev->_link; + prev->_link = END_TSO_QUEUE; for (; t != END_TSO_QUEUE; t = next) { - next = t->link; - t->link = END_TSO_QUEUE; + next = t->_link; + t->_link = END_TSO_QUEUE; if (t->what_next == ThreadRelocated || t->bound == task // don't move my bound thread || tsoLocked(t)) { // don't move a locked thread - prev->link = t; + setTSOLink(cap, prev, t); prev = t; } else if (i == n_free_caps) { pushed_to_all = rtsTrue; i = 0; // keep one for us - prev->link = t; + setTSOLink(cap, prev, t); prev = t; } else { debugTrace(DEBUG_sched, "pushing thread %lu to capability %d", (unsigned long)t->id, free_caps[i]->no); @@ -919,7 +913,7 @@ scheduleCheckWakeupThreads(Capability *cap USED_IF_THREADS) cap->run_queue_hd = cap->wakeup_queue_hd; cap->run_queue_tl = cap->wakeup_queue_tl; } else { - cap->run_queue_tl->link = cap->wakeup_queue_hd; + setTSOLink(cap, cap->run_queue_tl, cap->wakeup_queue_hd); cap->run_queue_tl = cap->wakeup_queue_tl; } cap->wakeup_queue_hd = cap->wakeup_queue_tl = END_TSO_QUEUE; @@ -1634,7 +1628,16 @@ scheduleHandleHeapOverflow( Capability *cap, StgTSO *t ) } #endif - pushOnRunQueue(cap,t); + if (context_switch) { + // Sometimes we miss a context switch, e.g. when calling + // primitives in a tight loop, MAYBE_GC() doesn't check the + // context switch flag, and we end up waiting for a GC. + // See #1984, and concurrent/should_run/1984 + context_switch = 0; + addToRunQueue(cap,t); + } else { + pushOnRunQueue(cap,t); + } return rtsTrue; /* actual GC is done at the end of the while loop in schedule() */ } @@ -1702,7 +1705,7 @@ scheduleHandleYield( Capability *cap, StgTSO *t, nat prev_what_next ) IF_DEBUG(sanity, //debugBelch("&& Doing sanity check on yielding TSO %ld.", t->id); checkTSO(t)); - ASSERT(t->link == END_TSO_QUEUE); + ASSERT(t->_link == END_TSO_QUEUE); // Shortcut if we're just switching evaluators: don't bother // doing stack squeezing (which can be expensive), just run the @@ -1889,7 +1892,7 @@ scheduleHandleThreadFinished (Capability *cap STG_UNUSED, Task *task, StgTSO *t) // point where we can deal with this. Leaving it on the run // queue also ensures that the garbage collector knows about // this thread and its return value (it gets dropped from the - // all_threads list so there's no other way to find it). + // step->threads list so there's no other way to find it). appendToRunQueue(cap,t); return rtsFalse; #else @@ -2007,10 +2010,12 @@ scheduleDoGC (Capability *cap, Task *task USED_IF_THREADS, rtsBool force_major) */ { StgTSO *next; + nat s; - for (t = all_threads; t != END_TSO_QUEUE; t = next) { + for (s = 0; s < total_steps; s++) { + for (t = all_steps[s].threads; t != END_TSO_QUEUE; t = next) { if (t->what_next == ThreadRelocated) { - next = t->link; + next = t->_link; } else { next = t->global_link; @@ -2043,6 +2048,7 @@ scheduleDoGC (Capability *cap, Task *task USED_IF_THREADS, rtsBool force_major) } } } + } } } @@ -2124,6 +2130,7 @@ forkProcess(HsStablePtr *entry pid_t pid; StgTSO* t,*next; Capability *cap; + nat s; #if defined(THREADED_RTS) if (RtsFlags.ParFlags.nNodes > 1) { @@ -2137,25 +2144,44 @@ forkProcess(HsStablePtr *entry // ToDo: for SMP, we should probably acquire *all* the capabilities cap = rts_lock(); + // no funny business: hold locks while we fork, otherwise if some + // other thread is holding a lock when the fork happens, the data + // structure protected by the lock will forever be in an + // inconsistent state in the child. See also #1391. + ACQUIRE_LOCK(&sched_mutex); + ACQUIRE_LOCK(&cap->lock); + ACQUIRE_LOCK(&cap->running_task->lock); + pid = fork(); if (pid) { // parent + RELEASE_LOCK(&sched_mutex); + RELEASE_LOCK(&cap->lock); + RELEASE_LOCK(&cap->running_task->lock); + // just return the pid rts_unlock(cap); return pid; } else { // child +#if defined(THREADED_RTS) + initMutex(&sched_mutex); + initMutex(&cap->lock); + initMutex(&cap->running_task->lock); +#endif + // Now, all OS threads except the thread that forked are // stopped. We need to stop all Haskell threads, including // those involved in foreign calls. Also we need to delete // all Tasks, because they correspond to OS threads that are // now gone. - for (t = all_threads; t != END_TSO_QUEUE; t = next) { + for (s = 0; s < total_steps; s++) { + for (t = all_steps[s].threads; t != END_TSO_QUEUE; t = next) { if (t->what_next == ThreadRelocated) { - next = t->link; + next = t->_link; } else { next = t->global_link; // don't allow threads to catch the ThreadKilled @@ -2163,6 +2189,7 @@ forkProcess(HsStablePtr *entry // threads may be evaluating thunks that we need later. deleteThread_(cap,t); } + } } // Empty the run queue. It seems tempting to let all the @@ -2176,14 +2203,19 @@ forkProcess(HsStablePtr *entry // don't exist now: cap->suspended_ccalling_tasks = NULL; - // Empty the all_threads list. Otherwise, the garbage + // Empty the threads lists. Otherwise, the garbage // collector may attempt to resurrect some of these threads. - all_threads = END_TSO_QUEUE; + for (s = 0; s < total_steps; s++) { + all_steps[s].threads = END_TSO_QUEUE; + } // Wipe the task list, except the current Task. ACQUIRE_LOCK(&sched_mutex); for (task = all_tasks; task != NULL; task=task->all_link) { if (task != cap->running_task) { +#if defined(THREADED_RTS) + initMutex(&task->lock); // see #1391 +#endif discardTask(task); } } @@ -2225,14 +2257,18 @@ deleteAllThreads ( Capability *cap ) // NOTE: only safe to call if we own all capabilities. StgTSO* t, *next; + nat s; + debugTrace(DEBUG_sched,"deleting all threads"); - for (t = all_threads; t != END_TSO_QUEUE; t = next) { + for (s = 0; s < total_steps; s++) { + for (t = all_steps[s].threads; t != END_TSO_QUEUE; t = next) { if (t->what_next == ThreadRelocated) { - next = t->link; + next = t->_link; } else { next = t->global_link; deleteThread(cap,t); } + } } // The run queue now contains a bunch of ThreadKilled threads. We @@ -2387,7 +2423,7 @@ resumeThread (void *task_) tso = task->suspended_tso; task->suspended_tso = NULL; - tso->link = END_TSO_QUEUE; + tso->_link = END_TSO_QUEUE; // no write barrier reqd debugTrace(DEBUG_sched, "thread %lu: re-entering RTS", (unsigned long)tso->id); if (tso->why_blocked == BlockedOnCCall) { @@ -2406,7 +2442,7 @@ resumeThread (void *task_) #endif /* We might have GC'd, mark the TSO dirty again */ - dirtyTSO(tso); + dirty_TSO(cap,tso); IF_DEBUG(sanity, checkTSO(tso)); @@ -2542,7 +2578,6 @@ initScheduler(void) #endif blackhole_queue = END_TSO_QUEUE; - all_threads = END_TSO_QUEUE; context_switch = 0; sched_state = SCHED_RUNNING; @@ -2643,87 +2678,6 @@ freeScheduler( void ) #endif } -/* --------------------------------------------------------------------------- - Where are the roots that we know about? - - - all the threads on the runnable queue - - all the threads on the blocked queue - - all the threads on the sleeping queue - - all the thread currently executing a _ccall_GC - - all the "main threads" - - ------------------------------------------------------------------------ */ - -/* This has to be protected either by the scheduler monitor, or by the - garbage collection monitor (probably the latter). - KH @ 25/10/99 -*/ - -void -GetRoots( evac_fn evac ) -{ - nat i; - Capability *cap; - Task *task; - -#if defined(GRAN) - for (i=0; i<=RtsFlags.GranFlags.proc; i++) { - if ((run_queue_hds[i] != END_TSO_QUEUE) && ((run_queue_hds[i] != NULL))) - evac((StgClosure **)&run_queue_hds[i]); - if ((run_queue_tls[i] != END_TSO_QUEUE) && ((run_queue_tls[i] != NULL))) - evac((StgClosure **)&run_queue_tls[i]); - - if ((blocked_queue_hds[i] != END_TSO_QUEUE) && ((blocked_queue_hds[i] != NULL))) - evac((StgClosure **)&blocked_queue_hds[i]); - if ((blocked_queue_tls[i] != END_TSO_QUEUE) && ((blocked_queue_tls[i] != NULL))) - evac((StgClosure **)&blocked_queue_tls[i]); - if ((ccalling_threadss[i] != END_TSO_QUEUE) && ((ccalling_threadss[i] != NULL))) - evac((StgClosure **)&ccalling_threads[i]); - } - - markEventQueue(); - -#else /* !GRAN */ - - for (i = 0; i < n_capabilities; i++) { - cap = &capabilities[i]; - evac((StgClosure **)(void *)&cap->run_queue_hd); - evac((StgClosure **)(void *)&cap->run_queue_tl); -#if defined(THREADED_RTS) - evac((StgClosure **)(void *)&cap->wakeup_queue_hd); - evac((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((StgClosure **)(void *)&task->suspended_tso); - } - - } - - -#if !defined(THREADED_RTS) - evac((StgClosure **)(void *)&blocked_queue_hd); - evac((StgClosure **)(void *)&blocked_queue_tl); - evac((StgClosure **)(void *)&sleeping_queue); -#endif -#endif - - // evac((StgClosure **)&blackhole_queue); - -#if defined(THREADED_RTS) || defined(PARALLEL_HASKELL) || defined(GRAN) - markSparkQueue(evac); -#endif - -#if defined(RTS_USER_SIGNALS) - // mark the signal handlers (signals should be already blocked) - if (RtsFlags.MiscFlags.install_signal_handlers) { - markSignalHandlers(evac); - } -#endif -} - /* ----------------------------------------------------------------------------- performGC @@ -2816,7 +2770,7 @@ threadStackOverflow(Capability *cap, StgTSO *tso) "increasing stack size from %ld words to %d.", (long)tso->stack_size, new_stack_size); - dest = (StgTSO *)allocate(new_tso_size); + dest = (StgTSO *)allocateLocal(cap,new_tso_size); TICK_ALLOC_TSO(new_stack_size,0); /* copy the TSO block and the old stack into the new area */ @@ -2837,7 +2791,7 @@ threadStackOverflow(Capability *cap, StgTSO *tso) * dead TSO's stack. */ tso->what_next = ThreadRelocated; - tso->link = dest; + setTSOLink(cap,tso,dest); tso->sp = (P_)&(tso->stack[tso->stack_size]); tso->why_blocked = NotBlocked; @@ -2859,6 +2813,54 @@ threadStackOverflow(Capability *cap, StgTSO *tso) return dest; } +static StgTSO * +threadStackUnderflow (Task *task STG_UNUSED, StgTSO *tso) +{ + bdescr *bd, *new_bd; + lnat new_tso_size_w, tso_size_w; + StgTSO *new_tso; + + tso_size_w = tso_sizeW(tso); + + if (tso_size_w < MBLOCK_SIZE_W || + (nat)(tso->stack + tso->stack_size - tso->sp) > tso->stack_size / 4) + { + return tso; + } + + // don't allow throwTo() to modify the blocked_exceptions queue + // while we are moving the TSO: + lockClosure((StgClosure *)tso); + + new_tso_size_w = round_to_mblocks(tso_size_w/2); + + debugTrace(DEBUG_sched, "thread %ld: reducing TSO size from %lu words to %lu", + tso->id, tso_size_w, new_tso_size_w); + + bd = Bdescr((StgPtr)tso); + new_bd = splitLargeBlock(bd, new_tso_size_w / BLOCK_SIZE_W); + + new_tso = (StgTSO *)new_bd->start; + memcpy(new_tso,tso,TSO_STRUCT_SIZE); + new_tso->stack_size = new_tso_size_w - TSO_STRUCT_SIZEW; + + tso->what_next = ThreadRelocated; + tso->_link = new_tso; // no write barrier reqd: same generation + + // The TSO attached to this Task may have moved, so update the + // pointer to it. + if (task->tso == tso) { + task->tso = new_tso; + } + + unlockTSO(new_tso); + unlockTSO(tso); + + IF_DEBUG(sanity,checkTSO(new_tso)); + + return new_tso; +} + /* --------------------------------------------------------------------------- Interrupt execution - usually called inside a signal handler so it mustn't do anything fancy. @@ -2928,7 +2930,7 @@ checkBlackHoles (Capability *cap) t = blackhole_queue; while (t != END_TSO_QUEUE) { ASSERT(t->why_blocked == BlockedOnBlackHole); - type = get_itbl(t->block_info.closure)->type; + type = get_itbl(UNTAG_CLOSURE(t->block_info.closure))->type; if (type != BLACKHOLE && type != CAF_BLACKHOLE) { IF_DEBUG(sanity,checkTSO(t)); t = unblockOne(cap, t); @@ -2937,8 +2939,8 @@ checkBlackHoles (Capability *cap) *prev = t; any_woke_up = rtsTrue; } else { - prev = &t->link; - t = t->link; + prev = &t->_link; + t = t->_link; } } @@ -3146,11 +3148,15 @@ resurrectThreads (StgTSO *threads) { StgTSO *tso, *next; Capability *cap; + step *step; for (tso = threads; tso != END_TSO_QUEUE; tso = next) { next = tso->global_link; - tso->global_link = all_threads; - all_threads = tso; + + step = Bdescr((P_)tso)->step; + tso->global_link = step->threads; + step->threads = tso; + debugTrace(DEBUG_sched, "resurrecting thread %lu", (unsigned long)tso->id); // Wake up the thread on the Capability it was last on