clnt_vc.c revision 184588
1/*	$NetBSD: clnt_vc.c,v 1.4 2000/07/14 08:40:42 fvdl Exp $	*/
2
3/*
4 * Sun RPC is a product of Sun Microsystems, Inc. and is provided for
5 * unrestricted use provided that this legend is included on all tape
6 * media and as a part of the software program in whole or part.  Users
7 * may copy or modify Sun RPC without charge, but are not authorized
8 * to license or distribute it to anyone else except as part of a product or
9 * program developed by the user.
10 *
11 * SUN RPC IS PROVIDED AS IS WITH NO WARRANTIES OF ANY KIND INCLUDING THE
12 * WARRANTIES OF DESIGN, MERCHANTIBILITY AND FITNESS FOR A PARTICULAR
13 * PURPOSE, OR ARISING FROM A COURSE OF DEALING, USAGE OR TRADE PRACTICE.
14 *
15 * Sun RPC is provided with no support and without any obligation on the
16 * part of Sun Microsystems, Inc. to assist in its use, correction,
17 * modification or enhancement.
18 *
19 * SUN MICROSYSTEMS, INC. SHALL HAVE NO LIABILITY WITH RESPECT TO THE
20 * INFRINGEMENT OF COPYRIGHTS, TRADE SECRETS OR ANY PATENTS BY SUN RPC
21 * OR ANY PART THEREOF.
22 *
23 * In no event will Sun Microsystems, Inc. be liable for any lost revenue
24 * or profits or other special, indirect and consequential damages, even if
25 * Sun has been advised of the possibility of such damages.
26 *
27 * Sun Microsystems, Inc.
28 * 2550 Garcia Avenue
29 * Mountain View, California  94043
30 */
31
32#if defined(LIBC_SCCS) && !defined(lint)
33static char *sccsid2 = "@(#)clnt_tcp.c 1.37 87/10/05 Copyr 1984 Sun Micro";
34static char *sccsid = "@(#)clnt_tcp.c	2.2 88/08/01 4.0 RPCSRC";
35static char sccsid3[] = "@(#)clnt_vc.c 1.19 89/03/16 Copyr 1988 Sun Micro";
36#endif
37#include <sys/cdefs.h>
38__FBSDID("$FreeBSD: head/sys/rpc/clnt_vc.c 184588 2008-11-03 10:38:00Z dfr $");
39
40/*
41 * clnt_tcp.c, Implements a TCP/IP based, client side RPC.
42 *
43 * Copyright (C) 1984, Sun Microsystems, Inc.
44 *
45 * TCP based RPC supports 'batched calls'.
46 * A sequence of calls may be batched-up in a send buffer.  The rpc call
47 * return immediately to the client even though the call was not necessarily
48 * sent.  The batching occurs if the results' xdr routine is NULL (0) AND
49 * the rpc timeout value is zero (see clnt.h, rpc).
50 *
51 * Clients should NOT casually batch calls that in fact return results; that is,
52 * the server side should be aware that a call is batched and not produce any
53 * return message.  Batched calls that produce many result messages can
54 * deadlock (netlock) the client and the server....
55 *
56 * Now go hang yourself.
57 */
58
59#include <sys/param.h>
60#include <sys/systm.h>
61#include <sys/lock.h>
62#include <sys/malloc.h>
63#include <sys/mbuf.h>
64#include <sys/mutex.h>
65#include <sys/pcpu.h>
66#include <sys/proc.h>
67#include <sys/protosw.h>
68#include <sys/socket.h>
69#include <sys/socketvar.h>
70#include <sys/syslog.h>
71#include <sys/time.h>
72#include <sys/uio.h>
73#include <netinet/tcp.h>
74
75#include <rpc/rpc.h>
76#include <rpc/rpc_com.h>
77
78#define MCALL_MSG_SIZE 24
79
80struct cmessage {
81        struct cmsghdr cmsg;
82        struct cmsgcred cmcred;
83};
84
85static enum clnt_stat clnt_vc_call(CLIENT *, struct rpc_callextra *,
86    rpcproc_t, struct mbuf *, struct mbuf **, struct timeval);
87static void clnt_vc_geterr(CLIENT *, struct rpc_err *);
88static bool_t clnt_vc_freeres(CLIENT *, xdrproc_t, void *);
89static void clnt_vc_abort(CLIENT *);
90static bool_t clnt_vc_control(CLIENT *, u_int, void *);
91static void clnt_vc_close(CLIENT *);
92static void clnt_vc_destroy(CLIENT *);
93static bool_t time_not_ok(struct timeval *);
94static void clnt_vc_soupcall(struct socket *so, void *arg, int waitflag);
95
96static struct clnt_ops clnt_vc_ops = {
97	.cl_call =	clnt_vc_call,
98	.cl_abort =	clnt_vc_abort,
99	.cl_geterr =	clnt_vc_geterr,
100	.cl_freeres =	clnt_vc_freeres,
101	.cl_close =	clnt_vc_close,
102	.cl_destroy =	clnt_vc_destroy,
103	.cl_control =	clnt_vc_control
104};
105
106/*
107 * A pending RPC request which awaits a reply. Requests which have
108 * received their reply will have cr_xid set to zero and cr_mrep to
109 * the mbuf chain of the reply.
110 */
111struct ct_request {
112	TAILQ_ENTRY(ct_request) cr_link;
113	uint32_t		cr_xid;		/* XID of request */
114	struct mbuf		*cr_mrep;	/* reply received by upcall */
115	int			cr_error;	/* any error from upcall */
116	char			cr_verf[MAX_AUTH_BYTES]; /* reply verf */
117};
118
119TAILQ_HEAD(ct_request_list, ct_request);
120
121struct ct_data {
122	struct mtx	ct_lock;
123	int		ct_threads;	/* number of threads in clnt_vc_call */
124	bool_t		ct_closing;	/* TRUE if we are closing */
125	bool_t		ct_closed;	/* TRUE if we are closed */
126	struct socket	*ct_socket;	/* connection socket */
127	bool_t		ct_closeit;	/* close it on destroy */
128	struct timeval	ct_wait;	/* wait interval in milliseconds */
129	struct sockaddr_storage	ct_addr; /* remote addr */
130	struct rpc_err	ct_error;
131	uint32_t	ct_xid;
132	char		ct_mcallc[MCALL_MSG_SIZE]; /* marshalled callmsg */
133	size_t		ct_mpos;	/* pos after marshal */
134	const char	*ct_waitchan;
135	int		ct_waitflag;
136	struct mbuf	*ct_record;	/* current reply record */
137	size_t		ct_record_resid; /* how much left of reply to read */
138	bool_t		ct_record_eor;	 /* true if reading last fragment */
139	struct ct_request_list ct_pending;
140};
141
142static const char clnt_vc_errstr[] = "%s : %s";
143static const char clnt_vc_str[] = "clnt_vc_create";
144static const char clnt_read_vc_str[] = "read_vc";
145static const char __no_mem_str[] = "out of memory";
146
147/*
148 * Create a client handle for a connection.
149 * Default options are set, which the user can change using clnt_control()'s.
150 * The rpc/vc package does buffering similar to stdio, so the client
151 * must pick send and receive buffer sizes, 0 => use the default.
152 * NB: fd is copied into a private area.
153 * NB: The rpch->cl_auth is set null authentication. Caller may wish to
154 * set this something more useful.
155 *
156 * fd should be an open socket
157 */
158CLIENT *
159clnt_vc_create(
160	struct socket *so,		/* open file descriptor */
161	struct sockaddr *raddr,		/* servers address */
162	const rpcprog_t prog,		/* program number */
163	const rpcvers_t vers,		/* version number */
164	size_t sendsz,			/* buffer recv size */
165	size_t recvsz)			/* buffer send size */
166{
167	CLIENT *cl;			/* client handle */
168	struct ct_data *ct = NULL;	/* client handle */
169	struct timeval now;
170	struct rpc_msg call_msg;
171	static uint32_t disrupt;
172	struct __rpc_sockinfo si;
173	XDR xdrs;
174	int error, interrupted, one = 1;
175	struct sockopt sopt;
176
177	if (disrupt == 0)
178		disrupt = (uint32_t)(long)raddr;
179
180	cl = (CLIENT *)mem_alloc(sizeof (*cl));
181	ct = (struct ct_data *)mem_alloc(sizeof (*ct));
182
183	mtx_init(&ct->ct_lock, "ct->ct_lock", NULL, MTX_DEF);
184	ct->ct_threads = 0;
185	ct->ct_closing = FALSE;
186	ct->ct_closed = FALSE;
187
188	if ((so->so_state & (SS_ISCONNECTED|SS_ISCONFIRMING)) == 0) {
189		error = soconnect(so, raddr, curthread);
190		SOCK_LOCK(so);
191		interrupted = 0;
192		while ((so->so_state & SS_ISCONNECTING)
193		    && so->so_error == 0) {
194			error = msleep(&so->so_timeo, SOCK_MTX(so),
195			    PSOCK | PCATCH, "connec", 0);
196			if (error) {
197				if (error == EINTR || error == ERESTART)
198					interrupted = 1;
199				break;
200			}
201		}
202		if (error == 0) {
203			error = so->so_error;
204			so->so_error = 0;
205		}
206		SOCK_UNLOCK(so);
207		if (error) {
208			if (!interrupted)
209				so->so_state &= ~SS_ISCONNECTING;
210			rpc_createerr.cf_stat = RPC_SYSTEMERROR;
211			rpc_createerr.cf_error.re_errno = error;
212			goto err;
213		}
214	}
215
216	if (!__rpc_socket2sockinfo(so, &si))
217		goto err;
218
219	if (so->so_proto->pr_flags & PR_CONNREQUIRED) {
220		bzero(&sopt, sizeof(sopt));
221		sopt.sopt_dir = SOPT_SET;
222		sopt.sopt_level = SOL_SOCKET;
223		sopt.sopt_name = SO_KEEPALIVE;
224		sopt.sopt_val = &one;
225		sopt.sopt_valsize = sizeof(one);
226		sosetopt(so, &sopt);
227	}
228
229	if (so->so_proto->pr_protocol == IPPROTO_TCP) {
230		bzero(&sopt, sizeof(sopt));
231		sopt.sopt_dir = SOPT_SET;
232		sopt.sopt_level = IPPROTO_TCP;
233		sopt.sopt_name = TCP_NODELAY;
234		sopt.sopt_val = &one;
235		sopt.sopt_valsize = sizeof(one);
236		sosetopt(so, &sopt);
237	}
238
239	ct->ct_closeit = FALSE;
240
241	/*
242	 * Set up private data struct
243	 */
244	ct->ct_socket = so;
245	ct->ct_wait.tv_sec = -1;
246	ct->ct_wait.tv_usec = -1;
247	memcpy(&ct->ct_addr, raddr, raddr->sa_len);
248
249	/*
250	 * Initialize call message
251	 */
252	getmicrotime(&now);
253	ct->ct_xid = ((uint32_t)++disrupt) ^ __RPC_GETXID(&now);
254	call_msg.rm_xid = ct->ct_xid;
255	call_msg.rm_direction = CALL;
256	call_msg.rm_call.cb_rpcvers = RPC_MSG_VERSION;
257	call_msg.rm_call.cb_prog = (uint32_t)prog;
258	call_msg.rm_call.cb_vers = (uint32_t)vers;
259
260	/*
261	 * pre-serialize the static part of the call msg and stash it away
262	 */
263	xdrmem_create(&xdrs, ct->ct_mcallc, MCALL_MSG_SIZE,
264	    XDR_ENCODE);
265	if (! xdr_callhdr(&xdrs, &call_msg)) {
266		if (ct->ct_closeit) {
267			soclose(ct->ct_socket);
268		}
269		goto err;
270	}
271	ct->ct_mpos = XDR_GETPOS(&xdrs);
272	XDR_DESTROY(&xdrs);
273	ct->ct_waitchan = "rpcrecv";
274	ct->ct_waitflag = 0;
275
276	/*
277	 * Create a client handle which uses xdrrec for serialization
278	 * and authnone for authentication.
279	 */
280	cl->cl_refs = 1;
281	cl->cl_ops = &clnt_vc_ops;
282	cl->cl_private = ct;
283	cl->cl_auth = authnone_create();
284	sendsz = __rpc_get_t_size(si.si_af, si.si_proto, (int)sendsz);
285	recvsz = __rpc_get_t_size(si.si_af, si.si_proto, (int)recvsz);
286	soreserve(ct->ct_socket, sendsz, recvsz);
287
288	SOCKBUF_LOCK(&ct->ct_socket->so_rcv);
289	ct->ct_socket->so_upcallarg = ct;
290	ct->ct_socket->so_upcall = clnt_vc_soupcall;
291	ct->ct_socket->so_rcv.sb_flags |= SB_UPCALL;
292	SOCKBUF_UNLOCK(&ct->ct_socket->so_rcv);
293
294	ct->ct_record = NULL;
295	ct->ct_record_resid = 0;
296	TAILQ_INIT(&ct->ct_pending);
297	return (cl);
298
299err:
300	if (cl) {
301		if (ct) {
302			mem_free(ct, sizeof (struct ct_data));
303		}
304		if (cl)
305			mem_free(cl, sizeof (CLIENT));
306	}
307	return ((CLIENT *)NULL);
308}
309
310static enum clnt_stat
311clnt_vc_call(
312	CLIENT		*cl,		/* client handle */
313	struct rpc_callextra *ext,	/* call metadata */
314	rpcproc_t	proc,		/* procedure number */
315	struct mbuf	*args,		/* pointer to args */
316	struct mbuf	**resultsp,	/* pointer to results */
317	struct timeval	utimeout)
318{
319	struct ct_data *ct = (struct ct_data *) cl->cl_private;
320	AUTH *auth;
321	struct rpc_err *errp;
322	enum clnt_stat stat;
323	XDR xdrs;
324	struct rpc_msg reply_msg;
325	bool_t ok;
326	int nrefreshes = 2;		/* number of times to refresh cred */
327	struct timeval timeout;
328	uint32_t xid;
329	struct mbuf *mreq = NULL, *results;
330	struct ct_request *cr;
331	int error;
332
333	cr = malloc(sizeof(struct ct_request), M_RPC, M_WAITOK);
334
335	mtx_lock(&ct->ct_lock);
336
337	if (ct->ct_closing || ct->ct_closed) {
338		mtx_unlock(&ct->ct_lock);
339		free(cr, M_RPC);
340		return (RPC_CANTSEND);
341	}
342	ct->ct_threads++;
343
344	if (ext) {
345		auth = ext->rc_auth;
346		errp = &ext->rc_err;
347	} else {
348		auth = cl->cl_auth;
349		errp = &ct->ct_error;
350	}
351
352	cr->cr_mrep = NULL;
353	cr->cr_error = 0;
354
355	if (ct->ct_wait.tv_usec == -1) {
356		timeout = utimeout;	/* use supplied timeout */
357	} else {
358		timeout = ct->ct_wait;	/* use default timeout */
359	}
360
361call_again:
362	mtx_assert(&ct->ct_lock, MA_OWNED);
363
364	ct->ct_xid++;
365	xid = ct->ct_xid;
366
367	mtx_unlock(&ct->ct_lock);
368
369	/*
370	 * Leave space to pre-pend the record mark.
371	 */
372	MGETHDR(mreq, M_WAIT, MT_DATA);
373	mreq->m_data += sizeof(uint32_t);
374	KASSERT(ct->ct_mpos + sizeof(uint32_t) <= MHLEN,
375	    ("RPC header too big"));
376	bcopy(ct->ct_mcallc, mreq->m_data, ct->ct_mpos);
377	mreq->m_len = ct->ct_mpos;
378
379	/*
380	 * The XID is the first thing in the request.
381	 */
382	*mtod(mreq, uint32_t *) = htonl(xid);
383
384	xdrmbuf_create(&xdrs, mreq, XDR_ENCODE);
385
386	errp->re_status = stat = RPC_SUCCESS;
387
388	if ((! XDR_PUTINT32(&xdrs, &proc)) ||
389	    (! AUTH_MARSHALL(auth, xid, &xdrs,
390		m_copym(args, 0, M_COPYALL, M_WAITOK)))) {
391		errp->re_status = stat = RPC_CANTENCODEARGS;
392		mtx_lock(&ct->ct_lock);
393		goto out;
394	}
395	mreq->m_pkthdr.len = m_length(mreq, NULL);
396
397	/*
398	 * Prepend a record marker containing the packet length.
399	 */
400	M_PREPEND(mreq, sizeof(uint32_t), M_WAIT);
401	*mtod(mreq, uint32_t *) =
402		htonl(0x80000000 | (mreq->m_pkthdr.len - sizeof(uint32_t)));
403
404	cr->cr_xid = xid;
405	mtx_lock(&ct->ct_lock);
406	TAILQ_INSERT_TAIL(&ct->ct_pending, cr, cr_link);
407	mtx_unlock(&ct->ct_lock);
408
409	/*
410	 * sosend consumes mreq.
411	 */
412	error = sosend(ct->ct_socket, NULL, NULL, mreq, NULL, 0, curthread);
413	mreq = NULL;
414	if (error == EMSGSIZE) {
415		SOCKBUF_LOCK(&ct->ct_socket->so_snd);
416		sbwait(&ct->ct_socket->so_snd);
417		SOCKBUF_UNLOCK(&ct->ct_socket->so_snd);
418		AUTH_VALIDATE(auth, xid, NULL, NULL);
419		mtx_lock(&ct->ct_lock);
420		TAILQ_REMOVE(&ct->ct_pending, cr, cr_link);
421		goto call_again;
422	}
423
424	reply_msg.acpted_rply.ar_verf.oa_flavor = AUTH_NULL;
425	reply_msg.acpted_rply.ar_verf.oa_base = cr->cr_verf;
426	reply_msg.acpted_rply.ar_verf.oa_length = 0;
427	reply_msg.acpted_rply.ar_results.where = NULL;
428	reply_msg.acpted_rply.ar_results.proc = (xdrproc_t)xdr_void;
429
430	mtx_lock(&ct->ct_lock);
431	if (error) {
432		TAILQ_REMOVE(&ct->ct_pending, cr, cr_link);
433		errp->re_errno = error;
434		errp->re_status = stat = RPC_CANTSEND;
435		goto out;
436	}
437
438	/*
439	 * Check to see if we got an upcall while waiting for the
440	 * lock. In both these cases, the request has been removed
441	 * from ct->ct_pending.
442	 */
443	if (cr->cr_error) {
444		TAILQ_REMOVE(&ct->ct_pending, cr, cr_link);
445		errp->re_errno = cr->cr_error;
446		errp->re_status = stat = RPC_CANTRECV;
447		goto out;
448	}
449	if (cr->cr_mrep) {
450		TAILQ_REMOVE(&ct->ct_pending, cr, cr_link);
451		goto got_reply;
452	}
453
454	/*
455	 * Hack to provide rpc-based message passing
456	 */
457	if (timeout.tv_sec == 0 && timeout.tv_usec == 0) {
458		TAILQ_REMOVE(&ct->ct_pending, cr, cr_link);
459		errp->re_status = stat = RPC_TIMEDOUT;
460		goto out;
461	}
462
463	error = msleep(cr, &ct->ct_lock, ct->ct_waitflag, ct->ct_waitchan,
464	    tvtohz(&timeout));
465
466	TAILQ_REMOVE(&ct->ct_pending, cr, cr_link);
467
468	if (error) {
469		/*
470		 * The sleep returned an error so our request is still
471		 * on the list. Turn the error code into an
472		 * appropriate client status.
473		 */
474		errp->re_errno = error;
475		switch (error) {
476		case EINTR:
477			stat = RPC_INTR;
478			break;
479		case EWOULDBLOCK:
480			stat = RPC_TIMEDOUT;
481			break;
482		default:
483			stat = RPC_CANTRECV;
484		}
485		errp->re_status = stat;
486		goto out;
487	} else {
488		/*
489		 * We were woken up by the upcall.  If the
490		 * upcall had a receive error, report that,
491		 * otherwise we have a reply.
492		 */
493		if (cr->cr_error) {
494			errp->re_errno = cr->cr_error;
495			errp->re_status = stat = RPC_CANTRECV;
496			goto out;
497		}
498	}
499
500got_reply:
501	/*
502	 * Now decode and validate the response. We need to drop the
503	 * lock since xdr_replymsg may end up sleeping in malloc.
504	 */
505	mtx_unlock(&ct->ct_lock);
506
507	if (ext && ext->rc_feedback)
508		ext->rc_feedback(FEEDBACK_OK, proc, ext->rc_feedback_arg);
509
510	xdrmbuf_create(&xdrs, cr->cr_mrep, XDR_DECODE);
511	ok = xdr_replymsg(&xdrs, &reply_msg);
512	cr->cr_mrep = NULL;
513
514	if (ok) {
515		if ((reply_msg.rm_reply.rp_stat == MSG_ACCEPTED) &&
516		    (reply_msg.acpted_rply.ar_stat == SUCCESS))
517			errp->re_status = stat = RPC_SUCCESS;
518		else
519			stat = _seterr_reply(&reply_msg, errp);
520
521		if (stat == RPC_SUCCESS) {
522			results = xdrmbuf_getall(&xdrs);
523			if (!AUTH_VALIDATE(auth, xid,
524				&reply_msg.acpted_rply.ar_verf,
525				&results)) {
526				errp->re_status = stat = RPC_AUTHERROR;
527				errp->re_why = AUTH_INVALIDRESP;
528			} else {
529				KASSERT(results,
530				    ("auth validated but no result"));
531				*resultsp = results;
532			}
533		}		/* end successful completion */
534		/*
535		 * If unsuccesful AND error is an authentication error
536		 * then refresh credentials and try again, else break
537		 */
538		else if (stat == RPC_AUTHERROR)
539			/* maybe our credentials need to be refreshed ... */
540			if (nrefreshes > 0 &&
541			    AUTH_REFRESH(auth, &reply_msg)) {
542				nrefreshes--;
543				XDR_DESTROY(&xdrs);
544				mtx_lock(&ct->ct_lock);
545				goto call_again;
546			}
547		/* end of unsuccessful completion */
548	}	/* end of valid reply message */
549	else {
550		errp->re_status = stat = RPC_CANTDECODERES;
551	}
552	XDR_DESTROY(&xdrs);
553	mtx_lock(&ct->ct_lock);
554out:
555	mtx_assert(&ct->ct_lock, MA_OWNED);
556
557	KASSERT(stat != RPC_SUCCESS || *resultsp,
558	    ("RPC_SUCCESS without reply"));
559
560	if (mreq)
561		m_freem(mreq);
562	if (cr->cr_mrep)
563		m_freem(cr->cr_mrep);
564
565	ct->ct_threads--;
566	if (ct->ct_closing)
567		wakeup(ct);
568
569	mtx_unlock(&ct->ct_lock);
570
571	if (auth && stat != RPC_SUCCESS)
572		AUTH_VALIDATE(auth, xid, NULL, NULL);
573
574	free(cr, M_RPC);
575
576	return (stat);
577}
578
579static void
580clnt_vc_geterr(CLIENT *cl, struct rpc_err *errp)
581{
582	struct ct_data *ct = (struct ct_data *) cl->cl_private;
583
584	*errp = ct->ct_error;
585}
586
587static bool_t
588clnt_vc_freeres(CLIENT *cl, xdrproc_t xdr_res, void *res_ptr)
589{
590	XDR xdrs;
591	bool_t dummy;
592
593	xdrs.x_op = XDR_FREE;
594	dummy = (*xdr_res)(&xdrs, res_ptr);
595
596	return (dummy);
597}
598
599/*ARGSUSED*/
600static void
601clnt_vc_abort(CLIENT *cl)
602{
603}
604
605static bool_t
606clnt_vc_control(CLIENT *cl, u_int request, void *info)
607{
608	struct ct_data *ct = (struct ct_data *)cl->cl_private;
609	void *infop = info;
610
611	mtx_lock(&ct->ct_lock);
612
613	switch (request) {
614	case CLSET_FD_CLOSE:
615		ct->ct_closeit = TRUE;
616		mtx_unlock(&ct->ct_lock);
617		return (TRUE);
618	case CLSET_FD_NCLOSE:
619		ct->ct_closeit = FALSE;
620		mtx_unlock(&ct->ct_lock);
621		return (TRUE);
622	default:
623		break;
624	}
625
626	/* for other requests which use info */
627	if (info == NULL) {
628		mtx_unlock(&ct->ct_lock);
629		return (FALSE);
630	}
631	switch (request) {
632	case CLSET_TIMEOUT:
633		if (time_not_ok((struct timeval *)info)) {
634			mtx_unlock(&ct->ct_lock);
635			return (FALSE);
636		}
637		ct->ct_wait = *(struct timeval *)infop;
638		break;
639	case CLGET_TIMEOUT:
640		*(struct timeval *)infop = ct->ct_wait;
641		break;
642	case CLGET_SERVER_ADDR:
643		(void) memcpy(info, &ct->ct_addr, (size_t)ct->ct_addr.ss_len);
644		break;
645	case CLGET_SVC_ADDR:
646		/*
647		 * Slightly different semantics to userland - we use
648		 * sockaddr instead of netbuf.
649		 */
650		memcpy(info, &ct->ct_addr, ct->ct_addr.ss_len);
651		break;
652	case CLSET_SVC_ADDR:		/* set to new address */
653		mtx_unlock(&ct->ct_lock);
654		return (FALSE);
655	case CLGET_XID:
656		*(uint32_t *)info = ct->ct_xid;
657		break;
658	case CLSET_XID:
659		/* This will set the xid of the NEXT call */
660		/* decrement by 1 as clnt_vc_call() increments once */
661		ct->ct_xid = *(uint32_t *)info - 1;
662		break;
663	case CLGET_VERS:
664		/*
665		 * This RELIES on the information that, in the call body,
666		 * the version number field is the fifth field from the
667		 * begining of the RPC header. MUST be changed if the
668		 * call_struct is changed
669		 */
670		*(uint32_t *)info =
671		    ntohl(*(uint32_t *)(void *)(ct->ct_mcallc +
672		    4 * BYTES_PER_XDR_UNIT));
673		break;
674
675	case CLSET_VERS:
676		*(uint32_t *)(void *)(ct->ct_mcallc +
677		    4 * BYTES_PER_XDR_UNIT) =
678		    htonl(*(uint32_t *)info);
679		break;
680
681	case CLGET_PROG:
682		/*
683		 * This RELIES on the information that, in the call body,
684		 * the program number field is the fourth field from the
685		 * begining of the RPC header. MUST be changed if the
686		 * call_struct is changed
687		 */
688		*(uint32_t *)info =
689		    ntohl(*(uint32_t *)(void *)(ct->ct_mcallc +
690		    3 * BYTES_PER_XDR_UNIT));
691		break;
692
693	case CLSET_PROG:
694		*(uint32_t *)(void *)(ct->ct_mcallc +
695		    3 * BYTES_PER_XDR_UNIT) =
696		    htonl(*(uint32_t *)info);
697		break;
698
699	case CLSET_WAITCHAN:
700		ct->ct_waitchan = (const char *)info;
701		break;
702
703	case CLGET_WAITCHAN:
704		*(const char **) info = ct->ct_waitchan;
705		break;
706
707	case CLSET_INTERRUPTIBLE:
708		if (*(int *) info)
709			ct->ct_waitflag = PCATCH;
710		else
711			ct->ct_waitflag = 0;
712		break;
713
714	case CLGET_INTERRUPTIBLE:
715		if (ct->ct_waitflag)
716			*(int *) info = TRUE;
717		else
718			*(int *) info = FALSE;
719		break;
720
721	default:
722		mtx_unlock(&ct->ct_lock);
723		return (FALSE);
724	}
725
726	mtx_unlock(&ct->ct_lock);
727	return (TRUE);
728}
729
730static void
731clnt_vc_close(CLIENT *cl)
732{
733	struct ct_data *ct = (struct ct_data *) cl->cl_private;
734	struct ct_request *cr;
735
736	mtx_lock(&ct->ct_lock);
737
738	if (ct->ct_closed) {
739		mtx_unlock(&ct->ct_lock);
740		return;
741	}
742
743	if (ct->ct_closing) {
744		while (ct->ct_closing)
745			msleep(ct, &ct->ct_lock, 0, "rpcclose", 0);
746		KASSERT(ct->ct_closed, ("client should be closed"));
747		mtx_unlock(&ct->ct_lock);
748		return;
749	}
750
751	if (ct->ct_socket) {
752		SOCKBUF_LOCK(&ct->ct_socket->so_rcv);
753		ct->ct_socket->so_upcallarg = NULL;
754		ct->ct_socket->so_upcall = NULL;
755		ct->ct_socket->so_rcv.sb_flags &= ~SB_UPCALL;
756		SOCKBUF_UNLOCK(&ct->ct_socket->so_rcv);
757
758		/*
759		 * Abort any pending requests and wait until everyone
760		 * has finished with clnt_vc_call.
761		 */
762		ct->ct_closing = TRUE;
763		TAILQ_FOREACH(cr, &ct->ct_pending, cr_link) {
764			cr->cr_xid = 0;
765			cr->cr_error = ESHUTDOWN;
766			wakeup(cr);
767		}
768
769		while (ct->ct_threads)
770			msleep(ct, &ct->ct_lock, 0, "rpcclose", 0);
771	}
772
773	ct->ct_closing = FALSE;
774	ct->ct_closed = TRUE;
775	mtx_unlock(&ct->ct_lock);
776	wakeup(ct);
777}
778
779static void
780clnt_vc_destroy(CLIENT *cl)
781{
782	struct ct_data *ct = (struct ct_data *) cl->cl_private;
783	struct socket *so = NULL;
784
785	clnt_vc_close(cl);
786
787	mtx_lock(&ct->ct_lock);
788
789	if (ct->ct_socket) {
790		if (ct->ct_closeit) {
791			so = ct->ct_socket;
792		}
793	}
794
795	mtx_unlock(&ct->ct_lock);
796
797	mtx_destroy(&ct->ct_lock);
798	if (so) {
799		soshutdown(so, SHUT_WR);
800		soclose(so);
801	}
802	mem_free(ct, sizeof(struct ct_data));
803	mem_free(cl, sizeof(CLIENT));
804}
805
806/*
807 * Make sure that the time is not garbage.   -1 value is disallowed.
808 * Note this is different from time_not_ok in clnt_dg.c
809 */
810static bool_t
811time_not_ok(struct timeval *t)
812{
813	return (t->tv_sec <= -1 || t->tv_sec > 100000000 ||
814		t->tv_usec <= -1 || t->tv_usec > 1000000);
815}
816
817void
818clnt_vc_soupcall(struct socket *so, void *arg, int waitflag)
819{
820	struct ct_data *ct = (struct ct_data *) arg;
821	struct uio uio;
822	struct mbuf *m;
823	struct ct_request *cr;
824	int error, rcvflag, foundreq;
825	uint32_t xid, header;
826	bool_t do_read;
827
828	uio.uio_td = curthread;
829	do {
830		/*
831		 * If ct_record_resid is zero, we are waiting for a
832		 * record mark.
833		 */
834		if (ct->ct_record_resid == 0) {
835
836			/*
837			 * Make sure there is either a whole record
838			 * mark in the buffer or there is some other
839			 * error condition
840			 */
841			do_read = FALSE;
842			SOCKBUF_LOCK(&so->so_rcv);
843			if (so->so_rcv.sb_cc >= sizeof(uint32_t)
844			    || (so->so_rcv.sb_state & SBS_CANTRCVMORE)
845			    || so->so_error)
846				do_read = TRUE;
847			SOCKBUF_UNLOCK(&so->so_rcv);
848
849			if (!do_read)
850				return;
851
852			uio.uio_resid = sizeof(uint32_t);
853			m = NULL;
854			rcvflag = MSG_DONTWAIT | MSG_SOCALLBCK;
855			error = soreceive(so, NULL, &uio, &m, NULL, &rcvflag);
856
857			if (error == EWOULDBLOCK)
858				break;
859
860			/*
861			 * If there was an error, wake up all pending
862			 * requests.
863			 */
864			if (error || uio.uio_resid > 0) {
865			wakeup_all:
866				mtx_lock(&ct->ct_lock);
867				if (!error) {
868					/*
869					 * We must have got EOF trying
870					 * to read from the stream.
871					 */
872					error = ECONNRESET;
873				}
874				ct->ct_error.re_status = RPC_CANTRECV;
875				ct->ct_error.re_errno = error;
876				TAILQ_FOREACH(cr, &ct->ct_pending, cr_link) {
877					cr->cr_error = error;
878					wakeup(cr);
879				}
880				mtx_unlock(&ct->ct_lock);
881				break;
882			}
883			bcopy(mtod(m, uint32_t *), &header, sizeof(uint32_t));
884			header = ntohl(header);
885			ct->ct_record = NULL;
886			ct->ct_record_resid = header & 0x7fffffff;
887			ct->ct_record_eor = ((header & 0x80000000) != 0);
888			m_freem(m);
889		} else {
890			/*
891			 * Wait until the socket has the whole record
892			 * buffered.
893			 */
894			do_read = FALSE;
895			SOCKBUF_LOCK(&so->so_rcv);
896			if (so->so_rcv.sb_cc >= ct->ct_record_resid
897			    || (so->so_rcv.sb_state & SBS_CANTRCVMORE)
898			    || so->so_error)
899				do_read = TRUE;
900			SOCKBUF_UNLOCK(&so->so_rcv);
901
902			if (!do_read)
903				return;
904
905			/*
906			 * We have the record mark. Read as much as
907			 * the socket has buffered up to the end of
908			 * this record.
909			 */
910			uio.uio_resid = ct->ct_record_resid;
911			m = NULL;
912			rcvflag = MSG_DONTWAIT | MSG_SOCALLBCK;
913			error = soreceive(so, NULL, &uio, &m, NULL, &rcvflag);
914
915			if (error == EWOULDBLOCK)
916				break;
917
918			if (error || uio.uio_resid == ct->ct_record_resid)
919				goto wakeup_all;
920
921			/*
922			 * If we have part of the record already,
923			 * chain this bit onto the end.
924			 */
925			if (ct->ct_record)
926				m_last(ct->ct_record)->m_next = m;
927			else
928				ct->ct_record = m;
929
930			ct->ct_record_resid = uio.uio_resid;
931
932			/*
933			 * If we have the entire record, see if we can
934			 * match it to a request.
935			 */
936			if (ct->ct_record_resid == 0
937			    && ct->ct_record_eor) {
938				/*
939				 * The XID is in the first uint32_t of
940				 * the reply.
941				 */
942				if (ct->ct_record->m_len < sizeof(xid))
943					ct->ct_record =
944						m_pullup(ct->ct_record,
945						    sizeof(xid));
946				if (!ct->ct_record)
947					break;
948				bcopy(mtod(ct->ct_record, uint32_t *),
949				    &xid, sizeof(uint32_t));
950				xid = ntohl(xid);
951
952				mtx_lock(&ct->ct_lock);
953				foundreq = 0;
954				TAILQ_FOREACH(cr, &ct->ct_pending, cr_link) {
955					if (cr->cr_xid == xid) {
956						/*
957						 * This one
958						 * matches. We leave
959						 * the reply mbuf in
960						 * cr->cr_mrep. Set
961						 * the XID to zero so
962						 * that we will ignore
963						 * any duplicaed
964						 * replies.
965						 */
966						cr->cr_xid = 0;
967						cr->cr_mrep = ct->ct_record;
968						cr->cr_error = 0;
969						foundreq = 1;
970						wakeup(cr);
971						break;
972					}
973				}
974				mtx_unlock(&ct->ct_lock);
975
976				if (!foundreq)
977					m_freem(ct->ct_record);
978				ct->ct_record = NULL;
979			}
980		}
981	} while (m);
982}
983