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     Task *task;
  72
  73     debugTrace(DEBUG_sched,
  74                "stopping task manager, %d tasks still running",
  75                tasksRunning);
  76
  77     ACQUIRE_LOCK(&sched_mutex);
  78     for (task = task_free_list; task != NULL; task = task->next) {
  79 #if defined(THREADED_RTS)
  80         closeCondition(&task->cond);
  81         closeMutex(&task->lock);
  82 #endif
  83     }
  84     RELEASE_LOCK(&sched_mutex);
  85 }
  86
  87
  88 void
  89 freeTaskManager (void)
  90 {
  91     Task *task, *next;
  92
  93     debugTrace(DEBUG_sched, "freeing task manager");
  94
  95     ACQUIRE_LOCK(&sched_mutex);
  96     for (task = task_free_list; task != NULL; task = next) {
  97         next = task->next;
  98         stgFree(task);
  99     }
 100     task_free_list = NULL;
 101     RELEASE_LOCK(&sched_mutex);
 102 }
 103
 104
 105 static Task*
 106 newTask (void)
 107 {
 108 #if defined(THREADED_RTS)
 109     Ticks currentElapsedTime, currentUserTime;
 110 #endif
 111     Task *task;
 112
 113     task = stgMallocBytes(sizeof(Task), "newTask");
 114
 115     task->cap  = NULL;
 116     task->stopped = rtsFalse;
 117     task->suspended_tso = NULL;
 118     task->tso  = NULL;
 119     task->stat = NoStatus;
 120     task->ret  = NULL;
 121
 122 #if defined(THREADED_RTS)
 123     initCondition(&task->cond);
 124     initMutex(&task->lock);
 125     task->wakeup = rtsFalse;
 126 #endif
 127
 128 #if defined(THREADED_RTS)
 129     currentUserTime = getThreadCPUTime();
 130     currentElapsedTime = getProcessElapsedTime();
 131     task->mut_time = 0;
 132     task->mut_etime = 0;
 133     task->gc_time = 0;
 134     task->gc_etime = 0;
 135     task->muttimestart = currentUserTime;
 136     task->elapsedtimestart = currentElapsedTime;
 137 #endif
 138
 139     task->prev = NULL;
 140     task->next = NULL;
 141     task->return_link = NULL;
 142
 143     task->all_link = all_tasks;
 144     all_tasks = task;
 145
 146     taskCount++;
 147     workerCount++;
 148
 149     return task;
 150 }
 151
 152 Task *
 153 newBoundTask (void)
 154 {
 155     Task *task;
 156
 157     ASSERT_LOCK_HELD(&sched_mutex);
 158     if (task_free_list == NULL) {
 159         task = newTask();
 160     } else {
 161         task = task_free_list;
 162         task_free_list = task->next;
 163         task->next = NULL;
 164         task->prev = NULL;
 165         task->stopped = rtsFalse;
 166     }
 167 #if defined(THREADED_RTS)
 168     task->id = osThreadId();
 169 #endif
 170     ASSERT(task->cap == NULL);
 171
 172     tasksRunning++;
 173
 174     taskEnter(task);
 175
 176     debugTrace(DEBUG_sched, "new task (taskCount: %d)", taskCount);
 177     return task;
 178 }
 179
 180 void
 181 boundTaskExiting (Task *task)
 182 {
 183     task->stopped = rtsTrue;
 184     task->cap = NULL;
 185
 186 #if defined(THREADED_RTS)
 187     ASSERT(osThreadId() == task->id);
 188 #endif
 189     ASSERT(myTask() == task);
 190     setMyTask(task->prev_stack);
 191
 192     tasksRunning--;
 193
 194     // sadly, we need a lock around the free task list. Todo: eliminate.
 195     ACQUIRE_LOCK(&sched_mutex);
 196     task->next = task_free_list;
 197     task_free_list = task;
 198     RELEASE_LOCK(&sched_mutex);
 199
 200     debugTrace(DEBUG_sched, "task exiting");
 201 }
 202
 203 #ifdef THREADED_RTS
 204 #define TASK_ID(t) (t)->id
 205 #else
 206 #define TASK_ID(t) (t)
 207 #endif
 208
 209 void
 210 discardTask (Task *task)
 211 {
 212     ASSERT_LOCK_HELD(&sched_mutex);
 213     if (!task->stopped) {
 214         debugTrace(DEBUG_sched, "discarding task %ld", (long)TASK_ID(task));
 215         task->cap = NULL;
 216         task->tso = NULL;
 217         task->stopped = rtsTrue;
 218         tasksRunning--;
 219         task->next = task_free_list;
 220         task_free_list = task;
 221     }
 222 }
 223
 224 void
 225 taskTimeStamp (Task *task USED_IF_THREADS)
 226 {
 227 #if defined(THREADED_RTS)
 228     Ticks currentElapsedTime, currentUserTime, elapsedGCTime;
 229
 230     currentUserTime = getThreadCPUTime();
 231     currentElapsedTime = getProcessElapsedTime();
 232
 233     // XXX this is wrong; we want elapsed GC time since the
 234     // Task started.
 235     elapsedGCTime = stat_getElapsedGCTime();
 236
 237     task->mut_time =
 238         currentUserTime - task->muttimestart - task->gc_time;
 239     task->mut_etime =
 240         currentElapsedTime - task->elapsedtimestart - elapsedGCTime;
 241
 242     if (task->mut_time  < 0) { task->mut_time  = 0; }
 243     if (task->mut_etime < 0) { task->mut_etime = 0; }
 244 #endif
 245 }
 246
 247 void
 248 workerTaskStop (Task *task)
 249 {
 250 #if defined(THREADED_RTS)
 251     OSThreadId id;
 252     id = osThreadId();
 253     ASSERT(task->id == id);
 254     ASSERT(myTask() == task);
 255 #endif
 256
 257     taskTimeStamp(task);
 258     task->stopped = rtsTrue;
 259     tasksRunning--;
 260 }
 261
 262 void
 263 resetTaskManagerAfterFork (void)
 264 {
 265     // TODO!
 266     taskCount = 0;
 267 }
 268
 269 #if defined(THREADED_RTS)
 270
 271 void
 272 startWorkerTask (Capability *cap,
 273                  void OSThreadProcAttr (*taskStart)(Task *task))
 274 {
 275   int r;
 276   OSThreadId tid;
 277   Task *task;
 278
 279   workerCount++;
 280
 281   // A worker always gets a fresh Task structure.
 282   task = newTask();
 283
 284   tasksRunning++;
 285
 286   // The lock here is to synchronise with taskStart(), to make sure
 287   // that we have finished setting up the Task structure before the
 288   // worker thread reads it.
 289   ACQUIRE_LOCK(&task->lock);
 290
 291   task->cap = cap;
 292
 293   // Give the capability directly to the worker; we can't let anyone
 294   // else get in, because the new worker Task has nowhere to go to
 295   // sleep so that it could be woken up again.
 296   ASSERT_LOCK_HELD(&cap->lock);
 297   cap->running_task = task;
 298
 299   r = createOSThread(&tid, (OSThreadProc *)taskStart, task);
 300   if (r != 0) {
 301     sysErrorBelch("failed to create OS thread");
 302     stg_exit(EXIT_FAILURE);
 303   }
 304
 305   debugTrace(DEBUG_sched, "new worker task (taskCount: %d)", taskCount);
 306
 307   task->id = tid;
 308
 309   // ok, finished with the Task struct.
 310   RELEASE_LOCK(&task->lock);
 311 }
 312
 313 #endif /* THREADED_RTS */
 314
 315 #ifdef DEBUG
 316
 317 static void *taskId(Task *task)
 318 {
 319 #ifdef THREADED_RTS
 320     return (void *)task->id;
 321 #else
 322     return (void *)task;
 323 #endif
 324 }
 325
 326 void printAllTasks(void);
 327
 328 void
 329 printAllTasks(void)
 330 {
 331     Task *task;
 332     for (task = all_tasks; task != NULL; task = task->all_link) {
 333         debugBelch("task %p is %s, ", taskId(task), task->stopped ? "stopped" : "alive");
 334         if (!task->stopped) {
 335             if (task->cap) {
 336                 debugBelch("on capability %d, ", task->cap->no);
 337             }
 338             if (task->tso) {
 339               debugBelch("bound to thread %lu", (unsigned long)task->tso->id);
 340             } else {
 341                 debugBelch("worker");
 342             }
 343         }
 344         debugBelch("\n");
 345     }
 346 }
 347
 348 #endif
 349