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         prodWorker();
  82         yieldThread();
  83     }
  84     IF_DEBUG(scheduler, sched_belch("%d tasks still running, exiting anyway", tasksRunning));
  85
  86     /*
  87        OLD CODE follows:
  88     */
  89 #if old_code
  90     /* Send 'em all a SIGHUP.  That should shut 'em up. */
  91     awaitDeath = taskCount==0 ? 0 : taskCount-1;
  92     for (i = 0; i < taskCount; i++) {
  93         /* don't cancel the thread running this piece of code. */
  94         if ( taskTable[i].id != tid ) {
  95             pthread_kill(taskTable[i].id,SIGTERM);
  96         }
  97     }
  98     while (awaitDeath > 0) {
  99         sched_yield();
 100     }
 101 #endif // old_code
 102 }
 103
 104
 105 rtsBool
 106 startTasks (nat num, void (*taskStart)(void))
 107 {
 108     nat i;
 109     for (i = 0; i < num; i++) {
 110         if (!startTask(taskStart)) {
 111             return rtsFalse;
 112         }
 113     }
 114     return rtsTrue;
 115 }
 116
 117 static TaskInfo*
 118 newTask (OSThreadId id, rtsBool is_worker)
 119 {
 120     long currentElapsedTime, currentUserTime, elapsedGCTime;
 121     TaskInfo *task_info;
 122
 123     if (taskCount >= taskTableSize) {
 124         expandTaskTable();
 125     }
 126
 127     insertHashTable( taskHash, id, &(taskTable[taskCount]) );
 128
 129     stat_getTimes(&currentElapsedTime, &currentUserTime, &elapsedGCTime);
 130
 131     task_info = &taskTable[taskCount];
 132
 133     task_info->id = id;
 134     task_info->is_worker = is_worker;
 135     task_info->stopped = rtsFalse;
 136     task_info->mut_time = 0.0;
 137     task_info->mut_etime = 0.0;
 138     task_info->gc_time = 0.0;
 139     task_info->gc_etime = 0.0;
 140     task_info->muttimestart = currentUserTime;
 141     task_info->elapsedtimestart = currentElapsedTime;
 142
 143     taskCount++;
 144     workerCount++;
 145     tasksRunning++;
 146
 147     IF_DEBUG(scheduler,sched_belch("startTask: new task %ld (total_count: %d; waiting: %d)\n", id, taskCount, rts_n_waiting_tasks););
 148
 149     return task_info;
 150 }
 151
 152 rtsBool
 153 startTask (void (*taskStart)(void))
 154 {
 155   int r;
 156   OSThreadId tid;
 157
 158   r = createOSThread(&tid,taskStart);
 159   if (r != 0) {
 160     barf("startTask: Can't create new task");
 161   }
 162   newTask (tid, rtsTrue);
 163   return rtsTrue;
 164 }
 165
 166 TaskInfo *
 167 threadIsTask (OSThreadId id)
 168 {
 169     TaskInfo *task_info;
 170
 171     task_info = lookupHashTable(taskHash, id);
 172     if (task_info != NULL) {
 173         if (task_info->stopped) {
 174             task_info->stopped = rtsFalse;
 175         }
 176         return task_info;
 177     }
 178
 179     return newTask(id, rtsFalse);
 180 }
 181
 182 TaskInfo *
 183 taskOfId (OSThreadId id)
 184 {
 185     return lookupHashTable(taskHash, id);
 186 }
 187
 188 void
 189 taskStop (void)
 190 {
 191     OSThreadId id;
 192     long currentElapsedTime, currentUserTime, elapsedGCTime;
 193     TaskInfo *task_info;
 194
 195     id = osThreadId();
 196     task_info = taskOfId(id);
 197     if (task_info == NULL) {
 198         debugBelch("taskStop: not a task");
 199         return;
 200     }
 201     ASSERT(task_info->id == id);
 202
 203     task_info->stopped = rtsTrue;
 204     tasksRunning--;
 205
 206     stat_getTimes(&currentElapsedTime, &currentUserTime, &elapsedGCTime);
 207
 208     task_info->mut_time =
 209         currentUserTime - task_info->muttimestart - task_info->gc_time;
 210     task_info->mut_etime =
 211         currentElapsedTime - task_info->elapsedtimestart - elapsedGCTime;
 212
 213     if (task_info->mut_time < 0.0)  { task_info->mut_time = 0.0;  }
 214     if (task_info->mut_etime < 0.0) { task_info->mut_etime = 0.0; }
 215 }
 216
 217 void
 218 resetTaskManagerAfterFork (void)
 219 {
 220     rts_n_waiting_tasks = 0;
 221     taskCount = 0;
 222 }
 223
 224 rtsBool
 225 maybeStartNewWorker (void (*taskStart)(void))
 226 {
 227     /*
 228      * If more than one worker thread is known to be blocked waiting
 229      * on thread_ready_cond, don't create a new one.
 230      */
 231     if ( rts_n_waiting_tasks > 0) {
 232         IF_DEBUG(scheduler,sched_belch(
 233                      "startTask: %d tasks waiting, not creating new one",
 234                      rts_n_waiting_tasks););
 235         // the task will run as soon as a capability is available,
 236         // so there's no need to wake it.
 237         return rtsFalse;
 238     }
 239
 240     /* If the task limit has been reached, just return. */
 241     if (maxWorkers > 0 && workerCount >= maxWorkers) {
 242         IF_DEBUG(scheduler,sched_belch("startTask: worker limit (%d) reached, not creating new one",maxWorkers));
 243         return rtsFalse;
 244     }
 245
 246     return startTask(taskStart);
 247 }
 248
 249 #endif /* RTS_SUPPORTS_THREADS */