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_EAGER_BLACKHOLE_info &&
392 i != &stg_WHITEHOLE_info) {
393 updateWithIndirection(cap, thunk, val);
397 v = ((StgInd*)thunk)->indirectee;
399 updateWithIndirection(cap, thunk, val);
402 if (i == &stg_TSO_info) {
403 owner = deRefTSO((StgTSO*)v);
405 checkBlockingQueues(cap, tso);
410 if (i != &stg_BLOCKING_QUEUE_CLEAN_info &&
411 i != &stg_BLOCKING_QUEUE_DIRTY_info) {
412 checkBlockingQueues(cap, tso);
416 owner = deRefTSO(((StgBlockingQueue*)v)->owner);
419 checkBlockingQueues(cap, tso);
421 wakeBlockingQueue(cap, (StgBlockingQueue*)v);
425 /* ---------------------------------------------------------------------------
426 * rtsSupportsBoundThreads(): is the RTS built to support bound threads?
427 * used by Control.Concurrent for error checking.
428 * ------------------------------------------------------------------------- */
431 rtsSupportsBoundThreads(void)
433 #if defined(THREADED_RTS)
436 return HS_BOOL_FALSE;
440 /* ---------------------------------------------------------------------------
441 * isThreadBound(tso): check whether tso is bound to an OS thread.
442 * ------------------------------------------------------------------------- */
445 isThreadBound(StgTSO* tso USED_IF_THREADS)
447 #if defined(THREADED_RTS)
448 return (tso->bound != NULL);
453 /* ----------------------------------------------------------------------------
454 * Debugging: why is a thread blocked
455 * ------------------------------------------------------------------------- */
459 printThreadBlockage(StgTSO *tso)
461 switch (tso->why_blocked) {
463 debugBelch("is blocked on read from fd %d", (int)(tso->block_info.fd));
466 debugBelch("is blocked on write to fd %d", (int)(tso->block_info.fd));
468 #if defined(mingw32_HOST_OS)
469 case BlockedOnDoProc:
470 debugBelch("is blocked on proc (request: %u)", tso->block_info.async_result->reqID);
474 debugBelch("is blocked until %ld", (long)(tso->block_info.target));
477 debugBelch("is blocked on an MVar @ %p", tso->block_info.closure);
479 case BlockedOnBlackHole:
480 debugBelch("is blocked on a black hole %p",
481 ((StgBlockingQueue*)tso->block_info.bh->bh));
483 case BlockedOnMsgThrowTo:
484 debugBelch("is blocked on a throwto message");
487 debugBelch("is not blocked");
489 case ThreadMigrating:
490 debugBelch("is runnable, but not on the run queue");
493 debugBelch("is blocked on an external call");
495 case BlockedOnCCall_NoUnblockExc:
496 debugBelch("is blocked on an external call (exceptions were already blocked)");
499 debugBelch("is blocked on an STM operation");
502 barf("printThreadBlockage: strange tso->why_blocked: %d for TSO %d (%d)",
503 tso->why_blocked, tso->id, tso);
509 printThreadStatus(StgTSO *t)
511 debugBelch("\tthread %4lu @ %p ", (unsigned long)t->id, (void *)t);
513 void *label = lookupThreadLabel(t->id);
514 if (label) debugBelch("[\"%s\"] ",(char *)label);
516 if (t->what_next == ThreadRelocated) {
517 debugBelch("has been relocated...\n");
519 switch (t->what_next) {
521 debugBelch("has been killed");
524 debugBelch("has completed");
527 printThreadBlockage(t);
530 debugBelch(" (TSO_DIRTY)");
531 } else if (t->flags & TSO_LINK_DIRTY) {
532 debugBelch(" (TSO_LINK_DIRTY)");
539 printAllThreads(void)
545 debugBelch("all threads:\n");
547 for (i = 0; i < n_capabilities; i++) {
548 cap = &capabilities[i];
549 debugBelch("threads on capability %d:\n", cap->no);
550 for (t = cap->run_queue_hd; t != END_TSO_QUEUE; t = t->_link) {
551 printThreadStatus(t);
555 debugBelch("other threads:\n");
556 for (g = 0; g < RtsFlags.GcFlags.generations; g++) {
557 for (t = generations[g].threads; t != END_TSO_QUEUE; t = next) {
558 if (t->why_blocked != NotBlocked) {
559 printThreadStatus(t);
561 if (t->what_next == ThreadRelocated) {
564 next = t->global_link;
572 printThreadQueue(StgTSO *t)
575 for (; t != END_TSO_QUEUE; t = t->_link) {
576 printThreadStatus(t);
579 debugBelch("%d threads on queue\n", i);