Windows: use a thread-local variable for myTask()
[ghc.git] / 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 "PosixSource.h"
12 #include "Rts.h"
13
14 #include "RtsUtils.h"
15 #include "Task.h"
16 #include "Capability.h"
17 #include "Stats.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: all_tasks_mutex.
28 Task *all_tasks = NULL;
29 static nat taskCount;
30 static int tasksInitialized = 0;
31
32 static void freeTask (Task *task);
33 static Task * allocTask (void);
34 static Task * newTask (rtsBool);
35
36 #if defined(THREADED_RTS)
37 static Mutex all_tasks_mutex;
38 #endif
39
40 /* -----------------------------------------------------------------------------
41 * Remembering the current thread's Task
42 * -------------------------------------------------------------------------- */
43
44 // A thread-local-storage key that we can use to get access to the
45 // current thread's Task structure.
46 #if defined(THREADED_RTS)
47 # if defined(MYTASK_USE_TLV)
48 __thread Task *my_task;
49 # else
50 ThreadLocalKey currentTaskKey;
51 # endif
52 #else
53 Task *my_task;
54 #endif
55
56 /* -----------------------------------------------------------------------------
57 * Rest of the Task API
58 * -------------------------------------------------------------------------- */
59
60 void
61 initTaskManager (void)
62 {
63 if (!tasksInitialized) {
64 taskCount = 0;
65 tasksInitialized = 1;
66 #if defined(THREADED_RTS)
67 #if !defined(MYTASK_USE_TLV)
68 newThreadLocalKey(&currentTaskKey);
69 #endif
70 initMutex(&all_tasks_mutex);
71 #endif
72 }
73 }
74
75 nat
76 freeTaskManager (void)
77 {
78 Task *task, *next;
79 nat tasksRunning = 0;
80
81 ACQUIRE_LOCK(&all_tasks_mutex);
82
83 for (task = all_tasks; task != NULL; task = next) {
84 next = task->all_link;
85 if (task->stopped) {
86 freeTask(task);
87 } else {
88 tasksRunning++;
89 }
90 }
91
92 debugTrace(DEBUG_sched, "freeing task manager, %d tasks still running",
93 tasksRunning);
94
95 all_tasks = NULL;
96
97 RELEASE_LOCK(&all_tasks_mutex);
98
99 #if defined(THREADED_RTS) && !defined(MYTASK_USE_TLV)
100 closeMutex(&all_tasks_mutex);
101 freeThreadLocalKey(&currentTaskKey);
102 #endif
103
104 tasksInitialized = 0;
105
106 return tasksRunning;
107 }
108
109 static Task *
110 allocTask (void)
111 {
112 Task *task;
113
114 task = myTask();
115 if (task != NULL) {
116 return task;
117 } else {
118 task = newTask(rtsFalse);
119 #if defined(THREADED_RTS)
120 task->id = osThreadId();
121 #endif
122 setMyTask(task);
123 return task;
124 }
125 }
126
127 static void
128 freeTask (Task *task)
129 {
130 InCall *incall, *next;
131
132 // We only free resources if the Task is not in use. A
133 // Task may still be in use if we have a Haskell thread in
134 // a foreign call while we are attempting to shut down the
135 // RTS (see conc059).
136 #if defined(THREADED_RTS)
137 closeCondition(&task->cond);
138 closeMutex(&task->lock);
139 #endif
140
141 for (incall = task->incall; incall != NULL; incall = next) {
142 next = incall->prev_stack;
143 stgFree(incall);
144 }
145 for (incall = task->spare_incalls; incall != NULL; incall = next) {
146 next = incall->next;
147 stgFree(incall);
148 }
149
150 stgFree(task);
151 }
152
153 static Task*
154 newTask (rtsBool worker)
155 {
156 #if defined(THREADED_RTS)
157 Ticks currentElapsedTime, currentUserTime;
158 #endif
159 Task *task;
160
161 #define ROUND_TO_CACHE_LINE(x) ((((x)+63) / 64) * 64)
162 task = stgMallocBytes(ROUND_TO_CACHE_LINE(sizeof(Task)), "newTask");
163
164 task->cap = NULL;
165 task->worker = worker;
166 task->stopped = rtsFalse;
167 task->running_finalizers = rtsFalse;
168 task->n_spare_incalls = 0;
169 task->spare_incalls = NULL;
170 task->incall = NULL;
171
172 #if defined(THREADED_RTS)
173 initCondition(&task->cond);
174 initMutex(&task->lock);
175 task->wakeup = rtsFalse;
176 #endif
177
178 #if defined(THREADED_RTS)
179 currentUserTime = getThreadCPUTime();
180 currentElapsedTime = getProcessElapsedTime();
181 task->mut_time = 0;
182 task->mut_etime = 0;
183 task->gc_time = 0;
184 task->gc_etime = 0;
185 task->muttimestart = currentUserTime;
186 task->elapsedtimestart = currentElapsedTime;
187 #endif
188
189 task->next = NULL;
190
191 ACQUIRE_LOCK(&all_tasks_mutex);
192
193 task->all_link = all_tasks;
194 all_tasks = task;
195
196 taskCount++;
197
198 RELEASE_LOCK(&all_tasks_mutex);
199
200 return task;
201 }
202
203 // avoid the spare_incalls list growing unboundedly
204 #define MAX_SPARE_INCALLS 8
205
206 static void
207 newInCall (Task *task)
208 {
209 InCall *incall;
210
211 if (task->spare_incalls != NULL) {
212 incall = task->spare_incalls;
213 task->spare_incalls = incall->next;
214 task->n_spare_incalls--;
215 } else {
216 incall = stgMallocBytes((sizeof(InCall)), "newBoundTask");
217 }
218
219 incall->tso = NULL;
220 incall->task = task;
221 incall->suspended_tso = NULL;
222 incall->suspended_cap = NULL;
223 incall->stat = NoStatus;
224 incall->ret = NULL;
225 incall->next = NULL;
226 incall->prev = NULL;
227 incall->prev_stack = task->incall;
228 task->incall = incall;
229 }
230
231 static void
232 endInCall (Task *task)
233 {
234 InCall *incall;
235
236 incall = task->incall;
237 incall->tso = NULL;
238 task->incall = task->incall->prev_stack;
239
240 if (task->n_spare_incalls >= MAX_SPARE_INCALLS) {
241 stgFree(incall);
242 } else {
243 incall->next = task->spare_incalls;
244 task->spare_incalls = incall;
245 task->n_spare_incalls++;
246 }
247 }
248
249
250 Task *
251 newBoundTask (void)
252 {
253 Task *task;
254
255 if (!tasksInitialized) {
256 errorBelch("newBoundTask: RTS is not initialised; call hs_init() first");
257 stg_exit(EXIT_FAILURE);
258 }
259
260 task = allocTask();
261
262 task->stopped = rtsFalse;
263
264 newInCall(task);
265
266 debugTrace(DEBUG_sched, "new task (taskCount: %d)", taskCount);
267 return task;
268 }
269
270 void
271 boundTaskExiting (Task *task)
272 {
273 task->stopped = rtsTrue;
274
275 #if defined(THREADED_RTS)
276 ASSERT(osThreadId() == task->id);
277 #endif
278 ASSERT(myTask() == task);
279
280 endInCall(task);
281
282 debugTrace(DEBUG_sched, "task exiting");
283 }
284
285
286 #ifdef THREADED_RTS
287 #define TASK_ID(t) (t)->id
288 #else
289 #define TASK_ID(t) (t)
290 #endif
291
292 void
293 discardTasksExcept (Task *keep)
294 {
295 Task *task, *next;
296
297 // Wipe the task list, except the current Task.
298 ACQUIRE_LOCK(&all_tasks_mutex);
299 for (task = all_tasks; task != NULL; task=next) {
300 next = task->all_link;
301 if (task != keep) {
302 debugTrace(DEBUG_sched, "discarding task %ld", (long)TASK_ID(task));
303 freeTask(task);
304 }
305 }
306 all_tasks = keep;
307 keep->all_link = NULL;
308 RELEASE_LOCK(&all_tasks_mutex);
309 }
310
311 void
312 taskTimeStamp (Task *task USED_IF_THREADS)
313 {
314 #if defined(THREADED_RTS)
315 Ticks currentElapsedTime, currentUserTime, elapsedGCTime;
316
317 currentUserTime = getThreadCPUTime();
318 currentElapsedTime = getProcessElapsedTime();
319
320 // XXX this is wrong; we want elapsed GC time since the
321 // Task started.
322 elapsedGCTime = stat_getElapsedGCTime();
323
324 task->mut_time =
325 currentUserTime - task->muttimestart - task->gc_time;
326 task->mut_etime =
327 currentElapsedTime - task->elapsedtimestart - elapsedGCTime;
328
329 if (task->mut_time < 0) { task->mut_time = 0; }
330 if (task->mut_etime < 0) { task->mut_etime = 0; }
331 #endif
332 }
333
334 #if defined(THREADED_RTS)
335
336 void
337 workerTaskStop (Task *task)
338 {
339 OSThreadId id;
340 id = osThreadId();
341 ASSERT(task->id == id);
342 ASSERT(myTask() == task);
343
344 task->cap = NULL;
345 taskTimeStamp(task);
346 task->stopped = rtsTrue;
347 }
348
349 #endif
350
351 #if defined(THREADED_RTS)
352
353 static void OSThreadProcAttr
354 workerStart(Task *task)
355 {
356 Capability *cap;
357
358 // See startWorkerTask().
359 ACQUIRE_LOCK(&task->lock);
360 cap = task->cap;
361 RELEASE_LOCK(&task->lock);
362
363 if (RtsFlags.ParFlags.setAffinity) {
364 setThreadAffinity(cap->no, n_capabilities);
365 }
366
367 // set the thread-local pointer to the Task:
368 setMyTask(task);
369
370 newInCall(task);
371
372 scheduleWorker(cap,task);
373 }
374
375 void
376 startWorkerTask (Capability *cap)
377 {
378 int r;
379 OSThreadId tid;
380 Task *task;
381
382 // A worker always gets a fresh Task structure.
383 task = newTask(rtsTrue);
384
385 // The lock here is to synchronise with taskStart(), to make sure
386 // that we have finished setting up the Task structure before the
387 // worker thread reads it.
388 ACQUIRE_LOCK(&task->lock);
389
390 task->cap = cap;
391
392 // Give the capability directly to the worker; we can't let anyone
393 // else get in, because the new worker Task has nowhere to go to
394 // sleep so that it could be woken up again.
395 ASSERT_LOCK_HELD(&cap->lock);
396 cap->running_task = task;
397
398 r = createOSThread(&tid, (OSThreadProc*)workerStart, task);
399 if (r != 0) {
400 sysErrorBelch("failed to create OS thread");
401 stg_exit(EXIT_FAILURE);
402 }
403
404 debugTrace(DEBUG_sched, "new worker task (taskCount: %d)", taskCount);
405
406 task->id = tid;
407
408 // ok, finished with the Task struct.
409 RELEASE_LOCK(&task->lock);
410 }
411
412 #endif /* THREADED_RTS */
413
414 #ifdef DEBUG
415
416 static void *taskId(Task *task)
417 {
418 #ifdef THREADED_RTS
419 return (void *)task->id;
420 #else
421 return (void *)task;
422 #endif
423 }
424
425 void printAllTasks(void);
426
427 void
428 printAllTasks(void)
429 {
430 Task *task;
431 for (task = all_tasks; task != NULL; task = task->all_link) {
432 debugBelch("task %p is %s, ", taskId(task), task->stopped ? "stopped" : "alive");
433 if (!task->stopped) {
434 if (task->cap) {
435 debugBelch("on capability %d, ", task->cap->no);
436 }
437 if (task->incall->tso) {
438 debugBelch("bound to thread %lu",
439 (unsigned long)task->incall->tso->id);
440 } else {
441 debugBelch("worker");
442 }
443 }
444 debugBelch("\n");
445 }
446 }
447
448 #endif
449