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