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