- Locks: sched_mutex is held upon entry and exit.
-
-*/
-static void
-unblockThread(Capability *cap, StgTSO *tso)
-{
- StgBlockingQueueElement *t, **last;
-
- switch (tso->why_blocked) {
-
- case NotBlocked:
- return; /* not blocked */
-
- case BlockedOnSTM:
- // Be careful: nothing to do here! We tell the scheduler that the thread
- // is runnable and we leave it to the stack-walking code to abort the
- // transaction while unwinding the stack. We should perhaps have a debugging
- // test to make sure that this really happens and that the 'zombie' transaction
- // does not get committed.
- goto done;
-
- case BlockedOnMVar:
- ASSERT(get_itbl(tso->block_info.closure)->type == MVAR);
- {
- StgBlockingQueueElement *last_tso = END_BQ_QUEUE;
- StgMVar *mvar = (StgMVar *)(tso->block_info.closure);
-
- last = (StgBlockingQueueElement **)&mvar->head;
- for (t = (StgBlockingQueueElement *)mvar->head;
- t != END_BQ_QUEUE;
- last = &t->link, last_tso = t, t = t->link) {
- if (t == (StgBlockingQueueElement *)tso) {
- *last = (StgBlockingQueueElement *)tso->link;
- if (mvar->tail == tso) {
- mvar->tail = (StgTSO *)last_tso;
- }
- goto done;
- }
- }
- barf("unblockThread (MVAR): TSO not found");
- }
-
- case BlockedOnBlackHole:
- ASSERT(get_itbl(tso->block_info.closure)->type == BLACKHOLE_BQ);
- {
- StgBlockingQueue *bq = (StgBlockingQueue *)(tso->block_info.closure);
-
- last = &bq->blocking_queue;
- for (t = bq->blocking_queue;
- t != END_BQ_QUEUE;
- last = &t->link, t = t->link) {
- if (t == (StgBlockingQueueElement *)tso) {
- *last = (StgBlockingQueueElement *)tso->link;
- goto done;
- }
- }
- barf("unblockThread (BLACKHOLE): TSO not found");
- }
-
- case BlockedOnException:
- {
- StgTSO *target = tso->block_info.tso;
-
- ASSERT(get_itbl(target)->type == TSO);
-
- if (target->what_next == ThreadRelocated) {
- target = target->link;
- ASSERT(get_itbl(target)->type == TSO);
- }
-
- ASSERT(target->blocked_exceptions != NULL);
-
- last = (StgBlockingQueueElement **)&target->blocked_exceptions;
- for (t = (StgBlockingQueueElement *)target->blocked_exceptions;
- t != END_BQ_QUEUE;
- last = &t->link, t = t->link) {
- ASSERT(get_itbl(t)->type == TSO);
- if (t == (StgBlockingQueueElement *)tso) {
- *last = (StgBlockingQueueElement *)tso->link;
- goto done;
- }
- }
- barf("unblockThread (Exception): TSO not found");
- }
-
- case BlockedOnRead:
- case BlockedOnWrite:
-#if defined(mingw32_HOST_OS)
- case BlockedOnDoProc:
-#endif
- {
- /* take TSO off blocked_queue */
- StgBlockingQueueElement *prev = NULL;
- for (t = (StgBlockingQueueElement *)blocked_queue_hd; t != END_BQ_QUEUE;
- prev = t, t = t->link) {
- if (t == (StgBlockingQueueElement *)tso) {
- if (prev == NULL) {
- blocked_queue_hd = (StgTSO *)t->link;
- if ((StgBlockingQueueElement *)blocked_queue_tl == t) {
- blocked_queue_tl = END_TSO_QUEUE;
- }
- } else {
- prev->link = t->link;
- if ((StgBlockingQueueElement *)blocked_queue_tl == t) {
- blocked_queue_tl = (StgTSO *)prev;
- }
- }
-#if defined(mingw32_HOST_OS)
- /* (Cooperatively) signal that the worker thread should abort
- * the request.
- */
- abandonWorkRequest(tso->block_info.async_result->reqID);
-#endif
- goto done;
- }
- }
- barf("unblockThread (I/O): TSO not found");
- }
-
- case BlockedOnDelay:
- {
- /* take TSO off sleeping_queue */
- StgBlockingQueueElement *prev = NULL;
- for (t = (StgBlockingQueueElement *)sleeping_queue; t != END_BQ_QUEUE;
- prev = t, t = t->link) {
- if (t == (StgBlockingQueueElement *)tso) {
- if (prev == NULL) {
- sleeping_queue = (StgTSO *)t->link;
- } else {
- prev->link = t->link;
- }
- goto done;
- }
- }
- barf("unblockThread (delay): TSO not found");
- }
-
- default:
- barf("unblockThread");
- }
-
- done:
- tso->link = END_TSO_QUEUE;
- tso->why_blocked = NotBlocked;
- tso->block_info.closure = NULL;
- pushOnRunQueue(cap,tso);
-}
-#else
-static void
-unblockThread(Capability *cap, StgTSO *tso)
-{
- StgTSO *t, **last;
-
- /* To avoid locking unnecessarily. */
- if (tso->why_blocked == NotBlocked) {
- return;
- }
-
- switch (tso->why_blocked) {
-
- case BlockedOnSTM:
- // Be careful: nothing to do here! We tell the scheduler that the thread
- // is runnable and we leave it to the stack-walking code to abort the
- // transaction while unwinding the stack. We should perhaps have a debugging
- // test to make sure that this really happens and that the 'zombie' transaction
- // does not get committed.
- goto done;
-
- case BlockedOnMVar:
- ASSERT(get_itbl(tso->block_info.closure)->type == MVAR);
- {
- StgTSO *last_tso = END_TSO_QUEUE;
- StgMVar *mvar = (StgMVar *)(tso->block_info.closure);
-
- last = &mvar->head;
- for (t = mvar->head; t != END_TSO_QUEUE;
- last = &t->link, last_tso = t, t = t->link) {
- if (t == tso) {
- *last = tso->link;
- if (mvar->tail == tso) {
- mvar->tail = last_tso;
- }
- goto done;
- }
- }
- barf("unblockThread (MVAR): TSO not found");
- }
-
- case BlockedOnBlackHole:
- {
- last = &blackhole_queue;
- for (t = blackhole_queue; t != END_TSO_QUEUE;
- last = &t->link, t = t->link) {
- if (t == tso) {
- *last = tso->link;
- goto done;
- }
- }
- barf("unblockThread (BLACKHOLE): TSO not found");
- }
-
- case BlockedOnException:
- {
- StgTSO *target = tso->block_info.tso;
-
- ASSERT(get_itbl(target)->type == TSO);
-
- while (target->what_next == ThreadRelocated) {
- target = target->link;
- ASSERT(get_itbl(target)->type == TSO);
- }
-
- ASSERT(target->blocked_exceptions != NULL);
-
- last = &target->blocked_exceptions;
- for (t = target->blocked_exceptions; t != END_TSO_QUEUE;
- last = &t->link, t = t->link) {
- ASSERT(get_itbl(t)->type == TSO);
- if (t == tso) {
- *last = tso->link;
- goto done;
- }
- }
- barf("unblockThread (Exception): TSO not found");
- }
-
-#if !defined(THREADED_RTS)
- case BlockedOnRead:
- case BlockedOnWrite:
-#if defined(mingw32_HOST_OS)
- case BlockedOnDoProc:
-#endif
- {
- StgTSO *prev = NULL;
- for (t = blocked_queue_hd; t != END_TSO_QUEUE;
- prev = t, t = t->link) {
- if (t == tso) {
- if (prev == NULL) {
- blocked_queue_hd = t->link;
- if (blocked_queue_tl == t) {
- blocked_queue_tl = END_TSO_QUEUE;
- }
- } else {
- prev->link = t->link;
- if (blocked_queue_tl == t) {
- blocked_queue_tl = prev;
- }
- }
-#if defined(mingw32_HOST_OS)
- /* (Cooperatively) signal that the worker thread should abort
- * the request.
- */
- abandonWorkRequest(tso->block_info.async_result->reqID);
-#endif
- goto done;
- }
- }
- barf("unblockThread (I/O): TSO not found");
- }
-
- case BlockedOnDelay:
- {
- StgTSO *prev = NULL;
- for (t = sleeping_queue; t != END_TSO_QUEUE;
- prev = t, t = t->link) {
- if (t == tso) {
- if (prev == NULL) {
- sleeping_queue = t->link;
- } else {
- prev->link = t->link;
- }
- goto done;
- }
- }
- barf("unblockThread (delay): TSO not found");
- }
-#endif
-
- default:
- barf("unblockThread");
- }
-
- done:
- tso->link = END_TSO_QUEUE;
- tso->why_blocked = NotBlocked;
- tso->block_info.closure = NULL;
- appendToRunQueue(cap,tso);
-
- // We might have just migrated this TSO to our Capability:
- if (tso->bound) {
- tso->bound->cap = cap;
- }
- tso->cap = cap;
-}
-#endif
-
-/* -----------------------------------------------------------------------------
- * checkBlackHoles()
- *
- * Check the blackhole_queue for threads that can be woken up. We do
- * this periodically: before every GC, and whenever the run queue is
- * empty.
- *
- * An elegant solution might be to just wake up all the blocked
- * threads with awakenBlockedQueue occasionally: they'll go back to
- * sleep again if the object is still a BLACKHOLE. Unfortunately this
- * doesn't give us a way to tell whether we've actually managed to
- * wake up any threads, so we would be busy-waiting.
- *
- * -------------------------------------------------------------------------- */
-
-static rtsBool
-checkBlackHoles (Capability *cap)
-{
- StgTSO **prev, *t;
- rtsBool any_woke_up = rtsFalse;
- StgHalfWord type;
-
- // blackhole_queue is global:
- ASSERT_LOCK_HELD(&sched_mutex);
-
- debugTrace(DEBUG_sched, "checking threads blocked on black holes");
-
- // ASSUMES: sched_mutex
- prev = &blackhole_queue;
- t = blackhole_queue;
- while (t != END_TSO_QUEUE) {
- ASSERT(t->why_blocked == BlockedOnBlackHole);
- type = get_itbl(t->block_info.closure)->type;
- if (type != BLACKHOLE && type != CAF_BLACKHOLE) {
- IF_DEBUG(sanity,checkTSO(t));
- t = unblockOne(cap, t);
- // urk, the threads migrate to the current capability
- // here, but we'd like to keep them on the original one.
- *prev = t;
- any_woke_up = rtsTrue;
- } else {
- prev = &t->link;
- t = t->link;
- }
- }
-
- return any_woke_up;
-}
-
-/* -----------------------------------------------------------------------------
- * raiseAsync()
- *
- * The following function implements the magic for raising an
- * asynchronous exception in an existing thread.
- *
- * We first remove the thread from any queue on which it might be
- * blocked. The possible blockages are MVARs and BLACKHOLE_BQs.
- *
- * We strip the stack down to the innermost CATCH_FRAME, building
- * thunks in the heap for all the active computations, so they can
- * be restarted if necessary. When we reach a CATCH_FRAME, we build
- * an application of the handler to the exception, and push it on
- * the top of the stack.
- *
- * How exactly do we save all the active computations? We create an
- * AP_STACK for every UpdateFrame on the stack. Entering one of these
- * AP_STACKs pushes everything from the corresponding update frame
- * upwards onto the stack. (Actually, it pushes everything up to the
- * next update frame plus a pointer to the next AP_STACK object.
- * Entering the next AP_STACK object pushes more onto the stack until we
- * reach the last AP_STACK object - at which point the stack should look
- * exactly as it did when we killed the TSO and we can continue
- * execution by entering the closure on top of the stack.
- *
- * We can also kill a thread entirely - this happens if either (a) the
- * exception passed to raiseAsync is NULL, or (b) there's no
- * CATCH_FRAME on the stack. In either case, we strip the entire
- * stack and replace the thread with a zombie.
- *
- * ToDo: in THREADED_RTS mode, this function is only safe if either
- * (a) we hold all the Capabilities (eg. in GC, or if there is only
- * one Capability), or (b) we own the Capability that the TSO is
- * currently blocked on or on the run queue of.
- *
- * -------------------------------------------------------------------------- */
-
-void
-raiseAsync(Capability *cap, StgTSO *tso, StgClosure *exception)
-{
- raiseAsync_(cap, tso, exception, rtsFalse, NULL);
-}
-
-void
-suspendComputation(Capability *cap, StgTSO *tso, StgPtr stop_here)
-{
- raiseAsync_(cap, tso, NULL, rtsFalse, stop_here);
-}
-
-static void
-raiseAsync_(Capability *cap, StgTSO *tso, StgClosure *exception,
- rtsBool stop_at_atomically, StgPtr stop_here)
-{
- StgRetInfoTable *info;
- StgPtr sp, frame;
- nat i;
-
- // Thread already dead?
- if (tso->what_next == ThreadComplete || tso->what_next == ThreadKilled) {
- return;