rts/Threads.c

   1 /* ---------------------------------------------------------------------------
   2  *
   3  * (c) The GHC Team, 2006
   4  *
   5  * Thread-related functionality
   6  *
   7  * --------------------------------------------------------------------------*/
   8
   9 #include "PosixSource.h"
  10 #include "Rts.h"
  11 #include "SchedAPI.h"
  12 #include "Storage.h"
  13 #include "Threads.h"
  14 #include "RtsFlags.h"
  15 #include "STM.h"
  16 #include "Schedule.h"
  17 #include "Trace.h"
  18 #include "ThreadLabels.h"
  19
  20 /* Next thread ID to allocate.
  21  * LOCK: sched_mutex
  22  */
  23 static StgThreadID next_thread_id = 1;
  24
  25 /* The smallest stack size that makes any sense is:
  26  *    RESERVED_STACK_WORDS    (so we can get back from the stack overflow)
  27  *  + sizeofW(StgStopFrame)   (the stg_stop_thread_info frame)
  28  *  + 1                       (the closure to enter)
  29  *  + 1                       (stg_ap_v_ret)
  30  *  + 1                       (spare slot req'd by stg_ap_v_ret)
  31  *
  32  * A thread with this stack will bomb immediately with a stack
  33  * overflow, which will increase its stack size.
  34  */
  35 #define MIN_STACK_WORDS (RESERVED_STACK_WORDS + sizeofW(StgStopFrame) + 3)
  36
  37 /* ---------------------------------------------------------------------------
  38    Create a new thread.
  39
  40    The new thread starts with the given stack size.  Before the
  41    scheduler can run, however, this thread needs to have a closure
  42    (and possibly some arguments) pushed on its stack.  See
  43    pushClosure() in Schedule.h.
  44
  45    createGenThread() and createIOThread() (in SchedAPI.h) are
  46    convenient packaged versions of this function.
  47
  48    currently pri (priority) is only used in a GRAN setup -- HWL
  49    ------------------------------------------------------------------------ */
  50 StgTSO *
  51 createThread(Capability *cap, nat size)
  52 {
  53     StgTSO *tso;
  54     nat stack_size;
  55
  56     /* sched_mutex is *not* required */
  57
  58     /* First check whether we should create a thread at all */
  59
  60     // ToDo: check whether size = stack_size - TSO_STRUCT_SIZEW
  61
  62     /* catch ridiculously small stack sizes */
  63     if (size < MIN_STACK_WORDS + TSO_STRUCT_SIZEW) {
  64         size = MIN_STACK_WORDS + TSO_STRUCT_SIZEW;
  65     }
  66
  67     size = round_to_mblocks(size);
  68     tso = (StgTSO *)allocateLocal(cap, size);
  69
  70     stack_size = size - TSO_STRUCT_SIZEW;
  71     TICK_ALLOC_TSO(stack_size, 0);
  72
  73     SET_HDR(tso, &stg_TSO_info, CCS_SYSTEM);
  74
  75     // Always start with the compiled code evaluator
  76     tso->what_next = ThreadRunGHC;
  77
  78     tso->why_blocked  = NotBlocked;
  79     tso->blocked_exceptions = END_TSO_QUEUE;
  80     tso->flags = TSO_DIRTY;
  81
  82     tso->saved_errno = 0;
  83     tso->bound = NULL;
  84     tso->cap = cap;
  85
  86     tso->stack_size     = stack_size;
  87     tso->max_stack_size = round_to_mblocks(RtsFlags.GcFlags.maxStkSize)
  88                           - TSO_STRUCT_SIZEW;
  89     tso->sp             = (P_)&(tso->stack) + stack_size;
  90
  91     tso->trec = NO_TREC;
  92
  93 #ifdef PROFILING
  94     tso->prof.CCCS = CCS_MAIN;
  95 #endif
  96
  97   /* put a stop frame on the stack */
  98     tso->sp -= sizeofW(StgStopFrame);
  99     SET_HDR((StgClosure*)tso->sp,(StgInfoTable *)&stg_stop_thread_info,CCS_SYSTEM);
 100     tso->_link = END_TSO_QUEUE;
 101
 102     /* Link the new thread on the global thread list.
 103      */
 104     ACQUIRE_LOCK(&sched_mutex);
 105     tso->id = next_thread_id++;  // while we have the mutex
 106     tso->global_link = g0s0->threads;
 107     g0s0->threads = tso;
 108     RELEASE_LOCK(&sched_mutex);
 109
 110     postEvent (cap, EVENT_CREATE_THREAD, tso->id, 0);
 111
 112     debugTrace(DEBUG_sched,
 113                "created thread %ld, stack size = %lx words",
 114                (long)tso->id, (long)tso->stack_size);
 115     return tso;
 116 }
 117
 118 /* ---------------------------------------------------------------------------
 119  * Comparing Thread ids.
 120  *
 121  * This is used from STG land in the implementation of the
 122  * instances of Eq/Ord for ThreadIds.
 123  * ------------------------------------------------------------------------ */
 124
 125 int
 126 cmp_thread(StgPtr tso1, StgPtr tso2)
 127 {
 128   StgThreadID id1 = ((StgTSO *)tso1)->id;
 129   StgThreadID id2 = ((StgTSO *)tso2)->id;
 130
 131   if (id1 < id2) return (-1);
 132   if (id1 > id2) return 1;
 133   return 0;
 134 }
 135
 136 /* ---------------------------------------------------------------------------
 137  * Fetching the ThreadID from an StgTSO.
 138  *
 139  * This is used in the implementation of Show for ThreadIds.
 140  * ------------------------------------------------------------------------ */
 141 int
 142 rts_getThreadId(StgPtr tso)
 143 {
 144   return ((StgTSO *)tso)->id;
 145 }
 146
 147 /* -----------------------------------------------------------------------------
 148    Remove a thread from a queue.
 149    Fails fatally if the TSO is not on the queue.
 150    -------------------------------------------------------------------------- */
 151
 152 void
 153 removeThreadFromQueue (Capability *cap, StgTSO **queue, StgTSO *tso)
 154 {
 155     StgTSO *t, *prev;
 156
 157     prev = NULL;
 158     for (t = *queue; t != END_TSO_QUEUE; prev = t, t = t->_link) {
 159         if (t == tso) {
 160             if (prev) {
 161                 setTSOLink(cap,prev,t->_link);
 162             } else {
 163                 *queue = t->_link;
 164             }
 165             return;
 166         }
 167     }
 168     barf("removeThreadFromQueue: not found");
 169 }
 170
 171 void
 172 removeThreadFromDeQueue (Capability *cap,
 173                          StgTSO **head, StgTSO **tail, StgTSO *tso)
 174 {
 175     StgTSO *t, *prev;
 176
 177     prev = NULL;
 178     for (t = *head; t != END_TSO_QUEUE; prev = t, t = t->_link) {
 179         if (t == tso) {
 180             if (prev) {
 181                 setTSOLink(cap,prev,t->_link);
 182             } else {
 183                 *head = t->_link;
 184             }
 185             if (*tail == tso) {
 186                 if (prev) {
 187                     *tail = prev;
 188                 } else {
 189                     *tail = END_TSO_QUEUE;
 190                 }
 191             }
 192             return;
 193         }
 194     }
 195     barf("removeThreadFromMVarQueue: not found");
 196 }
 197
 198 void
 199 removeThreadFromMVarQueue (Capability *cap, StgMVar *mvar, StgTSO *tso)
 200 {
 201     removeThreadFromDeQueue (cap, &mvar->head, &mvar->tail, tso);
 202 }
 203
 204 /* ----------------------------------------------------------------------------
 205    unblockOne()
 206
 207    unblock a single thread.
 208    ------------------------------------------------------------------------- */
 209
 210 StgTSO *
 211 unblockOne (Capability *cap, StgTSO *tso)
 212 {
 213     return unblockOne_(cap,tso,rtsTrue); // allow migration
 214 }
 215
 216 StgTSO *
 217 unblockOne_ (Capability *cap, StgTSO *tso,
 218              rtsBool allow_migrate USED_IF_THREADS)
 219 {
 220   StgTSO *next;
 221
 222   // NO, might be a WHITEHOLE: ASSERT(get_itbl(tso)->type == TSO);
 223   ASSERT(tso->why_blocked != NotBlocked);
 224
 225   tso->why_blocked = NotBlocked;
 226   next = tso->_link;
 227   tso->_link = END_TSO_QUEUE;
 228
 229 #if defined(THREADED_RTS)
 230   if (tso->cap == cap || (!tsoLocked(tso) &&
 231                           allow_migrate &&
 232                           RtsFlags.ParFlags.wakeupMigrate)) {
 233       // We are waking up this thread on the current Capability, which
 234       // might involve migrating it from the Capability it was last on.
 235       if (tso->bound) {
 236           ASSERT(tso->bound->cap == tso->cap);
 237           tso->bound->cap = cap;
 238       }
 239
 240       tso->cap = cap;
 241       appendToRunQueue(cap,tso);
 242
 243       // context-switch soonish so we can migrate the new thread if
 244       // necessary.  NB. not contextSwitchCapability(cap), which would
 245       // force a context switch immediately.
 246       cap->context_switch = 1;
 247   } else {
 248       // we'll try to wake it up on the Capability it was last on.
 249       wakeupThreadOnCapability(cap, tso->cap, tso);
 250   }
 251 #else
 252   appendToRunQueue(cap,tso);
 253
 254   // context-switch soonish so we can migrate the new thread if
 255   // necessary.  NB. not contextSwitchCapability(cap), which would
 256   // force a context switch immediately.
 257   cap->context_switch = 1;
 258 #endif
 259
 260   postEvent (cap, EVENT_THREAD_WAKEUP, tso->id, tso->cap->no);
 261
 262   debugTrace(DEBUG_sched, "waking up thread %ld on cap %d",
 263              (long)tso->id, tso->cap->no);
 264
 265   return next;
 266 }
 267
 268 /* ----------------------------------------------------------------------------
 269    awakenBlockedQueue
 270
 271    wakes up all the threads on the specified queue.
 272    ------------------------------------------------------------------------- */
 273
 274 void
 275 awakenBlockedQueue(Capability *cap, StgTSO *tso)
 276 {
 277     while (tso != END_TSO_QUEUE) {
 278         tso = unblockOne(cap,tso);
 279     }
 280 }
 281
 282 /* ---------------------------------------------------------------------------
 283  * rtsSupportsBoundThreads(): is the RTS built to support bound threads?
 284  * used by Control.Concurrent for error checking.
 285  * ------------------------------------------------------------------------- */
 286
 287 HsBool
 288 rtsSupportsBoundThreads(void)
 289 {
 290 #if defined(THREADED_RTS)
 291   return HS_BOOL_TRUE;
 292 #else
 293   return HS_BOOL_FALSE;
 294 #endif
 295 }
 296
 297 /* ---------------------------------------------------------------------------
 298  * isThreadBound(tso): check whether tso is bound to an OS thread.
 299  * ------------------------------------------------------------------------- */
 300
 301 StgBool
 302 isThreadBound(StgTSO* tso USED_IF_THREADS)
 303 {
 304 #if defined(THREADED_RTS)
 305   return (tso->bound != NULL);
 306 #endif
 307   return rtsFalse;
 308 }
 309
 310 /* ----------------------------------------------------------------------------
 311  * Debugging: why is a thread blocked
 312  * ------------------------------------------------------------------------- */
 313
 314 #if DEBUG
 315 void
 316 printThreadBlockage(StgTSO *tso)
 317 {
 318   switch (tso->why_blocked) {
 319   case BlockedOnRead:
 320     debugBelch("is blocked on read from fd %d", (int)(tso->block_info.fd));
 321     break;
 322   case BlockedOnWrite:
 323     debugBelch("is blocked on write to fd %d", (int)(tso->block_info.fd));
 324     break;
 325 #if defined(mingw32_HOST_OS)
 326     case BlockedOnDoProc:
 327     debugBelch("is blocked on proc (request: %u)", tso->block_info.async_result->reqID);
 328     break;
 329 #endif
 330   case BlockedOnDelay:
 331     debugBelch("is blocked until %ld", (long)(tso->block_info.target));
 332     break;
 333   case BlockedOnMVar:
 334     debugBelch("is blocked on an MVar @ %p", tso->block_info.closure);
 335     break;
 336   case BlockedOnException:
 337     debugBelch("is blocked on delivering an exception to thread %lu",
 338                (unsigned long)tso->block_info.tso->id);
 339     break;
 340   case BlockedOnBlackHole:
 341     debugBelch("is blocked on a black hole");
 342     break;
 343   case NotBlocked:
 344     debugBelch("is not blocked");
 345     break;
 346   case BlockedOnCCall:
 347     debugBelch("is blocked on an external call");
 348     break;
 349   case BlockedOnCCall_NoUnblockExc:
 350     debugBelch("is blocked on an external call (exceptions were already blocked)");
 351     break;
 352   case BlockedOnSTM:
 353     debugBelch("is blocked on an STM operation");
 354     break;
 355   default:
 356     barf("printThreadBlockage: strange tso->why_blocked: %d for TSO %d (%d)",
 357          tso->why_blocked, tso->id, tso);
 358   }
 359 }
 360
 361 void
 362 printThreadStatus(StgTSO *t)
 363 {
 364   debugBelch("\tthread %4lu @ %p ", (unsigned long)t->id, (void *)t);
 365     {
 366       void *label = lookupThreadLabel(t->id);
 367       if (label) debugBelch("[\"%s\"] ",(char *)label);
 368     }
 369     if (t->what_next == ThreadRelocated) {
 370         debugBelch("has been relocated...\n");
 371     } else {
 372         switch (t->what_next) {
 373         case ThreadKilled:
 374             debugBelch("has been killed");
 375             break;
 376         case ThreadComplete:
 377             debugBelch("has completed");
 378             break;
 379         default:
 380             printThreadBlockage(t);
 381         }
 382         if (t->flags & TSO_DIRTY) {
 383             debugBelch(" (TSO_DIRTY)");
 384         } else if (t->flags & TSO_LINK_DIRTY) {
 385             debugBelch(" (TSO_LINK_DIRTY)");
 386         }
 387         debugBelch("\n");
 388     }
 389 }
 390
 391 void
 392 printAllThreads(void)
 393 {
 394   StgTSO *t, *next;
 395   nat i, s;
 396   Capability *cap;
 397
 398 # if defined(GRAN)
 399   char time_string[TIME_STR_LEN], node_str[NODE_STR_LEN];
 400   ullong_format_string(TIME_ON_PROC(CurrentProc),
 401                        time_string, rtsFalse/*no commas!*/);
 402
 403   debugBelch("all threads at [%s]:\n", time_string);
 404 # elif defined(PARALLEL_HASKELL)
 405   char time_string[TIME_STR_LEN], node_str[NODE_STR_LEN];
 406   ullong_format_string(CURRENT_TIME,
 407                        time_string, rtsFalse/*no commas!*/);
 408
 409   debugBelch("all threads at [%s]:\n", time_string);
 410 # else
 411   debugBelch("all threads:\n");
 412 # endif
 413
 414   for (i = 0; i < n_capabilities; i++) {
 415       cap = &capabilities[i];
 416       debugBelch("threads on capability %d:\n", cap->no);
 417       for (t = cap->run_queue_hd; t != END_TSO_QUEUE; t = t->_link) {
 418           printThreadStatus(t);
 419       }
 420   }
 421
 422   debugBelch("other threads:\n");
 423   for (s = 0; s < total_steps; s++) {
 424     for (t = all_steps[s].threads; t != END_TSO_QUEUE; t = next) {
 425       if (t->why_blocked != NotBlocked) {
 426           printThreadStatus(t);
 427       }
 428       if (t->what_next == ThreadRelocated) {
 429           next = t->_link;
 430       } else {
 431           next = t->global_link;
 432       }
 433     }
 434   }
 435 }
 436
 437 // useful from gdb
 438 void
 439 printThreadQueue(StgTSO *t)
 440 {
 441     nat i = 0;
 442     for (; t != END_TSO_QUEUE; t = t->_link) {
 443         printThreadStatus(t);
 444         i++;
 445     }
 446     debugBelch("%d threads on queue\n", i);
 447 }
 448
 449 #endif /* DEBUG */