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     keep->all_link = NULL;
 296     RELEASE_LOCK(&sched_mutex);
 297 }
 298
 299 void
 300 taskTimeStamp (Task *task USED_IF_THREADS)
 301 {
 302 #if defined(THREADED_RTS)
 303     Ticks currentElapsedTime, currentUserTime, elapsedGCTime;
 304
 305     currentUserTime = getThreadCPUTime();
 306     currentElapsedTime = getProcessElapsedTime();
 307
 308     // XXX this is wrong; we want elapsed GC time since the
 309     // Task started.
 310     elapsedGCTime = stat_getElapsedGCTime();
 311
 312     task->mut_time =
 313         currentUserTime - task->muttimestart - task->gc_time;
 314     task->mut_etime =
 315         currentElapsedTime - task->elapsedtimestart - elapsedGCTime;
 316
 317     if (task->mut_time  < 0) { task->mut_time  = 0; }
 318     if (task->mut_etime < 0) { task->mut_etime = 0; }
 319 #endif
 320 }
 321
 322 #if defined(THREADED_RTS)
 323
 324 void
 325 workerTaskStop (Task *task)
 326 {
 327     OSThreadId id;
 328     id = osThreadId();
 329     ASSERT(task->id == id);
 330     ASSERT(myTask() == task);
 331
 332     task->cap = NULL;
 333     taskTimeStamp(task);
 334     task->stopped = rtsTrue;
 335 }
 336
 337 #endif
 338
 339 #if defined(THREADED_RTS)
 340
 341 static void OSThreadProcAttr
 342 workerStart(Task *task)
 343 {
 344     Capability *cap;
 345
 346     // See startWorkerTask().
 347     ACQUIRE_LOCK(&task->lock);
 348     cap = task->cap;
 349     RELEASE_LOCK(&task->lock);
 350
 351     if (RtsFlags.ParFlags.setAffinity) {
 352         setThreadAffinity(cap->no, n_capabilities);
 353     }
 354
 355     // set the thread-local pointer to the Task:
 356     setMyTask(task);
 357
 358     newInCall(task);
 359
 360     scheduleWorker(cap,task);
 361 }
 362
 363 void
 364 startWorkerTask (Capability *cap)
 365 {
 366   int r;
 367   OSThreadId tid;
 368   Task *task;
 369
 370   // A worker always gets a fresh Task structure.
 371   task = newTask(rtsTrue);
 372
 373   // The lock here is to synchronise with taskStart(), to make sure
 374   // that we have finished setting up the Task structure before the
 375   // worker thread reads it.
 376   ACQUIRE_LOCK(&task->lock);
 377
 378   task->cap = cap;
 379
 380   // Give the capability directly to the worker; we can't let anyone
 381   // else get in, because the new worker Task has nowhere to go to
 382   // sleep so that it could be woken up again.
 383   ASSERT_LOCK_HELD(&cap->lock);
 384   cap->running_task = task;
 385
 386   r = createOSThread(&tid, (OSThreadProc*)workerStart, task);
 387   if (r != 0) {
 388     sysErrorBelch("failed to create OS thread");
 389     stg_exit(EXIT_FAILURE);
 390   }
 391
 392   debugTrace(DEBUG_sched, "new worker task (taskCount: %d)", taskCount);
 393
 394   task->id = tid;
 395
 396   // ok, finished with the Task struct.
 397   RELEASE_LOCK(&task->lock);
 398 }
 399
 400 #endif /* THREADED_RTS */
 401
 402 #ifdef DEBUG
 403
 404 static void *taskId(Task *task)
 405 {
 406 #ifdef THREADED_RTS
 407     return (void *)task->id;
 408 #else
 409     return (void *)task;
 410 #endif
 411 }
 412
 413 void printAllTasks(void);
 414
 415 void
 416 printAllTasks(void)
 417 {
 418     Task *task;
 419     for (task = all_tasks; task != NULL; task = task->all_link) {
 420         debugBelch("task %p is %s, ", taskId(task), task->stopped ? "stopped" : "alive");
 421         if (!task->stopped) {
 422             if (task->cap) {
 423                 debugBelch("on capability %d, ", task->cap->no);
 424             }
 425             if (task->incall->tso) {
 426               debugBelch("bound to thread %lu",
 427                          (unsigned long)task->incall->tso->id);
 428             } else {
 429                 debugBelch("worker");
 430             }
 431         }
 432         debugBelch("\n");
 433     }
 434 }
 435
 436 #endif
 437