clnt_kudp.c

<B>Digital的Unix操作系统VAX 4.2源码</B>

	smp_lock(&p->cku_lk_outbuf, LK_RETRY);

	while (p->cku_flags & CKU_BUFBUSY) {
		p->cku_flags |= CKU_BUFWANTED;
		/*
		 * This is a kludge to avoid deadlock in the case of a
		 * loop-back call.  The client can block wainting for
		 * the server to free the mbuf while the server is blocked
		 * waiting for the client to free the reply mbuf.  Avoid
		 * this by flushing the input queue every once in a while
		 * while we are waiting.
		 */
		timeout(wakeup, (caddr_t)&p->cku_outbuf, hz);
		sleep_unlock((caddr_t)&p->cku_outbuf, PZERO-3, &p->cku_lk_outbuf);
		untimeout(wakeup, (caddr_t)&p->cku_outbuf);
		/* SMP lock socket during flush			*/
		SO_LOCK(so);
		/* smp_lock(&so->lk_socket, LK_RETRY); */
		sbflush(&so->so_rcv);
		smp_unlock(&so->lk_socket);
		smp_lock(&p->cku_lk_outbuf, LK_RETRY);
	}
	p->cku_flags |= CKU_BUFBUSY;
	smp_unlock(&p->cku_lk_outbuf);
	(void) splx(s);
	m = mclgetx(buffree, (caddr_t)p, p->cku_outbuf, UDPMSGSIZE, M_WAIT);
	if (m == NULL) {
		p->cku_err.re_status = RPC_SYSTEMERROR;
		p->cku_err.re_errno = ENOBUFS;
		buffree(p);
		goto done;
	}

	xdrs = &p->cku_outxdr;
	/*
	 * The transaction id is the first thing in the
	 * preserialized output buffer.
	 */
	(*(u_long *)(p->cku_outbuf)) = h->cl_xid;

	xdrmbuf_init(xdrs, m, XDR_ENCODE);

	if (rempos != 0) {
		XDR_SETPOS(xdrs, rempos);
	} else {
		/*
		 * Serialize dynamic stuff into the output buffer.
		 */
		XDR_SETPOS(xdrs, p->cku_outpos);
		if ((! XDR_PUTLONG(xdrs, (long *)&procnum)) ||
		    (! AUTH_MARSHALL(h->cl_auth, xdrs)) ||
		    (! (*xdr_args)(xdrs, argsp))) {
			p->cku_err.re_status = RPC_CANTENCODEARGS;
			p->cku_err.re_errno = EIO;
			(void) m_freem(m);
			goto done;
		}
		rempos = XDR_GETPOS(xdrs);
	}
	m->m_len = rempos;

	if ((p->cku_err.re_errno = ku_sendto_mbuf(so, m, &p->cku_addr)) != 0) {
		p->cku_err.re_status = RPC_CANTSEND;
		p->cku_err.re_errno = EIO;
		goto done;
	}

	for (rtries = RECVTRIES; rtries; rtries--) {
		s = splnet();
		/* SMP lock socket while doing setting up to sleep */
		/* smp_lock(&so->lk_socket, LK_RETRY); */
		SO_LOCK(so);
		while (so->so_rcv.sb_cc == 0) {
			/*
			 * Set timeout then wait for input, timeout,
 			 * or interrupt.
			 */
#ifdef RPCDEBUG
			rpc_debug(3, "callit: waiting %d, sec %d usec %d\n",
				timohz, time.tv_sec, time.tv_usec);
#endif
			timeout(ckuwakeup, (caddr_t)p, timohz);
			so->so_rcv.sb_flags |= SB_WAIT;

			if (!(pp->p_vm & SNFSPGN)) {
				/*
				 * If not doing page-in,
				 * allow interruptions for hangup, interrupt,
				 * quit, kill, and termination.
				 */
				smask = pp->p_sigmask;
				pp->p_sigmask |=
					~(sigmask(SIGHUP) |
					  sigmask(SIGINT) | sigmask(SIGQUIT) |
					  sigmask(SIGKILL) | sigmask(SIGTERM));
				interrupted =
				       sleep_unlock((caddr_t)&so->so_rcv.sb_cc,
						    PZERO+1 | PCATCH,
						    &so->lk_socket);
				pp->p_sigmask = smask;
				untimeout(ckuwakeup, (caddr_t)p);
				if (interrupted) {
					(void) splx(s);
					p->cku_err.re_status = RPC_INTR;
					p->cku_err.re_errno = EINTR;
					goto done;
				}
			} else {
				sleep_unlock((caddr_t)&so->so_rcv.sb_cc,
						    PRIBIO,
						    &so->lk_socket);
				untimeout(ckuwakeup, (caddr_t)p);
			}

			if (p->cku_flags & CKU_TIMEDOUT) {
				p->cku_flags &= ~CKU_TIMEDOUT;
				/*
				 * Avoid long sleep on m above (call_again)
				 * if the server failed to arp resolve and m
				 * is being held ransom (actually for three
				 * minutes). 
				 * The arphasmbuf() call will force it free,
				 * and buffree() will be called as a result.
				 */
				if (p->cku_flags & CKU_BUFBUSY)
					arphasmbuf(m);
				(void) splx(s);
				p->cku_err.re_status = RPC_TIMEDOUT;
				p->cku_err.re_errno = ETIMEDOUT;
				smp_lock(&lk_nfsstat, LK_RETRY);
				rcstat.rctimeouts++;
				smp_unlock(&lk_nfsstat);
				goto done;
			}
		SO_LOCK(so);
		/* smp_lock(&so->lk_socket, LK_RETRY); */
		}

		if (so->so_error) {
			so->so_error = 0;
			smp_unlock(&so->lk_socket);
			(void) splx(s);
			continue;
		}

		p->cku_inmbuf = ku_recvfrom(so, &from);
		smp_unlock(&so->lk_socket);
		(void) splx(s);
		if (p->cku_inmbuf == NULL) {
			continue;
		}
		p->cku_inbuf = mtod(p->cku_inmbuf, char *);

		if (p->cku_inmbuf->m_len < sizeof(u_long)) {
			m_freem(p->cku_inmbuf);
			continue;
		}
		/*
		 * If reply transaction id matches id sent
		 * we have a good packet.
		 */
		if (*((u_long *)(p->cku_inbuf))
		    != *((u_long *)(p->cku_outbuf))) {
			smp_lock(&lk_nfsstat, LK_RETRY);
			rcstat.rcbadxids++;
			smp_unlock(&lk_nfsstat);
			m_freem(p->cku_inmbuf);
			continue;
		}
		/*
		 * Flush the rest of the stuff on the input queue
		 * for the socket.
		 */
		s = splnet();
		SO_LOCK(so);
		/* smp_lock(&so->lk_socket, LK_RETRY); */
		sbflush(&so->so_rcv);
		smp_unlock(&so->lk_socket);
		(void) splx(s);
		break;
	} 

	if (rtries == 0) {
		p->cku_err.re_status = RPC_CANTRECV;
		p->cku_err.re_errno = EIO;
		goto done;
	}

	/*
	 * Process reply
	 */

	xdrs = &(p->cku_inxdr);
	xdrmbuf_init(xdrs, p->cku_inmbuf, XDR_DECODE);

	{
		/*
		 * Declare this variable here to have smaller
		 * demand for stack space in this procedure.
		 */
		struct rpc_msg		   reply_msg;

		reply_msg.acpted_rply.ar_verf = _null_auth;
		reply_msg.acpted_rply.ar_results.where = resultsp;
		reply_msg.acpted_rply.ar_results.proc = xdr_results;

		/*
		 * Decode and validate the response.
		 */
		if (xdr_replymsg(xdrs, &reply_msg)) {
			_seterr_reply(&reply_msg, &(p->cku_err));

			if (p->cku_err.re_status == RPC_SUCCESS) {
				/*
				 * Reply is good, check auth.
				 */
				if (! AUTH_VALIDATE(h->cl_auth,
				    &reply_msg.acpted_rply.ar_verf)) {
					p->cku_err.re_status = RPC_AUTHERROR;
					p->cku_err.re_why = AUTH_INVALIDRESP;
					rcstat.rcbadverfs++;
				}
				if (reply_msg.acpted_rply.ar_verf.oa_base !=
				    NULL) {
					/* free auth handle */
					xdrs->x_op = XDR_FREE;
					(void) xdr_opaque_auth(xdrs,
					    &(reply_msg.acpted_rply.ar_verf));
				}
			} else {
				/*
				 * Maybe our credential needs refreshed
				 */
				if (refreshes > 0 && AUTH_REFRESH(h->cl_auth)) {
					refreshes--;
					rcstat.rcnewcreds++;
					rempos = 0;
				}
			}
		} else {
			p->cku_err.re_status = RPC_CANTDECODERES;
		}
	}

	m_freem(p->cku_inmbuf);

#ifdef RPCDEBUG
	rpc_debug(4, "cku_callit done\n");
#endif
done:
	if ((p->cku_err.re_status != RPC_SUCCESS) &&
 	    (p->cku_err.re_status != RPC_INTR) &&
	    (p->cku_err.re_status != RPC_CANTENCODEARGS) &&
	    (--stries > 0) ) {
		smp_lock(&lk_nfsstat, LK_RETRY);
		rcstat.rcretrans++;
		smp_unlock(&lk_nfsstat);
		timohz = backoff(timohz);
		if (p->cku_err.re_status == RPC_SYSTEMERROR ||
		    p->cku_err.re_status == RPC_CANTSEND) {
			/*
			 * Errors due to lack o resources, wait a bit
			 * and try again.
			 */
			(void) sleep((caddr_t)&lbolt, PZERO-4);
		}
		goto call_again;
	}
	u.u_cred = tmpcred;
	p->cku_flags &= ~CKU_BUSY;
	if (p->cku_flags & CKU_WANTED) {
		p->cku_flags &= ~CKU_WANTED;
		wakeup((caddr_t)h);
	}
	if (p->cku_err.re_status != RPC_SUCCESS) {
		smp_lock(&lk_nfsstat, LK_RETRY);
		rcstat.rcbadcalls++;
		smp_unlock(&lk_nfsstat);
	}
	return (p->cku_err.re_status);
}

/*
 * Wake up client waiting for a reply.
 */
ckuwakeup(p)
	register struct cku_private *p;
{

#ifdef RPCDEBUG
	rpc_debug(4, "cku_timeout\n");
#endif
	p->cku_flags |= CKU_TIMEDOUT;
	sbwakeup(&p->cku_sock->so_rcv);
}

/*
 * Return error info on this handle.
 */
void
clntkudp_error(h, err)
	CLIENT *h;
	struct rpc_err *err;
{
	register struct cku_private *p = htop(h);

	*err = p->cku_err;
}

/* static */ bool_t
clntkudp_freeres(cl, xdr_res, res_ptr)
	CLIENT *cl;
	xdrproc_t xdr_res;
	caddr_t res_ptr;
{
	register struct cku_private *p = (struct cku_private *)cl->cl_private;
	register XDR *xdrs = &(p->cku_outxdr);

	xdrs->x_op = XDR_FREE;
	return ((*xdr_res)(xdrs, res_ptr));
}

void 
clntkudp_abort()
{
}

bool_t
clntkudp_control()
{
	return (FALSE);
}


/*
 * Destroy rpc handle.
 * Frees the space used for output buffer, private data, and handle
 * structure, and closes the socket for this handle.
 */
void
clntkudp_destroy(h)
	CLIENT *h;
{
	register struct cku_private *p = htop(h);

#ifdef RPCDEBUG
	rpc_debug(4, "cku_destroy %x\n", h);
#endif
	(void) soclose(p->cku_sock);
	kmem_free((caddr_t)(caddr_t)p->cku_outbuf, KM_RPC);
	kmem_free((caddr_t)(caddr_t)p, KM_RPC);
}

/*
 * try to bind to a reserved port
 */
bindresvport(so)
	struct socket *so;
{
	struct sockaddr_in *sin;
	struct mbuf *m;
	u_short i;
	int error;
	struct ucred *tmpcred;
	struct ucred *savecred;

#	define MAX_PRIV	(IPPORT_RESERVED-1)
#	define MIN_PRIV	(IPPORT_RESERVED/2)

	m = m_get(M_WAIT, MT_SONAME);
	if (m == NULL) {
		printf("bindresvport: couldn't alloc mbuf");
		return(ENOBUFS);
	}

	sin = mtod(m, struct sockaddr_in *);
	sin->sin_family = AF_INET;
	sin->sin_addr.s_addr = INADDR_ANY;
	m->m_len = sizeof(struct sockaddr_in);

	/*
	 * use our own cred for sobind call.
	 */
	tmpcred = crdup(u.u_cred);
	savecred = u.u_cred;
	u.u_cred = tmpcred;
	u.u_uid = 0;
	
	error = EADDRINUSE;
	for (i = MAX_PRIV; error == EADDRINUSE && i >= MIN_PRIV; i--) {
		sin->sin_port = htons(i);
		error = sobind(so, m);
	}
	(void) m_freem(m);
	u.u_cred = savecred;
	crfree(tmpcred);
	return (error);
}


/* Called from main in sys/init_main.c to setup rpc smp locks	*/
rpcinit()
	{

	lockinit(&lk_rpcxid, &lock_rpcxid_d);
	lockinit(&lk_rpcrqcred, &lock_rpcrqcred_d);
	lockinit(&lk_rpccallout, &lock_rpccallout_d);
	lockinit(&lk_rpcdupreq, &lock_rpcdupreq_d);

	}