condition.c

pebble

/* 
 * Copyright 1999, 2000, 2001, 2002 Lucent Technologies Inc.
 * All Rights Reserved.
 * Information Sciences Research Center, Bell Labs.
 *
 * LUCENT TECHNOLOGIES DOES NOT CLAIM MERCHANTABILITY OF THIS SOFTWARE 
 * OR THE SUITABILITY OF THIS SOFTWARE FOR ANY PARTICULAR PURPOSE. The
 * software is provided "as is" without expressed or implied warranty 
 * of any kind.
 *
 * These notices must be retained in any copies of any part of this
 * software.
 *
 */

/*
 * User-level condition manager.
 * The conditions mananger keeps the list of waiting threads on each condition.
 *
 * *** important ***
 * The condition manager must be initialized before any use of the conditions.
 */

#include <unistd.h>		/* for write() */
#include "diag.h"
#include "pebble.h"
#include "mem.h"
#include "synch.h"
#include "lock_free.h"
#include "signal.h"
#include "nucleus.h"
#include <assert.h>

enum	{
	N = 257,	/* number of bins in hash table */
};

/* per thread condition information */
typedef	struct ThreadCond ThreadCond;

/* the thread information is linked into two hash tables: */
/* the threads list, which is pointed by thread_table[], */
/* and the condition list, which is pointed by cond_table[] */
/* the thread information is always a member of the threads list and it */
/* is never removed from this list for the lifetime of the thread. */
/* The thread information is added and removed from the condition list */
/* as the thread is waiting on various conditions */
/* Note: there is a single element per thread, since a given thread */
/* may wait on a single condition at any given time */

struct	ThreadCond {
	void	*tid;		/* the thread */
	int	asid;		/* the protection domain that holds the */
				/* condition variable if it is allocated */
				/* in private memory. (not sharable) */
				/* zero indicates that the condition variable */
				/* is on the heap */
	void	*addr;		/* the address of the condition variable */
	int	sem;		/* the thread's semaphore */
	int	npost;		/* number of outstanding post requests on */
				/* semaphore */
	ThreadCond *next_thread;/* next thread in threads hash list */
	ThreadCond *prev_cond;	/* previous thread in condition list */
	ThreadCond *next_cond;	/* next thread in condition list */
};

/* doubly-linked list of threads */
typedef	struct	{
	Lock	lock;
	ThreadCond	*head, *tail;
} List;


static List thread_table[N];
static List cond_table[N];


static const char INIT[] = "condition initialization failed";
static const char NO_INIT[] = "condition not initialized";
static const char BAD_INIT[] = "condition initialized for a different variable";
static const char INV_TID[] = "thread not found";


/* add the thread condition to the tail of the specified conditions list */
/* assumption: the conditions list is locked */
void
add(int bin, ThreadCond *p)
{
	if (cond_table[bin].tail == NULL) {
		cond_table[bin].head = p;
		p->prev_cond = NULL;
	} else	{
		cond_table[bin].tail->next_cond = p;
		p->prev_cond = cond_table[bin].tail;
	}
	cond_table[bin].tail = p;
	p->next_cond = NULL;
}


/* remove the thread condition from the specified conditions list */
/* assumption: the conditions list is locked */
void
delete(int bin, ThreadCond *p)
{
	/* detect duplicate removal */
	if (p->prev_cond == NULL && p->next_cond == NULL &&
	    cond_table[bin].head != p)
		return;

	if (p->prev_cond == NULL)
		cond_table[bin].head = p->next_cond;
	else
		p->prev_cond->next_cond = p->next_cond;

	if (p->next_cond == NULL)
		cond_table[bin].tail = p->prev_cond;
	else
		p->next_cond->prev_cond = p->prev_cond;

	p->prev_cond = p->next_cond = NULL;
}


int
sys_cond_mgr_init(int asid, void *addr)
{
	void *tid;
	int bin;
	ThreadCond *p;

	tid = (void *)get_thread();
	if (on_heap(addr))
		asid = 0;

	DIAG(COND_DIAG, ("cond_mgr_init: tid=%p asid=%d addr=%p\n",
		tid, asid, addr));
	
	/* find the element corresponding to the thread */
	bin = (((ulong)tid)/sizeof(double)) %  N;
	mutex_lock(&thread_table[bin].lock);
	for (p = thread_table[bin].head; p != NULL && p->tid != tid;
	     p = p->next_thread)
		;

	if (p == NULL) {
		/* no element for the thread; must allocate a new one */
		if ((p = (ThreadCond *)calloc(1, sizeof(ThreadCond))) == NULL) {
			mutex_unlock(&thread_table[bin].lock);
			error(INIT);
		}
		DIAG(COND_DIAG,
			("cond_mgr_init: allocated new element %p for tid=%p\n",
			p, tid));
		p->tid = tid;
		if ((p->sem = sem_create(0)) < 0) {
			mutex_unlock(&thread_table[bin].lock);
			free(p);
			error(INIT);
		}

		/* add new element to the head of the threads list */
		p->next_thread = thread_table[bin].head;
		thread_table[bin].head = p;
		mutex_unlock(&thread_table[bin].lock);
		DIAG(COND_DIAG,
		("sys_cond_mgr_init: add element bin=%d p=%p p->tid=%p\n",
			bin, p, p->tid));
	} else	{
		/* found an element for the thread */
		mutex_unlock(&thread_table[bin].lock);

		DIAG(COND_DIAG,
			("cond_mgr_init: found element %p for tid=%p\n",
			p, tid));
		/* make sure that the semaphore value is zero */
		/* it could be that cond_signal() was sent after the thread */
		/* woke already from cond_twait() */
		while (p->npost > 0) {
			DIAG(COND_DIAG,
		("cond_mgr_init: thread %p sem_wait_caller_pri count=%d\n",
				tid, p->npost));
			if (sem_wait_caller_pri(p->sem) < 0)
				panic("sem_wait_caller_pri:");
			fadd(&p->npost, -1);
		}
	}

	/* place new condition information in element */
	p->asid = asid;
	p->addr = addr;

	/* add element to conditions list */
	bin = (((ulong)addr)/sizeof(int)) %  N;
	mutex_lock(&cond_table[bin].lock);
	add(bin, p);
	mutex_unlock(&cond_table[bin].lock);

	DIAG(COND_DIAG, ("cond_mgr_init: tid=%p addr=%p before return\n",
		tid, addr));
	
	return 0;
}


int
sys_cond_mgr_twait(int asid, void *addr, Time when)
{
	void *tid;
	int bin;
	ThreadCond *p;
	int code;


	tid = (void *)get_thread();
	if (on_heap(addr))
		asid = 0;

	DIAG(COND_DIAG,
		("cond_mgr_twait: asid=%d addr=%p tid=%p when=%08x %08x\n",
		asid, addr, tid, (int)(when>>32), (int)when));
	
	/* find the element corresponding to the thread */
	bin = (((ulong)tid)/sizeof(double)) % N;
	for (p = thread_table[bin].head; p != NULL && p->tid != tid;
	     p = p->next_thread)
		;

	if (p == NULL) {
		printf("conditions manager before %s error: bin=%d tid=%p\n",
			NO_INIT, bin, tid);
		for (p = thread_table[bin].head; p != NULL; p = p->next_thread)
			printf("p=%p p->tid=%p p->asid=%d p->addr=%p\n",
				p, p->tid, p->asid, p->addr);
		error(NO_INIT);
	}

	if (p->asid != asid || p->addr != addr)
		error(BAD_INIT);

	code = sem_twait(p->sem, when);
	fadd(&p->npost, -1);

	if (code < 0) {
		/* remove element from condition list, because timeout */
		/* occured first */
		DIAG(COND_DIAG,
		("cond_mgr_twait: removing element %p from conditions list\n",
			p));

		bin = (((ulong)addr)/sizeof(int)) % N;
		mutex_lock(&cond_table[bin].lock);
		delete(bin, p);
		mutex_unlock(&cond_table[bin].lock);
	}

	DIAG(COND_DIAG, ("cond_mgr_twait: tid=%p return code=%d\n",
		tid, code));
	
	return code;
}


int
sys_cond_mgr_post(int asid, void *addr, void *tid)
{
	int bin;
	ThreadCond *p, *next_p;

	DIAG(COND_DIAG, ("cond_mgr_post: asid=%d addr=%p tid=%p\n",
		asid, addr, tid));

	if (on_heap(addr))
		asid = 0;

	bin = (((ulong)addr)/sizeof(int)) %  N;
	mutex_lock(&cond_table[bin].lock);
	if (tid == (void *)0) {
		/* wakeup first waiting thread */
		for (p = cond_table[bin].head; p != NULL; p = p->next_cond)
			if (p->asid == asid && p->addr == addr) {
				DIAG(COND_DIAG,
					("cond_mgr_post: waking thread %p\n",
					p->tid));
				fadd(&p->npost, 1);
				sem_post(p->sem);
				delete(bin, p);
				break;
			}
	} else if (tid == (void *)-1) {
		/* wakeup all waiting threads */
		for (p = cond_table[bin].head; p != NULL; p = next_p) {
			DIAG(COND_DIAG, ("cond_mgr_post: p=%p\n", p));
			next_p = p->next_cond;
			if (p->asid == asid && p->addr == addr) {
				DIAG(COND_DIAG,
					("cond_mgr_post: waking thread %p\n",
					p->tid));
				fadd(&p->npost, 1);
				sem_post(p->sem);
				delete(bin, p);
			}
		}
	} else	{
		/* wakeup a specific thread */
		for (p = cond_table[bin].head; p != NULL; p = p->next_cond)
			if (p->asid == asid && p->addr == addr && p->tid ==tid){
				DIAG(COND_DIAG,
					("cond_mgr_post: waking thread %p\n",
					p->tid));
				fadd(&p->npost, 1);
				sem_post(p->sem);
				delete(bin, p);
				break;
			}

		if (p == NULL) {
			mutex_unlock(&cond_table[bin].lock);
			error(INV_TID);
		}
	}
	mutex_unlock(&cond_table[bin].lock);

	DIAG(COND_DIAG, ("cond_mgr_post(%p, %p) returns\n", addr, tid));
	return 0;
}


/* conditions manager initialization */
int main()
{
	printf("conditions manager is active\n");

	/* verify that we are running in user mode with interrupts enabled */
	if (!check_psw(1,1)) {
		DIAG(LOCK_DIAG,
			("signals: invalid processor status: %08lx\n",
			get_psw()));
		task_exit(1);
	}

	if (portal_create(SYS_COND_MGR_INIT, "spaiii", 0,
	    (Func) sys_cond_mgr_init, 0) < 0)
		panic("portal_create for cond_mgr_init():");
	if (portal_create(SYS_COND_MGR_TWAIT, "spaiii", 0,
	    (Func) sys_cond_mgr_twait, 0) < 0)
		panic("portal_create for cond_mgr_twait():");
	if (portal_create(SYS_COND_MGR_POST, "spaiii", 0,
	    (Func) sys_cond_mgr_post, 0) < 0)
		panic("portal_create for cond_mgr_post():");

	DIAG(COND_DIAG, ("before returning to initialization code\n"));

	/*
	 * return to initialization code.
	 * cannot just "return", since the startup code (crt0.S) calls
	 * exit when main routine terminates.
	 */
	call_portal(SYS_RTN_RPC);
	return(1);
}