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 "Rts.h"
  12 #include "RtsUtils.h"
  13 #include "OSThreads.h"
  14 #include "Task.h"
  15 #include "Capability.h"
  16 #include "Stats.h"
  17 #include "RtsFlags.h"
  18 #include "Storage.h"
  19 #include "Schedule.h"
  20 #include "Hash.h"
  21 #include "Trace.h"
  22
  23 #if HAVE_SIGNAL_H
  24 #include <signal.h>
  25 #endif
  26
  27 // Task lists and global counters.
  28 // Locks required: sched_mutex.
  29 Task *all_tasks = NULL;
  30 static Task *task_free_list = NULL; // singly-linked
  31 static nat taskCount;
  32 static nat tasksRunning;
  33 static nat workerCount;
  34
  35 /* -----------------------------------------------------------------------------
  36  * Remembering the current thread's Task
  37  * -------------------------------------------------------------------------- */
  38
  39 // A thread-local-storage key that we can use to get access to the
  40 // current thread's Task structure.
  41 #if defined(THREADED_RTS)
  42 ThreadLocalKey currentTaskKey;
  43 #else
  44 Task *my_task;
  45 #endif
  46
  47 /* -----------------------------------------------------------------------------
  48  * Rest of the Task API
  49  * -------------------------------------------------------------------------- */
  50
  51 void
  52 initTaskManager (void)
  53 {
  54     static int initialized = 0;
  55
  56     if (!initialized) {
  57         taskCount = 0;
  58         workerCount = 0;
  59         tasksRunning = 0;
  60         initialized = 1;
  61 #if defined(THREADED_RTS)
  62         newThreadLocalKey(&currentTaskKey);
  63 #endif
  64     }
  65 }
  66
  67
  68 void
  69 stopTaskManager (void)
  70 {
  71     debugTrace(DEBUG_sched,
  72                "stopping task manager, %d tasks still running",
  73                tasksRunning);
  74     /* nothing to do */
  75 }
  76
  77
  78 void
  79 freeTaskManager (void)
  80 {
  81     Task *task, *next;
  82
  83     debugTrace(DEBUG_sched, "freeing task manager");
  84
  85     ACQUIRE_LOCK(&sched_mutex);
  86     for (task = all_tasks; task != NULL; task = next) {
  87         next = task->all_link;
  88         if (task->stopped) {
  89             // We only free resources if the Task is not in use.  A
  90             // Task may still be in use if we have a Haskell thread in
  91             // a foreign call while we are attempting to shut down the
  92             // RTS (see conc059).
  93 #if defined(THREADED_RTS)
  94             closeCondition(&task->cond);
  95             closeMutex(&task->lock);
  96 #endif
  97             stgFree(task);
  98         }
  99     }
 100     all_tasks = NULL;
 101     task_free_list = NULL;
 102 #if defined(THREADED_RTS)
 103     freeThreadLocalKey(&currentTaskKey);
 104 #endif
 105     RELEASE_LOCK(&sched_mutex);
 106 }
 107
 108
 109 static Task*
 110 newTask (void)
 111 {
 112 #if defined(THREADED_RTS)
 113     Ticks currentElapsedTime, currentUserTime;
 114 #endif
 115     Task *task;
 116
 117 #define ROUND_TO_CACHE_LINE(x) ((((x)+63) / 64) * 64)
 118     task = stgMallocBytes(ROUND_TO_CACHE_LINE(sizeof(Task)), "newTask");
 119
 120     task->cap  = NULL;
 121     task->stopped = rtsFalse;
 122     task->suspended_tso = NULL;
 123     task->tso  = NULL;
 124     task->stat = NoStatus;
 125     task->ret  = NULL;
 126
 127 #if defined(THREADED_RTS)
 128     initCondition(&task->cond);
 129     initMutex(&task->lock);
 130     task->wakeup = rtsFalse;
 131 #endif
 132
 133 #if defined(THREADED_RTS)
 134     currentUserTime = getThreadCPUTime();
 135     currentElapsedTime = getProcessElapsedTime();
 136     task->mut_time = 0;
 137     task->mut_etime = 0;
 138     task->gc_time = 0;
 139     task->gc_etime = 0;
 140     task->muttimestart = currentUserTime;
 141     task->elapsedtimestart = currentElapsedTime;
 142 #endif
 143
 144     task->prev = NULL;
 145     task->next = NULL;
 146     task->return_link = NULL;
 147
 148     task->all_link = all_tasks;
 149     all_tasks = task;
 150
 151     taskCount++;
 152     workerCount++;
 153
 154     return task;
 155 }
 156
 157 Task *
 158 newBoundTask (void)
 159 {
 160     Task *task;
 161
 162     ASSERT_LOCK_HELD(&sched_mutex);
 163     if (task_free_list == NULL) {
 164         task = newTask();
 165     } else {
 166         task = task_free_list;
 167         task_free_list = task->next;
 168         task->next = NULL;
 169         task->prev = NULL;
 170         task->stopped = rtsFalse;
 171     }
 172 #if defined(THREADED_RTS)
 173     task->id = osThreadId();
 174 #endif
 175     ASSERT(task->cap == NULL);
 176
 177     tasksRunning++;
 178
 179     taskEnter(task);
 180
 181     debugTrace(DEBUG_sched, "new task (taskCount: %d)", taskCount);
 182     return task;
 183 }
 184
 185 void
 186 boundTaskExiting (Task *task)
 187 {
 188     task->stopped = rtsTrue;
 189     task->cap = NULL;
 190
 191 #if defined(THREADED_RTS)
 192     ASSERT(osThreadId() == task->id);
 193 #endif
 194     ASSERT(myTask() == task);
 195     setMyTask(task->prev_stack);
 196
 197     tasksRunning--;
 198
 199     // sadly, we need a lock around the free task list. Todo: eliminate.
 200     ACQUIRE_LOCK(&sched_mutex);
 201     task->next = task_free_list;
 202     task_free_list = task;
 203     RELEASE_LOCK(&sched_mutex);
 204
 205     debugTrace(DEBUG_sched, "task exiting");
 206 }
 207
 208 #ifdef THREADED_RTS
 209 #define TASK_ID(t) (t)->id
 210 #else
 211 #define TASK_ID(t) (t)
 212 #endif
 213
 214 void
 215 discardTask (Task *task)
 216 {
 217     ASSERT_LOCK_HELD(&sched_mutex);
 218     if (!task->stopped) {
 219         debugTrace(DEBUG_sched, "discarding task %ld", (long)TASK_ID(task));
 220         task->cap = NULL;
 221         task->tso = NULL;
 222         task->stopped = rtsTrue;
 223         tasksRunning--;
 224         task->next = task_free_list;
 225         task_free_list = task;
 226     }
 227 }
 228
 229 void
 230 taskTimeStamp (Task *task USED_IF_THREADS)
 231 {
 232 #if defined(THREADED_RTS)
 233     Ticks currentElapsedTime, currentUserTime, elapsedGCTime;
 234
 235     currentUserTime = getThreadCPUTime();
 236     currentElapsedTime = getProcessElapsedTime();
 237
 238     // XXX this is wrong; we want elapsed GC time since the
 239     // Task started.
 240     elapsedGCTime = stat_getElapsedGCTime();
 241
 242     task->mut_time =
 243         currentUserTime - task->muttimestart - task->gc_time;
 244     task->mut_etime =
 245         currentElapsedTime - task->elapsedtimestart - elapsedGCTime;
 246
 247     if (task->mut_time  < 0) { task->mut_time  = 0; }
 248     if (task->mut_etime < 0) { task->mut_etime = 0; }
 249 #endif
 250 }
 251
 252 void
 253 workerTaskStop (Task *task)
 254 {
 255 #if defined(THREADED_RTS)
 256     OSThreadId id;
 257     id = osThreadId();
 258     ASSERT(task->id == id);
 259     ASSERT(myTask() == task);
 260 #endif
 261
 262     task->cap = NULL;
 263     taskTimeStamp(task);
 264     task->stopped = rtsTrue;
 265     tasksRunning--;
 266
 267     ACQUIRE_LOCK(&sched_mutex);
 268     task->next = task_free_list;
 269     task_free_list = task;
 270     RELEASE_LOCK(&sched_mutex);
 271 }
 272
 273 void
 274 resetTaskManagerAfterFork (void)
 275 {
 276     // TODO!
 277     taskCount = 0;
 278 }
 279
 280 #if defined(THREADED_RTS)
 281
 282 void
 283 startWorkerTask (Capability *cap,
 284                  void OSThreadProcAttr (*taskStart)(Task *task))
 285 {
 286   int r;
 287   OSThreadId tid;
 288   Task *task;
 289
 290   workerCount++;
 291
 292   // A worker always gets a fresh Task structure.
 293   task = newTask();
 294
 295   tasksRunning++;
 296
 297   // The lock here is to synchronise with taskStart(), to make sure
 298   // that we have finished setting up the Task structure before the
 299   // worker thread reads it.
 300   ACQUIRE_LOCK(&task->lock);
 301
 302   task->cap = cap;
 303
 304   // Give the capability directly to the worker; we can't let anyone
 305   // else get in, because the new worker Task has nowhere to go to
 306   // sleep so that it could be woken up again.
 307   ASSERT_LOCK_HELD(&cap->lock);
 308   cap->running_task = task;
 309
 310   r = createOSThread(&tid, (OSThreadProc *)taskStart, task);
 311   if (r != 0) {
 312     sysErrorBelch("failed to create OS thread");
 313     stg_exit(EXIT_FAILURE);
 314   }
 315
 316   debugTrace(DEBUG_sched, "new worker task (taskCount: %d)", taskCount);
 317
 318   task->id = tid;
 319
 320   // ok, finished with the Task struct.
 321   RELEASE_LOCK(&task->lock);
 322 }
 323
 324 #endif /* THREADED_RTS */
 325
 326 #ifdef DEBUG
 327
 328 static void *taskId(Task *task)
 329 {
 330 #ifdef THREADED_RTS
 331     return (void *)task->id;
 332 #else
 333     return (void *)task;
 334 #endif
 335 }
 336
 337 void printAllTasks(void);
 338
 339 void
 340 printAllTasks(void)
 341 {
 342     Task *task;
 343     for (task = all_tasks; task != NULL; task = task->all_link) {
 344         debugBelch("task %p is %s, ", taskId(task), task->stopped ? "stopped" : "alive");
 345         if (!task->stopped) {
 346             if (task->cap) {
 347                 debugBelch("on capability %d, ", task->cap->no);
 348             }
 349             if (task->tso) {
 350               debugBelch("bound to thread %lu", (unsigned long)task->tso->id);
 351             } else {
 352                 debugBelch("worker");
 353             }
 354         }
 355         debugBelch("\n");
 356     }
 357 }
 358
 359 #endif
 360