rts/Task.c

   1 /* -----------------------------------------------------------------------------
   2  *
   3  * (c) The GHC Team 2001-2005
   4  *
   5  * The task manager subsystem.  Tasks execute STG code, with this
   6  * module providing the API which the Scheduler uses to control their
   7  * creation and destruction.
   8  *
   9  * -------------------------------------------------------------------------*/
  10
  11 #include "PosixSource.h"
  12 #include "Rts.h"
  13
  14 #include "RtsUtils.h"
  15 #include "Task.h"
  16 #include "Capability.h"
  17 #include "Stats.h"
  18 #include "Schedule.h"
  19 #include "Hash.h"
  20 #include "Trace.h"
  21
  22 #if HAVE_SIGNAL_H
  23 #include <signal.h>
  24 #endif
  25
  26 // Task lists and global counters.
  27 // Locks required: sched_mutex.
  28 Task *all_tasks = NULL;
  29 static nat taskCount;
  30 static int tasksInitialized = 0;
  31
  32 static void   freeTask  (Task *task);
  33 static Task * allocTask (void);
  34 static Task * newTask   (rtsBool);
  35
  36 /* -----------------------------------------------------------------------------
  37  * Remembering the current thread's Task
  38  * -------------------------------------------------------------------------- */
  39
  40 // A thread-local-storage key that we can use to get access to the
  41 // current thread's Task structure.
  42 #if defined(THREADED_RTS)
  43 # if defined(MYTASK_USE_TLV)
  44 __thread Task *my_task;
  45 # else
  46 ThreadLocalKey currentTaskKey;
  47 # endif
  48 #else
  49 Task *my_task;
  50 #endif
  51
  52 /* -----------------------------------------------------------------------------
  53  * Rest of the Task API
  54  * -------------------------------------------------------------------------- */
  55
  56 void
  57 initTaskManager (void)
  58 {
  59     if (!tasksInitialized) {
  60         taskCount = 0;
  61         tasksInitialized = 1;
  62 #if defined(THREADED_RTS) && !defined(MYTASK_USE_TLV)
  63         newThreadLocalKey(&currentTaskKey);
  64 #endif
  65     }
  66 }
  67
  68 nat
  69 freeTaskManager (void)
  70 {
  71     Task *task, *next;
  72     nat tasksRunning = 0;
  73
  74     ASSERT_LOCK_HELD(&sched_mutex);
  75
  76     for (task = all_tasks; task != NULL; task = next) {
  77         next = task->all_link;
  78         if (task->stopped) {
  79             freeTask(task);
  80         } else {
  81             tasksRunning++;
  82         }
  83     }
  84
  85     debugTrace(DEBUG_sched, "freeing task manager, %d tasks still running",
  86                tasksRunning);
  87
  88     all_tasks = NULL;
  89 #if defined(THREADED_RTS) && !defined(MYTASK_USE_TLV)
  90     freeThreadLocalKey(&currentTaskKey);
  91 #endif
  92
  93     tasksInitialized = 0;
  94
  95     return tasksRunning;
  96 }
  97
  98 static Task *
  99 allocTask (void)
 100 {
 101     Task *task;
 102
 103     task = myTask();
 104     if (task != NULL) {
 105         return task;
 106     } else {
 107         task = newTask(rtsFalse);
 108 #if defined(THREADED_RTS)
 109         task->id = osThreadId();
 110 #endif
 111         setMyTask(task);
 112         return task;
 113     }
 114 }
 115
 116 static void
 117 freeTask (Task *task)
 118 {
 119     InCall *incall, *next;
 120
 121     // We only free resources if the Task is not in use.  A
 122     // Task may still be in use if we have a Haskell thread in
 123     // a foreign call while we are attempting to shut down the
 124     // RTS (see conc059).
 125 #if defined(THREADED_RTS)
 126     closeCondition(&task->cond);
 127     closeMutex(&task->lock);
 128 #endif
 129
 130     for (incall = task->incall; incall != NULL; incall = next) {
 131         next = incall->prev_stack;
 132         stgFree(incall);
 133     }
 134     for (incall = task->spare_incalls; incall != NULL; incall = next) {
 135         next = incall->next;
 136         stgFree(incall);
 137     }
 138
 139     stgFree(task);
 140 }
 141
 142 static Task*
 143 newTask (rtsBool worker)
 144 {
 145 #if defined(THREADED_RTS)
 146     Ticks currentElapsedTime, currentUserTime;
 147 #endif
 148     Task *task;
 149
 150 #define ROUND_TO_CACHE_LINE(x) ((((x)+63) / 64) * 64)
 151     task = stgMallocBytes(ROUND_TO_CACHE_LINE(sizeof(Task)), "newTask");
 152
 153     task->cap           = NULL;
 154     task->worker        = worker;
 155     task->stopped       = rtsFalse;
 156     task->stat          = NoStatus;
 157     task->ret           = NULL;
 158     task->n_spare_incalls = 0;
 159     task->spare_incalls = NULL;
 160     task->incall        = NULL;
 161
 162 #if defined(THREADED_RTS)
 163     initCondition(&task->cond);
 164     initMutex(&task->lock);
 165     task->wakeup = rtsFalse;
 166 #endif
 167
 168 #if defined(THREADED_RTS)
 169     currentUserTime = getThreadCPUTime();
 170     currentElapsedTime = getProcessElapsedTime();
 171     task->mut_time = 0;
 172     task->mut_etime = 0;
 173     task->gc_time = 0;
 174     task->gc_etime = 0;
 175     task->muttimestart = currentUserTime;
 176     task->elapsedtimestart = currentElapsedTime;
 177 #endif
 178
 179     task->next = NULL;
 180
 181     ACQUIRE_LOCK(&sched_mutex);
 182
 183     task->all_link = all_tasks;
 184     all_tasks = task;
 185
 186     taskCount++;
 187
 188     RELEASE_LOCK(&sched_mutex);
 189
 190     return task;
 191 }
 192
 193 // avoid the spare_incalls list growing unboundedly
 194 #define MAX_SPARE_INCALLS 8
 195
 196 static void
 197 newInCall (Task *task)
 198 {
 199     InCall *incall;
 200
 201     if (task->spare_incalls != NULL) {
 202         incall = task->spare_incalls;
 203         task->spare_incalls = incall->next;
 204         task->n_spare_incalls--;
 205     } else {
 206         incall = stgMallocBytes((sizeof(InCall)), "newBoundTask");
 207     }
 208
 209     incall->tso = NULL;
 210     incall->task = task;
 211     incall->suspended_tso = NULL;
 212     incall->suspended_cap = NULL;
 213     incall->next = NULL;
 214     incall->prev = NULL;
 215     incall->prev_stack = task->incall;
 216     task->incall = incall;
 217 }
 218
 219 static void
 220 endInCall (Task *task)
 221 {
 222     InCall *incall;
 223
 224     incall = task->incall;
 225     incall->tso = NULL;
 226     task->incall = task->incall->prev_stack;
 227
 228     if (task->n_spare_incalls >= MAX_SPARE_INCALLS) {
 229         stgFree(incall);
 230     } else {
 231         incall->next = task->spare_incalls;
 232         task->spare_incalls = incall;
 233         task->n_spare_incalls++;
 234     }
 235 }
 236
 237
 238 Task *
 239 newBoundTask (void)
 240 {
 241     Task *task;
 242
 243     if (!tasksInitialized) {
 244         errorBelch("newBoundTask: RTS is not initialised; call hs_init() first");
 245         stg_exit(EXIT_FAILURE);
 246     }
 247
 248     task = allocTask();
 249
 250     task->stopped = rtsFalse;
 251
 252     newInCall(task);
 253
 254     debugTrace(DEBUG_sched, "new task (taskCount: %d)", taskCount);
 255     return task;
 256 }
 257
 258 void
 259 boundTaskExiting (Task *task)
 260 {
 261     task->stopped = rtsTrue;
 262
 263 #if defined(THREADED_RTS)
 264     ASSERT(osThreadId() == task->id);
 265 #endif
 266     ASSERT(myTask() == task);
 267
 268     endInCall(task);
 269
 270     debugTrace(DEBUG_sched, "task exiting");
 271 }
 272
 273
 274 #ifdef THREADED_RTS
 275 #define TASK_ID(t) (t)->id
 276 #else
 277 #define TASK_ID(t) (t)
 278 #endif
 279
 280 void
 281 discardTasksExcept (Task *keep)
 282 {
 283     Task *task, *next;
 284
 285     // Wipe the task list, except the current Task.
 286     ACQUIRE_LOCK(&sched_mutex);
 287     for (task = all_tasks; task != NULL; task=next) {
 288         next = task->all_link;
 289         if (task != keep) {
 290             debugTrace(DEBUG_sched, "discarding task %ld", (long)TASK_ID(task));
 291             freeTask(task);
 292         }
 293     }
 294     all_tasks = keep;
 295     RELEASE_LOCK(&sched_mutex);
 296 }
 297
 298 void
 299 taskTimeStamp (Task *task USED_IF_THREADS)
 300 {
 301 #if defined(THREADED_RTS)
 302     Ticks currentElapsedTime, currentUserTime, elapsedGCTime;
 303
 304     currentUserTime = getThreadCPUTime();
 305     currentElapsedTime = getProcessElapsedTime();
 306
 307     // XXX this is wrong; we want elapsed GC time since the
 308     // Task started.
 309     elapsedGCTime = stat_getElapsedGCTime();
 310
 311     task->mut_time =
 312         currentUserTime - task->muttimestart - task->gc_time;
 313     task->mut_etime =
 314         currentElapsedTime - task->elapsedtimestart - elapsedGCTime;
 315
 316     if (task->mut_time  < 0) { task->mut_time  = 0; }
 317     if (task->mut_etime < 0) { task->mut_etime = 0; }
 318 #endif
 319 }
 320
 321 #if defined(THREADED_RTS)
 322
 323 void
 324 workerTaskStop (Task *task)
 325 {
 326     OSThreadId id;
 327     id = osThreadId();
 328     ASSERT(task->id == id);
 329     ASSERT(myTask() == task);
 330
 331     task->cap = NULL;
 332     taskTimeStamp(task);
 333     task->stopped = rtsTrue;
 334 }
 335
 336 #endif
 337
 338 #if defined(THREADED_RTS)
 339
 340 static void OSThreadProcAttr
 341 workerStart(Task *task)
 342 {
 343     Capability *cap;
 344
 345     // See startWorkerTask().
 346     ACQUIRE_LOCK(&task->lock);
 347     cap = task->cap;
 348     RELEASE_LOCK(&task->lock);
 349
 350     if (RtsFlags.ParFlags.setAffinity) {
 351         setThreadAffinity(cap->no, n_capabilities);
 352     }
 353
 354     // set the thread-local pointer to the Task:
 355     setMyTask(task);
 356
 357     newInCall(task);
 358
 359     scheduleWorker(cap,task);
 360 }
 361
 362 void
 363 startWorkerTask (Capability *cap)
 364 {
 365   int r;
 366   OSThreadId tid;
 367   Task *task;
 368
 369   // A worker always gets a fresh Task structure.
 370   task = newTask(rtsTrue);
 371
 372   // The lock here is to synchronise with taskStart(), to make sure
 373   // that we have finished setting up the Task structure before the
 374   // worker thread reads it.
 375   ACQUIRE_LOCK(&task->lock);
 376
 377   task->cap = cap;
 378
 379   // Give the capability directly to the worker; we can't let anyone
 380   // else get in, because the new worker Task has nowhere to go to
 381   // sleep so that it could be woken up again.
 382   ASSERT_LOCK_HELD(&cap->lock);
 383   cap->running_task = task;
 384
 385   r = createOSThread(&tid, (OSThreadProc*)workerStart, task);
 386   if (r != 0) {
 387     sysErrorBelch("failed to create OS thread");
 388     stg_exit(EXIT_FAILURE);
 389   }
 390
 391   debugTrace(DEBUG_sched, "new worker task (taskCount: %d)", taskCount);
 392
 393   task->id = tid;
 394
 395   // ok, finished with the Task struct.
 396   RELEASE_LOCK(&task->lock);
 397 }
 398
 399 #endif /* THREADED_RTS */
 400
 401 #ifdef DEBUG
 402
 403 static void *taskId(Task *task)
 404 {
 405 #ifdef THREADED_RTS
 406     return (void *)task->id;
 407 #else
 408     return (void *)task;
 409 #endif
 410 }
 411
 412 void printAllTasks(void);
 413
 414 void
 415 printAllTasks(void)
 416 {
 417     Task *task;
 418     for (task = all_tasks; task != NULL; task = task->all_link) {
 419         debugBelch("task %p is %s, ", taskId(task), task->stopped ? "stopped" : "alive");
 420         if (!task->stopped) {
 421             if (task->cap) {
 422                 debugBelch("on capability %d, ", task->cap->no);
 423             }
 424             if (task->incall->tso) {
 425               debugBelch("bound to thread %lu",
 426                          (unsigned long)task->incall->tso->id);
 427             } else {
 428                 debugBelch("worker");
 429             }
 430         }
 431         debugBelch("\n");
 432     }
 433 }
 434
 435 #endif
 436