+ break;
+
+ case ThreadYielding:
+#if defined(GRAN)
+ IF_DEBUG(gran,
+ DumpGranEvent(GR_DESCHEDULE, t));
+ globalGranStats.tot_yields++;
+#elif defined(PAR)
+ IF_DEBUG(par,
+ DumpGranEvent(GR_DESCHEDULE, t));
+#endif
+ /* put the thread back on the run queue. Then, if we're ready to
+ * GC, check whether this is the last task to stop. If so, wake
+ * up the GC thread. getThread will block during a GC until the
+ * GC is finished.
+ */
+ IF_DEBUG(scheduler,
+ if (t->what_next == ThreadEnterHugs) {
+ /* ToDo: or maybe a timer expired when we were in Hugs?
+ * or maybe someone hit ctrl-C
+ */
+ belch("--<< thread %ld (%p; %s) stopped to switch to Hugs",
+ t->id, t, whatNext_strs[t->what_next]);
+ } else {
+ belch("--<< thread %ld (%p; %s) stopped, yielding",
+ t->id, t, whatNext_strs[t->what_next]);
+ }
+ );
+
+ threadPaused(t);
+
+ IF_DEBUG(sanity,
+ //belch("&& Doing sanity check on yielding TSO %ld.", t->id);
+ checkTSO(t));
+ ASSERT(t->link == END_TSO_QUEUE);
+#if defined(GRAN)
+ ASSERT(!is_on_queue(t,CurrentProc));
+
+ IF_DEBUG(sanity,
+ //belch("&& Doing sanity check on all ThreadQueues (and their TSOs).");
+ checkThreadQsSanity(rtsTrue));
+#endif
+ APPEND_TO_RUN_QUEUE(t);
+#if defined(GRAN)
+ /* add a ContinueThread event to actually process the thread */
+ new_event(CurrentProc, CurrentProc, CurrentTime[CurrentProc],
+ ContinueThread,
+ t, (StgClosure*)NULL, (rtsSpark*)NULL);
+ IF_GRAN_DEBUG(bq,
+ belch("GRAN: eventq and runnableq after adding yielded thread to queue again:");
+ G_EVENTQ(0);
+ G_CURR_THREADQ(0))
+#endif /* GRAN */
+ break;
+
+ case ThreadBlocked:
+#if defined(GRAN)
+ IF_DEBUG(scheduler,
+ belch("--<< thread %ld (%p; %s) stopped, blocking on node %p [PE %d] with BQ: ",
+ t->id, t, whatNext_strs[t->what_next], t->block_info.closure, (t->block_info.closure==(StgClosure*)NULL ? 99 : where_is(t->block_info.closure)));
+ if (t->block_info.closure!=(StgClosure*)NULL) print_bq(t->block_info.closure));
+
+ // ??? needed; should emit block before
+ IF_DEBUG(gran,
+ DumpGranEvent(GR_DESCHEDULE, t));
+ prune_eventq(t, (StgClosure *)NULL); // prune ContinueThreads for t
+ /*
+ ngoq Dogh!
+ ASSERT(procStatus[CurrentProc]==Busy ||
+ ((procStatus[CurrentProc]==Fetching) &&
+ (t->block_info.closure!=(StgClosure*)NULL)));
+ if (run_queue_hds[CurrentProc] == END_TSO_QUEUE &&
+ !(!RtsFlags.GranFlags.DoAsyncFetch &&
+ procStatus[CurrentProc]==Fetching))
+ procStatus[CurrentProc] = Idle;
+ */
+#elif defined(PAR)
+ IF_DEBUG(par,
+ DumpGranEvent(GR_DESCHEDULE, t));
+
+ /* Send a fetch (if BlockedOnGA) and dump event to log file */
+ blockThread(t);
+
+ IF_DEBUG(scheduler,
+ belch("--<< thread %ld (%p; %s) stopped, blocking on node %p with BQ: ",
+ t->id, t, whatNext_strs[t->what_next], t->block_info.closure);
+ if (t->block_info.closure!=(StgClosure*)NULL) print_bq(t->block_info.closure));
+
+#else /* !GRAN */
+ /* don't need to do anything. Either the thread is blocked on
+ * I/O, in which case we'll have called addToBlockedQueue
+ * previously, or it's blocked on an MVar or Blackhole, in which
+ * case it'll be on the relevant queue already.
+ */
+ IF_DEBUG(scheduler,
+ fprintf(stderr, "--<< thread %d (%p) stopped: ", t->id, t);
+ printThreadBlockage(t);
+ fprintf(stderr, "\n"));
+
+ /* Only for dumping event to log file
+ ToDo: do I need this in GranSim, too?
+ blockThread(t);
+ */
+#endif
+ threadPaused(t);
+ break;
+
+ case ThreadFinished:
+ /* Need to check whether this was a main thread, and if so, signal
+ * the task that started it with the return value. If we have no
+ * more main threads, we probably need to stop all the tasks until
+ * we get a new one.
+ */
+ /* We also end up here if the thread kills itself with an
+ * uncaught exception, see Exception.hc.
+ */
+ IF_DEBUG(scheduler,belch("--++ thread %d (%p) finished", t->id, t));
+#if defined(GRAN)
+ endThread(t, CurrentProc); // clean-up the thread
+#elif defined(PAR)
+ advisory_thread_count--;
+ if (RtsFlags.ParFlags.ParStats.Full)
+ DumpEndEvent(CURRENT_PROC, t, rtsFalse /* not mandatory */);
+#endif
+ break;
+
+ default:
+ barf("schedule: invalid thread return code %d", (int)ret);
+ }
+
+#ifdef SMP
+ cap->link = free_capabilities;
+ free_capabilities = cap;
+ n_free_capabilities++;
+#endif
+
+#ifdef SMP
+ if (ready_to_gc && n_free_capabilities == RtsFlags.ParFlags.nNodes)
+#else
+ if (ready_to_gc)
+#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_DEBUG(scheduler,sched_belch("doing GC"));
+#endif
+ GarbageCollect(GetRoots,rtsFalse);
+ ready_to_gc = rtsFalse;
+#ifdef SMP
+ pthread_cond_broadcast(&gc_pending_cond);
+#endif
+#if defined(GRAN)
+ /* add a ContinueThread event to continue execution of current thread */
+ new_event(CurrentProc, CurrentProc, CurrentTime[CurrentProc],
+ ContinueThread,
+ t, (StgClosure*)NULL, (rtsSpark*)NULL);
+ IF_GRAN_DEBUG(bq,
+ fprintf(stderr, "GRAN: eventq and runnableq after Garbage collection:\n");
+ G_EVENTQ(0);
+ G_CURR_THREADQ(0))
+#endif /* GRAN */
+ }
+#if defined(GRAN)
+ next_thread:
+ IF_GRAN_DEBUG(unused,
+ print_eventq(EventHd));
+
+ event = get_next_event();
+
+#elif defined(PAR)
+ next_thread:
+ /* ToDo: wait for next message to arrive rather than busy wait */
+
+#else /* GRAN */
+ /* not any more
+ next_thread:
+ t = take_off_run_queue(END_TSO_QUEUE);
+ */
+#endif /* GRAN */
+ } /* end of while(1) */
+}
+
+/* A hack for Hugs concurrency support. Needs sanitisation (?) */
+void deleteAllThreads ( void )
+{
+ StgTSO* t;
+ IF_DEBUG(scheduler,sched_belch("deleteAllThreads()"));
+ 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);
+ }
+ run_queue_hd = run_queue_tl = END_TSO_QUEUE;
+ blocked_queue_hd = blocked_queue_tl = 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
+
+/* ---------------------------------------------------------------------------
+ * Suspending & resuming Haskell threads.
+ *
+ * When making a "safe" call to C (aka _ccall_GC), the task gives back
+ * its capability before calling the C function. This allows another
+ * task to pick up the capability and carry on running Haskell
+ * threads. It also means that if the C call blocks, it won't lock
+ * the whole system.
+ *
+ * The Haskell thread making the C call is put to sleep for the
+ * duration of the call, on the susepended_ccalling_threads queue. We
+ * give out a token to the task, which it can use to resume the thread
+ * on return from the C function.
+ * ------------------------------------------------------------------------- */
+
+StgInt
+suspendThread( Capability *cap )
+{
+ nat tok;
+
+ ACQUIRE_LOCK(&sched_mutex);
+
+ IF_DEBUG(scheduler,
+ sched_belch("thread %d did a _ccall_gc\n", cap->rCurrentTSO->id));
+
+ threadPaused(cap->rCurrentTSO);
+ cap->rCurrentTSO->link = suspended_ccalling_threads;
+ suspended_ccalling_threads = cap->rCurrentTSO;
+
+ /* Use the thread ID as the token; it should be unique */
+ tok = cap->rCurrentTSO->id;
+
+#ifdef SMP
+ cap->link = free_capabilities;
+ free_capabilities = cap;
+ n_free_capabilities++;
+#endif
+
+ RELEASE_LOCK(&sched_mutex);
+ return tok;
+}
+
+Capability *
+resumeThread( StgInt tok )
+{
+ StgTSO *tso, **prev;
+ Capability *cap;
+
+ ACQUIRE_LOCK(&sched_mutex);
+
+ prev = &suspended_ccalling_threads;
+ for (tso = suspended_ccalling_threads;
+ tso != END_TSO_QUEUE;
+ prev = &tso->link, tso = tso->link) {
+ if (tso->id == (StgThreadID)tok) {
+ *prev = tso->link;
+ break;
+ }
+ }
+ if (tso == END_TSO_QUEUE) {
+ barf("resumeThread: thread not found");
+ }
+
+#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 = &MainRegTable;
+#endif
+
+ cap->rCurrentTSO = tso;
+
+ RELEASE_LOCK(&sched_mutex);
+ return cap;
+}
+
+
+/* ---------------------------------------------------------------------------
+ * Static functions
+ * ------------------------------------------------------------------------ */
+static void unblockThread(StgTSO *tso);
+
+/* ---------------------------------------------------------------------------
+ * Comparing Thread ids.
+ *
+ * This is used from STG land in the implementation of the
+ * instances of Eq/Ord for ThreadIds.
+ * ------------------------------------------------------------------------ */
+
+int cmp_thread(const StgTSO *tso1, const StgTSO *tso2)
+{
+ StgThreadID id1 = tso1->id;
+ StgThreadID id2 = tso2->id;
+
+ if (id1 < id2) return (-1);
+ if (id1 > id2) return 1;
+ return 0;
+}
+
+/* ---------------------------------------------------------------------------
+ Create a new thread.
+
+ The new thread starts with the given stack size. Before the
+ scheduler can run, however, this thread needs to have a closure
+ (and possibly some arguments) pushed on its stack. See
+ pushClosure() in Schedule.h.
+
+ createGenThread() and createIOThread() (in SchedAPI.h) are
+ convenient packaged versions of this function.
+
+ currently pri (priority) is only used in a GRAN setup -- HWL
+ ------------------------------------------------------------------------ */
+//@cindex createThread
+#if defined(GRAN)
+/* currently pri (priority) is only used in a GRAN setup -- HWL */
+StgTSO *
+createThread(nat stack_size, StgInt pri)
+{
+ return createThread_(stack_size, rtsFalse, pri);
+}
+
+static StgTSO *
+createThread_(nat size, rtsBool have_lock, StgInt pri)
+{
+#else
+StgTSO *
+createThread(nat stack_size)
+{
+ return createThread_(stack_size, rtsFalse);
+}
+
+static StgTSO *
+createThread_(nat size, rtsBool have_lock)
+{
+#endif
+
+ StgTSO *tso;
+ nat stack_size;
+
+ /* First check whether we should create a thread at all */
+#if defined(PAR)
+ /* check that no more than RtsFlags.ParFlags.maxThreads threads are created */
+ if (advisory_thread_count >= RtsFlags.ParFlags.maxThreads) {
+ threadsIgnored++;
+ belch("{createThread}Daq ghuH: refusing to create another thread; no more than %d threads allowed (currently %d)",
+ RtsFlags.ParFlags.maxThreads, advisory_thread_count);
+ return END_TSO_QUEUE;
+ }
+ threadsCreated++;
+#endif
+
+#if defined(GRAN)
+ ASSERT(!RtsFlags.GranFlags.Light || CurrentProc==0);
+#endif
+
+ // ToDo: check whether size = stack_size - TSO_STRUCT_SIZEW
+
+ /* catch ridiculously small stack sizes */
+ if (size < MIN_STACK_WORDS + TSO_STRUCT_SIZEW) {
+ size = MIN_STACK_WORDS + TSO_STRUCT_SIZEW;
+ }
+
+ stack_size = size - TSO_STRUCT_SIZEW;
+
+ tso = (StgTSO *)allocate(size);
+ TICK_ALLOC_TSO(size-TSO_STRUCT_SIZEW, 0);
+
+ SET_HDR(tso, &TSO_info, CCS_SYSTEM);
+#if defined(GRAN)
+ SET_GRAN_HDR(tso, ThisPE);
+#endif
+ tso->what_next = ThreadEnterGHC;
+
+ /* tso->id needs to be unique. For now we use a heavyweight mutex to
+ * protect the increment operation on next_thread_id.
+ * In future, we could use an atomic increment instead.
+ */
+ if (!have_lock) { ACQUIRE_LOCK(&sched_mutex); }
+ tso->id = next_thread_id++;
+ if (!have_lock) { RELEASE_LOCK(&sched_mutex); }
+
+ tso->why_blocked = NotBlocked;
+ tso->blocked_exceptions = NULL;
+
+ tso->splim = (P_)&(tso->stack) + RESERVED_STACK_WORDS;
+ tso->stack_size = stack_size;
+ tso->max_stack_size = round_to_mblocks(RtsFlags.GcFlags.maxStkSize)
+ - TSO_STRUCT_SIZEW;
+ tso->sp = (P_)&(tso->stack) + stack_size;
+
+#ifdef PROFILING
+ tso->prof.CCCS = CCS_MAIN;
+#endif
+
+ /* put a stop frame on the stack */
+ tso->sp -= sizeofW(StgStopFrame);
+ SET_HDR((StgClosure*)tso->sp,(StgInfoTable *)&stg_stop_thread_info,CCS_SYSTEM);
+ tso->su = (StgUpdateFrame*)tso->sp;
+
+ // ToDo: check this
+#if defined(GRAN)
+ tso->link = END_TSO_QUEUE;
+ /* uses more flexible routine in GranSim */
+ insertThread(tso, CurrentProc);
+#else
+ /* In a non-GranSim setup the pushing of a TSO onto the runq is separated
+ * from its creation
+ */
+#endif
+
+#if defined(GRAN) || defined(PAR)
+ DumpGranEvent(GR_START,tso);
+#endif
+
+ /* Link the new thread on the global thread list.
+ */
+ tso->global_link = all_threads;
+ all_threads = tso;
+
+#if defined(GRAN)
+ tso->gran.pri = pri;
+# if defined(DEBUG)
+ tso->gran.magic = TSO_MAGIC; // debugging only
+# endif
+ tso->gran.sparkname = 0;
+ tso->gran.startedat = CURRENT_TIME;
+ tso->gran.exported = 0;
+ tso->gran.basicblocks = 0;
+ tso->gran.allocs = 0;
+ tso->gran.exectime = 0;
+ tso->gran.fetchtime = 0;
+ tso->gran.fetchcount = 0;
+ tso->gran.blocktime = 0;
+ tso->gran.blockcount = 0;
+ tso->gran.blockedat = 0;
+ tso->gran.globalsparks = 0;
+ tso->gran.localsparks = 0;
+ if (RtsFlags.GranFlags.Light)
+ tso->gran.clock = Now; /* local clock */
+ else
+ tso->gran.clock = 0;
+
+ IF_DEBUG(gran,printTSO(tso));
+#elif defined(PAR)
+# if defined(DEBUG)
+ tso->par.magic = TSO_MAGIC; // debugging only
+# endif
+ tso->par.sparkname = 0;
+ tso->par.startedat = CURRENT_TIME;
+ tso->par.exported = 0;
+ tso->par.basicblocks = 0;
+ tso->par.allocs = 0;
+ tso->par.exectime = 0;
+ tso->par.fetchtime = 0;
+ tso->par.fetchcount = 0;
+ tso->par.blocktime = 0;
+ tso->par.blockcount = 0;
+ tso->par.blockedat = 0;
+ tso->par.globalsparks = 0;
+ tso->par.localsparks = 0;
+#endif
+
+#if defined(GRAN)
+ globalGranStats.tot_threads_created++;
+ globalGranStats.threads_created_on_PE[CurrentProc]++;
+ globalGranStats.tot_sq_len += spark_queue_len(CurrentProc);
+ globalGranStats.tot_sq_probes++;
+#endif
+
+#if defined(GRAN)
+ IF_GRAN_DEBUG(pri,
+ belch("==__ schedule: Created TSO %d (%p);",
+ CurrentProc, tso, tso->id));
+#elif defined(PAR)
+ IF_PAR_DEBUG(verbose,
+ belch("==__ schedule: Created TSO %d (%p); %d threads active",
+ tso->id, tso, advisory_thread_count));
+#else
+ IF_DEBUG(scheduler,sched_belch("created thread %ld, stack size = %lx words",
+ tso->id, tso->stack_size));
+#endif
+ return tso;
+}
+
+/*
+ Turn a spark into a thread.
+ ToDo: fix for SMP (needs to acquire SCHED_MUTEX!)
+*/
+#if defined(PAR)
+//@cindex activateSpark
+StgTSO *
+activateSpark (rtsSpark spark)
+{
+ StgTSO *tso;
+
+ ASSERT(spark != (rtsSpark)NULL);
+ tso = createThread_(RtsFlags.GcFlags.initialStkSize, rtsTrue);
+ if (tso!=END_TSO_QUEUE) {
+ pushClosure(tso,spark);
+ PUSH_ON_RUN_QUEUE(tso);
+ advisory_thread_count++;
+
+ if (RtsFlags.ParFlags.ParStats.Full) {
+ //ASSERT(run_queue_hd == END_TSO_QUEUE); // I think ...
+ IF_PAR_DEBUG(verbose,
+ belch("==^^ activateSpark: turning spark of closure %p (%s) into a thread",
+ (StgClosure *)spark, info_type((StgClosure *)spark)));
+ }
+ } else {
+ barf("activateSpark: Cannot create TSO");
+ }
+ // ToDo: fwd info on local/global spark to thread -- HWL
+ // tso->gran.exported = spark->exported;
+ // tso->gran.locked = !spark->global;
+ // tso->gran.sparkname = spark->name;
+
+ return tso;
+}
+#endif
+
+/* ---------------------------------------------------------------------------
+ * scheduleThread()
+ *
+ * scheduleThread puts a thread on the head of the runnable queue.
+ * This will usually be done immediately after a thread is created.
+ * The caller of scheduleThread must create the thread using e.g.
+ * createThread and push an appropriate closure
+ * on this thread's stack before the scheduler is invoked.
+ * ------------------------------------------------------------------------ */
+
+void
+scheduleThread(StgTSO *tso)
+{
+ if (tso==END_TSO_QUEUE){
+ schedule();
+ return;
+ }
+
+ ACQUIRE_LOCK(&sched_mutex);
+
+ /* Put the new thread on the head of the runnable queue. The caller
+ * better push an appropriate closure on this thread's stack
+ * beforehand. In the SMP case, the thread may start running as
+ * soon as we release the scheduler lock below.
+ */
+ PUSH_ON_RUN_QUEUE(tso);
+ THREAD_RUNNABLE();
+
+#if 0
+ IF_DEBUG(scheduler,printTSO(tso));
+#endif
+ 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
+ * ------------------------------------------------------------------------ */
+
+#if defined(PAR) || defined(SMP)
+void *
+taskStart( void *arg STG_UNUSED )
+{
+ rts_evalNothing(NULL);
+}
+#endif
+
+/* ---------------------------------------------------------------------------
+ * initScheduler()
+ *
+ * Initialise the scheduler. This resets all the queues - if the
+ * 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
+static void
+term_handler(int sig STG_UNUSED)
+{
+ stat_workerStop();
+ ACQUIRE_LOCK(&term_mutex);
+ await_death--;
+ RELEASE_LOCK(&term_mutex);
+ pthread_exit(NULL);
+}
+#endif
+
+//@cindex initScheduler
+void
+initScheduler(void)
+{
+#if defined(GRAN)
+ nat i;
+
+ for (i=0; i<=MAX_PROC; i++) {
+ run_queue_hds[i] = END_TSO_QUEUE;
+ run_queue_tls[i] = END_TSO_QUEUE;
+ blocked_queue_hds[i] = END_TSO_QUEUE;
+ blocked_queue_tls[i] = END_TSO_QUEUE;
+ ccalling_threadss[i] = END_TSO_QUEUE;
+ }
+#else
+ run_queue_hd = END_TSO_QUEUE;
+ run_queue_tl = END_TSO_QUEUE;
+ blocked_queue_hd = END_TSO_QUEUE;
+ blocked_queue_tl = END_TSO_QUEUE;
+#endif
+
+ suspended_ccalling_threads = END_TSO_QUEUE;
+
+ main_threads = NULL;
+ all_threads = END_TSO_QUEUE;
+
+ context_switch = 0;
+ interrupted = 0;
+
+#ifdef INTERPRETER
+ ecafList = END_ECAF_LIST;
+ clearECafTable();
+#endif
+
+ /* Install the SIGHUP handler */
+#ifdef SMP
+ {
+ struct sigaction action,oact;
+
+ action.sa_handler = term_handler;
+ sigemptyset(&action.sa_mask);
+ action.sa_flags = 0;
+ if (sigaction(SIGTERM, &action, &oact) != 0) {
+ barf("can't install TERM handler");