StgTSOBlockInfo block_info;
StgThreadID id;
int saved_errno;
- struct Task_* bound;
+ struct InCall_* bound;
struct Capability_* cap;
struct StgTRecHeader_ * trec; /* STM transaction record */
newReturningTask (Capability *cap, Task *task)
{
ASSERT_LOCK_HELD(&cap->lock);
- ASSERT(task->return_link == NULL);
+ ASSERT(task->next == NULL);
if (cap->returning_tasks_hd) {
- ASSERT(cap->returning_tasks_tl->return_link == NULL);
- cap->returning_tasks_tl->return_link = task;
+ ASSERT(cap->returning_tasks_tl->next == NULL);
+ cap->returning_tasks_tl->next = task;
} else {
cap->returning_tasks_hd = task;
}
Task *task;
task = cap->returning_tasks_hd;
ASSERT(task);
- cap->returning_tasks_hd = task->return_link;
+ cap->returning_tasks_hd = task->next;
if (!cap->returning_tasks_hd) {
cap->returning_tasks_tl = NULL;
}
- task->return_link = NULL;
+ task->next = NULL;
return task;
}
#endif
initMutex(&cap->lock);
cap->running_task = NULL; // indicates cap is free
cap->spare_workers = NULL;
- cap->suspended_ccalling_tasks = NULL;
+ cap->suspended_ccalls = NULL;
cap->returning_tasks_hd = NULL;
cap->returning_tasks_tl = NULL;
cap->wakeup_queue_hd = END_TSO_QUEUE;
ASSERT_LOCK_HELD(&cap->lock);
ASSERT(task->cap == cap);
debugTrace(DEBUG_sched, "passing capability %d to %s %p",
- cap->no, task->tso ? "bound task" : "worker",
+ cap->no, task->incall->tso ? "bound task" : "worker",
(void *)task->id);
ACQUIRE_LOCK(&task->lock);
task->wakeup = rtsTrue;
// assertion is false: in schedule() we force a yield after
// ThreadBlocked, but the thread may be back on the run queue
// by now.
- task = cap->run_queue_hd->bound;
+ task = cap->run_queue_hd->bound->task;
giveCapabilityToTask(cap,task);
return;
}
if (sched_state < SCHED_SHUTTING_DOWN || !emptyRunQueue(cap)) {
debugTrace(DEBUG_sched,
"starting new worker on capability %d", cap->no);
- startWorkerTask(cap, workerStart);
+ startWorkerTask(cap);
return;
}
}
// in which case it is not replaced on the spare_worker queue.
// This happens when the system is shutting down (see
// Schedule.c:workerStart()).
- // Also, be careful to check that this task hasn't just exited
- // Haskell to do a foreign call (task->suspended_tso).
- if (!isBoundTask(task) && !task->stopped && !task->suspended_tso) {
+ if (!isBoundTask(task) && !task->stopped) {
task->next = cap->spare_workers;
cap->spare_workers = task;
}
continue;
}
- if (task->tso == NULL) {
+ if (task->incall->tso == NULL) {
ASSERT(cap->spare_workers != NULL);
// if we're not at the front of the queue, release it
// again. This is unlikely to happen.
// ASSUMES: cap->lock is held (asserted in wakeupThreadOnCapability)
if (tso->bound) {
- ASSERT(tso->bound->cap == tso->cap);
- tso->bound->cap = other_cap;
+ ASSERT(tso->bound->task->cap == tso->cap);
+ tso->bound->task->cap = other_cap;
}
tso->cap = other_cap;
- ASSERT(tso->bound ? tso->bound->cap == other_cap : 1);
+ ASSERT(tso->bound ? tso->bound->task->cap == other_cap : 1);
if (other_cap->running_task == NULL) {
// nobody is running this Capability, we can add our thread
// that will try to return to code that has been unloaded.
// We can be a bit more relaxed when this is a standalone
// program that is about to terminate, and let safe=false.
- if (cap->suspended_ccalling_tasks && safe) {
+ if (cap->suspended_ccalls && safe) {
debugTrace(DEBUG_sched,
"thread(s) are involved in foreign calls, yielding");
cap->running_task = NULL;
{
nat i;
Capability *cap;
- Task *task;
+ InCall *incall;
// Each GC thread is responsible for following roots from the
// Capability of the same number. There will usually be the same
evac(user, (StgClosure **)(void *)&cap->wakeup_queue_hd);
evac(user, (StgClosure **)(void *)&cap->wakeup_queue_tl);
#endif
- for (task = cap->suspended_ccalling_tasks; task != NULL;
- task=task->next) {
- evac(user, (StgClosure **)(void *)&task->suspended_tso);
+ for (incall = cap->suspended_ccalls; incall != NULL;
+ incall=incall->next) {
+ evac(user, (StgClosure **)(void *)&incall->suspended_tso);
}
#if defined(THREADED_RTS)
// the suspended TSOs easily. Hence, when migrating a Task from
// the returning_tasks list, we must also migrate its entry from
// this list.
- Task *suspended_ccalling_tasks;
+ InCall *suspended_ccalls;
// One mutable list per generation, so we don't need to take any
// locks when updating an old-generation thunk. This also lets us
// If we are a worker, just exit. If we're a bound thread
// then we will exit below when we've removed our TSO from
// the run queue.
- if (task->tso == NULL && emptyRunQueue(cap)) {
+ if (!isBoundTask(task) && emptyRunQueue(cap)) {
return cap;
}
break;
// Check whether we can run this thread in the current task.
// If not, we have to pass our capability to the right task.
{
- Task *bound = t->bound;
+ InCall *bound = t->bound;
if (bound) {
- if (bound == task) {
+ if (bound->task == task) {
// yes, the Haskell thread is bound to the current native thread
} else {
debugTrace(DEBUG_sched,
}
} else {
// The thread we want to run is unbound.
- if (task->tso) {
+ if (task->incall->tso) {
debugTrace(DEBUG_sched,
"this OS thread cannot run thread %lu",
(unsigned long)t->id);
ASSERT_FULL_CAPABILITY_INVARIANTS(cap,task);
ASSERT(t->cap == cap);
- ASSERT(t->bound ? t->bound->cap == cap : 1);
+ ASSERT(t->bound ? t->bound->task->cap == cap : 1);
prev_what_next = t->what_next;
// and this task it bound).
return (waiting_for_gc ||
cap->returning_tasks_hd != NULL ||
- (!emptyRunQueue(cap) && (task->tso == NULL
+ (!emptyRunQueue(cap) && (task->incall->tso == NULL
? cap->run_queue_hd->bound != NULL
- : cap->run_queue_hd->bound != task)));
+ : cap->run_queue_hd->bound != task->incall)));
}
// This is the single place where a Task goes to sleep. There are
next = t->_link;
t->_link = END_TSO_QUEUE;
if (t->what_next == ThreadRelocated
- || t->bound == task // don't move my bound thread
+ || t->bound == task->incall // don't move my bound thread
|| tsoLocked(t)) { // don't move a locked thread
setTSOLink(cap, prev, t);
prev = t;
} else {
appendToRunQueue(free_caps[i],t);
- traceEventMigrateThread (cap, t, free_caps[i]->no);
+ traceEventMigrateThread (cap, t, free_caps[i]->no);
- if (t->bound) { t->bound->cap = free_caps[i]; }
+ if (t->bound) { t->bound->task->cap = free_caps[i]; }
t->cap = free_caps[i];
i++;
}
/* Probably a real deadlock. Send the current main thread the
* Deadlock exception.
*/
- if (task->tso) {
- switch (task->tso->why_blocked) {
+ if (task->incall->tso) {
+ switch (task->incall->tso->why_blocked) {
case BlockedOnSTM:
case BlockedOnBlackHole:
case BlockedOnException:
case BlockedOnMVar:
- throwToSingleThreaded(cap, task->tso,
+ throwToSingleThreaded(cap, task->incall->tso,
(StgClosure *)nonTermination_closure);
return;
default:
/* The TSO attached to this Task may have moved, so update the
* pointer to it.
*/
- if (task->tso == t) {
- task->tso = new_t;
+ if (task->incall->tso == t) {
+ task->incall->tso = new_t;
}
pushOnRunQueue(cap,new_t);
}
if (t->bound) {
- if (t->bound != task) {
+ if (t->bound != task->incall) {
#if !defined(THREADED_RTS)
// Must be a bound thread that is not the topmost one. Leave
// it on the run queue until the stack has unwound to the
#endif
}
- ASSERT(task->tso == t);
+ ASSERT(task->incall->tso == t);
if (t->what_next == ThreadComplete) {
if (task->ret) {
// NOTE: return val is tso->sp[1] (see StgStartup.hc)
- *(task->ret) = (StgClosure *)task->tso->sp[1];
+ *(task->ret) = (StgClosure *)task->incall->tso->sp[1];
}
task->stat = Success;
} else {
}
}
#ifdef DEBUG
- removeThreadLabel((StgWord)task->tso->id);
+ removeThreadLabel((StgWord)task->incall->tso->id);
#endif
// We no longer consider this thread and task to be bound to
// re-used). This was a real bug: the GC updated
// tso->bound->tso which lead to a deadlock.
t->bound = NULL;
- task->tso = NULL;
+ task->incall->tso = NULL;
return rtsTrue; // tells schedule() to return
}
)
{
#ifdef FORKPROCESS_PRIMOP_SUPPORTED
- Task *task;
pid_t pid;
StgTSO* t,*next;
Capability *cap;
// Any suspended C-calling Tasks are no more, their OS threads
// don't exist now:
- cap->suspended_ccalling_tasks = NULL;
+ cap->suspended_ccalls = NULL;
// Empty the threads lists. Otherwise, the garbage
// collector may attempt to resurrect some of these threads.
generations[g].threads = END_TSO_QUEUE;
}
- // Wipe the task list, except the current Task.
- ACQUIRE_LOCK(&sched_mutex);
- for (task = all_tasks; task != NULL; task=task->all_link) {
- if (task != cap->running_task) {
-#if defined(THREADED_RTS)
- initMutex(&task->lock); // see #1391
-#endif
- discardTask(task);
- }
- }
- RELEASE_LOCK(&sched_mutex);
+ discardTasksExcept(cap->running_task);
#if defined(THREADED_RTS)
// Wipe our spare workers list, they no longer exist. New
}
/* -----------------------------------------------------------------------------
- Managing the suspended_ccalling_tasks list.
+ Managing the suspended_ccalls list.
Locks required: sched_mutex
-------------------------------------------------------------------------- */
STATIC_INLINE void
suspendTask (Capability *cap, Task *task)
{
- ASSERT(task->next == NULL && task->prev == NULL);
- task->next = cap->suspended_ccalling_tasks;
- task->prev = NULL;
- if (cap->suspended_ccalling_tasks) {
- cap->suspended_ccalling_tasks->prev = task;
- }
- cap->suspended_ccalling_tasks = task;
+ InCall *incall;
+
+ incall = task->incall;
+ ASSERT(incall->next == NULL && incall->prev == NULL);
+ incall->next = cap->suspended_ccalls;
+ incall->prev = NULL;
+ if (cap->suspended_ccalls) {
+ cap->suspended_ccalls->prev = incall;
+ }
+ cap->suspended_ccalls = incall;
}
STATIC_INLINE void
recoverSuspendedTask (Capability *cap, Task *task)
{
- if (task->prev) {
- task->prev->next = task->next;
+ InCall *incall;
+
+ incall = task->incall;
+ if (incall->prev) {
+ incall->prev->next = incall->next;
} else {
- ASSERT(cap->suspended_ccalling_tasks == task);
- cap->suspended_ccalling_tasks = task->next;
+ ASSERT(cap->suspended_ccalls == incall);
+ cap->suspended_ccalls = incall->next;
}
- if (task->next) {
- task->next->prev = task->prev;
+ if (incall->next) {
+ incall->next->prev = incall->prev;
}
- task->next = task->prev = NULL;
+ incall->next = incall->prev = NULL;
}
/* ---------------------------------------------------------------------------
}
// Hand back capability
- task->suspended_tso = tso;
+ task->incall->suspended_tso = tso;
+ task->incall->suspended_cap = cap;
ACQUIRE_LOCK(&cap->lock);
resumeThread (void *task_)
{
StgTSO *tso;
+ InCall *incall;
Capability *cap;
Task *task = task_;
int saved_errno;
saved_winerror = GetLastError();
#endif
- cap = task->cap;
+ incall = task->incall;
+ cap = incall->suspended_cap;
+ task->cap = cap;
+
// Wait for permission to re-enter the RTS with the result.
waitForReturnCapability(&cap,task);
// we might be on a different capability now... but if so, our
- // entry on the suspended_ccalling_tasks list will also have been
+ // entry on the suspended_ccalls list will also have been
// migrated.
// Remove the thread from the suspended list
recoverSuspendedTask(cap,task);
- tso = task->suspended_tso;
- task->suspended_tso = NULL;
+ tso = incall->suspended_tso;
+ incall->suspended_tso = NULL;
+ incall->suspended_cap = NULL;
tso->_link = END_TSO_QUEUE; // no write barrier reqd
traceEventRunThread(cap, tso);
// This TSO is now a bound thread; make the Task and TSO
// point to each other.
- tso->bound = task;
+ tso->bound = task->incall;
tso->cap = cap;
- task->tso = tso;
+ task->incall->tso = tso;
task->ret = ret;
task->stat = NoStatus;
* ------------------------------------------------------------------------- */
#if defined(THREADED_RTS)
-void OSThreadProcAttr
-workerStart(Task *task)
+void scheduleWorker (Capability *cap, Task *task)
{
- Capability *cap;
-
- // See startWorkerTask().
- ACQUIRE_LOCK(&task->lock);
- cap = task->cap;
- RELEASE_LOCK(&task->lock);
-
- if (RtsFlags.ParFlags.setAffinity) {
- setThreadAffinity(cap->no, n_capabilities);
- }
-
- // set the thread-local pointer to the Task:
- taskEnter(task);
-
// schedule() runs without a lock.
cap = schedule(cap,task);
initSparkPools();
#endif
+ RELEASE_LOCK(&sched_mutex);
+
#if defined(THREADED_RTS)
/*
* Eagerly start one worker to run each Capability, except for
for (i = 1; i < n_capabilities; i++) {
cap = &capabilities[i];
ACQUIRE_LOCK(&cap->lock);
- startWorkerTask(cap, workerStart);
+ startWorkerTask(cap);
RELEASE_LOCK(&cap->lock);
}
}
#endif
-
- RELEASE_LOCK(&sched_mutex);
}
void
sched_state = SCHED_INTERRUPTING;
waitForReturnCapability(&task->cap,task);
scheduleDoGC(task->cap,task,rtsFalse);
- ASSERT(task->tso == NULL);
+ ASSERT(task->incall->tso == NULL);
releaseCapability(task->cap);
}
sched_state = SCHED_SHUTTING_DOWN;
nat i;
for (i = 0; i < n_capabilities; i++) {
- ASSERT(task->tso == NULL);
+ ASSERT(task->incall->tso == NULL);
shutdownCapability(&capabilities[i], task, wait_foreign);
}
}
// We must grab a new Task here, because the existing Task may be
// associated with a particular Capability, and chained onto the
- // suspended_ccalling_tasks queue.
+ // suspended_ccalls queue.
task = newBoundTask();
waitForReturnCapability(&task->cap,task);
// The TSO attached to this Task may have moved, so update the
// pointer to it.
- if (task->tso == tso) {
- task->tso = new_tso;
+ if (task->incall->tso == tso) {
+ task->incall->tso = new_tso;
}
unlockTSO(new_tso);
/* findRetryFrameHelper */
StgWord findRetryFrameHelper (StgTSO *tso);
-/* workerStart()
- *
- * Entry point for a new worker task.
- * Called from STG : NO
- * Locks assumed : none
- */
-#if defined(THREADED_RTS)
-void OSThreadProcAttr workerStart(Task *task);
-#endif
+/* Entry point for a new worker */
+void scheduleWorker (Capability *cap, Task *task);
/* The state of the scheduler. This is used to control the sequence
* of events during shutdown, and when the runtime is interrupted
i++, task = task->all_link) {
statsPrintf(" Task %2d %-8s : %6.2fs (%6.2fs) %6.2fs (%6.2fs)\n",
i,
- (task->tso == NULL) ? "(worker)" : "(bound)",
+ (task->worker) ? "(worker)" : "(bound)",
TICK_TO_DBL(task->mut_time),
TICK_TO_DBL(task->mut_etime),
TICK_TO_DBL(task->gc_time),
// Task lists and global counters.
// Locks required: sched_mutex.
Task *all_tasks = NULL;
-static Task *task_free_list = NULL; // singly-linked
static nat taskCount;
-static nat tasksRunning;
-static nat workerCount;
static int tasksInitialized = 0;
+static void freeTask (Task *task);
+static Task * allocTask (void);
+static Task * newTask (rtsBool);
+
/* -----------------------------------------------------------------------------
* Remembering the current thread's Task
* -------------------------------------------------------------------------- */
// A thread-local-storage key that we can use to get access to the
// current thread's Task structure.
#if defined(THREADED_RTS)
+# if defined(MYTASK_USE_TLV)
+__thread Task *my_task;
+# else
ThreadLocalKey currentTaskKey;
+# endif
#else
Task *my_task;
#endif
{
if (!tasksInitialized) {
taskCount = 0;
- workerCount = 0;
- tasksRunning = 0;
tasksInitialized = 1;
-#if defined(THREADED_RTS)
+#if defined(THREADED_RTS) && !defined(MYTASK_USE_TLV)
newThreadLocalKey(¤tTaskKey);
#endif
}
freeTaskManager (void)
{
Task *task, *next;
+ nat tasksRunning = 0;
ASSERT_LOCK_HELD(&sched_mutex);
- debugTrace(DEBUG_sched, "freeing task manager, %d tasks still running",
- tasksRunning);
-
for (task = all_tasks; task != NULL; task = next) {
next = task->all_link;
if (task->stopped) {
- // We only free resources if the Task is not in use. A
- // Task may still be in use if we have a Haskell thread in
- // a foreign call while we are attempting to shut down the
- // RTS (see conc059).
-#if defined(THREADED_RTS)
- closeCondition(&task->cond);
- closeMutex(&task->lock);
-#endif
- stgFree(task);
+ freeTask(task);
+ } else {
+ tasksRunning++;
}
}
+
+ debugTrace(DEBUG_sched, "freeing task manager, %d tasks still running",
+ tasksRunning);
+
all_tasks = NULL;
- task_free_list = NULL;
-#if defined(THREADED_RTS)
+#if defined(THREADED_RTS) && !defined(MYTASK_USE_TLV)
freeThreadLocalKey(¤tTaskKey);
#endif
return tasksRunning;
}
+static Task *
+allocTask (void)
+{
+ Task *task;
+
+ task = myTask();
+ if (task != NULL) {
+ return task;
+ } else {
+ task = newTask(rtsFalse);
+#if defined(THREADED_RTS)
+ task->id = osThreadId();
+#endif
+ setMyTask(task);
+ return task;
+ }
+}
+
+static void
+freeTask (Task *task)
+{
+ InCall *incall, *next;
+
+ // We only free resources if the Task is not in use. A
+ // Task may still be in use if we have a Haskell thread in
+ // a foreign call while we are attempting to shut down the
+ // RTS (see conc059).
+#if defined(THREADED_RTS)
+ closeCondition(&task->cond);
+ closeMutex(&task->lock);
+#endif
+
+ for (incall = task->incall; incall != NULL; incall = next) {
+ next = incall->prev_stack;
+ stgFree(incall);
+ }
+ for (incall = task->spare_incalls; incall != NULL; incall = next) {
+ next = incall->next;
+ stgFree(incall);
+ }
+
+ stgFree(task);
+}
static Task*
-newTask (void)
+newTask (rtsBool worker)
{
#if defined(THREADED_RTS)
Ticks currentElapsedTime, currentUserTime;
#define ROUND_TO_CACHE_LINE(x) ((((x)+63) / 64) * 64)
task = stgMallocBytes(ROUND_TO_CACHE_LINE(sizeof(Task)), "newTask");
- task->cap = NULL;
- task->stopped = rtsFalse;
- task->suspended_tso = NULL;
- task->tso = NULL;
- task->stat = NoStatus;
- task->ret = NULL;
+ task->cap = NULL;
+ task->worker = worker;
+ task->stopped = rtsFalse;
+ task->stat = NoStatus;
+ task->ret = NULL;
+ task->n_spare_incalls = 0;
+ task->spare_incalls = NULL;
+ task->incall = NULL;
#if defined(THREADED_RTS)
initCondition(&task->cond);
task->elapsedtimestart = currentElapsedTime;
#endif
- task->prev = NULL;
task->next = NULL;
- task->return_link = NULL;
+
+ ACQUIRE_LOCK(&sched_mutex);
task->all_link = all_tasks;
all_tasks = task;
taskCount++;
+ RELEASE_LOCK(&sched_mutex);
+
return task;
}
+// avoid the spare_incalls list growing unboundedly
+#define MAX_SPARE_INCALLS 8
+
+static void
+newInCall (Task *task)
+{
+ InCall *incall;
+
+ if (task->spare_incalls != NULL) {
+ incall = task->spare_incalls;
+ task->spare_incalls = incall->next;
+ task->n_spare_incalls--;
+ } else {
+ incall = stgMallocBytes((sizeof(InCall)), "newBoundTask");
+ }
+
+ incall->tso = NULL;
+ incall->task = task;
+ incall->suspended_tso = NULL;
+ incall->suspended_cap = NULL;
+ incall->next = NULL;
+ incall->prev = NULL;
+ incall->prev_stack = task->incall;
+ task->incall = incall;
+}
+
+static void
+endInCall (Task *task)
+{
+ InCall *incall;
+
+ incall = task->incall;
+ incall->tso = NULL;
+ task->incall = task->incall->prev_stack;
+
+ if (task->n_spare_incalls >= MAX_SPARE_INCALLS) {
+ stgFree(incall);
+ } else {
+ incall->next = task->spare_incalls;
+ task->spare_incalls = incall;
+ task->n_spare_incalls++;
+ }
+}
+
+
Task *
newBoundTask (void)
{
stg_exit(EXIT_FAILURE);
}
- // ToDo: get rid of this lock in the common case. We could store
- // a free Task in thread-local storage, for example. That would
- // leave just one lock on the path into the RTS: cap->lock when
- // acquiring the Capability.
- ACQUIRE_LOCK(&sched_mutex);
-
- if (task_free_list == NULL) {
- task = newTask();
- } else {
- task = task_free_list;
- task_free_list = task->next;
- task->next = NULL;
- task->prev = NULL;
- task->stopped = rtsFalse;
- }
-#if defined(THREADED_RTS)
- task->id = osThreadId();
-#endif
- ASSERT(task->cap == NULL);
-
- tasksRunning++;
+ task = allocTask();
- taskEnter(task);
+ task->stopped = rtsFalse;
- RELEASE_LOCK(&sched_mutex);
+ newInCall(task);
debugTrace(DEBUG_sched, "new task (taskCount: %d)", taskCount);
return task;
void
boundTaskExiting (Task *task)
{
- task->tso = NULL;
task->stopped = rtsTrue;
- task->cap = NULL;
#if defined(THREADED_RTS)
ASSERT(osThreadId() == task->id);
#endif
ASSERT(myTask() == task);
- setMyTask(task->prev_stack);
-
- tasksRunning--;
- // sadly, we need a lock around the free task list. Todo: eliminate.
- ACQUIRE_LOCK(&sched_mutex);
- task->next = task_free_list;
- task_free_list = task;
- RELEASE_LOCK(&sched_mutex);
+ endInCall(task);
debugTrace(DEBUG_sched, "task exiting");
}
+
#ifdef THREADED_RTS
#define TASK_ID(t) (t)->id
#else
#endif
void
-discardTask (Task *task)
+discardTasksExcept (Task *keep)
{
- ASSERT_LOCK_HELD(&sched_mutex);
- if (!task->stopped) {
- debugTrace(DEBUG_sched, "discarding task %ld", (long)TASK_ID(task));
- task->cap = NULL;
- if (task->tso == NULL) {
- workerCount--;
- } else {
- task->tso = NULL;
+ Task *task;
+
+ // Wipe the task list, except the current Task.
+ ACQUIRE_LOCK(&sched_mutex);
+ for (task = all_tasks; task != NULL; task=task->all_link) {
+ if (task != keep) {
+ debugTrace(DEBUG_sched, "discarding task %ld", (long)TASK_ID(task));
+ freeTask(task);
}
- task->stopped = rtsTrue;
- tasksRunning--;
- task->next = task_free_list;
- task_free_list = task;
}
+ all_tasks = keep;
+ RELEASE_LOCK(&sched_mutex);
}
void
task->cap = NULL;
taskTimeStamp(task);
task->stopped = rtsTrue;
- tasksRunning--;
- workerCount--;
-
- ACQUIRE_LOCK(&sched_mutex);
- task->next = task_free_list;
- task_free_list = task;
- RELEASE_LOCK(&sched_mutex);
}
#endif
#if defined(THREADED_RTS)
+static void OSThreadProcAttr
+workerStart(Task *task)
+{
+ Capability *cap;
+
+ // See startWorkerTask().
+ ACQUIRE_LOCK(&task->lock);
+ cap = task->cap;
+ RELEASE_LOCK(&task->lock);
+
+ if (RtsFlags.ParFlags.setAffinity) {
+ setThreadAffinity(cap->no, n_capabilities);
+ }
+
+ // set the thread-local pointer to the Task:
+ setMyTask(task);
+
+ newInCall(task);
+
+ scheduleWorker(cap,task);
+}
+
void
-startWorkerTask (Capability *cap,
- void OSThreadProcAttr (*taskStart)(Task *task))
+startWorkerTask (Capability *cap)
{
int r;
OSThreadId tid;
Task *task;
- workerCount++;
-
// A worker always gets a fresh Task structure.
- task = newTask();
-
- tasksRunning++;
+ task = newTask(rtsTrue);
// The lock here is to synchronise with taskStart(), to make sure
// that we have finished setting up the Task structure before the
ASSERT_LOCK_HELD(&cap->lock);
cap->running_task = task;
- r = createOSThread(&tid, (OSThreadProc *)taskStart, task);
+ r = createOSThread(&tid, (OSThreadProc*)workerStart, task);
if (r != 0) {
sysErrorBelch("failed to create OS thread");
stg_exit(EXIT_FAILURE);
if (task->cap) {
debugBelch("on capability %d, ", task->cap->no);
}
- if (task->tso) {
- debugBelch("bound to thread %lu", (unsigned long)task->tso->id);
+ if (task->incall->tso) {
+ debugBelch("bound to thread %lu",
+ (unsigned long)task->incall->tso->id);
} else {
debugBelch("worker");
}
Definition of a Task
--------------------
- A task is an OSThread that runs Haskell code. Every OSThread
- created by the RTS for the purposes of running Haskell code is a
- Task, and OS threads that enter the Haskell RTS for the purposes of
- making a call-in are also Tasks.
+ A task is an OSThread that runs Haskell code. Every OSThread that
+ runs inside the RTS, whether as a worker created by the RTS or via
+ an in-call from C to Haskell, has an associated Task. The first
+ time an OS thread calls into Haskell it is allocated a Task, which
+ remains until the RTS is shut down.
+
+ There is a one-to-one relationship between OSThreads and Tasks.
+ The Task for an OSThread is kept in thread-local storage, and can
+ be retrieved at any time using myTask().
In the THREADED_RTS build, multiple Tasks may all be running
Haskell code simultaneously. A task relinquishes its Capability
when it is asked to evaluate an external (C) call.
- In general, there may be multiple Tasks associated with a given OS
- thread. A second Task is created when one Task makes a foreign
- call from Haskell, and subsequently calls back in to Haskell,
- creating a new bound thread.
-
- A particular Task structure can belong to more than one OS thread
- over its lifetime. This is to avoid creating an unbounded number
- of Task structures. The stats just accumulate.
-
Ownership of Task
-----------------
(1) a bound Task, the TSO will be on a queue somewhere
(2) a worker task, on the spare_workers queue of task->cap.
- (b) making a foreign call. The Task will be on the
- suspended_ccalling_tasks list.
+ (b) making a foreign call. The InCall will be on the
+ suspended_ccalls list.
We re-establish ownership in each case by respectively
ownership of the Task and a Capability.
*/
+// The InCall structure represents either a single in-call from C to
+// Haskell, or a worker thread.
+typedef struct InCall_ {
+ StgTSO * tso; // the bound TSO (or NULL for a worker)
+
+ StgTSO * suspended_tso; // the TSO is stashed here when we
+ // make a foreign call (NULL otherwise);
+
+ Capability *suspended_cap; // The capability that the
+ // suspended_tso is on, because
+ // we can't read this from the TSO
+ // without owning a Capability in the
+ // first place.
+
+ struct Task_ *task;
+
+ // When a Haskell thread makes a foreign call that re-enters
+ // Haskell, we end up with another Task associated with the
+ // current thread. We have to remember the whole stack of InCalls
+ // associated with the current Task so that we can correctly
+ // save & restore the InCall on entry to and exit from Haskell.
+ struct InCall_ *prev_stack;
+
+ // Links InCalls onto suspended_ccalls, spare_incalls
+ struct InCall_ *prev;
+ struct InCall_ *next;
+} InCall;
+
typedef struct Task_ {
#if defined(THREADED_RTS)
OSThreadId id; // The OS Thread ID of this task
+
+ Condition cond; // used for sleeping & waking up this task
+ Mutex lock; // lock for the condition variable
+
+ // this flag tells the task whether it should wait on task->cond
+ // or just continue immediately. It's a workaround for the fact
+ // that signalling a condition variable doesn't do anything if the
+ // thread is already running, but we want it to be sticky.
+ rtsBool wakeup;
#endif
// This points to the Capability that the Task "belongs" to. If
// must be held when modifying task->cap.
struct Capability_ *cap;
+ // The current top-of-stack InCall
+ struct InCall_ *incall;
+
+ nat n_spare_incalls;
+ struct InCall_ *spare_incalls;
+
+ rtsBool worker; // == rtsTrue if this is a worker Task
rtsBool stopped; // this task has stopped or exited Haskell
- StgTSO * suspended_tso; // the TSO is stashed here when we
- // make a foreign call (NULL otherwise);
- // The following 3 fields are used by bound threads:
- StgTSO * tso; // the bound TSO (or NULL)
SchedulerStatus stat; // return status
StgClosure ** ret; // return value
-#if defined(THREADED_RTS)
- Condition cond; // used for sleeping & waking up this task
- Mutex lock; // lock for the condition variable
-
- // this flag tells the task whether it should wait on task->cond
- // or just continue immediately. It's a workaround for the fact
- // that signalling a condition variable doesn't do anything if the
- // thread is already running, but we want it to be sticky.
- rtsBool wakeup;
-#endif
-
// Stats that we collect about this task
// ToDo: we probably want to put this in a separate TaskStats
// structure, so we can share it between multiple Tasks. We don't
Ticks gc_time;
Ticks gc_etime;
- // Links tasks onto various lists. (ToDo: do we need double
- // linking now?)
- struct Task_ *prev;
+ // Links tasks on the returning_tasks queue of a Capability, and
+ // on spare_workers.
struct Task_ *next;
- // Links tasks on the returning_tasks queue of a Capability.
- struct Task_ *return_link;
-
// Links tasks on the all_tasks list
struct Task_ *all_link;
- // When a Haskell thread makes a foreign call that re-enters
- // Haskell, we end up with another Task associated with the
- // current thread. We have to remember the whole stack of Tasks
- // associated with the current thread so that we can correctly
- // save & restore the thread-local current task pointer.
- struct Task_ *prev_stack;
} Task;
INLINE_HEADER rtsBool
isBoundTask (Task *task)
{
- return (task->tso != NULL);
+ return (task->incall->tso != NULL);
}
//
void boundTaskExiting (Task *task);
-// This must be called when a new Task is associated with the current
-// thread. It sets up the thread-local current task pointer so that
-// myTask() can work.
-INLINE_HEADER void taskEnter (Task *task);
-
// Notify the task manager that a task has stopped. This is used
// mainly for stats-gathering purposes.
// Requires: sched_mutex.
// Put the task back on the free list, mark it stopped. Used by
// forkProcess().
//
-void discardTask (Task *task);
+void discardTasksExcept (Task *keep);
// Get the Task associated with the current OS thread (or NULL if none).
//
// will become the running_task for that Capability.
// Requires: sched_mutex.
//
-void startWorkerTask (struct Capability_ *cap,
- void OSThreadProcAttr (*taskStart)(Task *task));
+void startWorkerTask (Capability *cap);
#endif /* THREADED_RTS */
// A thread-local-storage key that we can use to get access to the
// current thread's Task structure.
#if defined(THREADED_RTS)
+#if defined(linux_HOST_OS) && \
+ (defined(i386_HOST_ARCH) || defined(x86_64_HOST_ARCH))
+#define MYTASK_USE_TLV
+extern __thread Task *my_task;
+#else
extern ThreadLocalKey currentTaskKey;
+#endif
#else
extern Task *my_task;
#endif
INLINE_HEADER Task *
myTask (void)
{
-#if defined(THREADED_RTS)
+#if defined(THREADED_RTS) && !defined(MYTASK_USE_TLV)
return getThreadLocalVar(¤tTaskKey);
#else
return my_task;
INLINE_HEADER void
setMyTask (Task *task)
{
-#if defined(THREADED_RTS)
+#if defined(THREADED_RTS) && !defined(MYTASK_USE_TLV)
setThreadLocalVar(¤tTaskKey,task);
#else
my_task = task;
#endif
}
-// This must be called when a new Task is associated with the current
-// thread. It sets up the thread-local current task pointer so that
-// myTask() can work.
-INLINE_HEADER void
-taskEnter (Task *task)
-{
- // save the current value, just in case this Task has been created
- // as a result of re-entering the RTS (defaults to NULL):
- task->prev_stack = myTask();
- setMyTask(task);
-}
-
END_RTS_PRIVATE
#endif /* TASK_H */
// We are waking up this thread on the current Capability, which
// might involve migrating it from the Capability it was last on.
if (tso->bound) {
- ASSERT(tso->bound->cap == tso->cap);
- tso->bound->cap = cap;
+ ASSERT(tso->bound->task->cap == tso->cap);
+ tso->bound->task->cap = cap;
}
tso->cap = cap;
// the task list
{
Task *task;
+ InCall *incall;
for (task = all_tasks; task != NULL; task = task->all_link) {
- if (task->tso) {
- thread_(&task->tso);
- }
+ for (incall = task->incall; incall != NULL;
+ incall = incall->prev_stack) {
+ if (incall->tso) {
+ thread_(&incall->tso);
+ }
+ }
}
}
projects / ghc-hetmet.git / commitdiff