1 /* ---------------------------------------------------------------------------
3 * (c) The GHC Team, 2006
5 * Thread-related functionality
7 * --------------------------------------------------------------------------*/
9 #include "PosixSource.h"
12 #include "Capability.h"
18 #include "ThreadLabels.h"
21 #include "sm/Storage.h"
23 /* Next thread ID to allocate.
26 static StgThreadID next_thread_id = 1;
28 /* The smallest stack size that makes any sense is:
29 * RESERVED_STACK_WORDS (so we can get back from the stack overflow)
30 * + sizeofW(StgStopFrame) (the stg_stop_thread_info frame)
31 * + 1 (the closure to enter)
33 * + 1 (spare slot req'd by stg_ap_v_ret)
35 * A thread with this stack will bomb immediately with a stack
36 * overflow, which will increase its stack size.
38 #define MIN_STACK_WORDS (RESERVED_STACK_WORDS + sizeofW(StgStopFrame) + 3)
40 /* ---------------------------------------------------------------------------
43 The new thread starts with the given stack size. Before the
44 scheduler can run, however, this thread needs to have a closure
45 (and possibly some arguments) pushed on its stack. See
46 pushClosure() in Schedule.h.
48 createGenThread() and createIOThread() (in SchedAPI.h) are
49 convenient packaged versions of this function.
51 currently pri (priority) is only used in a GRAN setup -- HWL
52 ------------------------------------------------------------------------ */
54 createThread(Capability *cap, nat size)
59 /* sched_mutex is *not* required */
61 /* First check whether we should create a thread at all */
63 // ToDo: check whether size = stack_size - TSO_STRUCT_SIZEW
65 /* catch ridiculously small stack sizes */
66 if (size < MIN_STACK_WORDS + TSO_STRUCT_SIZEW) {
67 size = MIN_STACK_WORDS + TSO_STRUCT_SIZEW;
70 size = round_to_mblocks(size);
71 tso = (StgTSO *)allocate(cap, size);
73 stack_size = size - TSO_STRUCT_SIZEW;
74 TICK_ALLOC_TSO(stack_size, 0);
76 SET_HDR(tso, &stg_TSO_info, CCS_SYSTEM);
78 // Always start with the compiled code evaluator
79 tso->what_next = ThreadRunGHC;
81 tso->why_blocked = NotBlocked;
82 tso->block_info.closure = (StgClosure *)END_TSO_QUEUE;
83 tso->blocked_exceptions = END_BLOCKED_EXCEPTIONS_QUEUE;
84 tso->bq = (StgBlockingQueue *)END_TSO_QUEUE;
92 tso->stack_size = stack_size;
93 tso->max_stack_size = round_to_mblocks(RtsFlags.GcFlags.maxStkSize)
95 tso->sp = (P_)&(tso->stack) + stack_size;
100 tso->prof.CCCS = CCS_MAIN;
103 /* put a stop frame on the stack */
104 tso->sp -= sizeofW(StgStopFrame);
105 SET_HDR((StgClosure*)tso->sp,(StgInfoTable *)&stg_stop_thread_info,CCS_SYSTEM);
106 tso->_link = END_TSO_QUEUE;
108 /* Link the new thread on the global thread list.
110 ACQUIRE_LOCK(&sched_mutex);
111 tso->id = next_thread_id++; // while we have the mutex
112 tso->global_link = g0->threads;
114 RELEASE_LOCK(&sched_mutex);
116 // ToDo: report the stack size in the event?
117 traceEventCreateThread(cap, tso);
122 /* ---------------------------------------------------------------------------
123 * Comparing Thread ids.
125 * This is used from STG land in the implementation of the
126 * instances of Eq/Ord for ThreadIds.
127 * ------------------------------------------------------------------------ */
130 cmp_thread(StgPtr tso1, StgPtr tso2)
132 StgThreadID id1 = ((StgTSO *)tso1)->id;
133 StgThreadID id2 = ((StgTSO *)tso2)->id;
135 if (id1 < id2) return (-1);
136 if (id1 > id2) return 1;
140 /* ---------------------------------------------------------------------------
141 * Fetching the ThreadID from an StgTSO.
143 * This is used in the implementation of Show for ThreadIds.
144 * ------------------------------------------------------------------------ */
146 rts_getThreadId(StgPtr tso)
148 return ((StgTSO *)tso)->id;
151 /* -----------------------------------------------------------------------------
152 Remove a thread from a queue.
153 Fails fatally if the TSO is not on the queue.
154 -------------------------------------------------------------------------- */
156 rtsBool // returns True if we modified queue
157 removeThreadFromQueue (Capability *cap, StgTSO **queue, StgTSO *tso)
162 for (t = *queue; t != END_TSO_QUEUE; prev = t, t = t->_link) {
165 setTSOLink(cap,prev,t->_link);
173 barf("removeThreadFromQueue: not found");
176 rtsBool // returns True if we modified head or tail
177 removeThreadFromDeQueue (Capability *cap,
178 StgTSO **head, StgTSO **tail, StgTSO *tso)
181 rtsBool flag = rtsFalse;
184 for (t = *head; t != END_TSO_QUEUE; prev = t, t = t->_link) {
187 setTSOLink(cap,prev,t->_link);
197 *tail = END_TSO_QUEUE;
205 barf("removeThreadFromMVarQueue: not found");
208 /* ----------------------------------------------------------------------------
211 Attempt to wake up a thread. tryWakeupThread is idempotent: it is
212 always safe to call it too many times, but it is not safe in
213 general to omit a call.
215 ------------------------------------------------------------------------- */
218 tryWakeupThread (Capability *cap, StgTSO *tso)
220 tryWakeupThread_(cap, deRefTSO(tso));
224 tryWakeupThread_ (Capability *cap, StgTSO *tso)
226 traceEventThreadWakeup (cap, tso, tso->cap->no);
232 msg = (MessageWakeup *)allocate(cap,sizeofW(MessageWakeup));
233 SET_HDR(msg, &stg_MSG_TRY_WAKEUP_info, CCS_SYSTEM);
235 sendMessage(cap, tso->cap, (Message*)msg);
236 debugTraceCap(DEBUG_sched, cap, "message: try wakeup thread %ld on cap %d",
237 (lnat)tso->id, tso->cap->no);
242 switch (tso->why_blocked)
246 if (tso->_link == END_TSO_QUEUE) {
247 tso->block_info.closure = (StgClosure*)END_TSO_QUEUE;
254 case BlockedOnMsgThrowTo:
256 const StgInfoTable *i;
258 i = lockClosure(tso->block_info.closure);
259 unlockClosure(tso->block_info.closure, i);
260 if (i != &stg_MSG_NULL_info) {
261 debugTraceCap(DEBUG_sched, cap, "thread %ld still blocked on throwto (%p)",
262 (lnat)tso->id, tso->block_info.throwto->header.info);
266 // remove the block frame from the stack
267 ASSERT(tso->sp[0] == (StgWord)&stg_block_throwto_info);
272 case BlockedOnBlackHole:
274 case ThreadMigrating:
278 // otherwise, do nothing
283 // just run the thread now, if the BH is not really available,
284 // we'll block again.
285 tso->why_blocked = NotBlocked;
286 appendToRunQueue(cap,tso);
289 /* ----------------------------------------------------------------------------
291 ------------------------------------------------------------------------- */
294 migrateThread (Capability *from, StgTSO *tso, Capability *to)
296 traceEventMigrateThread (from, tso, to->no);
297 // ThreadMigrating tells the target cap that it needs to be added to
298 // the run queue when it receives the MSG_TRY_WAKEUP.
299 tso->why_blocked = ThreadMigrating;
301 tryWakeupThread(from, tso);
304 /* ----------------------------------------------------------------------------
307 wakes up all the threads on the specified queue.
308 ------------------------------------------------------------------------- */
311 wakeBlockingQueue(Capability *cap, StgBlockingQueue *bq)
313 MessageBlackHole *msg;
314 const StgInfoTable *i;
316 ASSERT(bq->header.info == &stg_BLOCKING_QUEUE_DIRTY_info ||
317 bq->header.info == &stg_BLOCKING_QUEUE_CLEAN_info );
319 for (msg = bq->queue; msg != (MessageBlackHole*)END_TSO_QUEUE;
321 i = msg->header.info;
322 if (i != &stg_IND_info) {
323 ASSERT(i == &stg_MSG_BLACKHOLE_info);
324 tryWakeupThread(cap,msg->tso);
328 // overwrite the BQ with an indirection so it will be
329 // collected at the next GC.
330 #if defined(DEBUG) && !defined(THREADED_RTS)
331 // XXX FILL_SLOP, but not if THREADED_RTS because in that case
332 // another thread might be looking at this BLOCKING_QUEUE and
333 // checking the owner field at the same time.
334 bq->bh = 0; bq->queue = 0; bq->owner = 0;
336 OVERWRITE_INFO(bq, &stg_IND_info);
339 // If we update a closure that we know we BLACKHOLE'd, and the closure
340 // no longer points to the current TSO as its owner, then there may be
341 // an orphaned BLOCKING_QUEUE closure with blocked threads attached to
342 // it. We therefore traverse the BLOCKING_QUEUEs attached to the
343 // current TSO to see if any can now be woken up.
345 checkBlockingQueues (Capability *cap, StgTSO *tso)
347 StgBlockingQueue *bq, *next;
350 debugTraceCap(DEBUG_sched, cap,
351 "collision occurred; checking blocking queues for thread %ld",
354 for (bq = tso->bq; bq != (StgBlockingQueue*)END_TSO_QUEUE; bq = next) {
357 if (bq->header.info == &stg_IND_info) {
358 // ToDo: could short it out right here, to avoid
359 // traversing this IND multiple times.
365 if (p->header.info != &stg_BLACKHOLE_info ||
366 ((StgInd *)p)->indirectee != (StgClosure*)bq)
368 wakeBlockingQueue(cap,bq);
373 /* ----------------------------------------------------------------------------
376 Update a thunk with a value. In order to do this, we need to know
377 which TSO owns (or is evaluating) the thunk, in case we need to
378 awaken any threads that are blocked on it.
379 ------------------------------------------------------------------------- */
382 updateThunk (Capability *cap, StgTSO *tso, StgClosure *thunk, StgClosure *val)
386 const StgInfoTable *i;
388 i = thunk->header.info;
389 if (i != &stg_BLACKHOLE_info &&
390 i != &stg_CAF_BLACKHOLE_info &&
391 i != &stg_WHITEHOLE_info) {
392 updateWithIndirection(cap, thunk, val);
396 v = ((StgInd*)thunk)->indirectee;
398 updateWithIndirection(cap, thunk, val);
401 if (i == &stg_TSO_info) {
402 owner = deRefTSO((StgTSO*)v);
404 checkBlockingQueues(cap, tso);
409 if (i != &stg_BLOCKING_QUEUE_CLEAN_info &&
410 i != &stg_BLOCKING_QUEUE_DIRTY_info) {
411 checkBlockingQueues(cap, tso);
415 owner = deRefTSO(((StgBlockingQueue*)v)->owner);
418 checkBlockingQueues(cap, tso);
420 wakeBlockingQueue(cap, (StgBlockingQueue*)v);
424 /* ---------------------------------------------------------------------------
425 * rtsSupportsBoundThreads(): is the RTS built to support bound threads?
426 * used by Control.Concurrent for error checking.
427 * ------------------------------------------------------------------------- */
430 rtsSupportsBoundThreads(void)
432 #if defined(THREADED_RTS)
435 return HS_BOOL_FALSE;
439 /* ---------------------------------------------------------------------------
440 * isThreadBound(tso): check whether tso is bound to an OS thread.
441 * ------------------------------------------------------------------------- */
444 isThreadBound(StgTSO* tso USED_IF_THREADS)
446 #if defined(THREADED_RTS)
447 return (tso->bound != NULL);
452 /* ----------------------------------------------------------------------------
453 * Debugging: why is a thread blocked
454 * ------------------------------------------------------------------------- */
458 printThreadBlockage(StgTSO *tso)
460 switch (tso->why_blocked) {
462 debugBelch("is blocked on read from fd %d", (int)(tso->block_info.fd));
465 debugBelch("is blocked on write to fd %d", (int)(tso->block_info.fd));
467 #if defined(mingw32_HOST_OS)
468 case BlockedOnDoProc:
469 debugBelch("is blocked on proc (request: %u)", tso->block_info.async_result->reqID);
473 debugBelch("is blocked until %ld", (long)(tso->block_info.target));
476 debugBelch("is blocked on an MVar @ %p", tso->block_info.closure);
478 case BlockedOnBlackHole:
479 debugBelch("is blocked on a black hole %p",
480 ((StgBlockingQueue*)tso->block_info.bh->bh));
482 case BlockedOnMsgThrowTo:
483 debugBelch("is blocked on a throwto message");
486 debugBelch("is not blocked");
488 case ThreadMigrating:
489 debugBelch("is runnable, but not on the run queue");
492 debugBelch("is blocked on an external call");
494 case BlockedOnCCall_NoUnblockExc:
495 debugBelch("is blocked on an external call (exceptions were already blocked)");
498 debugBelch("is blocked on an STM operation");
501 barf("printThreadBlockage: strange tso->why_blocked: %d for TSO %d (%d)",
502 tso->why_blocked, tso->id, tso);
508 printThreadStatus(StgTSO *t)
510 debugBelch("\tthread %4lu @ %p ", (unsigned long)t->id, (void *)t);
512 void *label = lookupThreadLabel(t->id);
513 if (label) debugBelch("[\"%s\"] ",(char *)label);
515 if (t->what_next == ThreadRelocated) {
516 debugBelch("has been relocated...\n");
518 switch (t->what_next) {
520 debugBelch("has been killed");
523 debugBelch("has completed");
526 printThreadBlockage(t);
529 debugBelch(" (TSO_DIRTY)");
530 } else if (t->flags & TSO_LINK_DIRTY) {
531 debugBelch(" (TSO_LINK_DIRTY)");
538 printAllThreads(void)
544 debugBelch("all threads:\n");
546 for (i = 0; i < n_capabilities; i++) {
547 cap = &capabilities[i];
548 debugBelch("threads on capability %d:\n", cap->no);
549 for (t = cap->run_queue_hd; t != END_TSO_QUEUE; t = t->_link) {
550 printThreadStatus(t);
554 debugBelch("other threads:\n");
555 for (g = 0; g < RtsFlags.GcFlags.generations; g++) {
556 for (t = generations[g].threads; t != END_TSO_QUEUE; t = next) {
557 if (t->why_blocked != NotBlocked) {
558 printThreadStatus(t);
560 if (t->what_next == ThreadRelocated) {
563 next = t->global_link;
571 printThreadQueue(StgTSO *t)
574 for (; t != END_TSO_QUEUE; t = t->_link) {
575 printThreadStatus(t);
578 debugBelch("%d threads on queue\n", i);