+ /*
+ * Detect deadlock: when we have no threads to run, there are no
+ * threads waiting on I/O or sleeping, and all the other tasks are
+ * waiting for work, we must have a deadlock of some description.
+ *
+ * We first try to find threads blocked on themselves (ie. black
+ * holes), and generate NonTermination exceptions where necessary.
+ *
+ * If no threads are black holed, we have a deadlock situation, so
+ * 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
+#ifdef SMP
+ && (n_free_capabilities == RtsFlags.ParFlags.nNodes)
+#endif
+ )
+ {
+ IF_DEBUG(scheduler, sched_belch("deadlocked, forcing major GC..."));
+ 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);
+ }
+ main_threads = NULL;
+#else
+ deleteThread(m->tso);
+ m->ret = NULL;
+ m->stat = Deadlock;
+ main_threads = m->link;
+ return;
+#endif
+ }
+ }
+ }
+#elif defined(PAR)
+ /* ToDo: add deadlock detection in GUM (similar to SMP) -- HWL */
+#endif
+
+#ifdef 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);
+ }
+
+ /* 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"));
+ }
+#endif
+
+#if defined(GRAN)
+
+ if (RtsFlags.GranFlags.Light)
+ GranSimLight_enter_system(event, &ActiveTSO); // adjust ActiveTSO etc
+
+ /* adjust time based on time-stamp */
+ if (event->time > CurrentTime[CurrentProc] &&
+ event->evttype != ContinueThread)
+ CurrentTime[CurrentProc] = event->time;
+
+ /* Deal with the idle PEs (may issue FindWork or MoveSpark events) */
+ if (!RtsFlags.GranFlags.Light)
+ handleIdlePEs();
+
+ IF_DEBUG(gran, fprintf(stderr, "GRAN: switch by event-type\n"));
+
+ /* main event dispatcher in GranSim */
+ switch (event->evttype) {
+ /* Should just be continuing execution */
+ case ContinueThread:
+ IF_DEBUG(gran, fprintf(stderr, "GRAN: doing ContinueThread\n"));
+ /* ToDo: check assertion
+ ASSERT(run_queue_hd != (StgTSO*)NULL &&
+ run_queue_hd != END_TSO_QUEUE);
+ */
+ /* Ignore ContinueThreads for fetching threads (if synchr comm) */
+ if (!RtsFlags.GranFlags.DoAsyncFetch &&
+ procStatus[CurrentProc]==Fetching) {
+ belch("ghuH: Spurious ContinueThread while Fetching ignored; TSO %d (%p) [PE %d]",
+ CurrentTSO->id, CurrentTSO, CurrentProc);
+ goto next_thread;
+ }
+ /* Ignore ContinueThreads for completed threads */
+ if (CurrentTSO->what_next == ThreadComplete) {
+ belch("ghuH: found a ContinueThread event for completed thread %d (%p) [PE %d] (ignoring ContinueThread)",
+ CurrentTSO->id, CurrentTSO, CurrentProc);
+ goto next_thread;
+ }
+ /* Ignore ContinueThreads for threads that are being migrated */
+ if (PROCS(CurrentTSO)==Nowhere) {
+ belch("ghuH: trying to run the migrating TSO %d (%p) [PE %d] (ignoring ContinueThread)",
+ CurrentTSO->id, CurrentTSO, CurrentProc);
+ goto next_thread;
+ }
+ /* The thread should be at the beginning of the run queue */
+ if (CurrentTSO!=run_queue_hds[CurrentProc]) {
+ belch("ghuH: TSO %d (%p) [PE %d] is not at the start of the run_queue when doing a ContinueThread",
+ CurrentTSO->id, CurrentTSO, CurrentProc);
+ break; // run the thread anyway
+ }
+ /*
+ new_event(proc, proc, CurrentTime[proc],
+ FindWork,
+ (StgTSO*)NULL, (StgClosure*)NULL, (rtsSpark*)NULL);
+ goto next_thread;
+ */ /* Catches superfluous CONTINUEs -- should be unnecessary */
+ break; // now actually run the thread; DaH Qu'vam yImuHbej
+
+ case FetchNode:
+ do_the_fetchnode(event);
+ goto next_thread; /* handle next event in event queue */
+
+ case GlobalBlock:
+ do_the_globalblock(event);
+ goto next_thread; /* handle next event in event queue */
+
+ case FetchReply:
+ do_the_fetchreply(event);
+ goto next_thread; /* handle next event in event queue */
+
+ case UnblockThread: /* Move from the blocked queue to the tail of */
+ do_the_unblock(event);
+ goto next_thread; /* handle next event in event queue */
+
+ case ResumeThread: /* Move from the blocked queue to the tail of */
+ /* the runnable queue ( i.e. Qu' SImqa'lu') */
+ event->tso->gran.blocktime +=
+ CurrentTime[CurrentProc] - event->tso->gran.blockedat;
+ do_the_startthread(event);
+ goto next_thread; /* handle next event in event queue */
+
+ case StartThread:
+ do_the_startthread(event);
+ goto next_thread; /* handle next event in event queue */
+
+ case MoveThread:
+ do_the_movethread(event);
+ goto next_thread; /* handle next event in event queue */
+
+ case MoveSpark:
+ do_the_movespark(event);
+ goto next_thread; /* handle next event in event queue */
+
+ case FindWork:
+ do_the_findwork(event);
+ goto next_thread; /* handle next event in event queue */
+
+ default:
+ barf("Illegal event type %u\n", event->evttype);
+ } /* switch */
+
+ /* This point was scheduler_loop in the old RTS */
+
+ IF_DEBUG(gran, belch("GRAN: after main switch"));
+
+ TimeOfLastEvent = CurrentTime[CurrentProc];
+ TimeOfNextEvent = get_time_of_next_event();
+ IgnoreEvents=(TimeOfNextEvent==0); // HWL HACK
+ // CurrentTSO = ThreadQueueHd;
+
+ IF_DEBUG(gran, belch("GRAN: time of next event is: %ld",
+ TimeOfNextEvent));
+
+ if (RtsFlags.GranFlags.Light)
+ GranSimLight_leave_system(event, &ActiveTSO);
+
+ EndOfTimeSlice = CurrentTime[CurrentProc]+RtsFlags.GranFlags.time_slice;
+
+ IF_DEBUG(gran,
+ belch("GRAN: end of time-slice is %#lx", EndOfTimeSlice));
+
+ /* in a GranSim setup the TSO stays on the run queue */
+ t = CurrentTSO;
+ /* Take a thread from the run queue. */
+ t = POP_RUN_QUEUE(); // take_off_run_queue(t);
+
+ IF_DEBUG(gran,
+ fprintf(stderr, "GRAN: About to run current thread, which is\n");
+ G_TSO(t,5));
+
+ context_switch = 0; // turned on via GranYield, checking events and time slice
+
+ IF_DEBUG(gran,
+ DumpGranEvent(GR_SCHEDULE, t));
+
+ procStatus[CurrentProc] = Busy;
+
+#elif defined(PAR)
+ if (PendingFetches != END_BF_QUEUE) {
+ processFetches();
+ }
+
+ /* ToDo: phps merge with spark activation above */
+ /* check whether we have local work and send requests if we have none */
+ if (EMPTY_RUN_QUEUE()) { /* no runnable threads */
+ /* :-[ no local threads => look out for local sparks */
+ /* the spark pool for the current PE */
+ pool = &(MainRegTable.rSparks); // generalise to cap = &MainRegTable
+ if (advisory_thread_count < RtsFlags.ParFlags.maxThreads &&
+ pool->hd < pool->tl) {
+ /*
+ * ToDo: add GC code check that we really have enough heap afterwards!!
+ * Old comment:
+ * If we're here (no runnable threads) and we have pending
+ * sparks, we must have a space problem. Get enough space
+ * to turn one of those pending sparks into a
+ * thread...
+ */
+
+ spark = findSpark(rtsFalse); /* get a spark */
+ if (spark != (rtsSpark) NULL) {
+ tso = activateSpark(spark); /* turn the spark into a thread */
+ IF_PAR_DEBUG(schedule,
+ belch("==== schedule: Created TSO %d (%p); %d threads active",
+ tso->id, tso, advisory_thread_count));
+
+ if (tso==END_TSO_QUEUE) { /* failed to activate spark->back to loop */
+ belch("==^^ failed to activate spark");
+ goto next_thread;
+ } /* otherwise fall through & pick-up new tso */
+ } else {
+ IF_PAR_DEBUG(verbose,
+ belch("==^^ no local sparks (spark pool contains only NFs: %d)",
+ spark_queue_len(pool)));
+ goto next_thread;
+ }
+ }
+
+ /* If we still have no work we need to send a FISH to get a spark
+ from another PE
+ */
+ if (EMPTY_RUN_QUEUE()) {
+ /* =8-[ no local sparks => look for work on other PEs */
+ /*
+ * We really have absolutely no work. Send out a fish
+ * (there may be some out there already), and wait for
+ * something to arrive. We clearly can't run any threads
+ * until a SCHEDULE or RESUME arrives, and so that's what
+ * we're hoping to see. (Of course, we still have to
+ * respond to other types of messages.)
+ */
+ TIME now = msTime() /*CURRENT_TIME*/;
+ IF_PAR_DEBUG(verbose,
+ belch("-- now=%ld", now));
+ IF_PAR_DEBUG(verbose,
+ if (outstandingFishes < RtsFlags.ParFlags.maxFishes &&
+ (last_fish_arrived_at!=0 &&
+ last_fish_arrived_at+RtsFlags.ParFlags.fishDelay > now)) {
+ belch("--$$ delaying FISH until %ld (last fish %ld, delay %ld, now %ld)",
+ last_fish_arrived_at+RtsFlags.ParFlags.fishDelay,
+ last_fish_arrived_at,
+ RtsFlags.ParFlags.fishDelay, now);
+ });
+
+ if (outstandingFishes < RtsFlags.ParFlags.maxFishes &&
+ (last_fish_arrived_at==0 ||
+ (last_fish_arrived_at+RtsFlags.ParFlags.fishDelay <= now))) {
+ /* outstandingFishes is set in sendFish, processFish;
+ avoid flooding system with fishes via delay */
+ pe = choosePE();
+ sendFish(pe, mytid, NEW_FISH_AGE, NEW_FISH_HISTORY,
+ NEW_FISH_HUNGER);
+
+ // Global statistics: count no. of fishes
+ if (RtsFlags.ParFlags.ParStats.Global &&
+ RtsFlags.GcFlags.giveStats > NO_GC_STATS) {
+ globalParStats.tot_fish_mess++;
+ }
+ }
+
+ receivedFinish = processMessages();
+ goto next_thread;
+ }
+ } else if (PacketsWaiting()) { /* Look for incoming messages */
+ receivedFinish = processMessages();
+ }
+
+ /* Now we are sure that we have some work available */
+ ASSERT(run_queue_hd != END_TSO_QUEUE);
+
+ /* Take a thread from the run queue, if we have work */
+ t = POP_RUN_QUEUE(); // take_off_run_queue(END_TSO_QUEUE);
+ IF_DEBUG(sanity,checkTSO(t));
+
+ /* ToDo: write something to the log-file
+ if (RTSflags.ParFlags.granSimStats && !sameThread)
+ DumpGranEvent(GR_SCHEDULE, RunnableThreadsHd);
+
+ CurrentTSO = t;
+ */
+ /* the spark pool for the current PE */
+ pool = &(MainRegTable.rSparks); // generalise to cap = &MainRegTable
+
+ IF_DEBUG(scheduler,
+ belch("--=^ %d threads, %d sparks on [%#x]",
+ run_queue_len(), spark_queue_len(pool), CURRENT_PROC));
+
+#if 1
+ if (0 && RtsFlags.ParFlags.ParStats.Full &&
+ t && LastTSO && t->id != LastTSO->id &&
+ LastTSO->why_blocked == NotBlocked &&
+ LastTSO->what_next != ThreadComplete) {
+ // if previously scheduled TSO not blocked we have to record the context switch
+ DumpVeryRawGranEvent(TimeOfLastYield, CURRENT_PROC, CURRENT_PROC,
+ GR_DESCHEDULE, LastTSO, (StgClosure *)NULL, 0, 0);
+ }
+
+ if (RtsFlags.ParFlags.ParStats.Full &&
+ (emitSchedule /* forced emit */ ||
+ (t && LastTSO && t->id != LastTSO->id))) {
+ /*
+ we are running a different TSO, so write a schedule event to log file
+ NB: If we use fair scheduling we also have to write a deschedule
+ event for LastTSO; with unfair scheduling we know that the
+ previous tso has blocked whenever we switch to another tso, so
+ we don't need it in GUM for now
+ */
+ DumpRawGranEvent(CURRENT_PROC, CURRENT_PROC,
+ GR_SCHEDULE, t, (StgClosure *)NULL, 0, 0);
+ emitSchedule = rtsFalse;
+ }
+
+#endif
+#else /* !GRAN && !PAR */
+
+ /* 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
+
+ cap->r.rCurrentTSO = t;
+
+ /* context switches are now initiated by the timer signal, unless
+ * the user specified "context switch as often as possible", with
+ * +RTS -C0
+ */
+ if (
+#ifdef PROFILING
+ RtsFlags.ProfFlags.profileInterval == 0 ||
+#endif
+ (RtsFlags.ConcFlags.ctxtSwitchTicks == 0
+ && (run_queue_hd != END_TSO_QUEUE
+ || blocked_queue_hd != END_TSO_QUEUE
+ || sleeping_queue != END_TSO_QUEUE)))
+ context_switch = 1;
+ else
+ context_switch = 0;
+
+ RELEASE_LOCK(&sched_mutex);
+
+ IF_DEBUG(scheduler, sched_belch("-->> Running TSO %ld (%p) %s ...",
+ t->id, t, whatNext_strs[t->what_next]));
+
+#ifdef PROFILING
+ startHeapProfTimer();
+#endif
+
+ /* +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ */
+ /* Run the current thread
+ */
+ switch (cap->r.rCurrentTSO->what_next) {
+ case ThreadKilled:
+ case ThreadComplete:
+ /* Thread already finished, return to scheduler. */
+ ret = ThreadFinished;
+ break;
+ case ThreadEnterGHC:
+ ret = StgRun((StgFunPtr) stg_enterStackTop, &cap->r);
+ break;
+ case ThreadRunGHC:
+ ret = StgRun((StgFunPtr) stg_returnToStackTop, &cap->r);
+ break;
+ case ThreadEnterInterp:
+ ret = interpretBCO(cap);
+ break;
+ default:
+ barf("schedule: invalid what_next field");
+ }
+ /* +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ */
+
+ /* Costs for the scheduler are assigned to CCS_SYSTEM */
+#ifdef PROFILING
+ stopHeapProfTimer();
+ CCCS = CCS_SYSTEM;
+#endif
+
+ ACQUIRE_LOCK(&sched_mutex);
+
+#ifdef SMP
+ IF_DEBUG(scheduler,fprintf(stderr,"scheduler (task %ld): ", pthread_self()););
+#elif !defined(GRAN) && !defined(PAR)
+ IF_DEBUG(scheduler,fprintf(stderr,"scheduler: "););
+#endif
+ t = cap->r.rCurrentTSO;
+
+#if defined(PAR)
+ /* HACK 675: if the last thread didn't yield, make sure to print a
+ SCHEDULE event to the log file when StgRunning the next thread, even
+ if it is the same one as before */
+ LastTSO = t;
+ TimeOfLastYield = CURRENT_TIME;
+#endif
+
+ switch (ret) {
+ case HeapOverflow:
+#if defined(GRAN)
+ IF_DEBUG(gran, DumpGranEvent(GR_DESCHEDULE, t));
+ globalGranStats.tot_heapover++;
+#elif defined(PAR)
+ globalParStats.tot_heapover++;
+#endif
+
+ // did the task ask for a large block?
+ if (cap->r.rHpAlloc > BLOCK_SIZE_W) {
+ // if so, get one and push it on the front of the nursery.
+ bdescr *bd;
+ nat blocks;
+
+ blocks = (nat)BLOCK_ROUND_UP(cap->r.rHpAlloc * sizeof(W_)) / BLOCK_SIZE;
+
+ IF_DEBUG(scheduler,belch("--<< thread %ld (%p; %s) stopped: requesting a large block (size %d)",
+ t->id, t,
+ whatNext_strs[t->what_next], blocks));
+
+ // don't do this if it would push us over the
+ // alloc_blocks_lim limit; we'll GC first.
+ if (alloc_blocks + blocks < alloc_blocks_lim) {
+
+ alloc_blocks += blocks;
+ bd = allocGroup( blocks );
+
+ // link the new group into the list
+ bd->link = cap->r.rCurrentNursery;
+ bd->u.back = cap->r.rCurrentNursery->u.back;
+ if (cap->r.rCurrentNursery->u.back != NULL) {
+ cap->r.rCurrentNursery->u.back->link = bd;
+ } else {
+ ASSERT(g0s0->blocks == cap->r.rCurrentNursery &&
+ g0s0->blocks == cap->r.rNursery);
+ cap->r.rNursery = g0s0->blocks = bd;
+ }
+ cap->r.rCurrentNursery->u.back = bd;
+
+ // initialise it as a nursery block
+ bd->step = g0s0;
+ bd->gen_no = 0;
+ bd->flags = 0;
+ bd->free = bd->start;
+
+ // don't forget to update the block count in g0s0.
+ g0s0->n_blocks += blocks;
+ ASSERT(countBlocks(g0s0->blocks) == g0s0->n_blocks);
+
+ // now update the nursery to point to the new block
+ cap->r.rCurrentNursery = bd;
+
+ // we might be unlucky and have another thread get on the
+ // run queue before us and steal the large block, but in that
+ // case the thread will just end up requesting another large
+ // block.
+ PUSH_ON_RUN_QUEUE(t);
+ break;
+ }
+ }
+
+ /* make all the running tasks block on a condition variable,
+ * maybe set context_switch and wait till they all pile in,
+ * then have them wait on a GC condition variable.
+ */
+ IF_DEBUG(scheduler,belch("--<< thread %ld (%p; %s) stopped: HeapOverflow",
+ t->id, t, whatNext_strs[t->what_next]));
+ threadPaused(t);
+#if defined(GRAN)
+ ASSERT(!is_on_queue(t,CurrentProc));
+#elif defined(PAR)
+ /* Currently we emit a DESCHEDULE event before GC in GUM.
+ ToDo: either add separate event to distinguish SYSTEM time from rest
+ or just nuke this DESCHEDULE (and the following SCHEDULE) */
+ if (0 && RtsFlags.ParFlags.ParStats.Full) {
+ DumpRawGranEvent(CURRENT_PROC, CURRENT_PROC,
+ GR_DESCHEDULE, t, (StgClosure *)NULL, 0, 0);
+ emitSchedule = rtsTrue;
+ }
+#endif
+
+ ready_to_gc = rtsTrue;
+ context_switch = 1; /* stop other threads ASAP */
+ PUSH_ON_RUN_QUEUE(t);
+ /* actual GC is done at the end of the while loop */
+ break;
+
+ case StackOverflow:
+#if defined(GRAN)
+ IF_DEBUG(gran,
+ DumpGranEvent(GR_DESCHEDULE, t));
+ globalGranStats.tot_stackover++;
+#elif defined(PAR)
+ // IF_DEBUG(par,
+ // DumpGranEvent(GR_DESCHEDULE, t);
+ globalParStats.tot_stackover++;
+#endif
+ IF_DEBUG(scheduler,belch("--<< thread %ld (%p; %s) stopped, StackOverflow",
+ t->id, t, whatNext_strs[t->what_next]));
+ /* just adjust the stack for this thread, then pop it back
+ * on the run queue.
+ */
+ threadPaused(t);
+ {
+ StgMainThread *m;
+ /* enlarge the stack */
+ StgTSO *new_t = threadStackOverflow(t);
+
+ /* This TSO has moved, so update any pointers to it from the
+ * main thread stack. It better not be on any other queues...
+ * (it shouldn't be).
+ */
+ for (m = main_threads; m != NULL; m = m->link) {
+ if (m->tso == t) {
+ m->tso = new_t;
+ }
+ }
+ threadPaused(new_t);
+ PUSH_ON_RUN_QUEUE(new_t);
+ }
+ 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);
+ globalParStats.tot_yields++;
+#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 == ThreadEnterInterp) {
+ /* 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
+#if defined(PAR)
+ if (RtsFlags.ParFlags.doFairScheduling) {
+ /* this does round-robin scheduling; good for concurrency */
+ APPEND_TO_RUN_QUEUE(t);
+ } else {
+ /* this does unfair scheduling; good for parallelism */
+ PUSH_ON_RUN_QUEUE(t);
+ }
+#else
+ /* this does round-robin scheduling; good for concurrency */
+ APPEND_TO_RUN_QUEUE(t);
+#endif
+#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(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_PAR_DEBUG(bq,
+
+ if (t->block_info.closure!=(StgClosure*)NULL)
+ print_bq(t->block_info.closure));
+
+ /* Send a fetch (if BlockedOnGA) and dump event to log file */
+ blockThread(t);
+
+ /* whatever we schedule next, we must log that schedule */
+ emitSchedule = rtsTrue;
+
+#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)
+ /* For now all are advisory -- HWL */
+ //if(t->priority==AdvisoryPriority) ??
+ advisory_thread_count--;
+
+# ifdef DIST
+ if(t->dist.priority==RevalPriority)
+ FinishReval(t);
+# endif
+
+ if (RtsFlags.ParFlags.ParStats.Full &&
+ !RtsFlags.ParFlags.ParStats.Suppressed)
+ 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 PROFILING
+ if (RtsFlags.ProfFlags.profileInterval==0 || performHeapProfile) {
+ if (RtsFlags.ProfFlags.doHeapProfile == HEAP_BY_RETAINER) {
+ //
+ // Note: currently retainer profiling is performed after
+ // a major garbage collection.
+ //
+ GarbageCollect(GetRoots, rtsTrue);
+ retainerProfile();
+ } else if (RtsFlags.ProfFlags.doHeapProfile == HEAP_BY_LDV) {
+ //
+ // We have LdvCensus() preceded by a major garbage
+ // collection because we don't want *genuinely* dead
+ // closures to be involved in LDV profiling. Another good
+ // reason is to produce consistent profiling results
+ // regardless of the interval at which GCs are performed.
+ // In other words, we want LDV profiling results to be
+ // completely independent of the GC interval.
+ //
+ GarbageCollect(GetRoots, rtsTrue);
+ LdvCensus();
+ } else {
+ //
+ // Normal creator-based heap profile
+ //
+ GarbageCollect(GetRoots, rtsTrue);
+ heapCensus();
+ }
+ performHeapProfile = rtsFalse;
+ ready_to_gc = rtsFalse; // we already GC'd
+ }
+#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 */
+#endif /* GRAN */
+
+ } /* end of while(1) */
+
+ IF_PAR_DEBUG(verbose,
+ belch("== Leaving schedule() after having received Finish"));
+}
+
+/* ---------------------------------------------------------------------------
+ * deleteAllThreads(): kill all the live threads.
+ *
+ * This is used when we catch a user interrupt (^C), before performing
+ * any necessary cleanups and running finalizers.
+ * ------------------------------------------------------------------------- */
+
+void deleteAllThreads ( void )
+{
+ StgTSO* t;
+ 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) {
+ 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
+
+/* ---------------------------------------------------------------------------
+ * 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( StgRegTable *reg )
+{
+ nat tok;
+ Capability *cap;
+
+ // assume that *reg is a pointer to the StgRegTable part of a Capability
+ cap = (Capability *)((void *)reg - sizeof(StgFunTable));
+
+ ACQUIRE_LOCK(&sched_mutex);
+
+ IF_DEBUG(scheduler,
+ sched_belch("thread %d did a _ccall_gc", cap->r.rCurrentTSO->id));
+
+ threadPaused(cap->r.rCurrentTSO);
+ cap->r.rCurrentTSO->link = suspended_ccalling_threads;
+ suspended_ccalling_threads = cap->r.rCurrentTSO;
+
+ /* 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++;
+#endif
+
+ RELEASE_LOCK(&sched_mutex);
+ return tok;
+}
+
+StgRegTable *
+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");
+ }
+ tso->link = END_TSO_QUEUE;
+
+#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
+
+ cap->r.rCurrentTSO = tso;
+
+ RELEASE_LOCK(&sched_mutex);
+ return &cap->r;
+}
+
+
+/* ---------------------------------------------------------------------------
+ * 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;
+}
+
+/* ---------------------------------------------------------------------------
+ * Fetching the ThreadID from an StgTSO.
+ *
+ * This is used in the implementation of Show for ThreadIds.
+ * ------------------------------------------------------------------------ */
+int rts_getThreadId(const StgTSO *tso)
+{
+ return tso->id;
+}
+
+/* ---------------------------------------------------------------------------
+ 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, &stg_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->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)
+ if (RtsFlags.GranFlags.GranSimStats.Full)
+ DumpGranEvent(GR_START,tso);
+#elif defined(PAR)
+ if (RtsFlags.ParFlags.ParStats.Full)
+ DumpGranEvent(GR_STARTQ,tso);
+ /* HACk to avoid SCHEDULE
+ LastTSO = tso; */
+#endif
+
+ /* Link the new thread on the global thread list.
+ */
+ tso->global_link = all_threads;
+ all_threads = tso;
+
+#if defined(DIST)
+ tso->dist.priority = MandatoryPriority; //by default that is...
+#endif
+
+#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++;
+#elif defined(PAR)
+ // collect parallel global statistics (currently done together with GC stats)
+ if (RtsFlags.ParFlags.ParStats.Global &&
+ RtsFlags.GcFlags.giveStats > NO_GC_STATS) {
+ //fprintf(stderr, "Creating thread %d @ %11.2f\n", tso->id, usertime());
+ globalParStats.tot_threads_created++;
+ }
+#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;
+}
+
+#if defined(PAR)
+/* RFP:
+ all parallel thread creation calls should fall through the following routine.
+*/
+StgTSO *
+createSparkThread(rtsSpark spark)
+{ StgTSO *tso;
+ ASSERT(spark != (rtsSpark)NULL);
+ if (advisory_thread_count >= RtsFlags.ParFlags.maxThreads)
+ { threadsIgnored++;
+ barf("{createSparkThread}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;
+ }
+ else
+ { threadsCreated++;
+ tso = createThread_(RtsFlags.GcFlags.initialStkSize, rtsTrue);
+ if (tso==END_TSO_QUEUE)
+ barf("createSparkThread: Cannot create TSO");
+#if defined(DIST)
+ tso->priority = AdvisoryPriority;
+#endif
+ pushClosure(tso,spark);
+ PUSH_ON_RUN_QUEUE(tso);
+ advisory_thread_count++;
+ }
+ return tso;
+}
+#endif
+
+/*
+ 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;
+
+ tso = createSparkThread(spark);
+ 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)));
+ }
+ // 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) /* ( void *arg STG_UNUSED) */
+{
+ scheduleThread(END_TSO_QUEUE);
+}
+#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
+
+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)
+{
+#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;
+ sleeping_queue = 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;
+ sleeping_queue = END_TSO_QUEUE;
+#endif
+
+ suspended_ccalling_threads = END_TSO_QUEUE;
+
+ main_threads = NULL;
+ all_threads = END_TSO_QUEUE;
+
+ context_switch = 0;
+ interrupted = 0;
+
+ RtsFlags.ConcFlags.ctxtSwitchTicks =
+ RtsFlags.ConcFlags.ctxtSwitchTime / TICK_MILLISECS;
+
+ /* 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");
+ }
+ }
+#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);
+#endif
+
+#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););
+ }
+}
+#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);