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