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