+/* Free capability list.
+ * Locks required: sched_mutex.
+ */
+#ifdef SMP
+Capability *free_capabilities; /* Available capabilities for running threads */
+nat n_free_capabilities; /* total number of available capabilities */
+#else
+Capability MainRegTable; /* for non-SMP, we have one global capability */
+#endif
+
+rtsBool ready_to_gc;
+
+/* All our current task ids, saved in case we need to kill them later.
+ */
+#ifdef SMP
+task_info *task_ids;
+#endif
+
+void addToBlockedQueue ( StgTSO *tso );
+
+static void schedule ( void );
+static void initThread ( StgTSO *tso, nat stack_size );
+ void interruptStgRts ( void );
+
+#ifdef SMP
+pthread_mutex_t sched_mutex = PTHREAD_MUTEX_INITIALIZER;
+pthread_mutex_t term_mutex = PTHREAD_MUTEX_INITIALIZER;
+pthread_cond_t thread_ready_cond = PTHREAD_COND_INITIALIZER;
+pthread_cond_t gc_pending_cond = PTHREAD_COND_INITIALIZER;
+
+nat await_death;
+#endif
+
+/* -----------------------------------------------------------------------------
+ Main scheduling loop.
+
+ We use round-robin scheduling, each thread returning to the
+ scheduler loop when one of these conditions is detected:
+
+ * out of heap space
+ * timer expires (thread yields)
+ * thread blocks
+ * thread ends
+ * stack overflow
+
+ Locking notes: we acquire the scheduler lock once at the beginning
+ of the scheduler loop, and release it when
+
+ * running a thread, or
+ * waiting for work, or
+ * waiting for a GC to complete.
+
+ -------------------------------------------------------------------------- */
+
+static void
+schedule( void )
+{
+ StgTSO *t;
+ Capability *cap;
+ StgThreadReturnCode ret;
+
+ ACQUIRE_LOCK(&sched_mutex);
+
+ while (1) {
+
+ /* Check whether any waiting threads need to be woken up.
+ * If the run queue is empty, we can wait indefinitely for
+ * something to happen.
+ */
+ if (blocked_queue_hd != END_TSO_QUEUE) {
+ awaitEvent(run_queue_hd == END_TSO_QUEUE);
+ }
+
+ /* check for signals each time around the scheduler */
+#ifndef __MINGW32__
+ if (signals_pending()) {
+ start_signal_handlers();
+ }
+#endif
+
+#ifdef SMP
+ /* If there's a GC pending, don't do anything until it has
+ * completed.
+ */
+ if (ready_to_gc) {
+ IF_DEBUG(scheduler,fprintf(stderr,"schedule (task %ld): waiting for GC\n",
+ pthread_self()););
+ pthread_cond_wait(&gc_pending_cond, &sched_mutex);
+ }
+
+ /* block until we've got a thread on the run queue and a free
+ * capability.
+ */
+ while (run_queue_hd == END_TSO_QUEUE || free_capabilities == NULL) {
+ IF_DEBUG(scheduler,
+ fprintf(stderr, "schedule (task %ld): waiting for work\n",
+ pthread_self()););
+ pthread_cond_wait(&thread_ready_cond, &sched_mutex);
+ IF_DEBUG(scheduler,
+ fprintf(stderr, "schedule (task %ld): work now available\n",
+ pthread_self()););
+ }
+#endif
+
+ /* grab a thread from the run queue
+ */
+ t = POP_RUN_QUEUE();
+
+ /* grab a capability
+ */
+#ifdef SMP
+ cap = free_capabilities;
+ free_capabilities = cap->link;
+ n_free_capabilities--;
+#else
+ cap = &MainRegTable;
+#endif
+
+ cap->rCurrentTSO = t;
+
+ /* set the context_switch flag
+ */
+ if (run_queue_hd == END_TSO_QUEUE)
+ context_switch = 0;
+ else
+ context_switch = 1;
+
+ RELEASE_LOCK(&sched_mutex);
+
+#ifdef SMP
+ IF_DEBUG(scheduler,fprintf(stderr,"schedule (task %ld): running thread %d\n", pthread_self(),t->id));
+#else
+ IF_DEBUG(scheduler,fprintf(stderr,"schedule: running thread %d\n",t->id));
+#endif
+
+ /* Run the current thread
+ */
+ switch (cap->rCurrentTSO->whatNext) {
+ case ThreadKilled:
+ case ThreadComplete:
+ /* Thread already finished, return to scheduler. */
+ ret = ThreadFinished;
+ break;
+ case ThreadEnterGHC:
+ ret = StgRun((StgFunPtr) stg_enterStackTop, cap);
+ break;
+ case ThreadRunGHC:
+ ret = StgRun((StgFunPtr) stg_returnToStackTop, cap);
+ break;
+ case ThreadEnterHugs:
+#ifdef INTERPRETER
+ {
+ StgClosure* c;
+ IF_DEBUG(scheduler,belch("schedule: entering Hugs"));
+ c = (StgClosure *)(cap->rCurrentTSO->sp[0]);
+ cap->rCurrentTSO->sp += 1;
+ ret = enter(cap,c);
+ break;
+ }
+#else
+ barf("Panic: entered a BCO but no bytecode interpreter in this build");
+#endif
+ default:
+ barf("schedule: invalid whatNext field");
+ }
+
+ /* Costs for the scheduler are assigned to CCS_SYSTEM */
+#ifdef PROFILING
+ CCCS = CCS_SYSTEM;
+#endif
+
+ ACQUIRE_LOCK(&sched_mutex);
+
+#ifdef SMP
+ IF_DEBUG(scheduler,fprintf(stderr,"schedule (task %ld): ", pthread_self()););
+#else
+ IF_DEBUG(scheduler,fprintf(stderr,"schedule: "););
+#endif
+ t = cap->rCurrentTSO;
+
+ switch (ret) {
+ case HeapOverflow:
+ /* make all the running tasks block on a condition variable,
+ * maybe set context_switch and wait till they all pile in,
+ * then have them wait on a GC condition variable.
+ */
+ IF_DEBUG(scheduler,belch("thread %ld stopped: HeapOverflow", t->id));
+ threadPaused(t);
+
+ ready_to_gc = rtsTrue;
+ context_switch = 1; /* stop other threads ASAP */
+ PUSH_ON_RUN_QUEUE(t);
+ break;
+
+ case StackOverflow:
+ /* just adjust the stack for this thread, then pop it back
+ * on the run queue.
+ */
+ IF_DEBUG(scheduler,belch("thread %ld stopped, StackOverflow", t->id));
+ threadPaused(t);
+ {
+ StgMainThread *m;
+ /* enlarge the stack */
+ StgTSO *new_t = threadStackOverflow(t);
+
+ /* This TSO has moved, so update any pointers to it from the
+ * main thread stack. It better not be on any other queues...
+ * (it shouldn't be)
+ */
+ for (m = main_threads; m != NULL; m = m->link) {
+ if (m->tso == t) {
+ m->tso = new_t;
+ }
+ }
+ PUSH_ON_RUN_QUEUE(new_t);
+ }
+ break;
+
+ case ThreadYielding:
+ /* put the thread back on the run queue. Then, if we're ready to
+ * GC, check whether this is the last task to stop. If so, wake
+ * up the GC thread. getThread will block during a GC until the
+ * GC is finished.
+ */
+ IF_DEBUG(scheduler,
+ if (t->whatNext == ThreadEnterHugs) {
+ /* ToDo: or maybe a timer expired when we were in Hugs?
+ * or maybe someone hit ctrl-C
+ */
+ belch("thread %ld stopped to switch to Hugs", t->id);
+ } else {
+ belch("thread %ld stopped, yielding", t->id);
+ }
+ );
+ threadPaused(t);
+ APPEND_TO_RUN_QUEUE(t);
+ break;
+
+ case ThreadBlocked:
+ /* don't need to do anything. Either the thread is blocked on
+ * I/O, in which case we'll have called addToBlockedQueue
+ * previously, or it's blocked on an MVar or Blackhole, in which
+ * case it'll be on the relevant queue already.
+ */
+ IF_DEBUG(scheduler,
+ fprintf(stderr, "thread %d stopped, ", t->id);
+ printThreadBlockage(t);
+ fprintf(stderr, "\n"));
+ threadPaused(t);
+ break;
+
+ case ThreadFinished:
+ /* Need to check whether this was a main thread, and if so, signal
+ * the task that started it with the return value. If we have no
+ * more main threads, we probably need to stop all the tasks until
+ * we get a new one.
+ */
+ IF_DEBUG(scheduler,belch("thread %ld finished", t->id));
+ t->whatNext = ThreadComplete;
+ break;
+
+ default:
+ barf("doneThread: invalid thread return code");
+ }
+
+#ifdef SMP
+ cap->link = free_capabilities;
+ free_capabilities = cap;
+ n_free_capabilities++;
+#endif
+
+#ifdef SMP
+ if (ready_to_gc && n_free_capabilities == RtsFlags.ConcFlags.nNodes) {
+#else
+ if (ready_to_gc) {
+#endif
+ /* everybody back, start the GC.
+ * Could do it in this thread, or signal a condition var
+ * to do it in another thread. Either way, we need to
+ * broadcast on gc_pending_cond afterward.
+ */
+#ifdef SMP
+ IF_DEBUG(scheduler,belch("schedule (task %ld): doing GC", pthread_self()));
+#endif
+ GarbageCollect(GetRoots);
+ ready_to_gc = rtsFalse;
+#ifdef SMP
+ pthread_cond_broadcast(&gc_pending_cond);
+#endif
+ }
+
+ /* Go through the list of main threads and wake up any
+ * clients whose computations have finished. ToDo: this
+ * should be done more efficiently without a linear scan
+ * of the main threads list, somehow...
+ */
+#ifdef SMP
+ {
+ StgMainThread *m, **prev;
+ prev = &main_threads;
+ for (m = main_threads; m != NULL; m = m->link) {
+ if (m->tso->whatNext == ThreadComplete) {
+ if (m->ret) {
+ *(m->ret) = (StgClosure *)m->tso->sp[0];
+ }
+ *prev = m->link;
+ m->stat = Success;
+ pthread_cond_broadcast(&m->wakeup);
+ }
+ if (m->tso->whatNext == ThreadKilled) {
+ *prev = m->link;
+ m->stat = Killed;
+ pthread_cond_broadcast(&m->wakeup);
+ }
+ }
+ }
+#else
+ /* If our main thread has finished or been killed, return.
+ * If we were re-entered as a result of a _ccall_gc, then
+ * pop the blocked thread off the ccalling_threads stack back
+ * into CurrentTSO.
+ */
+ {
+ StgMainThread *m = main_threads;
+ if (m->tso->whatNext == ThreadComplete
+ || m->tso->whatNext == ThreadKilled) {
+ main_threads = main_threads->link;
+ if (m->tso->whatNext == ThreadComplete) {
+ /* we finished successfully, fill in the return value */
+ if (m->ret) { *(m->ret) = (StgClosure *)m->tso->sp[0]; };
+ m->stat = Success;
+ return;
+ } else {
+ m->stat = Killed;
+ return;
+ }
+ }
+ }
+#endif
+
+ } /* end of while(1) */
+}
+
+/* -----------------------------------------------------------------------------
+ * Suspending & resuming Haskell threads.
+ *
+ * When making a "safe" call to C (aka _ccall_GC), the task gives back
+ * its capability before calling the C function. This allows another
+ * task to pick up the capability and carry on running Haskell
+ * threads. It also means that if the C call blocks, it won't lock
+ * the whole system.
+ *
+ * The Haskell thread making the C call is put to sleep for the
+ * duration of the call, on the susepended_ccalling_threads queue. We
+ * give out a token to the task, which it can use to resume the thread
+ * on return from the C function.
+ * -------------------------------------------------------------------------- */
+
+StgInt
+suspendThread( Capability *cap )
+{
+ nat tok;
+
+ ACQUIRE_LOCK(&sched_mutex);
+
+#ifdef SMP
+ IF_DEBUG(scheduler,
+ fprintf(stderr, "schedule (task %ld): thread %d did a _ccall_gc\n",
+ pthread_self(), cap->rCurrentTSO->id));
+#else
+ IF_DEBUG(scheduler,
+ fprintf(stderr, "schedule: thread %d did a _ccall_gc\n",
+ cap->rCurrentTSO->id));
+#endif
+
+ threadPaused(cap->rCurrentTSO);
+ cap->rCurrentTSO->link = suspended_ccalling_threads;
+ suspended_ccalling_threads = cap->rCurrentTSO;
+
+ /* Use the thread ID as the token; it should be unique */
+ tok = cap->rCurrentTSO->id;
+
+#ifdef SMP
+ cap->link = free_capabilities;
+ free_capabilities = cap;
+ n_free_capabilities++;
+#endif
+
+ RELEASE_LOCK(&sched_mutex);
+ return tok;
+}
+
+Capability *
+resumeThread( StgInt tok )
+{
+ StgTSO *tso, **prev;
+ Capability *cap;
+
+ ACQUIRE_LOCK(&sched_mutex);
+
+ prev = &suspended_ccalling_threads;
+ for (tso = suspended_ccalling_threads;
+ tso != END_TSO_QUEUE;
+ prev = &tso->link, tso = tso->link) {
+ if (tso->id == (StgThreadID)tok) {
+ *prev = tso->link;
+ break;
+ }
+ }
+ if (tso == END_TSO_QUEUE) {
+ barf("resumeThread: thread not found");
+ }
+
+#ifdef SMP
+ while (free_capabilities == NULL) {
+ IF_DEBUG(scheduler,
+ fprintf(stderr,"schedule (task %ld): waiting to resume\n",
+ pthread_self()));
+ pthread_cond_wait(&thread_ready_cond, &sched_mutex);
+ IF_DEBUG(scheduler,fprintf(stderr,
+ "schedule (task %ld): resuming thread %d\n",
+ pthread_self(), tso->id));
+ }
+ cap = free_capabilities;
+ free_capabilities = cap->link;
+ n_free_capabilities--;
+#else
+ cap = &MainRegTable;
+#endif
+
+ cap->rCurrentTSO = tso;
+
+ RELEASE_LOCK(&sched_mutex);
+ return cap;
+}
+
+/* -----------------------------------------------------------------------------
+ * Static functions
+ * -------------------------------------------------------------------------- */
+static void unblockThread(StgTSO *tso);
+
+/* -----------------------------------------------------------------------------
+ * Comparing Thread ids.
+ *
+ * This is used from STG land in the implementation of the
+ * instances of Eq/Ord for ThreadIds.
+ * -------------------------------------------------------------------------- */
+
+int cmp_thread(const StgTSO *tso1, const StgTSO *tso2)
+{
+ StgThreadID id1 = tso1->id;
+ StgThreadID id2 = tso2->id;
+
+ if (id1 < id2) return (-1);
+ if (id1 > id2) return 1;
+ return 0;
+}
+
projects / ghc-hetmet.git / blobdiff