New tracing interface
[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 "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