ghc/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 #if defined(RTS_SUPPORTS_THREADS) /* to the end */
  13 #include "RtsUtils.h"
  14 #include "OSThreads.h"
  15 #include "Task.h"
  16 #include "Stats.h"
  17 #include "RtsFlags.h"
  18 #include "Schedule.h"
  19 #include "Hash.h"
  20 #include "Capability.h"
  21
  22 #if HAVE_SIGNAL_H
  23 #include <signal.h>
  24 #endif
  25
  26 #define INIT_TASK_TABLE_SIZE 16
  27
  28 TaskInfo* taskTable;
  29 static nat taskTableSize;
  30
  31 HashTable *taskHash; // maps OSThreadID to TaskInfo*
  32
  33 nat taskCount;
  34 static nat tasksRunning;
  35 static nat workerCount;
  36
  37 #define DEFAULT_MAX_WORKERS 64
  38 nat maxWorkers; // we won't create more workers than this
  39
  40 void
  41 initTaskManager (void)
  42 {
  43     static int initialized = 0;
  44
  45     if (!initialized) {
  46         taskTableSize = INIT_TASK_TABLE_SIZE;
  47         taskTable = stgMallocBytes( taskTableSize * sizeof(TaskInfo),
  48                                     "initTaskManager");
  49
  50         taskCount = 0;
  51         workerCount = 0;
  52         tasksRunning = 0;
  53
  54         taskHash = allocHashTable();
  55
  56         maxWorkers = DEFAULT_MAX_WORKERS;
  57
  58         initialized = 1;
  59     }
  60 }
  61
  62 static void
  63 expandTaskTable (void)
  64 {
  65     taskTableSize *= 2;
  66     taskTable = stgReallocBytes(taskTable, taskTableSize * sizeof(TaskInfo),
  67                                 "expandTaskTable");
  68 }
  69
  70 void
  71 stopTaskManager (void)
  72 {
  73     nat i;
  74
  75     IF_DEBUG(scheduler, sched_belch("stopping task manager, %d tasks still running", tasksRunning));
  76     for (i = 1000; i > 0; i--) {
  77         if (tasksRunning == 0) {
  78             IF_DEBUG(scheduler, sched_belch("all tasks stopped"));
  79             return;
  80         }
  81         IF_DEBUG(scheduler, sched_belch("yielding"));
  82         prodWorker();
  83         yieldThread();
  84     }
  85     errorBelch("%d tasks still running, exiting anyway", tasksRunning);
  86
  87     /*
  88        OLD CODE follows:
  89     */
  90 #if old_code
  91     /* Send 'em all a SIGHUP.  That should shut 'em up. */
  92     awaitDeath = taskCount==0 ? 0 : taskCount-1;
  93     for (i = 0; i < taskCount; i++) {
  94         /* don't cancel the thread running this piece of code. */
  95         if ( taskTable[i].id != tid ) {
  96             pthread_kill(taskTable[i].id,SIGTERM);
  97         }
  98     }
  99     while (awaitDeath > 0) {
 100         sched_yield();
 101     }
 102 #endif // old_code
 103 }
 104
 105
 106 rtsBool
 107 startTasks (nat num, void (*taskStart)(void))
 108 {
 109     nat i;
 110     for (i = 0; i < num; i++) {
 111         if (!startTask(taskStart)) {
 112             return rtsFalse;
 113         }
 114     }
 115     return rtsTrue;
 116 }
 117
 118 static TaskInfo*
 119 newTask (OSThreadId id, rtsBool is_worker)
 120 {
 121     long currentElapsedTime, currentUserTime, elapsedGCTime;
 122     TaskInfo *task_info;
 123
 124     if (taskCount >= taskTableSize) {
 125         expandTaskTable();
 126     }
 127
 128     insertHashTable( taskHash, id, &(taskTable[taskCount]) );
 129
 130     stat_getTimes(&currentElapsedTime, &currentUserTime, &elapsedGCTime);
 131
 132     task_info = &taskTable[taskCount];
 133
 134     task_info->id = id;
 135     task_info->is_worker = is_worker;
 136     task_info->stopped = rtsFalse;
 137     task_info->mut_time = 0.0;
 138     task_info->mut_etime = 0.0;
 139     task_info->gc_time = 0.0;
 140     task_info->gc_etime = 0.0;
 141     task_info->muttimestart = currentUserTime;
 142     task_info->elapsedtimestart = currentElapsedTime;
 143
 144     taskCount++;
 145     workerCount++;
 146     tasksRunning++;
 147
 148     IF_DEBUG(scheduler,sched_belch("startTask: new task %ld (total_count: %d; waiting: %d)\n", id, taskCount, rts_n_waiting_tasks););
 149
 150     return task_info;
 151 }
 152
 153 rtsBool
 154 startTask (void (*taskStart)(void))
 155 {
 156   int r;
 157   OSThreadId tid;
 158
 159   r = createOSThread(&tid,taskStart);
 160   if (r != 0) {
 161     barf("startTask: Can't create new task");
 162   }
 163   newTask (tid, rtsTrue);
 164   return rtsTrue;
 165 }
 166
 167 TaskInfo *
 168 threadIsTask (OSThreadId id)
 169 {
 170     TaskInfo *task_info;
 171
 172     task_info = lookupHashTable(taskHash, id);
 173     if (task_info != NULL) {
 174         if (task_info->stopped) {
 175             task_info->stopped = rtsFalse;
 176         }
 177         return task_info;
 178     }
 179
 180     return newTask(id, rtsFalse);
 181 }
 182
 183 TaskInfo *
 184 taskOfId (OSThreadId id)
 185 {
 186     return lookupHashTable(taskHash, id);
 187 }
 188
 189 void
 190 taskStop (void)
 191 {
 192     OSThreadId id;
 193     long currentElapsedTime, currentUserTime, elapsedGCTime;
 194     TaskInfo *task_info;
 195
 196     id = osThreadId();
 197     task_info = taskOfId(id);
 198     if (task_info == NULL) {
 199         debugBelch("taskStop: not a task");
 200         return;
 201     }
 202     ASSERT(task_info->id == id);
 203
 204     stat_getTimes(&currentElapsedTime, &currentUserTime, &elapsedGCTime);
 205
 206     task_info->mut_time =
 207         currentUserTime - task_info->muttimestart - task_info->gc_time;
 208     task_info->mut_etime =
 209         currentElapsedTime - task_info->elapsedtimestart - elapsedGCTime;
 210
 211     if (task_info->mut_time < 0.0)  { task_info->mut_time = 0.0;  }
 212     if (task_info->mut_etime < 0.0) { task_info->mut_etime = 0.0; }
 213
 214     task_info->stopped = rtsTrue;
 215     tasksRunning--;
 216 }
 217
 218 void
 219 resetTaskManagerAfterFork (void)
 220 {
 221     rts_n_waiting_tasks = 0;
 222     taskCount = 0;
 223 }
 224
 225 rtsBool
 226 maybeStartNewWorker (void (*taskStart)(void))
 227 {
 228     /*
 229      * If more than one worker thread is known to be blocked waiting
 230      * on thread_ready_cond, don't create a new one.
 231      */
 232     if ( rts_n_waiting_tasks > 0) {
 233         IF_DEBUG(scheduler,sched_belch(
 234                      "startTask: %d tasks waiting, not creating new one",
 235                      rts_n_waiting_tasks););
 236         // the task will run as soon as a capability is available,
 237         // so there's no need to wake it.
 238         return rtsFalse;
 239     }
 240
 241     /* If the task limit has been reached, just return. */
 242     if (maxWorkers > 0 && workerCount >= maxWorkers) {
 243         IF_DEBUG(scheduler,sched_belch("startTask: worker limit (%d) reached, not creating new one",maxWorkers));
 244         return rtsFalse;
 245     }
 246
 247     return startTask(taskStart);
 248 }
 249
 250 #endif /* RTS_SUPPORTS_THREADS */