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