jthread.c

来自「kaffe Java 解释器语言,源码,Java的子集系统,开放源代码」· C语言 代码 · 共 2,882 行 · 第 1/5 页

        if (time == 0) {
                return; 
        }
        intsDisable();
	BLOCKED_ON_EXTERNAL(currentJThread);
        suspendOnQThread(currentJThread, 0, time);
        intsRestore();
}

/* 
 * Check whether a thread is alive.
 *
 * Note that threads executing their onstop function are not alive.
 */
int
jthread_alive(jthread *jtid)
{
        int status = true;
        intsDisable();
        if (jtid == 0
#if 0
	    /* this code makes kaffe behave like sun, but it means
	     * that Thread.join() after Thread.stop() is useless.
	     */
	    || (jtid->flags & (THREAD_FLAGS_KILLED | THREAD_FLAGS_DYING)) != 0 
#else
	    /* There seems to be a window in which death can be
	     * broadcast before it is waited for.  Basically,
	     * jthread_alive will be false immediately after
	     * Thread.stop(), unless stopping the thread was
	     * disabled.  Thread.alive() will become false as soon as
	     * the thread is on its way out.
	     */
	    || (jtid->flags & (THREAD_FLAGS_DYING | THREAD_FLAGS_EXITING))
#endif
	    || jtid->status == THREAD_DEAD)
                status = false;
        intsRestore();
        return status;
}

/*
 * Change thread priority.
 */
void
jthread_setpriority(jthread* jtid, int prio)
{
	KaffeNodeQueue** ntid;
	KaffeNodeQueue* last;
	KaffeNodeQueue* node;

	if (jtid->status == THREAD_SUSPENDED) {
		jtid->priority = (unsigned char)prio;
		return;
	}

	intsDisable();

	/* Remove from current thread list */
	last = NULL;
	node = NULL;
	for (ntid = &threadQhead[jtid->priority]; 
		*ntid != 0; 
		ntid = &(*ntid)->next) 
	{
		if (JTHREADQ(*ntid) == jtid) {
			node = *ntid;
			*ntid = node->next;
			if (*ntid == 0) {
				threadQtail[jtid->priority] = last;
			}
			break;
		}
		last = *ntid;
	}

	assert(node != NULL);

	/* Insert onto a new one */
	jtid->priority = (unsigned char)prio;
	if (threadQhead[prio] == 0) {
		threadQhead[prio] = node;
		threadQtail[prio] = node;
	}
	else {
		threadQtail[prio]->next = node;
		threadQtail[prio] = node;
	}
	node->next = NULL;

	/* If I was rescheduled, or something of greater priority was,
	 * insist on a reschedule.
	 */
	if (jtid == currentJThread || prio > currentJThread->priority) {
		needReschedule = true;
	}

	intsRestore();
}

/*
 * Stop a thread in its tracks.
 */
void
jthread_stop(jthread *jtid)
{
	intsDisable();
	/* No reason to hit a dead man over the head */
	if (jtid->status != THREAD_DEAD) {
		jtid->flags |= THREAD_FLAGS_KILLED;
	}

	/* if it's us, die */
	if (jtid == jthread_current() && 
		(jtid->flags & THREAD_FLAGS_DONTSTOP) != 0 && blockInts == 1)
		die();

        resumeThread(jtid);
	intsRestore();
}

/*
 * Have a thread exit.  This function does not return.
 */
void
jthread_exit(void)
{
	jthread* tid;
	KaffeNodeQueue *liveQ;

DBG(JTHREAD,
	dprintf("jthread_exit %p\n", currentJThread);		)

	jthread_disable_stop();
	jmutex_lock(&threadLock);

	talive--;
	if (currentJThread->daemon) {
		tdaemon--;
	}

	assert(!(currentJThread->flags & THREAD_FLAGS_EXITING));
	currentJThread->flags |= THREAD_FLAGS_EXITING;

	jmutex_unlock(&threadLock);
	jthread_enable_stop();

	/* If we only have daemons left, then we should exit. */
	if (talive == tdaemon) {
DBG(JTHREAD,
	dprintf("all done, closing shop\n");	)
		if (runOnExit != 0) {
		    runOnExit();
		}
		/* we disable interrupts while we go out to prevent a reschedule
		 * in killThread()
		 */
		intsDisable();

		for (liveQ = liveThreads; liveQ != 0; liveQ = liveQ->next) {
		        tid = JTHREADQ(liveQ);
		        /* The current thread is still on the live
			 * list, and we don't want to recursively
			 * suicide.
			 */
			if (!(tid->flags & THREAD_FLAGS_EXITING))
				killThread(tid);
		}
		EXIT(0);
	}
	/* we disable interrupts while we go out to prevent a reschedule
	 * in killThread()
	 */
	intsDisable();
	for (;;) {
		killThread(currentJThread);
		jthread_sleep(1000);
	}
}

/*
 * have main thread wait for all threads to finish
 */
void jthread_exit_when_done(void)
{
        while (talive > 1)
		jthread_yield();
	jthread_exit();
}

/*
 * Reschedule the thread.
 * Called whenever a change in the running thread is required.
 */
void
reschedule(void)
{
	int i;
	jthread* lastThread;
	int b;

	/* A reschedule in a non-blocked context is half way to hell */
	assert(intsDisabled());

	b = blockInts;

	for (;;) {
	        for (i = max_priority; i >= min_priority; i--) {
			if (threadQhead[i] == 0) 
			    continue;

			if (JTHREADQ(threadQhead[i]) != currentJThread) {
				lastThread = currentJThread;
				currentJThread = JTHREADQ(threadQhead[i]);

				{
					struct rusage ru;
					jlong ct;

					getrusage(RUSAGE_SELF, &ru);
					ct = ((jlong)ru.ru_utime.tv_sec *
					      1000000)
						+ ((jlong)ru.ru_utime.tv_usec);
					ct += ((jlong)ru.ru_stime.tv_sec *
					       1000000)
						+ ((jlong)ru.ru_stime.tv_usec);

					lastThread->totalUsed +=
						(ct - lastThread->startUsed);
					currentJThread->startUsed = ct;
				}
DBG(JTHREADDETAIL,
dprintf("switch from %p to %p\n", lastThread, currentJThread); )

				/* save and restore floating point state */
#if defined(SAVE_FP)
				SAVE_FP(lastThread->fpstate);
#endif
#if defined(CONTEXT_SWITCH)
				CONTEXT_SWITCH(lastThread, currentJThread);
#else
				if (JTHREAD_CONTEXT_SAVE(lastThread->env) == 0) {
				    lastThread->restorePoint = 
					GET_SP(lastThread->env);
				    JTHREAD_CONTEXT_RESTORE(currentJThread->env, 1);
				}
#endif
#if defined(LOAD_FP)
				LOAD_FP(currentJThread->fpstate);
#endif

				/* Restore ints */
				blockInts = b;

				assert(currentJThread == lastThread);

				/* Now handle external requests for cancelation
				 * We do not act upon them if:
				 * + The thread has the DONTSTOP flags set.
				 * + The threads is already exiting
				 */
				if ((currentJThread->flags & 
					THREAD_FLAGS_KILLED) != 0 && 
				    (currentJThread->flags & 
					THREAD_FLAGS_DONTSTOP) == 0 &&
				    (currentJThread->flags & 
					THREAD_FLAGS_EXITING) == 0 &&
				    blockInts == 1) 
				{
					die();
				}
			}
			/* Now kill the schedule */
			needReschedule = false;
			return;
		}

		/* since we set `wouldlosewakeup' first, we might write into
		 * the pipe but not go to handleIO at all.  That's okay ---
		 * all it means is that we'll return from the next select()
		 * for no reason.  handleIO will eventually drain the pipe.
		 */
		wouldlosewakeup = 1;
		if (sigPending) {
			wouldlosewakeup = 0;
			processSignals();
			continue;
		}

#if defined(DETECTDEADLOCK)
		if (tblocked_on_external == 0) {
			ondeadlock();
		}
#endif
		/* if we thought we should reschedule, but there's no thread
		 * currently runnable, reset needReschedule and wait for another
		 * event that will set it again.
		 */
		needReschedule = false;
		handleIO(true);
	}
}

/*============================================================================
 * 
 * I/O interrupt related functions
 *
 */

#if 0
static void
removeQueueFromBlockQueue(jthread *jtid, KaffeNodeQueue *queue)
{
  KaffeNodeQueue **thisQ;
  
  for (thisQ = &(jtid->blockqueue); *thisQ != 0; thisQ = &(*thisQ)->next) {
    KaffeNodeQueue *node = *thisQ;
    
    if (*((KaffeNodeQueue **)node->element) == queue) {
      *thisQ = node->next;
      KaffePoolReleaseNode(queuePool, node);
      break;
    }
  }
}
#endif

/*
 * resume all threads blocked on a given queue
 */
static void
resumeQueue(KaffeNodeQueue *queue)
{
	KaffeNodeQueue *tid;
	KaffeNodeQueue *ntid;
	
	for (tid = queue; tid != 0; tid = ntid) {
		ntid = tid->next;
		resumeThread(JTHREADQ(tid));
	}
}

/*
 * Process incoming SIGIO
 * return 1 if select was interrupted
 */
static
void
handleIO(int sleep)
{
	int r;
	/* NB: both pollarray and rd, wr are thread-local */
#if USE_POLL
	/* for poll(2) */
	int nfd;
#if DONT_USE_ALLOCA
	struct pollfd pollarray[FD_SETSIZE];	/* huge (use alloca?) */
#else
	struct pollfd *pollarray = alloca(sizeof(struct pollfd) * (maxFd+1));
#endif
#else
	/* for select(2) */
	fd_set rd;
	fd_set wr;
	struct timeval zero = { 0, 0 };
#endif
	int i, b = 0;

	assert(intsDisabled());

DBG(JTHREADDETAIL,
	dprintf("handleIO(sleep=%d)\n", sleep);		)

#if USE_POLL
	/* Build pollarray from fd_sets.
	 * This is probably not the most efficient way to handle this.
	 */
	for (nfd = 0, i = 0; i <= maxFd; i++) {
		short ev = 0;
		if (readQ[i] != 0) { 	/* FD_ISSET(i, &readsPending) */
			ev |= POLLIN;
		}
		if (writeQ[i] != 0) {   /* FD_ISSET(i, &writesPending) */
			ev |= POLLOUT;
			assert(FD_ISSET(i, &writesPending));
		}
		if (ev != 0) {
			pollarray[nfd].fd = i;
			pollarray[nfd].events = ev;
			nfd++;
		}
	}
#else
	FD_COPY(&readsPending, &rd);
	FD_COPY(&writesPending, &wr);
#endif

	/*
	 * figure out which fds are ready
	 */
retry:
	if (sleep) {
		b = blockInts;
		/* NB: BEGIN unprotected region */
		blockInts = 0;
		/* add sigpipe[0] if needed */
#if USE_POLL
		pollarray[nfd].fd = sigPipe[0];
		pollarray[nfd].events = POLLIN;
		nfd++;
#else
		FD_SET(sigPipe[0], &rd);
#endif
	}
#if USE_POLL
	r = poll(pollarray, nfd, sleep ? -1 : 0);
#else
	r = select(maxFd+1, &rd, &wr, 0, sleep ? 0 : &zero);
#endif
	/* Reset wouldlosewakeup here */
	wouldlosewakeup = 0; 

	if (sleep) {
		int can_read_from_pipe = 0;
		blockInts = b;
		/* NB: END unprotected region */

#if USE_POLL
		can_read_from_pipe = (pollarray[--nfd].revents & POLLIN);
#else
		can_read_from_pipe = FD_ISSET(sigPipe[0], &rd);
#endif
		/* drain helper pipe if a byte was written */
		if (r > 0 && can_read_from_pipe) {
			char c;

			/* NB: since "rd" is a thread-local variable, it can 
			 * still say that we should read from the pipe when 
			 * in fact another thread has already read from it.
			 * That's why we count how many bytes go in and out.
			 */
			if (bytesInPipe > 0) {
				read(sigPipe[0], &c, 1);
				bytesInPipe--;
			}
		}

		if (sigPending) {
			processSignals();
		}
	}
	if ((r < 0 && errno == EINTR) && !sleep) 
		goto retry;

	if (r <= 0)
		return;

DBG(JTHREADDETAIL,
	dprintf("Select returns %d\n", r);			)

#if USE_POLL
	for (i = 0; r > 0 && i < nfd; i++) {
		int fd;
		register short rev = pollarray[i].revents;
		if (rev == 0) {
			continue;
		}

		fd = pollarray[i].fd;
		needReschedule = true;
		r--;

		/* If there's an error, we don't know whether to wake
		 * up readers or writers.  So wake up both if so.
		 * Note that things such as failed connect attempts
		 * are reported as errors, not a read or write readiness.
		 */
		/* wake up readers when not just POLLOUT */
		if (rev != POLLOUT && readQ[fd] != 0) {
			resumeQueue(readQ[fd]);
			readQ[fd] = 0;
		}
		/* wake up writers when not just POLLIN */
		if (rev != POLLIN && writeQ[fd] != 0) {
			resumeQueue(writeQ[fd]);
			writeQ[fd] = 0;
		}
	}
#else
	for (i = 0; r > 0 && i <= maxFd; i++) {
		if (readQ[i] != 0 && FD_ISSET(i, &rd)) {
			needReschedule = true;
			resumeQueue(readQ[i]);
			readQ[i] = 0;
			r--;
		}
		if (writeQ[i] != 0 && FD_ISSET(i, &wr)) {
			needReschedule = true;
			resumeQueue(writeQ[i]);
			writeQ[i] = 0;
			r--;
		}
	}
#endif
	return;
}

/*
 * A file I/O operation could not be completed. Sleep until we are woken up
 * by the SIGIO handler.
 *
 * Interrupts are disabled on entry and exit.
 * fd is assumed to be valid.
 * Returns true if operation was interrupted.
 */
static int
blockOnFile(int fd, int op, int timeout)
{
	int rc = false;
DBG(JTHREAD,
	dprintf("blockOnFile(%d,%s)\n", fd, op == TH_READ ? "r":"w"); )

	assert(intsDisabled());
	BLOCKED_ON_EXTERNAL(currentJThread);

	if (fd > maxFd) {
		maxFd = fd;
	}
	if (op == TH_READ) {
		FD_SET(fd, &readsPending);
		rc = suspendOnQThread(currentJThread, &readQ[fd], timeout);
		FD_CLR(fd, &readsPending);
	}
	else {
		FD_SET(fd, &writesPending);
		rc = suspendOnQThread(currentJThread, &writeQ[fd], timeout);
		FD_CLR(fd, &writesPending);
	}
	return (rc);
}

/*============================================================================
 * 
 * locking subsystem
 *
 */

void 
jmutex_initialise(jmutex *lock)
{
	lock->holder = NULL;
	lock->waiting = NULL;
}

void
jmutex_lock(jmutex *lock)
{
DBG(JTHREAD,
	dprintf("jmutex_lock(%p)\n", lock); )
	intsDisable();
	jthread_current()->flags |= THREAD_FLAGS_WAIT_MUTEX;
	while (lock->holder != NULL)
		suspendOnQThread(jthread_current(), &lock->waiting, NOTIMEOUT);
	jthread_current()->flags &= ~THREAD_FLAGS_WAIT_MUTEX;

	lock->holder = jthread_current();			
	intsRestore();
}

void
jmutex_unlock(jmutex *lock)
{
DBG(JTHREAD,
	dprintf("jmutex_unlock(%p)\n", lock); )
	intsDisable();
	lock->holder = NULL;
	if (lock->waiting != 0) {