jthread.c

linux下建立JAVA虚拟机的源码KAFFE

	 *
	 * So we check periodically, every 0.2 seconds virtual time.
	 */
	if (++c % 20 == 0) {
		handleIO(false);
	}
}

/*
 * handle a SIGALRM alarm.
 */
static void 
alarmException(void)
{
	jthread* jtid;
	jlong curTime;

	/* Wake all the threads which need waking */
	curTime = currentTime();
	while (alarmList != 0 && JTHREADQ(alarmList)->time <= curTime) {
	        KaffeNodeQueue* node = alarmList;
		/* Restart thread - this will tidy up the alarm and blocked
		 * queues.
		 */
		jtid = JTHREADQ(node);
		alarmList = node->next;
		KaffePoolReleaseNode(queuePool, node);
		
		resumeThread(jtid);
	}

	/* Restart alarm */
	if (alarmList != 0) {
		MALARM(JTHREADQ(alarmList)->time - curTime);
	}
}

/*
 * print thread flags in pretty form.
 */
static char*
printflags(unsigned i)
{
	static char b[256];	/* plenty */
	struct {
		int flagvalue;
		const char *flagname;
	} flags[] = {
	    { THREAD_FLAGS_GENERAL, "GENERAL" },
	    { THREAD_FLAGS_NOSTACKALLOC, "NOSTACKALLOC" },
	    { THREAD_FLAGS_KILLED, "KILLED" },
	    { THREAD_FLAGS_ALARM, "ALARM" },
	    { THREAD_FLAGS_EXITING, "EXITING" },
	    { THREAD_FLAGS_DONTSTOP, "DONTSTOP" },
	    { THREAD_FLAGS_DYING, "DYING" },
	    { THREAD_FLAGS_BLOCKEDEXTERNAL, "BLOCKEDEXTERNAL" },
	    { THREAD_FLAGS_INTERRUPTED, "INTERRUPTED" },
	    { 0, NULL }
	}, *f = flags;

	b[0] = '\0';
	while (f->flagname) {
		if (i & f->flagvalue) {
			strcat(b, f->flagname);
			strcat(b, " ");
		}
		f++;
	}
	return b;
}

/* 
 * dump information about a thread to stderr
 */
void
jthread_dumpthreadinfo(jthread_t tid)
{
	dprintf("tid %p, status %s flags %s\n", tid, 
		tid->status == THREAD_SUSPENDED ? "SUSPENDED" :
		tid->status == THREAD_RUNNING ? "RUNNING" :
		tid->status == THREAD_DEAD ? "DEAD" : "UNKNOWN!!!", 
		printflags(tid->flags));
	if (tid->blockqueue != NULL) {
		int i;

		dprintf(" blocked");
		if (isOnList(waitForList, tid)) {
			dprintf(": waiting for children");
		}
#if 0
		/* XXX FIXME: alarmList uses nextalarm, but isOnList iterates
		 * using nextQ
		 */
		if (isOnList(alarmList, tid)) {
			dprintf(": sleeping");
		}
#endif
		for (i = 0; i < FD_SETSIZE; i++) {
			if (isOnList(readQ[i], tid)) {
				dprintf(": reading from fd %d ", i);
				break;
			}
			if (isOnList(writeQ[i], tid)) {
				dprintf(": writing to fd %d ", i);
				break;
			}
		}

#if 0
		dprintf("@%p (%p->", tid->blockqueue,
					     t = *tid->blockqueue);
		while (t && t->nextQ) {
			t = t->nextQ; 
			dprintf("%p->", t);
		}
		dprintf("|) ");
#endif
	}
}

/*
 * print info about a java thread (hopefully we'll get this to include more
 * detail, such as the actual stack traces for each thread)
 *
 */
static void
dumpJavaThreadLocal(jthread_t thread, UNUSED void *p)
{
	Hjava_lang_VMThread *tid = (Hjava_lang_VMThread *)jthread_get_data(thread)->jlThread;
	dprintf("`%s' ", nameThread(tid));
	jthread_dumpthreadinfo(thread);
	dprintf("\n");
}

static void
dumpThreadsLocal(void)
{
	dprintf("Dumping live threads:\n");
	jthread_walkLiveThreads(dumpJavaThreadLocal, NULL);
}


/*
 * handle an interrupt.
 * 
 * this function is either invoked from within a signal handler, or as the
 * result of intsRestore.
 */
static void 
handleInterrupt(int sig, SIGNAL_CONTEXT_POINTER(sc))
{
	switch(sig) {
	case SIGALRM:
		alarmException();
		break;

	case SIGUSR1:
		ondeadlock();
		break;

	case SIGUSR2:
		dumpThreadsLocal();
		break;

#if defined(SIGVTALRM)
	case SIGVTALRM:
		handleVtAlarm(sig, sc);
		break;
#endif

	case SIGCHLD:
		childDeath();
		break;

	case SIGIO:
		handleIO(false);
		break;

	default:
		dprintf("unknown signal %d\n", sig);
		exit(-1);
	}
}

/*============================================================================
 *
 * Functions related to run queue manipulation
 */


/*
 * Resume a thread running.
 * This routine has to be called only from locations which ensure
 * run / block queue consistency. There is no check for illegal resume
 * conditions (like explicitly resuming an IO blocked thread). 
 */
static void
resumeThread(jthread* jtid)
{
	KaffeNodeQueue** ntid;

DBG(JTHREAD,	dprintf("resumeThread %p\n", jtid);	);
	intsDisable();

	if (jtid->status != THREAD_RUNNING) {

		CLEAR_BLOCKED_ON_EXTERNAL(jtid);

		/* Remove from alarmQ if necessary */
		if ((jtid->flags & THREAD_FLAGS_ALARM) != 0) {
			removeFromAlarmQ(jtid);
		}
		/* Remove from lockQ if necessary */
		if (jtid->blockqueue != 0) {
		  KaffeNodeQueue **queue;

		  for (queue= &jtid->blockqueue;
		       *queue != 0;
		       queue = &(*queue)->next)
		    {
		      for (ntid = (KaffeNodeQueue **)((*queue)->element); 
			   *ntid != 0;
			   ntid = &(*ntid)->next) 
			{
			  if (JTHREADQ(*ntid) == jtid) {
			    KaffeNodeQueue *node = *ntid;
			    
			    *ntid = node->next;
			    KaffePoolReleaseNode(queuePool, node);
			    break;
			  }
			}
		    }
		    KaffePoolReleaseList(queuePool, jtid->blockqueue);
		    jtid->blockqueue = NULL;
		}

		jtid->status = THREAD_RUNNING;

		/* Place thread on the end of its queue */
		if (jtid->suspender != NULL) {
			/* Need to wait for the suspender to resume them. */
		}
		else if (threadQhead[jtid->priority] == 0) {
			threadQhead[jtid->priority] = KaffePoolNewNode(queuePool);
			threadQhead[jtid->priority]->element = jtid;
			threadQtail[jtid->priority] = threadQhead[jtid->priority];
			if (jtid->priority > currentJThread->priority) {
				needReschedule = true;
			}
		}
		else {
		        KaffeNodeQueue *queue = KaffePoolNewNode(queuePool);
			
			queue->element = jtid;
			threadQtail[jtid->priority]->next = queue;
			threadQtail[jtid->priority] = queue;
		}
	} else {
DBG(JTHREAD,		dprintf("Re-resuming %p\n", jtid); );
	}
	intsRestore();
}

/*
 * Add a new waiting queue for this thread.
 * Assert: ints must be disabled to avoid reschedule
 * while we set up the waiting queues.
 */
static void
addWaitQThread(jthread *jtid, KaffeNodeQueue **queue)
{
	KaffeNodeQueue *node;

	assert(intsDisabled()); 
	assert(queue != NULL);
	assert(jtid != NULL);
	
	/* Insert onto head of lock wait Q */
	node = KaffePoolNewNode(queuePool);
	node->next = *queue;
	node->element = jtid;
	*queue = node;
	
	/* Add the new queue to the list of registered blocking queues */
	node = KaffePoolNewNode(queuePool);
	node->next = jtid->blockqueue;
	node->element = queue;
	
	jtid->blockqueue = node;
}

static void
cleanupWaitQ(jthread *jtid)
{
	KaffeNodeQueue **ntid;
	
	if (jtid->blockqueue != 0) {
		KaffeNodeQueue **queue;
		
		for (queue= &jtid->blockqueue;
		     *queue != 0;
		     queue = &(*queue)->next)
			{
				KaffeNodeQueue **one_wait_queue = (KaffeNodeQueue **)((*queue)->element);
				
				for (ntid = one_wait_queue; 
				     *ntid != 0;
				     ntid = &(*ntid)->next) 
					{
						KaffeNodeQueue *node = *ntid;
						
						if (JTHREADQ(node) == jtid) {
							*ntid = node->next;
							KaffePoolReleaseNode(queuePool, node);
							break;
						}
					}
			}
		KaffePoolReleaseList(queuePool, jtid->blockqueue);
		jtid->blockqueue = NULL;
	}
}

/*
 * Suspend a thread on a queue.
 * Return true if thread was interrupted.
 */
static int
suspendOnQThread(jthread* jtid, KaffeNodeQueue** queue, long timeout)
{
	int rc = false;
	KaffeNodeQueue** ntid;
	KaffeNodeQueue* last;

DBG(JTHREAD,	dprintf("suspendOnQThread %p %p (%ld) bI %d\n",
	jtid, queue, timeout, blockInts); );

	assert(timeout >= 0 || timeout == NOTIMEOUT);
	assert(intsDisabled()); 

	if (timeout == 0)
		return false;

	if (jtid->status != THREAD_SUSPENDED) {
		jtid->status = THREAD_SUSPENDED;

#ifdef ENABLE_JVMPI
		FIRSTFRAME(currentJThread->localData.topFrame, 0);
#endif

		last = 0;
		for (ntid = &threadQhead[jtid->priority]; 
			*ntid != 0; 
			ntid = &(*ntid)->next) 
		{
			if (JTHREADQ(*ntid) == jtid) {
			        KaffeNodeQueue *node = (*ntid);
				
				/* Remove thread from runq */
				*ntid = node->next;
				KaffePoolReleaseNode(queuePool, node);
				if (*ntid == 0) {
					threadQtail[jtid->priority] = last;
				}

				/* Insert onto head of lock wait Q */
				if (queue != 0) {
				  addWaitQThread(jtid, queue);
				}

				/* If I have a timeout, insert into alarmq */
				if (timeout != NOTIMEOUT) {
				  addToAlarmQ(jtid, (jlong)timeout);
				}

				/* If I was running, reschedule */
				if (jtid == currentJThread) {
					reschedule();
					if (jtid->flags & THREAD_FLAGS_INTERRUPTED) {
						rc = true;
					}
				}
				break;
			}
			last = *ntid;
		}
	} else {
	DBG(JTHREAD,
		dprintf("Re-suspending %p on %p\n", jtid, *queue); );
	}
	return (rc);
}

/*
 * Kill thread.
 */
static void 
killThread(jthread *tid)
{
	KaffeNodeQueue **ntid;
	KaffeNodeQueue* last;

	intsDisable();

	/* allow thread to perform any action before being killed -
	 * such as notifying on the object 
	 */
	if (destructor1)
		(*destructor1)(tid->localData.jlThread);

DBG(JTHREAD,	
	dprintf("killThread %p kills %p\n", currentJThread, tid); );

	if (tid->status != THREAD_DEAD) {

		/* Get thread off runq (if it needs it) */
		if (tid->status == THREAD_RUNNING) {
			int pri = tid->priority;
			last = 0;

			for (ntid = &threadQhead[pri]; 
				*ntid != 0; 
				ntid = &(*ntid)->next) 
			{
				if (JTHREADQ(*ntid) == tid) {
				        KaffeNodeQueue *node = (*ntid);
					
					*ntid = node->next;
					KaffePoolReleaseNode(queuePool, node);
					if (*ntid == 0)
						threadQtail[pri] = last;	
					break;
				}
				last = *ntid;
			}
		}

		/* Run something else */
		if (currentJThread == tid) {
			needReschedule = true;
			blockInts = 1;
		}

		/* Now that we are off the runQ, it is safe to leave
		 * the list of live threads and be GCed.
		 */
		/* Remove thread from live list so it can be garbaged */
		for (ntid = &liveThreads; *ntid != 0; ntid =
			     &(*ntid)->next) { 
			if (tid == JTHREADQ(*ntid)) {
			  KaffeNodeQueue *node = (*ntid);
			  
			  (*ntid) = node->next;
			  KaffePoolReleaseNode(queuePool, node);
			  break;
			}
		}

		cleanupWaitQ(tid);

		/* Dead Jim - let the GC pick up the remains */
		tid->status = THREAD_DEAD;
	}

	intsRestore();
}



/*============================================================================
 *
 * Functions dealing with thread contexts and the garbage collection interface
 *
 */

/*
 * Allocate a new thread context and stack.
 */
static jthread*
newThreadCtx(size_t stackSize)
{
	jthread *ct;

	ct = KGC_malloc(threadCollector, sizeof(jthread) + 16 + stackSize, KGC_ALLOC_THREADCTX);
	if (ct == 0) {
		return 0;
	}

	KGC_addRef(threadCollector, ct);
#if defined(__ia64__)
	/* (gb) Align jmp_buf on 16-byte boundaries */
	ct = (jthread *)((((unsigned long)(ct)) & 15) ^ (unsigned long)(ct));
#endif
	ct->stackBase = (ct + 1);
	ct->stackEnd = (char *) ct->stackBase + stackSize;
	ct->restorePoint = ct->stackEnd;
	ct->status = THREAD_SUSPENDED;

DBG(JTHREAD,
	dprintf("allocating new thread, stack base %p-%p\n", 
	    ct->stackBase, ct->stackEnd); );

	return (ct);
}

/*
 * free a thread context and its stack
 */
void    
jthread_destroy(jthread *jtid)
{
	KaffeNodeQueue *x;

	if (DBGEXPR(JTHREAD, true, false)) {
		for (x = liveThreads; x; x = x->next)
			assert(JTHREADQ(x) != jtid);
	}
	/* We do not free explicitly as it should be done by the GC. */
	KGC_rmRef(threadCollector, jtid);
}

/*
 * iterate over all live threads
 */
void
jthread_walkLiveThreads(void (*func)(jthread_t,void*), void *priv)
{
        KaffeNodeQueue* liveQ;

        for (liveQ = liveThreads; liveQ != NULL; liveQ = liveQ->next) {
                func(JTHREADQ(liveQ), priv);
        }
}

void
jthread_walkLiveThreads_r(void (*func)(jthread_t,void*), void *priv)
{
	jthread_walkLiveThreads(func, priv);
}

/*
 * determine the interesting stack range for a conservative gc
 */
int
jthread_extract_stack(jthread *jtid, void **from, unsigned *len)
{
	assert(jtid != NULL);
#if defined(STACK_GROWS_UP)
	*from = jtid->stackBase;
	*len = jtid->restorePoint - jtid->stackBase;
#else   
	*from = jtid->restorePoint;
	*len = (char *) jtid->stackEnd - (char *) jtid->restorePoint;
#endif
	return (1);
}

/* 
 * XXX this is supposed to count the number of stack frames 
 */
int
jthread_frames(jthread *thrd UNUSED)
{
        return (0);
}

void jthread_suspend(jthread_t jt, void *suspender)
{
	assert(jt != jthread_current());
	
	intsDisable();
	if( jt->suspender == suspender )
	{