+ /* 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
+
+#ifdef THREADED_RTS
+ {
+ StgMainThread *m;
+ for(m = main_threads; m; m = m->link)
+ {
+ if(m->tso == t)
+ break;
+ }
+
+ if(m)
+ {
+ if(m == mainThread)
+ {
+ IF_DEBUG(scheduler,
+ fprintf(stderr,"### Running TSO %p in bound OS thread %u\n",
+ t, osThreadId()));
+ // yes, the Haskell thread is bound to the current native thread
+ }
+ else
+ {
+ IF_DEBUG(scheduler,
+ fprintf(stderr,"### TSO %p bound to other OS thread than %u\n",
+ t, osThreadId()));
+ // no, bound to a different Haskell thread: pass to that thread
+ PUSH_ON_RUN_QUEUE(t);
+ passCapability(&sched_mutex,cap,&m->bound_thread_cond);
+ cap = NULL;
+ continue;
+ }
+ }
+ else
+ {
+ // The thread we want to run is not bound.
+ if(mainThread == NULL)
+ {
+ IF_DEBUG(scheduler,
+ fprintf(stderr,"### Running TSO %p in worker OS thread %u\n",
+ t, osThreadId()));
+ // if we are a worker thread,
+ // we may run it here
+ }
+ else
+ {
+ IF_DEBUG(scheduler,
+ fprintf(stderr,"### TSO %p is not appropriate for main thread %p in OS thread %u\n",
+ t, mainThread, osThreadId()));
+ // no, the current native thread is bound to a different
+ // Haskell thread, so pass it to any worker thread
+ PUSH_ON_RUN_QUEUE(t);
+ passCapabilityToWorker(&sched_mutex, cap);
+ cap = NULL;
+ continue;
+ }
+ }
+ }
+#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 ((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;
+
+run_thread:
+
+ RELEASE_LOCK(&sched_mutex);
+
+ IF_DEBUG(scheduler, sched_belch("-->> running thread %ld %s ...",
+ t->id, whatNext_strs[t->what_next]));
+
+#ifdef PROFILING
+ startHeapProfTimer();
+#endif
+
+ /* +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ */
+ /* Run the current thread
+ */
+ prev_what_next = t->what_next;
+ switch (prev_what_next) {
+ case ThreadKilled:
+ case ThreadComplete:
+ /* Thread already finished, return to scheduler. */
+ ret = ThreadFinished;
+ break;
+ case ThreadRunGHC:
+ errno = t->saved_errno;
+ ret = StgRun((StgFunPtr) stg_returnToStackTop, &cap->r);
+ t->saved_errno = errno;
+ break;
+ case ThreadInterpret:
+ 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 RTS_SUPPORTS_THREADS
+ IF_DEBUG(scheduler,fprintf(stderr,"scheduler (task %p): ", osThreadId()););
+#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 (%s) stopped: requesting a large block (size %d)",
+ t->id, 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. We initialise the
+ // step, gen_no, and flags field of *every* sub-block in
+ // this large block, because this is easier than making
+ // sure that we always find the block head of a large
+ // block whenever we call Bdescr() (eg. evacuate() and
+ // isAlive() in the GC would both have to do this, at
+ // least).
+ {
+ bdescr *x;
+ for (x = bd; x < bd + blocks; x++) {
+ x->step = g0s0;
+ x->gen_no = 0;
+ x->flags = 0;
+ }
+ }
+
+ // don't forget to update the block count in g0s0.
+ g0s0->n_blocks += blocks;
+ // This assert can be a killer if the app is doing lots
+ // of large block allocations.
+ 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;
+ }
+ }
projects / ghc-hetmet.git / blobdiff