/*
 * This file contains the client end functions
 * of a remote procedure call.  Its responsibility
 * is to ensure reliable transmission of call
 * requests and to discard duplicate return
 * responses.
 *
 * jam 841026-27-29-30-31-1101
 * sas 851002	Fix to completely broken hash scheme
 */

#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 NCCBINS		32		/* Size of client class hash table */
#define NCSBINS		32		/* Size of client state hash table */

#define CCHASH(class)	&rpc_ccbins[binarymod(class, NCCBINS)]
#define CSHASH(clientId) &rpc_csbins[binarymod(clientId, NCSBINS)]

rpc_hash_t rpc_ccbins[NCCBINS];	/* Client class hash table */
rpc_hash_t rpc_csbins[NCSBINS];	/* Client state descriptor hash table */

/* Free client state descriptors */
rpc_hash_t rpc_cFreeStates;
rpc_hash_t *rpc_clientFreeStates = &rpc_cFreeStates; 

long	rpc_nextCallRetry = ETERNITY;	/* Next time retransmitter will run */

u_long	rpc_sequence;			/* Last sequence number used */

int rpc_callRetransmitter();
int rpc_clientException();
rpc_clientState_t *rpc_findClientState();
rpc_clientState_t *rpc_allocClientState();
caddr_t rpc_discardHeader();

/*
 * Initialize the hash tables
 */
rpc_initClientHash()
{
	register rpc_hash_t *h;
	for (h = rpc_ccbins; h < &rpc_ccbins[NCCBINS]; h++)
		h->forw = h->tail = (caddr_t)h;
	for (h = rpc_csbins; h < &rpc_csbins[NCSBINS]; h++)
		h->forw = h->tail = (caddr_t)h;
	rpc_clientFreeStates->forw = rpc_clientFreeStates;
	rpc_clientFreeStates->tail = rpc_clientFreeStates;

}
	
rpc_clientInit()
{
	return(conn_attach(rpc_clientException));
}

/*
 * Called by the client with a call packet and
 * a destination node to service the request.
 * This function will not block; instead the
 * caller will be notified by a call to either
 * the return function or the error function
 * for his class.  The caller MUST be prepared
 * to have the return (or error) function called
 * before this function returns.
 */
rpc_call(clientId, clientClassp, serverConn, serverClass, operation, params)
   u_long clientId;
   rpc_clientClass_t *clientClassp;
   connection_t *serverConn;
   u_short serverClass;
   u_short operation;
   caddr_t params;
{
	register rpc_clientState_t *cstate;
	register rpc_header_t *pkt;
	register rpc_hash_t *csbin;
	long retryTime;
	int s;

	/*
	 * Do not even attempt the call unless the
	 * destination is reachable.
	 */
	if (serverConn->state != CONN_ACTIVE) {
		rpc_freeParams(params);
		rpc_classError(clientClassp, clientId, EHOSTUNREACH);
		return;
	}

	/*
	 * Allocate a client state descriptor for
	 * this call.  If we cannot then the caller
	 * will be given an error.
	 */
	if ((cstate = rpc_allocClientState()) == NULL) {
		rpc_freeParams(params);
		rpc_classError(clientClassp, clientId, ELOCALRESOURCE);
		return;
	}

	/* BUG there is a race before we link it */
	conn_link(serverConn);

	pkt = rpc_buildHeader(params);
	RPC_INCMSTAT(callPackets);
	pkt->client = clientId;
	pkt->seqno = ++rpc_sequence;
	pkt->clientgen = myGeneration;
	pkt->type = RPCPKT_CALL;
	pkt->class = serverClass;
	pkt->operation = operation;

	/*
	 * Set the client state with the information
	 * necessary for a call and send the call
	 * packet for the first time.
	 */
	cstate->state = RPCSTATE_CALLING;
	cstate->clientId = clientId;
	cstate->classp = clientClassp;
	cstate->pkt = pkt;
	cstate->conn = serverConn;
	cstate->seqno = pkt->seqno;
	cstate->retries = 0;
	cstate->time = currentTime;

	s = RPC_SPL();

	csbin = CSHASH(pkt->client);
	rpc_addToTable(csbin, cstate);

	rpc_output(pkt, cstate->conn);

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

	splx(s);
}

/*
 * The call retransmitter is always scheduled to run
 * at the time the next call retry must be made;
 * therefore there is only one call retransmitter
 * scheduled at any time.  If the call list becomes
 * empty then the daemon is not scheduled.  Each
 * time the daemon runs it scans the call list
 * retransmitting calls which have not been
 * acknowledged.  The daemon then reschedules
 * itself to be run at the earliest time that a
 * call needs to be retransmitted.
 */
/*ARGSUSED*/
rpc_callRetransmitter(arg)
   int arg;
{
	register rpc_hash_t *csbin;
	register rpc_clientState_t *cstate;
	register rpc_clientState_t *tmpcstate;
	long minRetryTime = ETERNITY;
	int s = RPC_SPL();
	/* BUG we would prefer the RPC_SPL above to be inside the loops */

	RPC_INCSTAT(callDaemon);
	for (csbin = rpc_csbins; csbin < &rpc_csbins[NCSBINS]; ++csbin)
	  for (cstate = (rpc_clientState_t *)csbin->forw; 
		cstate != (rpc_clientState_t *)csbin; 
		cstate = (tmpcstate ? tmpcstate : cstate->forw)) {

		tmpcstate = (rpc_clientState_t *)0;

		/*
		 * If the client is no longer calling then
		 * go on to the next client.
		 */
		if (cstate->state != RPCSTATE_CALLING)
			continue;

		/*
		 * If the call's retry time has been reached then
		 * either retransmit it or assume that the call
		 * has timed out.
		 */
		if (currentTime >= cstate->time + rpc_retryTime(cstate)) {

			/*
			 * If the call has not exceeded its maximum
			 * number of retries then retransmit the
			 * call packet.
			 */
			if (cstate->retries < rpc_maxRetries(cstate)) {
				rpc_statlist.callRetriesCur.count =
								cstate->retries;
				RPC_INCMSTAT(callRetries);
				++cstate->retries;
				cstate->time = currentTime;
				rpc_output(cstate->pkt, cstate->conn);
			}

			/*
			 * The call has exceeded its maximum number of
			 * retries.  Free the whole thing up and send
			 * an error back to the caller.
			 */
			else {
			register connection_t *c = cstate->conn;
			extern int efs_sasDEBUG;
			
				RPC_INCSTAT(lostCalls);

	efsDEBUG("rpc_callRetransmitter/I","rpc call timeout: %s, state %d",
		c->name, c->state);
	if (efs_sasDEBUG)
		printf("rpc last heard from %s %d seconds ago\n", c->name, 
			time.tv_sec - boottime.tv_sec - c->time);
	
				cstate->state = RPCSTATE_IDLE;
				if (cstate->pkt) {
					rpc_freePkt(cstate->pkt);
					RPC_DECMSTAT(callPackets);
					cstate->pkt = NULL;
				}
				tmpcstate = cstate->forw;
				rpc_removeFromTable(csbin, cstate);
				rpc_classError(cstate->classp, cstate->clientId,
							ERPCCALLTIMEOUT);
				rpc_freeClientState(cstate);
				continue;
			}
		}

		/*
		 * Find the smallest amount of time until
		 * the next call retry so we can reschedule
		 * the call retransmitter.
		 */
		if (cstate->state == RPCSTATE_CALLING) {
			long retryTime = rpc_retryTime(cstate);

			if (retryTime < minRetryTime)
				minRetryTime = retryTime;
		}

	}

	splx(s);

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

/*
 * This function consumes all call acknowledge
 * packets and must be called with rpc interrupts
 * locked out.
 *
 * I think that this is a BUG either in the protocol or this implementation.
 * We take ourselves off the re-transmit queue assuming that we will
 * be shortly getting a response to our call.  By taking ourselves off the
 * re-xmt queue, we also cease "counting down" before initiating cleanup
 * in case the server drops.  We assume that CONN, that wonder of wonders,
 * will initiate cleanup or that the user will get bored and interrupt
 * the process (assuming that the signal sub-system works).  Oh well,
 * we shouldn't be surprised, should we?
 * sas 851002
 */
rpc_callAck(pkt)
   rpc_header_t *pkt;
{
	rpc_clientState_t *cstate;

	RPC_INCSTAT(callAcks);
	/*
	 * Find the client state descriptor for the
	 * client referred to in the call ack packet.
	 * If he is not found do nothing; otherwise ...
	 */
	if (cstate = rpc_findClientState(pkt->client)) {


		/*
		 * If the client was still in the calling state
		 * for this sequence number then we can discard
		 * the call packet and just wait for a return.
		 */
		if (cstate->state == RPCSTATE_CALLING &&
		    cstate->seqno == pkt->seqno) {
			if (cstate->pkt) {
				rpc_freePkt(cstate->pkt);
				RPC_DECMSTAT(callPackets);
				cstate->pkt = NULL;
			}
		}
	}

	rpc_freePkt(pkt);
}

/*
 * This function consumes return and error
 * packets; although it may actually pass the
 * packet along to the client to consume it.
 * This function must be called with rpc
 * interrupts locked out.
 */
rpc_returnReceived(pkt)
   register rpc_header_t *pkt;
{
	register rpc_clientState_t *cstate;
	register rpc_hash_t *csbin;

	/*
	 * If we found the client descriptor for this
	 * sequence number then we do at least some
	 * work ...
	 */
	if ((cstate = rpc_findClientState(pkt->client)) &&
	    cstate->seqno == pkt->seqno) {

		/*
		 * If the client still has his call packet
		 * then discard it.
		 */
		if (cstate->state == RPCSTATE_CALLING) {
			cstate->state = RPCSTATE_WANTRETURN;
			if (cstate->pkt) {
				rpc_freePkt(cstate->pkt);
				RPC_DECMSTAT(callPackets);
				cstate->pkt = NULL;
			}
		}

		csbin = CSHASH(cstate->clientId);
		rpc_removeFromTable(csbin, cstate);

		/*
		 * If the client is waiting for the return
		 * packet then pass it along to him (or, if
		 * it is an error, give him the error).
		 */
		if (pkt->type == RPCPKT_ERROR) {
			RPC_INCSTAT(errorsRecvd);
			rpc_classError(cstate->classp, cstate->clientId,
				       pkt->operation);
			rpc_freePkt(pkt);
		}
		else
			rpc_classReturn(cstate->classp, cstate->clientId,
					rpc_discardHeader(pkt));

		rpc_freeClientState(cstate);
	}

	/*
	 * Otherwise, send a return acknowledgement
	 * so that the server will stop sending return
	 * packets.
	 */
	else
	{
		register connection_t *conn = conn_findByAddress(&pkt->source);

		rpc_reply(pkt, conn, RPCPKT_RETURNACK);
		conn_free(conn);
		RPC_INCSTAT(returnAcksSent);
	}
}

/*
 * Called from the connection manager when
 * a node changes state (comes up, crashes,
 * etc.).
 */
rpc_clientException(conn, exception)
   connection_t  *conn;
   int exception;
{
	register rpc_hash_t *csbin;
	register rpc_clientState_t *cstate;
	register rpc_hash_t *ccbin;
	register rpc_clientClass_t *classp;
	rpc_clientState_t *tmpcstate;
	int breakall = 0;
	int s;

	if (exception == CONN_UP)
		RPC_INCMSTAT(nodesUp);
	else if (exception == CONN_DOWN)
		RPC_DECMSTAT(nodesUp);
	/*
	 * 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 server
	 * has crashed.  For each client with an
	 * outstanding call on the effected node(s)
	 * give the client an error.  All clients for
	 * the effected node(s) will be left in the
	 * idle state when we are through.
	 */
	for (csbin = rpc_csbins; csbin < &rpc_csbins[NCSBINS]; ++csbin)
	  for (cstate = (rpc_clientState_t *)csbin->forw; 
		cstate != (rpc_clientState_t *)csbin; cstate = tmpcstate) {

		/*
		 * 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.
		 */
		tmpcstate = cstate->forw;
		if (!breakall && cstate->conn != conn)
			continue;

		rpc_removeFromTable(csbin, cstate);
		RPC_INCSTAT(callsFlushed);

		/*
		 * If the client still calling then free the
		 * call packet.
		 */
		if (cstate->state == RPCSTATE_CALLING) {
			if (cstate->pkt) {
				rpc_freePkt(cstate->pkt);
				cstate->pkt = NULL;
				RPC_DECMSTAT(callPackets);
			}
		}


		/*
		 * If the client was waiting for a return then
		 * Give the client an error.
		 */
		rpc_classError(cstate->classp, cstate->clientId,
							ESERVERCRASHED);

		rpc_freeClientState(cstate);
	}

	splx(s);

handlers:
	for (ccbin = rpc_ccbins; ccbin < &rpc_ccbins[NCCBINS]; ++ccbin)
	  for (classp = (rpc_clientClass_t *)ccbin->forw; 
		classp != (rpc_clientClass_t *)ccbin; classp = classp->forw)
		rpc_classException(classp, conn, exception);
}

/*
 * Find the client state descriptor for a
 * particular client.  Return a pointer to
 * it or NULL if it is not found.
 */
rpc_clientState_t *
rpc_findClientState(clientId)
   u_long clientId;
{
	register rpc_hash_t *csbin = CSHASH(clientId);
	register rpc_clientState_t *cstate;

	/* BUG -- what if more than one state info per clientId */
	for (cstate = (rpc_clientState_t *)csbin->forw; 
	  cstate != (rpc_clientState_t *)csbin; cstate = cstate->forw)
		if (cstate->clientId == clientId)
			return(cstate);
	return(NULL);
}

/*
 * Attach a new client class to RPC.
 */
rpc_clientAttach(newclassp)
   rpc_clientClass_t *newclassp;
{
	register rpc_hash_t *ccbin = CCHASH(newclassp->class);
	register rpc_clientClass_t *classp;
	int s = RPC_SPL();
	
	/*
	 * First check to make sure that we aren't adding the same class twice
	 */
	for (classp = (rpc_clientClass_t *)ccbin->forw; 
		classp != (rpc_clientClass_t *)ccbin; classp = classp->forw) {
		if (classp->class == newclassp->class) {
			splx(s);
			return 1;
		}
	}
	rpc_addToTable(ccbin, newclassp);
	RPC_INCMSTAT(clientsAttached);
	splx(s);
	return(1);
}

/*
 * Detach a client class from RPC.
 */
rpc_clientDetach(classp)
   rpc_clientClass_t *classp;
{
	register rpc_clientClass_t *tmpclassp;
	register rpc_hash_t *ccbin = CCHASH(classp->class);
	int s = RPC_SPL();

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

/*
 * Allocate a client state descriptor.
 */
rpc_clientState_t *
rpc_allocClientState()
{
	register rpc_clientState_t *cstate;

	if (rpc_clientFreeStates != (rpc_hash_t *)rpc_clientFreeStates->forw) {
		int s = RPC_SPL();
		cstate = (rpc_clientState_t *)rpc_clientFreeStates->forw;
		rpc_removeFromTable(rpc_clientFreeStates, cstate);
		splx(s);
	} else
		cstate = (rpc_clientState_t *)
			calloc(sizeof(rpc_clientState_t), C_WAIT);
	RPC_INCMSTAT(calls);
	return(cstate);
}

/*
 * Free a state descriptor.  Put it on
 * the list of free state descriptors
 * so alloc above can just find it.
 */
rpc_freeClientState(cstate)
   register rpc_clientState_t *cstate;
{
	int s;

	conn_free(cstate->conn);
	if (cstate->forw || cstate->back)
		cdebugger("rpc_freeClientState");
	
	s = RPC_SPL();
	rpc_addToTable(rpc_clientFreeStates, cstate);
	splx(s);
	RPC_DECMSTAT(calls);
}

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