#include "PosixSource.h"
#define KEEP_LOCKCLOSURE
#include "Rts.h"
-#include "SchedAPI.h"
+
+#include "sm/Storage.h"
#include "RtsUtils.h"
-#include "RtsFlags.h"
-#include "OSThreads.h"
-#include "Storage.h"
#include "StgRun.h"
-#include "Hooks.h"
#include "Schedule.h"
-#include "StgMiscClosures.h"
#include "Interpreter.h"
#include "Printer.h"
#include "RtsSignals.h"
#include "Sanity.h"
#include "Stats.h"
#include "STM.h"
-#include "Timer.h"
#include "Prelude.h"
#include "ThreadLabels.h"
-#include "LdvProfile.h"
#include "Updates.h"
#include "Proftimer.h"
#include "ProfHeap.h"
-#include "GC.h"
#include "Weak.h"
-#include "EventLog.h"
-
-/* PARALLEL_HASKELL includes go here */
-
+#include "sm/GC.h" // waitForGcThreads, releaseGCThreads, N
#include "Sparks.h"
#include "Capability.h"
#include "Task.h"
#include "Trace.h"
#include "RaiseAsync.h"
#include "Threads.h"
-#include "ThrIOManager.h"
+#include "Timer.h"
+#include "ThreadPaused.h"
#ifdef HAVE_SYS_TYPES_H
#include <sys/types.h>
#include <errno.h>
#endif
-// Turn off inlining when debugging - it obfuscates things
-#ifdef DEBUG
-# undef STATIC_INLINE
-# define STATIC_INLINE static
-#endif
-
/* -----------------------------------------------------------------------------
* Global variables
* -------------------------------------------------------------------------- */
static void schedulePreLoop (void);
static void scheduleFindWork (Capability *cap);
#if defined(THREADED_RTS)
-static void scheduleYield (Capability **pcap, Task *task);
+static void scheduleYield (Capability **pcap, Task *task, rtsBool);
#endif
static void scheduleStartSignalHandlers (Capability *cap);
static void scheduleCheckBlockedThreads (Capability *cap);
static void scheduleCheckBlackHoles (Capability *cap);
static void scheduleDetectDeadlock (Capability *cap, Task *task);
static void schedulePushWork(Capability *cap, Task *task);
-#if defined(PARALLEL_HASKELL)
-static rtsBool scheduleGetRemoteWork(Capability *cap);
-static void scheduleSendPendingMessages(void);
-#endif
-#if defined(PARALLEL_HASKELL) || defined(THREADED_RTS)
+#if defined(THREADED_RTS)
static void scheduleActivateSpark(Capability *cap);
#endif
static void schedulePostRunThread(Capability *cap, StgTSO *t);
static void deleteThread_(Capability *cap, StgTSO *tso);
#endif
-#ifdef DEBUG
-static char *whatNext_strs[] = {
- "(unknown)",
- "ThreadRunGHC",
- "ThreadInterpret",
- "ThreadKilled",
- "ThreadRelocated",
- "ThreadComplete"
-};
-#endif
-
/* -----------------------------------------------------------------------------
* Putting a thread on the run queue: different scheduling policies
* -------------------------------------------------------------------------- */
STATIC_INLINE void
addToRunQueue( Capability *cap, StgTSO *t )
{
-#if defined(PARALLEL_HASKELL)
- if (RtsFlags.ParFlags.doFairScheduling) {
- // this does round-robin scheduling; good for concurrency
- appendToRunQueue(cap,t);
- } else {
- // this does unfair scheduling; good for parallelism
- pushOnRunQueue(cap,t);
- }
-#else
// this does round-robin scheduling; good for concurrency
appendToRunQueue(cap,t);
-#endif
}
/* ---------------------------------------------------------------------------
StgTSO *t;
Capability *cap;
StgThreadReturnCode ret;
-#if defined(PARALLEL_HASKELL)
- rtsBool receivedFinish = rtsFalse;
-#endif
nat prev_what_next;
rtsBool ready_to_gc;
#if defined(THREADED_RTS)
rtsBool first = rtsTrue;
+ rtsBool force_yield = rtsFalse;
#endif
cap = initialCapability;
// The sched_mutex is *NOT* held
// NB. on return, we still hold a capability.
- debugTrace (DEBUG_sched,
- "### NEW SCHEDULER LOOP (task: %p, cap: %p)",
- task, initialCapability);
-
- if (running_finalizers) {
- errorBelch("error: a C finalizer called back into Haskell.\n"
- " This was previously allowed, but is disallowed in GHC 6.10.2 and later.\n"
- " To create finalizers that may call back into Haskll, use\n"
- " Foreign.Concurrent.newForeignPtr instead of Foreign.newForeignPtr.");
- stg_exit(EXIT_FAILURE);
- }
+ debugTrace (DEBUG_sched, "cap %d: schedule()", initialCapability->no);
schedulePreLoop();
// -----------------------------------------------------------
// Scheduler loop starts here:
-#if defined(PARALLEL_HASKELL)
-#define TERMINATION_CONDITION (!receivedFinish)
-#else
-#define TERMINATION_CONDITION rtsTrue
-#endif
-
- while (TERMINATION_CONDITION) {
+ while (1) {
// Check whether we have re-entered the RTS from Haskell without
// going via suspendThread()/resumeThread (i.e. a 'safe' foreign
(pushes threads, wakes up idle capabilities for stealing) */
schedulePushWork(cap,task);
-#if defined(PARALLEL_HASKELL)
- /* since we perform a blocking receive and continue otherwise,
- either we never reach here or we definitely have work! */
- // from here: non-empty run queue
- ASSERT(!emptyRunQueue(cap));
-
- if (PacketsWaiting()) { /* now process incoming messages, if any
- pending...
-
- CAUTION: scheduleGetRemoteWork called
- above, waits for messages as well! */
- processMessages(cap, &receivedFinish);
- }
-#endif // PARALLEL_HASKELL: non-empty run queue!
-
scheduleDetectDeadlock(cap,task);
#if defined(THREADED_RTS)
}
yield:
- scheduleYield(&cap,task);
+ scheduleYield(&cap,task,force_yield);
+ force_yield = rtsFalse;
+
if (emptyRunQueue(cap)) continue; // look for work again
#endif
if (bound) {
if (bound == task) {
- debugTrace(DEBUG_sched,
- "### Running thread %lu in bound thread", (unsigned long)t->id);
// yes, the Haskell thread is bound to the current native thread
} else {
debugTrace(DEBUG_sched,
- "### thread %lu bound to another OS thread", (unsigned long)t->id);
+ "thread %lu bound to another OS thread",
+ (unsigned long)t->id);
// no, bound to a different Haskell thread: pass to that thread
pushOnRunQueue(cap,t);
continue;
// The thread we want to run is unbound.
if (task->tso) {
debugTrace(DEBUG_sched,
- "### this OS thread cannot run thread %lu", (unsigned long)t->id);
+ "this OS thread cannot run thread %lu",
+ (unsigned long)t->id);
// no, the current native thread is bound to a different
// Haskell thread, so pass it to any worker thread
pushOnRunQueue(cap,t);
// that.
cap->r.rCurrentTSO = t;
- debugTrace(DEBUG_sched, "-->> running thread %ld %s ...",
- (long)t->id, whatNext_strs[t->what_next]);
-
startHeapProfTimer();
// Check for exceptions blocked on this thread
}
#endif
- postEvent(cap, EVENT_RUN_THREAD, t->id, 0);
+ traceSchedEvent(cap, EVENT_RUN_THREAD, t, 0);
switch (prev_what_next) {
t->saved_winerror = GetLastError();
#endif
- postEvent (cap, EVENT_STOP_THREAD, t->id, ret);
+ traceSchedEvent (cap, EVENT_STOP_THREAD, t, ret);
#if defined(THREADED_RTS)
// If ret is ThreadBlocked, and this Task is bound to the TSO that
// that task->cap != cap. We better yield this Capability
// immediately and return to normaility.
if (ret == ThreadBlocked) {
- debugTrace(DEBUG_sched,
- "--<< thread %lu (%s) stopped: blocked",
- (unsigned long)t->id, whatNext_strs[t->what_next]);
+ force_yield = rtsTrue;
goto yield;
}
#endif
schedulePostRunThread(cap,t);
- t = threadStackUnderflow(task,t);
+ if (ret != StackOverflow) {
+ t = threadStackUnderflow(task,t);
+ }
ready_to_gc = rtsFalse;
scheduleCheckBlockedThreads(cap);
-#if defined(THREADED_RTS) || defined(PARALLEL_HASKELL)
+#if defined(THREADED_RTS)
if (emptyRunQueue(cap)) { scheduleActivateSpark(cap); }
#endif
-
-#if defined(PARALLEL_HASKELL)
- // if messages have been buffered...
- scheduleSendPendingMessages();
-#endif
-
-#if defined(PARALLEL_HASKELL)
- if (emptyRunQueue(cap)) {
- receivedFinish = scheduleGetRemoteWork(cap);
- continue; // a new round, (hopefully) with new work
- /*
- in GUM, this a) sends out a FISH and returns IF no fish is
- out already
- b) (blocking) awaits and receives messages
-
- in Eden, this is only the blocking receive, as b) in GUM.
- */
- }
-#endif
}
#if defined(THREADED_RTS)
// and also check the benchmarks in nofib/parallel for regressions.
static void
-scheduleYield (Capability **pcap, Task *task)
+scheduleYield (Capability **pcap, Task *task, rtsBool force_yield)
{
Capability *cap = *pcap;
// if we have work, and we don't need to give up the Capability, continue.
- if (!shouldYieldCapability(cap,task) &&
+ //
+ // The force_yield flag is used when a bound thread blocks. This
+ // is a particularly tricky situation: the current Task does not
+ // own the TSO any more, since it is on some queue somewhere, and
+ // might be woken up or manipulated by another thread at any time.
+ // The TSO and Task might be migrated to another Capability.
+ // Certain invariants might be in doubt, such as task->bound->cap
+ // == cap. We have to yield the current Capability immediately,
+ // no messing around.
+ //
+ if (!force_yield &&
+ !shouldYieldCapability(cap,task) &&
(!emptyRunQueue(cap) ||
!emptyWakeupQueue(cap) ||
blackholes_need_checking ||
debugTrace(DEBUG_sched, "pushing thread %lu to capability %d", (unsigned long)t->id, free_caps[i]->no);
appendToRunQueue(free_caps[i],t);
- postEvent (cap, EVENT_MIGRATE_THREAD, t->id, free_caps[i]->no);
+ traceSchedEvent (cap, EVENT_MIGRATE_THREAD, t, free_caps[i]->no);
if (t->bound) { t->bound->cap = free_caps[i]; }
t->cap = free_caps[i];
if (spark != NULL) {
debugTrace(DEBUG_sched, "pushing spark %p to capability %d", spark, free_caps[i]->no);
- postEvent(free_caps[i], EVENT_STEAL_SPARK, t->id, cap->no);
+ traceSchedEvent(free_caps[i], EVENT_STEAL_SPARK, t, cap->no);
newSpark(&(free_caps[i]->r), spark);
}
static void
scheduleDetectDeadlock (Capability *cap, Task *task)
{
-
-#if defined(PARALLEL_HASKELL)
- // ToDo: add deadlock detection in GUM (similar to THREADED_RTS) -- HWL
- return;
-#endif
-
/*
* Detect deadlock: when we have no threads to run, there are no
* threads blocked, waiting for I/O, or sleeping, and all the
* Activate spark threads (PARALLEL_HASKELL and THREADED_RTS)
* ------------------------------------------------------------------------- */
-#if defined(PARALLEL_HASKELL) || defined(THREADED_RTS)
+#if defined(THREADED_RTS)
static void
scheduleActivateSpark(Capability *cap)
{
#endif // PARALLEL_HASKELL || THREADED_RTS
/* ----------------------------------------------------------------------------
- * Get work from a remote node (PARALLEL_HASKELL only)
- * ------------------------------------------------------------------------- */
-
-#if defined(PARALLEL_HASKELL)
-static rtsBool /* return value used in PARALLEL_HASKELL only */
-scheduleGetRemoteWork (Capability *cap STG_UNUSED)
-{
-#if defined(PARALLEL_HASKELL)
- rtsBool receivedFinish = rtsFalse;
-
- // idle() , i.e. send all buffers, wait for work
- if (RtsFlags.ParFlags.BufferTime) {
- IF_PAR_DEBUG(verbose,
- debugBelch("...send all pending data,"));
- {
- nat i;
- for (i=1; i<=nPEs; i++)
- sendImmediately(i); // send all messages away immediately
- }
- }
-
- /* this would be the place for fishing in GUM...
-
- if (no-earlier-fish-around)
- sendFish(choosePe());
- */
-
- // Eden:just look for incoming messages (blocking receive)
- IF_PAR_DEBUG(verbose,
- debugBelch("...wait for incoming messages...\n"));
- processMessages(cap, &receivedFinish); // blocking receive...
-
-
- return receivedFinish;
- // reenter scheduling look after having received something
-
-#else /* !PARALLEL_HASKELL, i.e. THREADED_RTS */
-
- return rtsFalse; /* return value unused in THREADED_RTS */
-
-#endif /* PARALLEL_HASKELL */
-}
-#endif // PARALLEL_HASKELL || THREADED_RTS
-
-/* ----------------------------------------------------------------------------
* After running a thread...
* ------------------------------------------------------------------------- */
// partially-evaluated thunks on the heap.
throwToSingleThreaded_(cap, t, NULL, rtsTrue);
- ASSERT(get_itbl((StgClosure *)t->sp)->type == ATOMICALLY_FRAME);
+// ASSERT(get_itbl((StgClosure *)t->sp)->type == ATOMICALLY_FRAME);
}
}
debugTrace(DEBUG_sched,
"--<< thread %ld (%s) stopped: requesting a large block (size %ld)\n",
- (long)t->id, whatNext_strs[t->what_next], blocks);
+ (long)t->id, what_next_strs[t->what_next], blocks);
// don't do this if the nursery is (nearly) full, we'll GC first.
if (cap->r.rCurrentNursery->link != NULL ||
if (cap->r.rCurrentNursery->u.back != NULL) {
cap->r.rCurrentNursery->u.back->link = bd;
} else {
-#if !defined(THREADED_RTS)
- ASSERT(g0s0->blocks == cap->r.rCurrentNursery &&
- g0s0 == cap->r.rNursery);
-#endif
cap->r.rNursery->blocks = bd;
}
cap->r.rCurrentNursery->u.back = bd;
}
}
- debugTrace(DEBUG_sched,
- "--<< thread %ld (%s) stopped: HeapOverflow",
- (long)t->id, whatNext_strs[t->what_next]);
-
if (cap->r.rHpLim == NULL || cap->context_switch) {
// Sometimes we miss a context switch, e.g. when calling
// primitives in a tight loop, MAYBE_GC() doesn't check the
static void
scheduleHandleStackOverflow (Capability *cap, Task *task, StgTSO *t)
{
- debugTrace (DEBUG_sched,
- "--<< thread %ld (%s) stopped, StackOverflow",
- (long)t->id, whatNext_strs[t->what_next]);
-
/* just adjust the stack for this thread, then pop it back
* on the run queue.
*/
if (t->what_next != prev_what_next) {
debugTrace(DEBUG_sched,
"--<< thread %ld (%s) stopped to switch evaluators",
- (long)t->id, whatNext_strs[t->what_next]);
- } else {
- debugTrace(DEBUG_sched,
- "--<< thread %ld (%s) stopped, yielding",
- (long)t->id, whatNext_strs[t->what_next]);
+ (long)t->id, what_next_strs[t->what_next]);
}
#endif
static void
scheduleHandleThreadBlocked( StgTSO *t
-#if !defined(GRAN) && !defined(DEBUG)
+#if !defined(DEBUG)
STG_UNUSED
#endif
)
// exception, see maybePerformBlockedException().
#ifdef DEBUG
- if (traceClass(DEBUG_sched)) {
- debugTraceBegin("--<< thread %lu (%s) stopped: ",
- (unsigned long)t->id, whatNext_strs[t->what_next]);
- printThreadBlockage(t);
- debugTraceEnd();
- }
+ traceThreadStatus(DEBUG_sched, t);
#endif
}
* We also end up here if the thread kills itself with an
* uncaught exception, see Exception.cmm.
*/
- debugTrace(DEBUG_sched, "--++ thread %lu (%s) finished",
- (unsigned long)t->id, whatNext_strs[t->what_next]);
// blocked exceptions can now complete, even if the thread was in
// blocked mode (see #2910). This unconditionally calls
if (gc_type == PENDING_GC_SEQ)
{
- postEvent(cap, EVENT_REQUEST_SEQ_GC, 0, 0);
+ traceSchedEvent(cap, EVENT_REQUEST_SEQ_GC, 0, 0);
// single-threaded GC: grab all the capabilities
for (i=0; i < n_capabilities; i++) {
debugTrace(DEBUG_sched, "ready_to_gc, grabbing all the capabilies (%d/%d)", i, n_capabilities);
{
// multi-threaded GC: make sure all the Capabilities donate one
// GC thread each.
- postEvent(cap, EVENT_REQUEST_PAR_GC, 0, 0);
+ traceSchedEvent(cap, EVENT_REQUEST_PAR_GC, 0, 0);
debugTrace(DEBUG_sched, "ready_to_gc, grabbing GC threads");
waitForGcThreads(cap);
heap_census = scheduleNeedHeapProfile(rtsTrue);
#if defined(THREADED_RTS)
- postEvent(cap, EVENT_GC_START, 0, 0);
- debugTrace(DEBUG_sched, "doing GC");
+ traceSchedEvent(cap, EVENT_GC_START, 0, 0);
// reset waiting_for_gc *before* GC, so that when the GC threads
// emerge they don't immediately re-enter the GC.
waiting_for_gc = 0;
#else
GarbageCollect(force_major || heap_census, 0, cap);
#endif
- postEvent(cap, EVENT_GC_END, 0, 0);
+ traceSchedEvent(cap, EVENT_GC_END, 0, 0);
if (recent_activity == ACTIVITY_INACTIVE && force_major)
{
}
#else /* !FORKPROCESS_PRIMOP_SUPPORTED */
barf("forkProcess#: primop not supported on this platform, sorry!\n");
- return -1;
#endif
}
task = cap->running_task;
tso = cap->r.rCurrentTSO;
- postEvent(cap, EVENT_STOP_THREAD, tso->id, THREAD_SUSPENDED_FOREIGN_CALL);
- debugTrace(DEBUG_sched,
- "thread %lu did a safe foreign call",
- (unsigned long)cap->r.rCurrentTSO->id);
+ traceSchedEvent(cap, EVENT_STOP_THREAD, tso, THREAD_SUSPENDED_FOREIGN_CALL);
// XXX this might not be necessary --SDM
tso->what_next = ThreadRunGHC;
RELEASE_LOCK(&cap->lock);
-#if defined(THREADED_RTS)
- /* Preparing to leave the RTS, so ensure there's a native thread/task
- waiting to take over.
- */
- debugTrace(DEBUG_sched, "thread %lu: leaving RTS", (unsigned long)tso->id);
-#endif
-
errno = saved_errno;
#if mingw32_HOST_OS
SetLastError(saved_winerror);
task->suspended_tso = NULL;
tso->_link = END_TSO_QUEUE; // no write barrier reqd
- postEvent(cap, EVENT_RUN_THREAD, tso->id, 0);
- debugTrace(DEBUG_sched, "thread %lu: re-entering RTS", (unsigned long)tso->id);
+ traceSchedEvent(cap, EVENT_RUN_THREAD, tso, tso->what_next);
if (tso->why_blocked == BlockedOnCCall) {
// avoid locking the TSO if we don't have to
if (cpu == cap->no) {
appendToRunQueue(cap,tso);
} else {
- postEvent (cap, EVENT_MIGRATE_THREAD, tso->id, capabilities[cpu].no);
+ traceSchedEvent (cap, EVENT_MIGRATE_THREAD, tso, capabilities[cpu].no);
wakeupThreadOnCapability(cap, &capabilities[cpu], tso);
}
#else
initTaskManager();
-#if defined(THREADED_RTS) || defined(PARALLEL_HASKELL)
+#if defined(THREADED_RTS)
initSparkPools();
#endif
tso->sp = (P_)&(tso->stack[tso->stack_size]);
tso->why_blocked = NotBlocked;
- IF_PAR_DEBUG(verbose,
- debugBelch("@@ threadStackOverflow of TSO %d (now at %p): stack size increased to %ld\n",
- tso->id, tso, tso->stack_size);
- /* If we're debugging, just print out the top of the stack */
- printStackChunk(tso->sp, stg_min(tso->stack+tso->stack_size,
- tso->sp+64)));
-
unlockTSO(dest);
unlockTSO(tso);
tso_size_w = tso_sizeW(tso);
- if (tso_size_w < MBLOCK_SIZE_W ||
+ if (tso_size_w < MBLOCK_SIZE_W ||
+ // TSO is less than 2 mblocks (since the first mblock is
+ // shorter than MBLOCK_SIZE_W)
+ (tso_size_w - BLOCKS_PER_MBLOCK*BLOCK_SIZE_W) % MBLOCK_SIZE_W != 0 ||
+ // or TSO is not a whole number of megablocks (ensuring
+ // precondition of splitLargeBlock() below)
+ (tso_size_w <= round_up_to_mblocks(RtsFlags.GcFlags.initialStkSize)) ||
+ // or TSO is smaller than the minimum stack size (rounded up)
(nat)(tso->stack + tso->stack_size - tso->sp) > tso->stack_size / 4)
+ // or stack is using more than 1/4 of the available space
{
+ // then do nothing
return tso;
}
{
sched_state = SCHED_INTERRUPTING;
setContextSwitches();
+#if defined(THREADED_RTS)
wakeUpRts();
+#endif
}
/* -----------------------------------------------------------------------------
will have interrupted any blocking system call in progress anyway.
-------------------------------------------------------------------------- */
-void
-wakeUpRts(void)
-{
#if defined(THREADED_RTS)
+void wakeUpRts(void)
+{
// This forces the IO Manager thread to wakeup, which will
// in turn ensure that some OS thread wakes up and runs the
// scheduler loop, which will cause a GC and deadlock check.
ioManagerWakeup();
-#endif
}
+#endif
/* -----------------------------------------------------------------------------
* checkBlackHoles()
case CATCH_STM_FRAME: {
StgTRecHeader *trec = tso -> trec;
- StgTRecHeader *outer = stmGetEnclosingTRec(trec);
+ StgTRecHeader *outer = trec -> enclosing_trec;
debugTrace(DEBUG_stm,
"found CATCH_STM_FRAME at %p during retry", p);
debugTrace(DEBUG_stm, "trec=%p outer=%p", trec, outer);
switch (tso->why_blocked) {
case BlockedOnMVar:
- case BlockedOnException:
/* Called by GC - sched_mutex lock is currently held. */
throwToSingleThreaded(cap, tso,
- (StgClosure *)blockedOnDeadMVar_closure);
+ (StgClosure *)blockedIndefinitelyOnMVar_closure);
break;
case BlockedOnBlackHole:
throwToSingleThreaded(cap, tso,
break;
case BlockedOnSTM:
throwToSingleThreaded(cap, tso,
- (StgClosure *)blockedIndefinitely_closure);
+ (StgClosure *)blockedIndefinitelyOnSTM_closure);
break;
case NotBlocked:
/* This might happen if the thread was blocked on a black hole
* can wake up threads, remember...).
*/
continue;
+ case BlockedOnException:
+ // throwTo should never block indefinitely: if the target
+ // thread dies or completes, throwTo returns.
+ barf("resurrectThreads: thread BlockedOnException");
+ break;
default:
barf("resurrectThreads: thread blocked in a strange way");
}
{
StgTSO *tso, *next;
Capability *cap;
+ Task *task, *saved_task;;
step *step;
+ task = myTask();
+ cap = task->cap;
+
for (tso = threads; tso != END_TSO_QUEUE; tso = next) {
next = tso->global_link;
debugTrace(DEBUG_sched, "performing blocked throwTo to thread %lu", (unsigned long)tso->id);
- cap = tso->cap;
- maybePerformBlockedException(cap, tso);
- }
+ // We must pretend this Capability belongs to the current Task
+ // for the time being, as invariants will be broken otherwise.
+ // In fact the current Task has exclusive access to the systme
+ // at this point, so this is just bookkeeping:
+ task->cap = tso->cap;
+ saved_task = tso->cap->running_task;
+ tso->cap->running_task = task;
+ maybePerformBlockedException(tso->cap, tso);
+ tso->cap->running_task = saved_task;
+ }
+
+ // Restore our original Capability:
+ task->cap = cap;
}
projects / ghc-hetmet.git / blobdiff