/*
 * jam 841026-29-30-31-1101
 * sas 861005
 *	Aaargh!  Rewrote most of the code dealing with state descriptors
 *	and allocation/de-allocation thereof.  The point is that the
 *	ABORT state is being botched (in a subtle manner) because of
 *	the way that the state descriptors are being (mis-)managed.
 *	If the comments here sound rather bitter, well, it's just
 *	that a fundamental grounding in data structures by the implementors
 *	could have saved a lot of grief for myself and for the various
 *	Valid customers that have bashed into these problems.
 */

#include "../h/param.h"
#include "../h/time.h"
#include "../h/kernel.h"
#include "../h/errno.h"
#include "../h/systm.h"
#include "../vnet/vnet.h"
#include "../conn/conn.h"
#include "../rpc/rpc.h"
#include "../s32/kstat.h"
#include "../rpc/rpc_stat.h"

extern connection_t *conn_findByAddress();
extern rpc_header_t *rpc_buildHeader();

#define NSCBINS		32		/* Size of server class hash table */
#define NSSBINS		32		/* Size of server state hash table */

#define SCHASH(class)	 &rpc_scbins[binarymod(class, NSCBINS)]
#define SSHASH(clientId) &rpc_ssbins[binarymod(clientId, NSSBINS)]

rpc_hash_t rpc_scbins[NSCBINS];		/* Server class hash table */
rpc_hash_t rpc_ssbins[NSSBINS];		/* Server state descr hash table */

/* Free server state descriptors */
rpc_hash_t rpc_sFreeStates;
rpc_hash_t *rpc_serverFreeStates = &rpc_sFreeStates;

long	rpc_nextReturnRetry = ETERNITY;	/* Next time retransmitter will run */
long	rpc_nextCleanupTime = ETERNITY;	/* Next time cleanup daemon will run */

#define rpc_schedCleanup(sstate) \
	(sstate)->time = currentTime; \
	if (timeUntil(rpc_nextCleanupTime) > RPC_SERVERCLEANTIME) { \
		untimeout(rpc_serverDaemon, 0); \
		rpc_nextCleanupTime = currentTime + RPC_SERVERCLEANTIME; \
		timeout(rpc_serverDaemon, 0, RPC_SERVERCLEANTIME); \
	} \
	RPC_DECMSTAT(callsServiced);

rpc_serverState_t *rpc_sState();
rpc_serverState_t *rpc_sStateBySeqno();
rpc_serverState_t *rpc_sStateNoAbort();
rpc_serverState_t *rpc_sState1();
rpc_serverState_t *rpc_newServerState();
rpc_serverClass_t *rpc_findServerClass();
int rpc_returnRetransmitter();
int rpc_serverDaemon();
int rpc_serverException();

/*
 * Initialize the hash tables
 */
rpc_initServerHash()
{
	register rpc_hash_t *h;
	for (h = rpc_scbins; h < &rpc_scbins[NSCBINS]; h++)
		h->forw = h->tail = (caddr_t)h;
	for (h = rpc_ssbins; h < &rpc_ssbins[NSSBINS]; h++)
		h->forw = h->tail = (caddr_t)h;
	rpc_serverFreeStates->forw = (caddr_t)rpc_serverFreeStates;
	rpc_serverFreeStates->tail = (caddr_t)rpc_serverFreeStates;

}

rpc_serverInit()
{
	return(conn_attach(rpc_serverException));
}

/*
 * Consumer of call packets.  Finds the appropriate
 * function to handle calls and switches out to it.
 * Duplicate calls are dropped.  Must be called
 * with rpc interrupts locked out.
 */
rpc_callReceived(pkt)
   register rpc_header_t *pkt;
{
	register rpc_serverState_t *sstate;

	/*
	 * NOTE -- My rewrite of this section of code means that we
	 * will be looking through the state bins twice for each
	 * new connection.  While this is wasteful, I am willing
	 * to leave it for three reasons:
	 *	(1)  The bins rarely have more than one entry.
	 *	(2)  I would rather get this to work (properly) and then
	 *	     think about optimization.
	 *	(3)  This protocol is complete baloney anyways and the
	 *	     (currently scheduled) implementation of NFS will
	 *	     obviate the need for this code.
	 * sas 861004
	 */
	 
	if ((sstate = rpc_sStateBySeqno(&pkt->source,
					pkt->client,
					pkt->seqno)) == NULL) {
		
		/*
		 * We do not have any state for this connection -- find
		 * out if there is state for any connection from this
		 * client proc.
		 */
		if ((sstate = rpc_sStateNoAbort(&pkt->source,
						pkt->client)) == NULL){
			/*
			 * No state from this client proc ==> a new
			 * connection is created
			 */
			if ((sstate = rpc_newServerState(&pkt->source,
						pkt->client,
						pkt->seqno)) == NULL) {
				/*
				 * Whoops -- no more space in camap.
				 * Bummer.
				 */
				register connection_t *conn =
					conn_findByAddress(&pkt->source);
				
				printf("rpc_callReceived/E: no space\n");
				rpc_reply(pkt, conn, EREMOTERESOURCE);
				return;
			} else {
				goto nextlayer;
			}
		} else {
			/*
			 * We make it here if there is state from the
			 * client proc laying around...but it's not
			 * the same sequence number as the request
			 * that we just received.
			 *
			 * Also, the state is guaranteed NOT to be an
			 * ABORT state -- which forms the solution
			 * to the "Fluke Bug".
			 */
			 
			/*
			 * If this is a request with a sequence number
			 * older than the one we're currently working
			 * on, drop the packet.
			 *
			 * Implications abound as to what happens if
			 * the client botches its notion of the current
			 * sequence number.  It is an excercise left
			 * to the reader.
			 */
			if (pkt->seqno < sstate->seqno) {
				rpc_freePkt(pkt);
				return;
			}
			 
			/*
			 * If the server is working on an older call
			 * immediately abort the server state.  If the server
			 * aborts immediately and calls return or error,
			 * then he will end up leaving the aborted state
			 * and the call will be serviced below.
			 *
			 * ==> VERIFY THIS CLAIM <==
			 */
			if (sstate->state == RPCSTATE_WORKING) {
				sstate->state = RPCSTATE_ABORTED;
				rpc_classAbort(sstate->classp,
						sstate->conn,
						sstate->clientId,
						sstate->seqno);
				RPC_INCSTAT(callsAborted);
			}
			
			/*
			 * If the server was still re-xmiting a previous
			 * return packet then we free the return packet
			 * and set the state to idle so the call will
			 * serviced below.
			 */
			else if (sstate->state == RPCSTATE_RETURNED) {
				sstate->state = RPCSTATE_IDLE;
				rpc_freePkt(sstate->pkt);
				RPC_DECMSTAT(returnPackets);
				sstate->pkt = NULL;
				RPC_DECMSTAT(callsServiced);
			}
		}
	} else {
		/*
		 * We make it here if we've found a state in the bins which
		 * corresponds (sequence-number-wise) to the pakcet
		 * which we've received.
		 */
		
		/*
		 * If the server is currently working on the same call,
		 * send a call ack to the client will stop sending the
		 * request.
		 * 
		 * I think that this is the most broken part of the
		 * protocol.  We probably would want the client
		 * to continue re-xmiting so that it can gracefully
		 * clean up.  But, hey.  What do I know?
		 */
		if (sstate->state == RPCSTATE_WORKING) {
			RPC_INCSTAT(dupCallsAck);
			rpc_reply(pkt, sstate->conn, RPCPKT_CALLACK);
			return;
		}
		
		/*
		 * If the server has already sent a return to this call then
		 * retransmit the return and drop this call.
		 */
		else if (sstate->state = RPCSTATE_RETURNED) {
			RPC_INCSTAT(dupCalls);
			rpc_freePkt(pkt);
			rpc_output(sstate->pkt, sstate->conn);
			sstate->time = currentTime;
			return;
		}
	}
	
nextlayer:					
	/*
	 * If the server for this client is idle (or
	 * was made so above) and this is a new call,
	 * then service it by passing it to the server of 
	 * the appropriate class
	 */
	if (sstate->state == RPCSTATE_IDLE && pkt->seqno > sstate->seqno) {
		sstate->clientgen = pkt->sourcegen;
		sstate->seqno = pkt->seqno;
	
		if ((sstate->classp = rpc_findServerClass(pkt->class))
								== NULL) {
			rpc_replyError(pkt, sstate->conn, EREMOTECLASS);
			RPC_INCSTAT(unknownClass);
			return;
		}
				 
	
		/*
		 * Everything seems fine.  Send the call
		 * up to the higher level.
		 */
		sstate->state = RPCSTATE_WORKING;
		sstate->operation = pkt->operation;
		/* BUG if the contents begin in a different mbuf than the */
		/* BUG rpc header then the header mbuf will never be freed */
		rpc_classCall(sstate->classp, sstate->conn, sstate->clientId,
			      sstate->classp, pkt->operation,
			      rpc_discardHeader(pkt), sstate->seqno);
		RPC_INCMSTAT(callsServiced);
	} else
		rpc_freePkt(pkt);
}


/*
 * The server calls this function with the
 * return results to be sent to the client.
 */
rpc_return(clientConn, clientId, results, cookie)
   connection_t *clientConn;
   u_long clientId;
   caddr_t results;
   u_long cookie;
{
	rpc_serverState_t *sstate;
	rpc_header_t *pkt;
	long retryTime;
	int s = RPC_SPL();

	if ((sstate = rpc_sStateBySeqno(&clientConn->node,
			clientId,
			cookie)) == NULL) {
		rpc_serverBug("return for unknown client", 0, 0);
		rpc_freeResults(results);
		splx(s);
		return;
	}

	/*
	 * If the call has been aborted then just
	 * drop the return packet and set the state
	 * to idle.
	 */
	if (sstate->state == RPCSTATE_ABORTED) {
		sstate->state = RPCSTATE_IDLE;
		rpc_schedCleanup(sstate);
		rpc_freeResults(results);
		splx(s);
		return;
	}

	/*
	 * If the server is not in the working state
	 * then there is something seriously wrong
	 * and we print a message.
	 *
	 * stupid -- doesn't consider case in which we have allocated
	 * in the process if "finding".
	 */
	if (sstate->state != RPCSTATE_WORKING) {
		rpc_serverBug("return from bad state", sstate, 0);
		rpc_freeResults(results);
		splx(s);
		return;
	}

	pkt = rpc_buildHeader(results);
	RPC_INCMSTAT(returnPackets);
	pkt->client = sstate->clientId;
	pkt->seqno = sstate->seqno;
	pkt->clientgen = sstate->clientgen;
	pkt->type = RPCPKT_RETURN;
	pkt->class = sstate->classp->class;
	pkt->operation = sstate->operation;

	sstate->state = RPCSTATE_RETURNED;
	sstate->pkt = pkt;
	sstate->retries = 0;
	sstate->time = currentTime;
	rpc_output(pkt, sstate->conn);

	/*
	 * Schedule the return retransmitter to run at the
	 * time our call will next be retried if it is
	 * not already going to run sooner.
	 */
	rpc_setRetryInfo(sstate->classp->failrate, sstate->classp->maxwait,
			 &sstate->maxretries, &sstate->retryhz);
	retryTime = rpc_retryTime(sstate);
	if (timeUntil(rpc_nextReturnRetry) > retryTime)
	{
		untimeout(rpc_returnRetransmitter, 0);
		rpc_nextReturnRetry = retryTime + currentTime;
		timeout(rpc_returnRetransmitter, 0, retryTime*hz);
	}

	splx(s);
}

/*
 * The server calls this function with an
 * error to be sent to the client.
 */
rpc_error(clientConn, clientId, error, cookie)
   connection_t *clientConn;
   u_long clientId;
   int error;
   u_long cookie;
{
	rpc_serverState_t *sstate;
	rpc_header_t *pkt;
	long retryTime;
	int s = RPC_SPL();

	if ((sstate = rpc_sStateBySeqno(&clientConn->node,
					clientId,
					cookie)) == NULL) {
		rpc_serverBug("error for unknown client", 0, 0);
		splx(s);
		return;
	}

	/*
	 * If the call has been aborted then just
	 * drop set the state to idle.
	 */
	if (sstate->state == RPCSTATE_ABORTED) {
		sstate->state = RPCSTATE_IDLE;
		rpc_schedCleanup(sstate);
		splx(s);
		return;
	}

	/*
	 * If the server is not in the working state
	 * then there is something seriously wrong
	 * and we print a message.
	 */
	if (sstate->state != RPCSTATE_WORKING) {
		rpc_serverBug("error from bad state", sstate, 0);
		splx(s);
		return;
	}

	pkt = rpc_buildHeader(0);
	RPC_INCMSTAT(returnPackets);
	pkt->client = sstate->clientId;
	pkt->seqno = sstate->seqno;
	pkt->clientgen = sstate->clientgen;
	pkt->type = RPCPKT_ERROR;
	pkt->class = sstate->classp->class;
	pkt->operation = error;

	sstate->state = RPCSTATE_RETURNED;
	sstate->pkt = pkt;
	sstate->retries = 0;
	sstate->time = currentTime;
	rpc_output(pkt, sstate->conn);

	/*
	 * Schedule the return retransmitter to run at the
	 * time our call will next be retried if it is
	 * not already going to run sooner.
	 */
	rpc_setRetryInfo(sstate->classp->failrate, sstate->classp->maxwait,
			 &sstate->maxretries, &sstate->retryhz);
	retryTime = rpc_retryTime(sstate);
	if (timeUntil(rpc_nextReturnRetry) > retryTime)
	{
		untimeout(rpc_returnRetransmitter, 0);
		rpc_nextReturnRetry = retryTime + currentTime;
		timeout(rpc_returnRetransmitter, 0, retryTime*hz);
	}

	splx(s);
}

/*
 * The return retransmitter is scheduled to
 * run by rpc_return() and by itself.  It
 * is always scheduled to run at the next time
 * a return needs to be retransmitted.
 */
/*ARGSUSED*/
rpc_returnRetransmitter(arg)
   int arg;
{
	rpc_hash_t *ssbin;
	rpc_serverState_t *sstate;
	long minRetryTime = ETERNITY;
	int s = RPC_SPL();
	/* BUG we would prefer the RPC_SPL above to be inside the loops */

	RPC_INCSTAT(returnDaemon);
	for (ssbin = rpc_ssbins; ssbin < &rpc_ssbins[NSSBINS]; ++ssbin)
	  for (sstate = (rpc_serverState_t *)ssbin->forw;
	    sstate != (rpc_serverState_t *)ssbin;
	    sstate = sstate->forw) {

		/*
		 * If the server is no longer returning then
		 * go on to the next server.
		 */
		if (sstate->state != RPCSTATE_RETURNED)
			continue;

		/*
		 * The return's retry time has been reached.
		 * Either retransmit the return or drop it if
		 * it has timed out.
		 */
		if (currentTime >= sstate->time + rpc_retryTime(sstate)) {

			/*
			 * The return has not exceeded its maximum
			 * retry count.  Send the return packet again.
			 */
			if (sstate->retries < rpc_maxRetries(sstate)) {
#ifndef NO_RPC_STAT
				rpc_statlist.returnRetriesCur.count =
								sstate->retries;
				RPC_INCMSTAT(returnRetries);
#endif NO_RPC_STAT
				++sstate->retries;
				sstate->time = currentTime;
				rpc_output(sstate->pkt, sstate->conn);
			}

			/*
			 * The return has exceeded its maximum number of
			 * retries.  Free the whole thing up.
			 */
			else {
				RPC_INCSTAT(lostReturns);
				sstate->state = RPCSTATE_IDLE;
				rpc_freePkt(sstate->pkt);
				RPC_DECMSTAT(returnPackets);
				sstate->pkt = NULL;
				rpc_schedCleanup(sstate);
				continue;
			}
		}

		/*
		 * Find the shortest amount of time until
		 * the next return retry so we can reschedule
		 * the return retransmitter.
		 */
		if (sstate->state == RPCSTATE_RETURNED) {
			long retryTime = rpc_retryTime(sstate);

			if (retryTime < minRetryTime)
				minRetryTime = retryTime;
		}

	}

	splx(s);

	/*
	 * Reschedule the return retransmitter to run
	 * later if there are any clients still in
	 * the returned state.  If there aren't any
	 * then the retransmitter will be reshceduled
	 * at the next return.
	 */
	if (minRetryTime != ETERNITY) {
		timeout(rpc_returnRetransmitter, 0, minRetryTime*hz);
		rpc_nextReturnRetry = currentTime + minRetryTime;
	}
	else
		rpc_nextReturnRetry = ETERNITY;
}

/*
 * Return acknowledge consumer.  A return
 * ack stops retransmission of a return.
 * Must be called with rpc interrupts
 * locked out.
 */
rpc_returnAck(pkt)
   rpc_header_t *pkt;
{
	rpc_serverState_t *sstate;

	RPC_INCSTAT(returnAcksRecvd);
	if (sstate = rpc_sState(&pkt->source, pkt->client)) {

		/*
		 * If the server was retransmitting return
		 * packets then free the return packet and
		 * make the server idle.
		 */
		if (sstate->state == RPCSTATE_RETURNED &&
		    pkt->seqno == sstate->seqno) {
			sstate->state = RPCSTATE_IDLE;
			rpc_freePkt(sstate->pkt);
			RPC_DECMSTAT(returnPackets);
			sstate->pkt = NULL;
			rpc_schedCleanup(sstate);
		}

		/*
		 * We should never get a return acknowledge
		 * unless we have sent the return already.
		 */
		else if ((sstate->state == RPCSTATE_WORKING ||
			  sstate->state == RPCSTATE_ABORTED) &&
		         pkt->seqno == sstate->seqno)
			rpc_serverBug("return ack, but no return sent", sstate, pkt);
	}

	rpc_freePkt(pkt);
}

/*
 * The server daemon frees server state
 * descriptors which have been idle for a
 * long time.  This is the only way a server
 * state descriptor is freed.
 *
 * State descriptor deallocation is the flaming hub cap of
 * this "Plan 9 From Outer Space" sequel.
 */
/*ARGSUSED*/
rpc_serverDaemon(arg)
   int arg;
{
	rpc_hash_t *ssbin;
	rpc_serverState_t *sstate;
	long minRetryTime = ETERNITY;
	rpc_serverState_t *tmpsstate;
	int s = RPC_SPL();
	/* BUG we would prefer the RPC_SPL above to be inside the loops */

	RPC_INCSTAT(serverDaemon);
	for (ssbin = rpc_ssbins; ssbin < &rpc_ssbins[NSSBINS]; ++ssbin)
	  for (sstate = (rpc_serverState_t *)ssbin->forw; 
	    sstate != (rpc_serverState_t *)ssbin;
	    sstate = (tmpsstate ? tmpsstate : sstate->forw)) {
	  	
		tmpsstate = (rpc_serverState_t *)0;

		/*
		 * If the server is not idle then go on
		 * to the next server.
		 */
		if (sstate->state != RPCSTATE_IDLE &&
				sstate->state != RPCSTATE_ABORTED)
			continue;
						
		/*
		 * If the server has been idle long enough
		 * then free the server descriptor.
		 */
		if (currentTime >= sstate->time + RPC_SERVERCLEANTIME) {
			tmpsstate = sstate->forw;
			rpc_freeServerState(sstate);
		}

		/*
		 * Otherwise find the minimum time to wait
		 * until another server descriptor can be
		 * freed.
		 */
		else {
			long retryTime = rpc_retryTime(sstate);

			if (retryTime < minRetryTime)
				minRetryTime = retryTime;
		}

	}

	splx(s);

	/*
	 * Reschedule the server daemon to run
	 * later if there are any servers still in
	 * the idle state.  If there aren't any
	 * then the daemon will be reshceduled
	 * next time a server becomes idle.
	 */
	if (minRetryTime != ETERNITY) {
		timeout(rpc_serverDaemon, 0, minRetryTime*hz);
		rpc_nextCleanupTime = currentTime + minRetryTime;
	}
	else
		rpc_nextCleanupTime = ETERNITY;
}

/*
 * Called from the connection manager when
 * a node changes state (come up, crashes,
 * etc.).
 */
rpc_serverException(conn, exception)
   connection_t *conn;
   int exception;
{
	rpc_hash_t *ssbin;
	rpc_serverState_t *sstate;
	rpc_hash_t *scbin;
	rpc_serverClass_t *classp;
	rpc_serverState_t *tmpsstate;
	int breakall = 0;
	int s;

	/*
	 * If another node comes up we don't really
	 * care much, although a higher layer might.
	 * If our node comes up then we enable rpc.
	 */
	if (exception == CONN_UP && nodecmp(conn->node, myNode)) {
		/* BUG enable rpc */
	}

	if (exception != CONN_DOWN)
		goto handlers;

	/*
	 * If this node is shutting down then we will
	 * terminate all outstanding calls.
	 */
	if (exception == CONN_DOWN && nodecmp(conn->node, myNode)) {
		/* BUG disable rpc */
		breakall = 1;
	}

	s = RPC_SPL();
	/* BUG we would prefer the RPC_SPL above to be inside the loops */

	/*
	 * Either we are shutting down or a client
	 * has crashed.  For each server with an
	 * outstanding call on the affected node(s)
	 * abort the server.  All servers for
	 * the affected node(s) will be left in the
	 * idle state when we are through.
	 */
	for (ssbin = rpc_ssbins; ssbin < &rpc_ssbins[NSSBINS]; ++ssbin)
	  for (sstate = (rpc_serverState_t *)ssbin->forw;
	    sstate != (rpc_serverState_t *)ssbin;
	    sstate = tmpsstate) {

		/*
		 * If we are not aborting all calls, and
		 * if we are not aborting calls on the node
		 * for the particular client then go on to
		 * the next descriptor.
		 */
		tmpsstate = sstate->forw;
		
		if (!breakall && !nodecmp(sstate->conn->node, conn->node))
			continue;

		rpc_removeFromTable(ssbin, sstate);

		/*
		 * If the server is working on a call then
		 * abort the call.

		 */
		if (sstate->state == RPCSTATE_WORKING) {
			sstate->state = RPCSTATE_ABORTED;
			rpc_classAbort(sstate->classp, sstate->conn,
					sstate->clientId, sstate->seqno);
			RPC_INCSTAT(callsAborted);
		}

		/*
		 * If the server has sent a return the free
		 * the return packet and put the server
		 * in the idle state.
		 */
		else if (sstate->state == RPCSTATE_RETURNED) {
			sstate->state = RPCSTATE_IDLE;
			rpc_freePkt(sstate->pkt);
			sstate->pkt = NULL;
			rpc_schedCleanup(sstate);
		}

	}

	splx(s);

handlers:
	for (scbin = rpc_scbins; scbin < &rpc_scbins[NSCBINS]; ++scbin)
	  for (classp = (rpc_serverClass_t *)scbin->forw;
	    classp != (rpc_serverClass_t *)scbin;
	    classp = classp->forw)
		rpc_classException(classp, conn, exception);
}

/*
 * Find the server state descriptor for a
 * particular client.  Return a pointer to
 * it or NULL if it is not found.
 */
 
enum seqno_t { USE_SEQNO, DONT_USE_SEQNO, NO_ABORT };

rpc_serverState_t *
rpc_sStateBySeqno(node, clientId, cookie)
	node_t *node;
	u_long clientId;
	u_long cookie;
{
	return rpc_sState1(node, clientId, cookie, USE_SEQNO);
}

rpc_serverState_t *
rpc_sState(node, clientId)
	node_t *node;
	u_long clientId;
{
	return rpc_sState1(node, clientId, -1, DONT_USE_SEQNO);
}

rpc_serverState_t *
rpc_sStateNoAbort(node, clientId)
	node_t *node;
	u_long clientId;
{
	return rpc_sState1(node, clientId, -1, NO_ABORT);
}

rpc_serverState_t *
rpc_sState1(node, clientId, cookie, useflag)
	node_t *node;
	u_long clientId;
	u_long cookie;
	enum seqno_t useflag;
{
	register rpc_serverState_t *sstate;
	register rpc_hash_t *ssbin = (rpc_hash_t *)SSHASH(clientId);

	switch (useflag) {
	case USE_SEQNO:
		for (sstate = (rpc_serverState_t *)ssbin->forw;
		  sstate != (rpc_serverState_t *)ssbin;
		  sstate = sstate->forw)
			if (sstate->clientId == clientId &&
			    nodecmp(sstate->conn->node, *node) &&
			    sstate->seqno == cookie)
				return(sstate);
		break;
	case DONT_USE_SEQNO:
		for (sstate = (rpc_serverState_t *)ssbin->forw;
		  sstate != (rpc_serverState_t *)ssbin;
		  sstate = sstate->forw)
			if (sstate->clientId == clientId &&
			    nodecmp(sstate->conn->node, *node))
				return(sstate);
		break;
	case NO_ABORT:
		for (sstate = (rpc_serverState_t *)ssbin->forw;
		  sstate != (rpc_serverState_t *)ssbin;
		  sstate = sstate->forw)
			if (sstate->clientId == clientId &&
			    nodecmp(sstate->conn->node, *node) &&
			    sstate->state != RPCSTATE_ABORTED)
				return(sstate);
		break;
	default:
		printf("rpc_sState1/I: Unknown useflag!\n");
		break;
	}
	return (rpc_serverState_t *)NULL;	
}

rpc_serverState_t *
rpc_newServerState(node, clientId, seqno)
	register connection_t *node;
	u_long clientId;
	u_long seqno;
{
	register rpc_serverState_t *sstate;
	register rpc_hash_t *ssbin = (rpc_hash_t *)SSHASH(clientId);
	register connection_t *conn;
	int s;

	if ((conn = conn_findByAddress(node)) == NULL)
		return(NULL);

	if (rpc_serverFreeStates !=
				(rpc_hash_t *)rpc_serverFreeStates->forw) {
		s = RPC_SPL();
		sstate = (rpc_serverState_t *)rpc_serverFreeStates->forw;
		rpc_removeFromTable(rpc_serverFreeStates, sstate);
		splx(s);
	}
	else
		sstate = (rpc_serverState_t *)
				calloc(sizeof(rpc_serverState_t), C_WAIT);
	sstate->clientId = clientId;
	sstate->seqno = 0;
	sstate->conn = conn;
	sstate->state = RPCSTATE_IDLE;
	sstate->pkt = NULL;
	sstate->retries = 0;
	
	s = RPC_SPL();	
	ssbin = (rpc_hash_t *)SSHASH(clientId);
	rpc_addToTable(ssbin, sstate);
	splx(s);
	
	RPC_INCMSTAT(remoteClients);
	return(sstate);
}	

/*
 * Free a state descriptor.  Must be
 * called with rpc interrupts locked
 * out.
 */
rpc_freeServerState(sstate)
   register rpc_serverState_t *sstate;
{
	register rpc_serverState_t *tmpstatep;
	register rpc_hash_t *ssbin = (rpc_hash_t *)SSHASH(sstate->clientId);
	int s;

	/*
	 * Remove the descriptor from the hash
	 * list first.
	 */
	for (tmpstatep = (rpc_serverState_t *)ssbin->forw;
	  tmpstatep != (rpc_serverState_t *)ssbin;
	  tmpstatep = tmpstatep->forw)
		;
	if (tmpstatep == (rpc_serverState_t *)NULL) {
		rpc_serverBug("server state not in hash table", sstate, 0);
		return;
	}
	
	s = RPC_SPL();
	
	rpc_removeFromTable(ssbin, sstate);

	conn_free(sstate->conn);

	/*
	 * Maintain a list of free state
	 * descriptors.
	 */
	sstate->forw = (rpc_hash_t *)rpc_serverFreeStates;
	rpc_addToTable(rpc_serverFreeStates, sstate);
	RPC_DECMSTAT(remoteClients);
	splx(s);
}

/*
 * Attach a new server class to RPC.
 */
rpc_serverAttach(newclassp)
   rpc_serverClass_t *newclassp;
{
	register rpc_serverClass_t *tmpclassp;
	register rpc_hash_t *scbin = (rpc_hash_t *)SCHASH(newclassp->class);
	int s = RPC_SPL();

	for (tmpclassp = (rpc_serverClass_t *)scbin->forw;
	  tmpclassp != (rpc_serverClass_t *)scbin;
	  tmpclassp = tmpclassp->forw)
		if (tmpclassp->class == newclassp->class) {
			splx(s);
			return(0);
		}
	rpc_addToTable(scbin, newclassp);
	splx(s);
	RPC_INCMSTAT(serversAttached);
	return(1);
}

/*
 * Detach a server class from RPC.
 */
rpc_serverDetach(classp)
   rpc_serverClass_t *classp;
{
	register rpc_serverClass_t *tmpclassp;
	register rpc_hash_t *scbin = (rpc_hash_t *)SCHASH(classp->class);
	int s = RPC_SPL();

	for (tmpclassp = (rpc_serverClass_t *)scbin->forw;
	  tmpclassp != (rpc_serverClass_t *)scbin;
          tmpclassp = tmpclassp->forw);
		if (tmpclassp->class == classp->class) {
			rpc_removeFromTable(scbin, classp);
			splx(s);
			RPC_DECMSTAT(serversAttached);
			return(1);
		}
	splx(s);
	return(0);
}

/*
 * Find the class descriptor for a particular
 * class number.  Return a pointer to it or
 * NULL if it is not found.
 */
rpc_serverClass_t *
rpc_findServerClass(class)
   register int class;
{
	register rpc_serverClass_t *classp;
	register rpc_hash_t *scbin = (rpc_hash_t *)SCHASH(class);

	for (classp = (rpc_serverClass_t *)scbin->forw;
	  classp != (rpc_serverClass_t *)scbin;
	  classp = classp->forw)
		if (classp->class == class)
			return(classp);
	return(NULL);
}

rpc_serverBug(msg, sstate, pkt)
   char *msg;
   rpc_serverState_t *sstate;
   rpc_header_t *pkt;
{
	printf("rpc server error: %s\n", msg);
	if (sstate) {
		printf("  state at 0x%X: state %d, clientId 0x%X, seqno 0x%x\n",
			sstate, sstate->state, sstate->clientId, sstate->seqno);
		printf("      server class %d \"%s\", op %d, retries %d, time left %d\n",
			sstate->classp->class,
			(sstate->classp->name ? sstate->classp->name : ""),
			sstate->operation,
			sstate->retries, timeUntil(sstate->time));
		printf("      connection %04x %06X-%06X \"%s\", gen 0x%X\n",
			sstate->conn->node.net,
			sstate->conn->node.host.high,
			sstate->conn->node.host.low,
			sstate->conn->name, sstate->conn->generation);
		if (sstate->pkt) {
			printf("    return pkt at 0x%x:\n", sstate->pkt);
			rpc_showPkt(sstate->pkt);
		}
	}
	if (pkt) {
		printf("  packet at 0x%x:\n", pkt);
		rpc_showPkt(pkt);
	}
}
