/* ---------------------------------------------------------------------------
- * $Id: Schedule.c,v 1.110 2001/12/18 12:33:45 simonmar Exp $
+ * $Id: Schedule.c,v 1.134 2002/03/12 13:57:11 simonmar Exp $
*
* (c) The GHC Team, 1998-2000
*
*
* WAY Name CPP flag What's it for
* --------------------------------------
- * mp GUM PAR Parallel execution on a distributed memory machine
- * s SMP SMP Parallel execution on a shared memory machine
- * mg GranSim GRAN Simulation of parallel execution
- * md GUM/GdH DIST Distributed execution (based on GUM)
+ * mp GUM PAR Parallel execution on a distributed memory machine
+ * s SMP SMP Parallel execution on a shared memory machine
+ * mg GranSim GRAN Simulation of parallel execution
+ * md GUM/GdH DIST Distributed execution (based on GUM)
+ *
* --------------------------------------------------------------------------*/
//@node Main scheduling code, , ,
# include "HLC.h"
#endif
#include "Sparks.h"
+#include "Capability.h"
+#include "OSThreads.h"
+#include "Task.h"
#include <stdarg.h>
//@node Variables and Data structures, Prototypes, Includes, Main scheduling code
//@subsection Variables and Data structures
-/* Main threads:
- *
- * These are the threads which clients have requested that we run.
- *
- * In an SMP build, we might have several concurrent clients all
- * waiting for results, and each one will wait on a condition variable
- * until the result is available.
- *
- * In non-SMP, clients are strictly nested: the first client calls
- * into the RTS, which might call out again to C with a _ccall_GC, and
- * eventually re-enter the RTS.
- *
- * Main threads information is kept in a linked list:
- */
-//@cindex StgMainThread
-typedef struct StgMainThread_ {
- StgTSO * tso;
- SchedulerStatus stat;
- StgClosure ** ret;
-#ifdef SMP
- pthread_cond_t wakeup;
-#endif
- struct StgMainThread_ *link;
-} StgMainThread;
-
/* Main thread queue.
* Locks required: sched_mutex.
*/
-static StgMainThread *main_threads;
+StgMainThread *main_threads;
/* Thread queues.
* Locks required: sched_mutex.
/* rtsTime TimeOfNextEvent, EndOfTimeSlice; now in GranSim.c */
/*
- In GranSim we have a runable and a blocked queue for each processor.
+ In GranSim we have a runnable and a blocked queue for each processor.
In order to minimise code changes new arrays run_queue_hds/tls
are created. run_queue_hd is then a short cut (macro) for
run_queue_hds[CurrentProc] (see GranSim.h).
*/
StgTSO *all_threads;
-/* Threads suspended in _ccall_GC.
+/* When a thread performs a safe C call (_ccall_GC, using old
+ * terminology), it gets put on the suspended_ccalling_threads
+ * list. Used by the garbage collector.
*/
static StgTSO *suspended_ccalling_threads;
#define MIN_STACK_WORDS (RESERVED_STACK_WORDS + sizeofW(StgStopFrame) + 2)
-/* Free capability list.
- * Locks required: sched_mutex.
- */
-#ifdef SMP
-Capability *free_capabilities; /* Available capabilities for running threads */
-nat n_free_capabilities; /* total number of available capabilities */
-#else
-Capability MainCapability; /* for non-SMP, we have one global capability */
-#endif
#if defined(GRAN)
StgTSO *CurrentTSO;
rtsBool ready_to_gc;
-/* All our current task ids, saved in case we need to kill them later.
- */
-#ifdef SMP
-//@cindex task_ids
-task_info *task_ids;
-#endif
-
void addToBlockedQueue ( StgTSO *tso );
static void schedule ( void );
static void sched_belch(char *s, ...);
#endif
-#ifdef SMP
-//@cindex sched_mutex
-//@cindex term_mutex
-//@cindex thread_ready_cond
-//@cindex gc_pending_cond
-pthread_mutex_t sched_mutex = PTHREAD_MUTEX_INITIALIZER;
-pthread_mutex_t term_mutex = PTHREAD_MUTEX_INITIALIZER;
-pthread_cond_t thread_ready_cond = PTHREAD_COND_INITIALIZER;
-pthread_cond_t gc_pending_cond = PTHREAD_COND_INITIALIZER;
+#if defined(RTS_SUPPORTS_THREADS)
+/* ToDo: carefully document the invariants that go together
+ * with these synchronisation objects.
+ */
+Mutex sched_mutex = INIT_MUTEX_VAR;
+Mutex term_mutex = INIT_MUTEX_VAR;
+# if defined(SMP)
+static Condition gc_pending_cond = INIT_COND_VAR;
nat await_death;
-#endif
+# endif
+
+#endif /* RTS_SUPPORTS_THREADS */
#if defined(PAR)
StgTSO *LastTSO;
};
#endif
-#ifdef PAR
+#if defined(PAR)
StgTSO * createSparkThread(rtsSpark spark);
StgTSO * activateSpark (rtsSpark spark);
#endif
StgTSO *MainTSO;
*/
+#if defined(PAR) || defined(RTS_SUPPORTS_THREADS)
+static void taskStart(void);
+static void
+taskStart(void)
+{
+ schedule();
+}
+#endif
+
+
+
+
//@node Main scheduling loop, Suspend and Resume, Prototypes, Main scheduling code
//@subsection Main scheduling loop
rtsBool was_interrupted = rtsFalse;
ACQUIRE_LOCK(&sched_mutex);
+
+#if defined(RTS_SUPPORTS_THREADS)
+ /* Check to see whether there are any worker threads
+ waiting to deposit external call results. If so,
+ yield our capability */
+ yieldToReturningWorker(&sched_mutex, cap);
-#if defined(GRAN)
+ waitForWorkCapability(&sched_mutex, &cap, rtsFalse);
+#endif
+#if defined(GRAN)
/* set up first event to get things going */
/* ToDo: assign costs for system setup and init MainTSO ! */
new_event(CurrentProc, CurrentProc, CurrentTime[CurrentProc],
* should be done more efficiently without a linear scan
* of the main threads list, somehow...
*/
-#ifdef SMP
+#if defined(RTS_SUPPORTS_THREADS)
{
StgMainThread *m, **prev;
prev = &main_threads;
}
*prev = m->link;
m->stat = Success;
- pthread_cond_broadcast(&m->wakeup);
+ broadcastCondition(&m->wakeup);
break;
case ThreadKilled:
if (m->ret) *(m->ret) = NULL;
} else {
m->stat = Killed;
}
- pthread_cond_broadcast(&m->wakeup);
+ broadcastCondition(&m->wakeup);
break;
default:
break;
}
}
-#else // not SMP
+#else /* not threaded */
# if defined(PAR)
/* in GUM do this only on the Main PE */
/* Top up the run queue from our spark pool. We try to make the
* number of threads in the run queue equal to the number of
* free capabilities.
+ *
+ * Disable spark support in SMP for now, non-essential & requires
+ * a little bit of work to make it compile cleanly. -- sof 1/02.
*/
-#if defined(SMP)
+#if 0 /* defined(SMP) */
{
- nat n = n_free_capabilities;
+ nat n = getFreeCapabilities();
StgTSO *tso = run_queue_hd;
/* Count the run queue */
/* We need to wake up the other tasks if we just created some
* work for them.
*/
- if (n_free_capabilities - n > 1) {
- pthread_cond_signal(&thread_ready_cond);
+ if (getFreeCapabilities() - n > 1) {
+ signalCondition( &thread_ready_cond );
}
}
#endif // SMP
* ToDo: what if another client comes along & requests another
* main thread?
*/
- if (blocked_queue_hd != END_TSO_QUEUE || sleeping_queue != END_TSO_QUEUE) {
- awaitEvent(
- (run_queue_hd == END_TSO_QUEUE)
-#ifdef SMP
- && (n_free_capabilities == RtsFlags.ParFlags.nNodes)
+ if ( !EMPTY_QUEUE(blocked_queue_hd) || !EMPTY_QUEUE(sleeping_queue) ) {
+ awaitEvent( EMPTY_RUN_QUEUE()
+#if defined(SMP)
+ && allFreeCapabilities()
#endif
);
}
* inform all the main threads.
*/
#ifndef PAR
- if (blocked_queue_hd == END_TSO_QUEUE
- && run_queue_hd == END_TSO_QUEUE
- && sleeping_queue == END_TSO_QUEUE
+ if ( EMPTY_THREAD_QUEUES()
+#if defined(RTS_SUPPORTS_THREADS)
+ && EMPTY_QUEUE(suspended_ccalling_threads)
+#endif
#ifdef SMP
- && (n_free_capabilities == RtsFlags.ParFlags.nNodes)
+ && allFreeCapabilities()
#endif
)
{
IF_DEBUG(scheduler, sched_belch("deadlocked, forcing major GC..."));
+#if defined(THREADED_RTS)
+ /* and SMP mode ..? */
+ releaseCapability(cap);
+#endif
+ // Garbage collection can release some new threads due to
+ // either (a) finalizers or (b) threads resurrected because
+ // they are about to be send BlockedOnDeadMVar. Any threads
+ // thus released will be immediately runnable.
GarbageCollect(GetRoots,rtsTrue);
- if (blocked_queue_hd == END_TSO_QUEUE
- && run_queue_hd == END_TSO_QUEUE
- && sleeping_queue == END_TSO_QUEUE) {
- IF_DEBUG(scheduler, sched_belch("still deadlocked, checking for black holes..."));
- detectBlackHoles();
- if (run_queue_hd == END_TSO_QUEUE) {
- StgMainThread *m = main_threads;
-#ifdef SMP
- for (; m != NULL; m = m->link) {
- deleteThread(m->tso);
- m->ret = NULL;
- m->stat = Deadlock;
- pthread_cond_broadcast(&m->wakeup);
+
+ if ( !EMPTY_RUN_QUEUE() ) { goto not_deadlocked; }
+
+ IF_DEBUG(scheduler,
+ sched_belch("still deadlocked, checking for black holes..."));
+ detectBlackHoles();
+
+ if ( !EMPTY_RUN_QUEUE() ) { goto not_deadlocked; }
+
+#ifndef mingw32_TARGET_OS
+ /* If we have user-installed signal handlers, then wait
+ * for signals to arrive rather then bombing out with a
+ * deadlock.
+ */
+ if ( anyUserHandlers() ) {
+ IF_DEBUG(scheduler,
+ sched_belch("still deadlocked, waiting for signals..."));
+
+ awaitUserSignals();
+
+ // we might be interrupted...
+ if (interrupted) { continue; }
+
+ if (signals_pending()) {
+ startSignalHandlers();
+ }
+ ASSERT(!EMPTY_RUN_QUEUE());
+ goto not_deadlocked;
+ }
+#endif
+
+ /* Probably a real deadlock. Send the current main thread the
+ * Deadlock exception (or in the SMP build, send *all* main
+ * threads the deadlock exception, since none of them can make
+ * progress).
+ */
+ {
+ StgMainThread *m;
+#if defined(RTS_SUPPORTS_THREADS)
+ for (m = main_threads; m != NULL; m = m->link) {
+ switch (m->tso->why_blocked) {
+ case BlockedOnBlackHole:
+ raiseAsync(m->tso, (StgClosure *)NonTermination_closure);
+ break;
+ case BlockedOnException:
+ case BlockedOnMVar:
+ raiseAsync(m->tso, (StgClosure *)Deadlock_closure);
+ break;
+ default:
+ barf("deadlock: main thread blocked in a strange way");
}
- main_threads = NULL;
+ }
#else
- deleteThread(m->tso);
- m->ret = NULL;
- m->stat = Deadlock;
- main_threads = m->link;
- return;
-#endif
+ m = main_threads;
+ switch (m->tso->why_blocked) {
+ case BlockedOnBlackHole:
+ raiseAsync(m->tso, (StgClosure *)NonTermination_closure);
+ break;
+ case BlockedOnException:
+ case BlockedOnMVar:
+ raiseAsync(m->tso, (StgClosure *)Deadlock_closure);
+ break;
+ default:
+ barf("deadlock: main thread blocked in a strange way");
}
+#endif
}
+
+#if defined(RTS_SUPPORTS_THREADS)
+ /* ToDo: revisit conditions (and mechanism) for shutting
+ down a multi-threaded world */
+ IF_DEBUG(scheduler, sched_belch("all done, i think...shutting down."));
+ shutdownHaskellAndExit(0);
+#endif
}
+ not_deadlocked:
+
#elif defined(PAR)
/* ToDo: add deadlock detection in GUM (similar to SMP) -- HWL */
#endif
-#ifdef SMP
+#if defined(SMP)
/* If there's a GC pending, don't do anything until it has
* completed.
*/
if (ready_to_gc) {
IF_DEBUG(scheduler,sched_belch("waiting for GC"));
- pthread_cond_wait(&gc_pending_cond, &sched_mutex);
+ waitCondition( &gc_pending_cond, &sched_mutex );
}
-
+#endif
+
+#if defined(RTS_SUPPORTS_THREADS)
/* block until we've got a thread on the run queue and a free
* capability.
+ *
*/
- while (run_queue_hd == END_TSO_QUEUE || free_capabilities == NULL) {
- IF_DEBUG(scheduler, sched_belch("waiting for work"));
- pthread_cond_wait(&thread_ready_cond, &sched_mutex);
- IF_DEBUG(scheduler, sched_belch("work now available"));
+ if ( EMPTY_RUN_QUEUE() ) {
+ /* Give up our capability */
+ releaseCapability(cap);
+ IF_DEBUG(scheduler, sched_belch("thread %d: waiting for work", osThreadId()));
+ waitForWorkCapability(&sched_mutex, &cap, rtsTrue);
+ IF_DEBUG(scheduler, sched_belch("thread %d: work now available", osThreadId()));
+#if 0
+ while ( EMPTY_RUN_QUEUE() ) {
+ waitForWorkCapability(&sched_mutex, &cap);
+ IF_DEBUG(scheduler, sched_belch("thread %d: work now available", osThreadId()));
+ }
+#endif
}
#endif
#if defined(GRAN)
-
if (RtsFlags.GranFlags.Light)
GranSimLight_enter_system(event, &ActiveTSO); // adjust ActiveTSO etc
belch("--=^ %d threads, %d sparks on [%#x]",
run_queue_len(), spark_queue_len(pool), CURRENT_PROC));
-#if 1
+# if 1
if (0 && RtsFlags.ParFlags.ParStats.Full &&
t && LastTSO && t->id != LastTSO->id &&
LastTSO->why_blocked == NotBlocked &&
emitSchedule = rtsFalse;
}
-#endif
+# endif
#else /* !GRAN && !PAR */
- /* grab a thread from the run queue
- */
+ /* grab a thread from the run queue */
ASSERT(run_queue_hd != END_TSO_QUEUE);
t = POP_RUN_QUEUE();
-
// Sanity check the thread we're about to run. This can be
// expensive if there is lots of thread switching going on...
IF_DEBUG(sanity,checkTSO(t));
-
#endif
- /* grab a capability
- */
-#ifdef SMP
- cap = free_capabilities;
- free_capabilities = cap->link;
- n_free_capabilities--;
-#else
- cap = &MainCapability;
-#endif
-
+ grabCapability(&cap);
cap->r.rCurrentTSO = t;
/* context switches are now initiated by the timer signal, unless
ACQUIRE_LOCK(&sched_mutex);
#ifdef SMP
- IF_DEBUG(scheduler,fprintf(stderr,"scheduler (task %ld): ", pthread_self()););
+ IF_DEBUG(scheduler,fprintf(stderr,"scheduler (task %ld): ", osThreadId()););
#elif !defined(GRAN) && !defined(PAR)
IF_DEBUG(scheduler,fprintf(stderr,"scheduler: "););
#endif
barf("schedule: invalid thread return code %d", (int)ret);
}
-#ifdef SMP
- cap->link = free_capabilities;
- free_capabilities = cap;
- n_free_capabilities++;
+#if defined(RTS_SUPPORTS_THREADS)
+ /* I don't understand what this re-grab is doing -- sof */
+ grabCapability(&cap);
#endif
#ifdef PROFILING
}
#endif
+ if (ready_to_gc
#ifdef SMP
- if (ready_to_gc && n_free_capabilities == RtsFlags.ParFlags.nNodes)
-#else
- if (ready_to_gc)
+ && allFreeCapabilities()
#endif
- {
+ ) {
/* everybody back, start the GC.
* Could do it in this thread, or signal a condition var
* to do it in another thread. Either way, we need to
* broadcast on gc_pending_cond afterward.
*/
-#ifdef SMP
+#if defined(RTS_SUPPORTS_THREADS)
IF_DEBUG(scheduler,sched_belch("doing GC"));
#endif
GarbageCollect(GetRoots,rtsFalse);
ready_to_gc = rtsFalse;
#ifdef SMP
- pthread_cond_broadcast(&gc_pending_cond);
+ broadcastCondition(&gc_pending_cond);
#endif
#if defined(GRAN)
/* add a ContinueThread event to continue execution of current thread */
void deleteAllThreads ( void )
{
- StgTSO* t;
+ StgTSO* t, *next;
IF_DEBUG(scheduler,sched_belch("deleting all threads"));
- for (t = run_queue_hd; t != END_TSO_QUEUE; t = t->link) {
- deleteThread(t);
- }
- for (t = blocked_queue_hd; t != END_TSO_QUEUE; t = t->link) {
- deleteThread(t);
- }
- for (t = sleeping_queue; t != END_TSO_QUEUE; t = t->link) {
+ for (t = all_threads; t != END_TSO_QUEUE; t = next) {
+ next = t->global_link;
deleteThread(t);
- }
+ }
run_queue_hd = run_queue_tl = END_TSO_QUEUE;
blocked_queue_hd = blocked_queue_tl = END_TSO_QUEUE;
sleeping_queue = END_TSO_QUEUE;
/* startThread and insertThread are now in GranSim.c -- HWL */
+
//@node Suspend and Resume, Run queue code, Main scheduling loop, Main scheduling code
//@subsection Suspend and Resume
* ------------------------------------------------------------------------- */
StgInt
-suspendThread( StgRegTable *reg )
+suspendThread( StgRegTable *reg,
+ rtsBool concCall
+#if !defined(RTS_SUPPORTS_THREADS)
+ STG_UNUSED
+#endif
+ )
{
nat tok;
Capability *cap;
- // assume that *reg is a pointer to the StgRegTable part of a Capability
+ /* assume that *reg is a pointer to the StgRegTable part
+ * of a Capability.
+ */
cap = (Capability *)((void *)reg - sizeof(StgFunTable));
ACQUIRE_LOCK(&sched_mutex);
cap->r.rCurrentTSO->link = suspended_ccalling_threads;
suspended_ccalling_threads = cap->r.rCurrentTSO;
+#if defined(RTS_SUPPORTS_THREADS)
+ cap->r.rCurrentTSO->why_blocked = BlockedOnCCall;
+#endif
+
/* Use the thread ID as the token; it should be unique */
tok = cap->r.rCurrentTSO->id;
-#ifdef SMP
- cap->link = free_capabilities;
- free_capabilities = cap;
- n_free_capabilities++;
+ /* Hand back capability */
+ releaseCapability(cap);
+
+#if defined(RTS_SUPPORTS_THREADS)
+ /* Preparing to leave the RTS, so ensure there's a native thread/task
+ waiting to take over.
+
+ ToDo: optimise this and only create a new task if there's a need
+ for one (i.e., if there's only one Concurrent Haskell thread alive,
+ there's no need to create a new task).
+ */
+ IF_DEBUG(scheduler, sched_belch("worker thread (%d): leaving RTS", tok));
+ if (concCall) {
+ startTask(taskStart);
+ }
#endif
+ /* Other threads _might_ be available for execution; signal this */
+ THREAD_RUNNABLE();
RELEASE_LOCK(&sched_mutex);
return tok;
}
StgRegTable *
-resumeThread( StgInt tok )
+resumeThread( StgInt tok,
+ rtsBool concCall
+#if !defined(RTS_SUPPORTS_THREADS)
+ STG_UNUSED
+#endif
+ )
{
StgTSO *tso, **prev;
Capability *cap;
- ACQUIRE_LOCK(&sched_mutex);
+#if defined(RTS_SUPPORTS_THREADS)
+ /* Wait for permission to re-enter the RTS with the result. */
+ if ( concCall ) {
+ grabReturnCapability(&sched_mutex, &cap);
+ } else {
+ grabCapability(&cap);
+ }
+#else
+ grabCapability(&cap);
+#endif
+ /* Remove the thread off of the suspended list */
prev = &suspended_ccalling_threads;
for (tso = suspended_ccalling_threads;
tso != END_TSO_QUEUE;
barf("resumeThread: thread not found");
}
tso->link = END_TSO_QUEUE;
+ /* Reset blocking status */
+ tso->why_blocked = NotBlocked;
-#ifdef SMP
- while (free_capabilities == NULL) {
- IF_DEBUG(scheduler, sched_belch("waiting to resume"));
- pthread_cond_wait(&thread_ready_cond, &sched_mutex);
- IF_DEBUG(scheduler, sched_belch("resuming thread %d", tso->id));
- }
- cap = free_capabilities;
- free_capabilities = cap->link;
- n_free_capabilities--;
-#else
- cap = &MainCapability;
-#endif
+ RELEASE_LOCK(&sched_mutex);
cap->r.rCurrentTSO = tso;
-
- RELEASE_LOCK(&sched_mutex);
return &cap->r;
}
* on this thread's stack before the scheduler is invoked.
* ------------------------------------------------------------------------ */
+static void scheduleThread_ (StgTSO* tso, rtsBool createTask);
+
void
-scheduleThread(StgTSO *tso)
+scheduleThread_(StgTSO *tso
+ , rtsBool createTask
+#if !defined(THREADED_RTS)
+ STG_UNUSED
+#endif
+ )
{
ACQUIRE_LOCK(&sched_mutex);
* soon as we release the scheduler lock below.
*/
PUSH_ON_RUN_QUEUE(tso);
+#if defined(THREADED_RTS)
+ /* If main() is scheduling a thread, don't bother creating a
+ * new task.
+ */
+ if ( createTask ) {
+ startTask(taskStart);
+ }
+#endif
THREAD_RUNNABLE();
#if 0
RELEASE_LOCK(&sched_mutex);
}
-/* ---------------------------------------------------------------------------
- * startTasks()
- *
- * Start up Posix threads to run each of the scheduler tasks.
- * I believe the task ids are not needed in the system as defined.
- * KH @ 25/10/99
- * ------------------------------------------------------------------------ */
+void scheduleThread(StgTSO* tso)
+{
+ return scheduleThread_(tso, rtsFalse);
+}
-#if defined(PAR) || defined(SMP)
-void
-taskStart(void) /* ( void *arg STG_UNUSED) */
+void scheduleExtThread(StgTSO* tso)
{
- schedule();
+ return scheduleThread_(tso, rtsTrue);
}
-#endif
/* ---------------------------------------------------------------------------
* initScheduler()
* queues contained any threads, they'll be garbage collected at the
* next pass.
*
- * This now calls startTasks(), so should only be called once! KH @ 25/10/99
* ------------------------------------------------------------------------ */
#ifdef SMP
ACQUIRE_LOCK(&term_mutex);
await_death--;
RELEASE_LOCK(&term_mutex);
- pthread_exit(NULL);
+ shutdownThread();
}
#endif
-static void
-initCapability( Capability *cap )
-{
- cap->f.stgChk0 = (F_)__stg_chk_0;
- cap->f.stgChk1 = (F_)__stg_chk_1;
- cap->f.stgGCEnter1 = (F_)__stg_gc_enter_1;
- cap->f.stgUpdatePAP = (F_)__stg_update_PAP;
-}
-
void
initScheduler(void)
{
RtsFlags.ConcFlags.ctxtSwitchTicks =
RtsFlags.ConcFlags.ctxtSwitchTime / TICK_MILLISECS;
+
+#if defined(RTS_SUPPORTS_THREADS)
+ /* Initialise the mutex and condition variables used by
+ * the scheduler. */
+ initMutex(&sched_mutex);
+ initMutex(&term_mutex);
+
+ initCondition(&thread_ready_cond);
+#endif
+
+#if defined(SMP)
+ initCondition(&gc_pending_cond);
+#endif
+
+#if defined(RTS_SUPPORTS_THREADS)
+ ACQUIRE_LOCK(&sched_mutex);
+#endif
/* Install the SIGHUP handler */
-#ifdef SMP
+#if defined(SMP)
{
struct sigaction action,oact;
}
#endif
-#ifdef SMP
- /* Allocate N Capabilities */
- {
- nat i;
- Capability *cap, *prev;
- cap = NULL;
- prev = NULL;
- for (i = 0; i < RtsFlags.ParFlags.nNodes; i++) {
- cap = stgMallocBytes(sizeof(Capability), "initScheduler:capabilities");
- initCapability(cap);
- cap->link = prev;
- prev = cap;
- }
- free_capabilities = cap;
- n_free_capabilities = RtsFlags.ParFlags.nNodes;
- }
- IF_DEBUG(scheduler,fprintf(stderr,"scheduler: Allocated %d capabilities\n",
- n_free_capabilities););
-#else
- initCapability(&MainCapability);
+ /* A capability holds the state a native thread needs in
+ * order to execute STG code. At least one capability is
+ * floating around (only SMP builds have more than one).
+ */
+ initCapabilities();
+
+#if defined(RTS_SUPPORTS_THREADS)
+ /* start our haskell execution tasks */
+# if defined(SMP)
+ startTaskManager(RtsFlags.ParFlags.nNodes, taskStart);
+# else
+ startTaskManager(0,taskStart);
+# endif
#endif
-#if defined(SMP) || defined(PAR)
+#if /* defined(SMP) ||*/ defined(PAR)
initSparkPools();
#endif
-}
-#ifdef SMP
-void
-startTasks( void )
-{
- nat i;
- int r;
- pthread_t tid;
-
- /* make some space for saving all the thread ids */
- task_ids = stgMallocBytes(RtsFlags.ParFlags.nNodes * sizeof(task_info),
- "initScheduler:task_ids");
-
- /* and create all the threads */
- for (i = 0; i < RtsFlags.ParFlags.nNodes; i++) {
- r = pthread_create(&tid,NULL,taskStart,NULL);
- if (r != 0) {
- barf("startTasks: Can't create new Posix thread");
- }
- task_ids[i].id = tid;
- task_ids[i].mut_time = 0.0;
- task_ids[i].mut_etime = 0.0;
- task_ids[i].gc_time = 0.0;
- task_ids[i].gc_etime = 0.0;
- task_ids[i].elapsedtimestart = elapsedtime();
- IF_DEBUG(scheduler,fprintf(stderr,"scheduler: Started task: %ld\n",tid););
- }
-}
+#if defined(RTS_SUPPORTS_THREADS)
+ RELEASE_LOCK(&sched_mutex);
#endif
+}
+
void
exitScheduler( void )
{
-#ifdef SMP
- nat i;
-
- /* Don't want to use pthread_cancel, since we'd have to install
- * these silly exception handlers (pthread_cleanup_{push,pop}) around
- * all our locks.
- */
-#if 0
- /* Cancel all our tasks */
- for (i = 0; i < RtsFlags.ParFlags.nNodes; i++) {
- pthread_cancel(task_ids[i].id);
- }
-
- /* Wait for all the tasks to terminate */
- for (i = 0; i < RtsFlags.ParFlags.nNodes; i++) {
- IF_DEBUG(scheduler,fprintf(stderr,"scheduler: waiting for task %ld\n",
- task_ids[i].id));
- pthread_join(task_ids[i].id, NULL);
- }
-#endif
-
- /* Send 'em all a SIGHUP. That should shut 'em up.
- */
- await_death = RtsFlags.ParFlags.nNodes;
- for (i = 0; i < RtsFlags.ParFlags.nNodes; i++) {
- pthread_kill(task_ids[i].id,SIGTERM);
- }
- while (await_death > 0) {
- sched_yield();
- }
+#if defined(RTS_SUPPORTS_THREADS)
+ stopTaskManager();
#endif
}
{
do {
while (run_queue_hd != END_TSO_QUEUE) {
- waitThread ( run_queue_hd, NULL );
+ waitThread ( run_queue_hd, NULL);
}
while (blocked_queue_hd != END_TSO_QUEUE) {
- waitThread ( blocked_queue_hd, NULL );
+ waitThread ( blocked_queue_hd, NULL);
}
while (sleeping_queue != END_TSO_QUEUE) {
- waitThread ( blocked_queue_hd, NULL );
+ waitThread ( blocked_queue_hd, NULL);
}
} while
(blocked_queue_hd != END_TSO_QUEUE ||
SchedulerStatus
waitThread(StgTSO *tso, /*out*/StgClosure **ret)
+{
+#if defined(THREADED_RTS)
+ return waitThread_(tso,ret, rtsFalse);
+#else
+ return waitThread_(tso,ret);
+#endif
+}
+
+SchedulerStatus
+waitThread_(StgTSO *tso,
+ /*out*/StgClosure **ret
+#if defined(THREADED_RTS)
+ , rtsBool blockWaiting
+#endif
+ )
{
StgMainThread *m;
SchedulerStatus stat;
m->tso = tso;
m->ret = ret;
m->stat = NoStatus;
-#ifdef SMP
- pthread_cond_init(&m->wakeup, NULL);
+#if defined(RTS_SUPPORTS_THREADS)
+ initCondition(&m->wakeup);
#endif
m->link = main_threads;
main_threads = m;
- IF_DEBUG(scheduler, fprintf(stderr, "== scheduler: new main thread (%d)\n",
- m->tso->id));
+ IF_DEBUG(scheduler, sched_belch("== scheduler: new main thread (%d)\n", m->tso->id));
-#ifdef SMP
- do {
- pthread_cond_wait(&m->wakeup, &sched_mutex);
- } while (m->stat == NoStatus);
+#if defined(RTS_SUPPORTS_THREADS)
+
+# if defined(THREADED_RTS)
+ if (!blockWaiting) {
+ /* In the threaded case, the OS thread that called main()
+ * gets to enter the RTS directly without going via another
+ * task/thread.
+ */
+ RELEASE_LOCK(&sched_mutex);
+ schedule();
+ ASSERT(m->stat != NoStatus);
+ } else
+# endif
+ {
+ IF_DEBUG(scheduler, sched_belch("sfoo"));
+ do {
+ waitCondition(&m->wakeup, &sched_mutex);
+ } while (m->stat == NoStatus);
+ }
#elif defined(GRAN)
/* GranSim specific init */
CurrentTSO = m->tso; // the TSO to run
schedule();
#else
+ RELEASE_LOCK(&sched_mutex);
schedule();
ASSERT(m->stat != NoStatus);
#endif
stat = m->stat;
-#ifdef SMP
- pthread_cond_destroy(&m->wakeup);
+#if defined(RTS_SUPPORTS_THREADS)
+ closeCondition(&m->wakeup);
#endif
IF_DEBUG(scheduler, fprintf(stderr, "== scheduler: main thread (%d) finished\n",
m->tso->id));
free(m);
- RELEASE_LOCK(&sched_mutex);
+#if defined(THREADED_RTS)
+ if (blockWaiting)
+#endif
+ RELEASE_LOCK(&sched_mutex);
return stat;
}
}
#endif
- for (m = main_threads; m != NULL; m = m->link) {
- evac((StgClosure **)&m->tso);
- }
if (suspended_ccalling_threads != END_TSO_QUEUE) {
evac((StgClosure **)&suspended_ccalling_threads);
}
-#if defined(SMP) || defined(PAR) || defined(GRAN)
+#if defined(PAR) || defined(GRAN)
markSparkQueue(evac);
#endif
}
void
performGC(void)
{
+ /* Obligated to hold this lock upon entry */
+ ACQUIRE_LOCK(&sched_mutex);
GarbageCollect(GetRoots,rtsFalse);
+ RELEASE_LOCK(&sched_mutex);
}
void
performMajorGC(void)
{
+ ACQUIRE_LOCK(&sched_mutex);
GarbageCollect(GetRoots,rtsTrue);
+ RELEASE_LOCK(&sched_mutex);
}
static void
void
performGCWithRoots(void (*get_roots)(evac_fn))
{
+ ACQUIRE_LOCK(&sched_mutex);
extra_roots = get_roots;
GarbageCollect(AllRoots,rtsFalse);
+ RELEASE_LOCK(&sched_mutex);
}
/* -----------------------------------------------------------------------------
StgAP_UPD * ap;
/* If we find a CATCH_FRAME, and we've got an exception to raise,
- * then build PAP(handler,exception,realworld#), and leave it on
- * top of the stack ready to enter.
+ * then build the THUNK raise(exception), and leave it on
+ * top of the CATCH_FRAME ready to enter.
*/
if (get_itbl(su)->type == CATCH_FRAME && exception != NULL) {
StgCatchFrame *cf = (StgCatchFrame *)su;
+ StgClosure *raise;
+
/* we've got an exception to raise, so let's pass it to the
* handler in this frame.
*/
- ap = (StgAP_UPD *)allocate(sizeofW(StgPAP) + 2);
- TICK_ALLOC_UPD_PAP(3,0);
- SET_HDR(ap,&stg_PAP_info,cf->header.prof.ccs);
-
- ap->n_args = 2;
- ap->fun = cf->handler; /* :: Exception -> IO a */
- ap->payload[0] = exception;
- ap->payload[1] = ARG_TAG(0); /* realworld token */
-
- /* throw away the stack from Sp up to and including the
- * CATCH_FRAME.
- */
- sp = (P_)su + sizeofW(StgCatchFrame) - 1;
- tso->su = cf->link;
-
- /* Restore the blocked/unblocked state for asynchronous exceptions
- * at the CATCH_FRAME.
- *
- * If exceptions were unblocked at the catch, arrange that they
- * are unblocked again after executing the handler by pushing an
- * unblockAsyncExceptions_ret stack frame.
+ raise = (StgClosure *)allocate(sizeofW(StgClosure)+1);
+ TICK_ALLOC_SE_THK(1,0);
+ SET_HDR(raise,&stg_raise_info,cf->header.prof.ccs);
+ raise->payload[0] = exception;
+
+ /* throw away the stack from Sp up to the CATCH_FRAME.
*/
- if (!cf->exceptions_blocked) {
- *(sp--) = (W_)&stg_unblockAsyncExceptionszh_ret_info;
- }
-
- /* Ensure that async exceptions are blocked when running the handler.
+ sp = (P_)su - 1;
+
+ /* Ensure that async excpetions are blocked now, so we don't get
+ * a surprise exception before we get around to executing the
+ * handler.
*/
if (tso->blocked_exceptions == NULL) {
- tso->blocked_exceptions = END_TSO_QUEUE;
+ tso->blocked_exceptions = END_TSO_QUEUE;
}
-
- /* Put the newly-built PAP on top of the stack, ready to execute
+
+ /* Put the newly-built THUNK on top of the stack, ready to execute
* when the thread restarts.
*/
- sp[0] = (W_)ap;
+ sp[0] = (W_)raise;
tso->sp = sp;
+ tso->su = su;
tso->what_next = ThreadEnterGHC;
IF_DEBUG(sanity, checkTSO(tso));
return;
tso->block_info.closure, info_type(tso->block_info.closure));
break;
#endif
+#if defined(RTS_SUPPORTS_THREADS)
+ case BlockedOnCCall:
+ fprintf(stderr,"is blocked on an external call");
+ break;
+#endif
default:
barf("printThreadBlockage: strange tso->why_blocked: %d for TSO %d (%d)",
tso->why_blocked, tso->id, tso);
va_list ap;
va_start(ap,s);
#ifdef SMP
- fprintf(stderr, "scheduler (task %ld): ", pthread_self());
+ fprintf(stderr, "scheduler (task %ld): ", osThreadId());
#elif defined(PAR)
fprintf(stderr, "== ");
#else
//@subsection Index
//@index
-//* MainRegTable:: @cindex\s-+MainRegTable
//* StgMainThread:: @cindex\s-+StgMainThread
//* awaken_blocked_queue:: @cindex\s-+awaken_blocked_queue
//* blocked_queue_hd:: @cindex\s-+blocked_queue_hd
//* blocked_queue_tl:: @cindex\s-+blocked_queue_tl
//* context_switch:: @cindex\s-+context_switch
//* createThread:: @cindex\s-+createThread
-//* free_capabilities:: @cindex\s-+free_capabilities
//* gc_pending_cond:: @cindex\s-+gc_pending_cond
//* initScheduler:: @cindex\s-+initScheduler
//* interrupted:: @cindex\s-+interrupted
-//* n_free_capabilities:: @cindex\s-+n_free_capabilities
//* next_thread_id:: @cindex\s-+next_thread_id
//* print_bq:: @cindex\s-+print_bq
//* run_queue_hd:: @cindex\s-+run_queue_hd
//* sched_mutex:: @cindex\s-+sched_mutex
//* schedule:: @cindex\s-+schedule
//* take_off_run_queue:: @cindex\s-+take_off_run_queue
-//* task_ids:: @cindex\s-+task_ids
//* term_mutex:: @cindex\s-+term_mutex
-//* thread_ready_cond:: @cindex\s-+thread_ready_cond
//@end index