+ * blockFetch blocks a BlockedFetch node on some kind of black hole.
+ */
+//@cindex blockFetch
+void
+blockFetch(StgBlockedFetch *bf, StgClosure *bh) {
+ bf->node = bh;
+ switch (get_itbl(bh)->type) {
+ case BLACKHOLE:
+ bf->link = END_BQ_QUEUE;
+ //((StgBlockingQueue *)bh)->header.info = &BLACKHOLE_BQ_info;
+ SET_INFO(bh, &BLACKHOLE_BQ_info); // turn closure into a blocking queue
+ ((StgBlockingQueue *)bh)->blocking_queue = (StgBlockingQueueElement *)bf;
+
+ // put bh on the mutables list
+ recordMutable((StgMutClosure *)bh);
+ break;
+
+ case BLACKHOLE_BQ:
+ /* enqueue bf on blocking queue of closure bh */
+ bf->link = ((StgBlockingQueue *)bh)->blocking_queue;
+ ((StgBlockingQueue *)bh)->blocking_queue = (StgBlockingQueueElement *)bf;
+
+ // put bh on the mutables list; ToDo: check
+ recordMutable((StgMutClosure *)bh);
+ break;
+
+ case FETCH_ME_BQ:
+ /* enqueue bf on blocking queue of closure bh */
+ bf->link = ((StgFetchMeBlockingQueue *)bh)->blocking_queue;
+ ((StgFetchMeBlockingQueue *)bh)->blocking_queue = (StgBlockingQueueElement *)bf;
+
+ // put bh on the mutables list; ToDo: check
+ recordMutable((StgMutClosure *)bh);
+ break;
+
+ case RBH:
+ /* enqueue bf on blocking queue of closure bh */
+ bf->link = ((StgRBH *)bh)->blocking_queue;
+ ((StgRBH *)bh)->blocking_queue = (StgBlockingQueueElement *)bf;
+
+ // put bh on the mutables list; ToDo: check
+ recordMutable((StgMutClosure *)bh);
+ break;
+
+ default:
+ barf("blockFetch: thought %p was a black hole (IP %#lx, %s)",
+ (StgClosure *)bh, get_itbl((StgClosure *)bh),
+ info_type((StgClosure *)bh));
+ }
+ IF_PAR_DEBUG(schedule,
+ belch("##++ blockFetch: after block the BQ of %p (%s) is:",
+ bh, info_type(bh));
+ print_bq(bh));
+}
+
+
+/*
+ blockThread is called from the main scheduler whenever tso returns with
+ a ThreadBlocked return code; tso has already been added to a blocking
+ queue (that's done in the entry code of the closure, because it is a
+ cheap operation we have to do in any case); the main purpose of this
+ routine is to send a Fetch message in case we are blocking on a FETCHME(_BQ)
+ closure, which is indicated by the tso.why_blocked field;
+ we also write an entry into the log file if we are generating one
+
+ Should update exectime etc in the entry code already; but we don't have
+ something like ``system time'' in the log file anyway, so this should
+ even out the inaccuracies.
+*/
+
+//@cindex blockThread
+void
+blockThread(StgTSO *tso)
+{
+ globalAddr *remote_ga;
+ globalAddr *local_ga;
+ globalAddr fmbq_ga;
+
+ // ASSERT(we are on some blocking queue)
+ ASSERT(tso->block_info.closure != (StgClosure *)NULL);
+
+ /*
+ We have to check why this thread has been blocked.
+ */
+ switch (tso->why_blocked) {
+ case BlockedOnGA:
+ /* the closure must be a FETCH_ME_BQ; tso came in here via
+ FETCH_ME entry code */
+ ASSERT(get_itbl(tso->block_info.closure)->type==FETCH_ME_BQ);
+
+ /* HACK: the link field is used to hold the GA between FETCH_ME_entry
+ end this point; if something (eg. GC) happens inbetween the whole
+ thing will blow up
+ The problem is that the ga field of the FETCH_ME has been overwritten
+ with the head of the blocking (which is tso).
+ */
+ //ASSERT(looks_like_ga((globalAddr *)tso->link));
+ ASSERT(tso->link!=END_TSO_QUEUE && tso->link!=NULL);
+ remote_ga = (globalAddr *)tso->link; // ((StgFetchMe *)tso->block_info.closure)->ga;
+ tso->link = END_BQ_QUEUE;
+ /* it was tso which turned node from FETCH_ME into FETCH_ME_BQ =>
+ we have to send a Fetch message here! */
+ if (RtsFlags.ParFlags.ParStats.Full) {
+ /* Note that CURRENT_TIME may perform an unsafe call */
+ //rtsTime now = CURRENT_TIME; /* Now */
+ tso->par.exectime += CURRENT_TIME - tso->par.blockedat;
+ tso->par.fetchcount++;
+ tso->par.blockedat = CURRENT_TIME;
+ /* we are about to send off a FETCH message, so dump a FETCH event */
+ DumpRawGranEvent(CURRENT_PROC,
+ taskIDtoPE(remote_ga->payload.gc.gtid),
+ GR_FETCH, tso, tso->block_info.closure, 0);
+ }
+ /* Phil T. claims that this was a workaround for a hard-to-find
+ * bug, hence I'm leaving it out for now --SDM
+ */
+ /* Assign a brand-new global address to the newly created FMBQ */
+ local_ga = makeGlobal(tso->block_info.closure, rtsFalse);
+ splitWeight(&fmbq_ga, local_ga);
+ ASSERT(fmbq_ga.weight == 1L << (BITS_IN(unsigned) - 1));
+
+ sendFetch(remote_ga, &fmbq_ga, 0/*load*/);
+
+ break;
+
+ case BlockedOnGA_NoSend:
+ /* the closure must be a FETCH_ME_BQ; tso came in here via
+ FETCH_ME_BQ entry code */
+ ASSERT(get_itbl(tso->block_info.closure)->type==FETCH_ME_BQ);
+
+ /* Fetch message has been sent already */
+ if (RtsFlags.ParFlags.ParStats.Full) {
+ /* Note that CURRENT_TIME may perform an unsafe call */
+ //rtsTime now = CURRENT_TIME; /* Now */
+ tso->par.exectime += CURRENT_TIME - tso->par.blockedat;
+ tso->par.blockcount++;
+ tso->par.blockedat = CURRENT_TIME;
+ /* dump a block event, because fetch has been sent already */
+ DumpRawGranEvent(CURRENT_PROC, thisPE,
+ GR_BLOCK, tso, tso->block_info.closure, 0);
+ }
+ break;
+
+ case BlockedOnBlackHole:
+ /* the closure must be a BLACKHOLE_BQ or an RBH; tso came in here via
+ BLACKHOLE(_BQ) or CAF_BLACKHOLE or RBH entry code */
+ ASSERT(get_itbl(tso->block_info.closure)->type==BLACKHOLE_BQ ||
+ get_itbl(tso->block_info.closure)->type==RBH);
+
+ /* if collecting stats update the execution time etc */
+ if (RtsFlags.ParFlags.ParStats.Full) {
+ /* Note that CURRENT_TIME may perform an unsafe call */
+ //rtsTime now = CURRENT_TIME; /* Now */
+ tso->par.exectime += CURRENT_TIME - tso->par.blockedat;
+ tso->par.blockcount++;
+ tso->par.blockedat = CURRENT_TIME;
+ DumpRawGranEvent(CURRENT_PROC, thisPE,
+ GR_BLOCK, tso, tso->block_info.closure, 0);
+ }
+ break;
+
+ default:
+ barf("blockThread: impossible why_blocked code %d for TSO %d",
+ tso->why_blocked, tso->id);
+ }
+
+ IF_PAR_DEBUG(schedule,
+ belch("##++ blockThread: TSO %d blocked on closure %p (%s)",
+ tso->id, tso->block_info.closure, info_type(tso->block_info.closure)));
+}
+
+/*