xref: /titanic_44/usr/src/uts/common/rpc/svc_cots.c (revision 6dfee4834394825da35b977ca71cdc965bc7b6a4)
1 /*
2  * CDDL HEADER START
3  *
4  * The contents of this file are subject to the terms of the
5  * Common Development and Distribution License, Version 1.0 only
6  * (the "License").  You may not use this file except in compliance
7  * with the License.
8  *
9  * You can obtain a copy of the license at usr/src/OPENSOLARIS.LICENSE
10  * or http://www.opensolaris.org/os/licensing.
11  * See the License for the specific language governing permissions
12  * and limitations under the License.
13  *
14  * When distributing Covered Code, include this CDDL HEADER in each
15  * file and include the License file at usr/src/OPENSOLARIS.LICENSE.
16  * If applicable, add the following below this CDDL HEADER, with the
17  * fields enclosed by brackets "[]" replaced with your own identifying
18  * information: Portions Copyright [yyyy] [name of copyright owner]
19  *
20  * CDDL HEADER END
21  */
22 /*
23  *  Copyright 2004 Sun Microsystems, Inc.  All rights reserved.
24  *  Use is subject to license terms.
25  */
26 
27 /*	Copyright (c) 1983, 1984, 1985, 1986, 1987, 1988, 1989 AT&T	*/
28 /*	  All Rights Reserved  	*/
29 
30 /*
31  * Portions of this source code were derived from Berkeley 4.3 BSD
32  * under license from the Regents of the University of California.
33  */
34 
35 #pragma ident	"%Z%%M%	%I%	%E% SMI"
36 
37 /*
38  * svc_cots.c
39  * Server side for connection-oriented RPC in the kernel.
40  *
41  */
42 
43 #include <sys/param.h>
44 #include <sys/types.h>
45 #include <sys/sysmacros.h>
46 #include <sys/file.h>
47 #include <sys/stream.h>
48 #include <sys/strsubr.h>
49 #include <sys/strsun.h>
50 #include <sys/stropts.h>
51 #include <sys/tiuser.h>
52 #include <sys/timod.h>
53 #include <sys/tihdr.h>
54 #include <sys/fcntl.h>
55 #include <sys/errno.h>
56 #include <sys/kmem.h>
57 #include <sys/systm.h>
58 #include <sys/debug.h>
59 #include <sys/cmn_err.h>
60 #include <sys/kstat.h>
61 #include <sys/vtrace.h>
62 
63 #include <rpc/types.h>
64 #include <rpc/xdr.h>
65 #include <rpc/auth.h>
66 #include <rpc/rpc_msg.h>
67 #include <rpc/svc.h>
68 
69 #define	COTS_MAX_ALLOCSIZE	2048
70 #define	MSG_OFFSET		128	/* offset of call into the mblk */
71 #define	RM_HDR_SIZE		4	/* record mark header size */
72 
73 /*
74  * Routines exported through ops vector.
75  */
76 static bool_t		svc_cots_krecv(SVCXPRT *, mblk_t *, struct rpc_msg *);
77 static bool_t		svc_cots_ksend(SVCXPRT *, struct rpc_msg *);
78 static bool_t		svc_cots_kgetargs(SVCXPRT *, xdrproc_t, caddr_t);
79 static bool_t		svc_cots_kfreeargs(SVCXPRT *, xdrproc_t, caddr_t);
80 static void		svc_cots_kdestroy(SVCMASTERXPRT *);
81 static int		svc_cots_kdup(struct svc_req *, caddr_t, int,
82 				struct dupreq **, bool_t *);
83 static void		svc_cots_kdupdone(struct dupreq *, caddr_t,
84 				void (*)(), int, int);
85 static int32_t		*svc_cots_kgetres(SVCXPRT *, int);
86 static void		svc_cots_kfreeres(SVCXPRT *);
87 static void		svc_cots_kclone_destroy(SVCXPRT *);
88 static void		svc_cots_kstart(SVCMASTERXPRT *);
89 
90 /*
91  * Server transport operations vector.
92  */
93 struct svc_ops svc_cots_op = {
94 	svc_cots_krecv,		/* Get requests */
95 	svc_cots_kgetargs,	/* Deserialize arguments */
96 	svc_cots_ksend,		/* Send reply */
97 	svc_cots_kfreeargs,	/* Free argument data space */
98 	svc_cots_kdestroy,	/* Destroy transport handle */
99 	svc_cots_kdup,		/* Check entry in dup req cache */
100 	svc_cots_kdupdone,	/* Mark entry in dup req cache as done */
101 	svc_cots_kgetres,	/* Get pointer to response buffer */
102 	svc_cots_kfreeres,	/* Destroy pre-serialized response header */
103 	svc_cots_kclone_destroy, /* Destroy a clone xprt */
104 	svc_cots_kstart		/* Tell `ready-to-receive' to rpcmod */
105 };
106 
107 /*
108  * Master transport private data.
109  * Kept in xprt->xp_p2.
110  */
111 struct cots_master_data {
112 	char	*cmd_src_addr;	/* client's address */
113 	int	cmd_xprt_started; /* flag for clone routine to call */
114 				/* rpcmod's start routine. */
115 	struct rpc_cots_server *cmd_stats;	/* stats for zone */
116 };
117 
118 /*
119  * Transport private data.
120  * Kept in clone_xprt->xp_p2buf.
121  */
122 typedef struct cots_data {
123 	mblk_t	*cd_mp;		/* pre-allocated reply message */
124 	mblk_t	*cd_req_mp;	/* request message */
125 } cots_data_t;
126 
127 /*
128  * Server statistics
129  * NOTE: This structure type is duplicated in the NFS fast path.
130  */
131 static const struct rpc_cots_server {
132 	kstat_named_t	rscalls;
133 	kstat_named_t	rsbadcalls;
134 	kstat_named_t	rsnullrecv;
135 	kstat_named_t	rsbadlen;
136 	kstat_named_t	rsxdrcall;
137 	kstat_named_t	rsdupchecks;
138 	kstat_named_t	rsdupreqs;
139 } cots_rsstat_tmpl = {
140 	{ "calls",	KSTAT_DATA_UINT64 },
141 	{ "badcalls",	KSTAT_DATA_UINT64 },
142 	{ "nullrecv",	KSTAT_DATA_UINT64 },
143 	{ "badlen",	KSTAT_DATA_UINT64 },
144 	{ "xdrcall",	KSTAT_DATA_UINT64 },
145 	{ "dupchecks",	KSTAT_DATA_UINT64 },
146 	{ "dupreqs",	KSTAT_DATA_UINT64 }
147 };
148 
149 #define	CLONE2STATS(clone_xprt)	\
150 	((struct cots_master_data *)(clone_xprt)->xp_master->xp_p2)->cmd_stats
151 #define	RSSTAT_INCR(s, x)	\
152 	atomic_add_64(&(s)->x.value.ui64, 1)
153 
154 /*
155  * Pointer to a transport specific `ready to receive' function in rpcmod
156  * (set from rpcmod).
157  */
158 void    (*mir_start)(queue_t *);
159 uint_t	*svc_max_msg_sizep;
160 
161 /*
162  * the address size of the underlying transport can sometimes be
163  * unknown (tinfo->ADDR_size == -1).  For this case, it is
164  * necessary to figure out what the size is so the correct amount
165  * of data is allocated.  This is an itterative process:
166  *	1. take a good guess (use T_MINADDRSIZE)
167  *	2. try it.
168  *	3. if it works then everything is ok
169  *	4. if the error is ENAMETOLONG, double the guess
170  *	5. go back to step 2.
171  */
172 #define	T_UNKNOWNADDRSIZE	(-1)
173 #define	T_MINADDRSIZE	32
174 
175 /*
176  * Create a transport record.
177  * The transport record, output buffer, and private data structure
178  * are allocated.  The output buffer is serialized into using xdrmem.
179  * There is one transport record per user process which implements a
180  * set of services.
181  */
182 static kmutex_t cots_kcreate_lock;
183 
184 int
185 svc_cots_kcreate(file_t *fp, uint_t max_msgsize, struct T_info_ack *tinfo,
186     SVCMASTERXPRT **nxprt)
187 {
188 	struct cots_master_data *cmd;
189 	int err;
190 	int retval;
191 	SVCMASTERXPRT *xprt;
192 	int addr_size;
193 	struct rpcstat *rpcstat;
194 
195 	if (nxprt == NULL)
196 		return (EINVAL);
197 
198 	rpcstat = zone_getspecific(rpcstat_zone_key, curproc->p_zone);
199 	ASSERT(rpcstat != NULL);
200 
201 	addr_size = tinfo->ADDR_size;
202 	if (addr_size == T_UNKNOWNADDRSIZE) {
203 	    addr_size = T_MINADDRSIZE;
204 	}
205 
206 allocate_space:
207 
208 	xprt = kmem_zalloc(sizeof (SVCMASTERXPRT), KM_SLEEP);
209 	cmd = kmem_zalloc(sizeof (*cmd) + addr_size, KM_SLEEP);
210 
211 	/* cd_src_addr is set to the end of cots_data_t struct */
212 	cmd->cmd_src_addr = (char *)&cmd[1];
213 
214 	if ((tinfo->TIDU_size > COTS_MAX_ALLOCSIZE) ||
215 	    (tinfo->TIDU_size <= 0))
216 		xprt->xp_msg_size = COTS_MAX_ALLOCSIZE;
217 	else {
218 		xprt->xp_msg_size = tinfo->TIDU_size -
219 			(tinfo->TIDU_size % BYTES_PER_XDR_UNIT);
220 	}
221 
222 	xprt->xp_ops = &svc_cots_op;
223 	xprt->xp_p2 = (caddr_t)cmd;
224 	cmd->cmd_xprt_started = 0;
225 	cmd->cmd_stats = rpcstat->rpc_cots_server;
226 
227 	xprt->xp_rtaddr.maxlen = addr_size;
228 	xprt->xp_rtaddr.len = 0;
229 	xprt->xp_rtaddr.buf = cmd->cmd_src_addr;
230 
231 	/*
232 	 * Get the address of the client for duplicate request
233 	 * cache processing. Note that the TI_GETPEERNAME ioctl should
234 	 * be replaced with a T_ADDR_REQ/T_ADDR_ACK handshake when
235 	 * TCP supports these standard TPI primitives.
236 	 */
237 	retval = 0;
238 	err = strioctl(fp->f_vnode, TI_GETPEERNAME,
239 	    (intptr_t)&xprt->xp_rtaddr, 0, K_TO_K, CRED(), &retval);
240 	if (err) {
241 		kmem_free(xprt, sizeof (SVCMASTERXPRT));
242 		kmem_free(cmd, sizeof (*cmd) + addr_size);
243 		if ((err == ENAMETOOLONG) &&
244 			(tinfo->ADDR_size == T_UNKNOWNADDRSIZE)) {
245 			addr_size *= 2;
246 			goto allocate_space;
247 		}
248 		return (err);
249 	}
250 
251 	/*
252 	 * If the current sanity check size in rpcmod is smaller
253 	 * than the size needed for this xprt, then increase
254 	 * the sanity check.
255 	 */
256 	if (max_msgsize != 0 && svc_max_msg_sizep &&
257 	    max_msgsize > *svc_max_msg_sizep) {
258 
259 		/* This check needs a lock */
260 		mutex_enter(&cots_kcreate_lock);
261 		if (svc_max_msg_sizep && max_msgsize > *svc_max_msg_sizep)
262 			*svc_max_msg_sizep = max_msgsize;
263 		mutex_exit(&cots_kcreate_lock);
264 	}
265 
266 	*nxprt = xprt;
267 	return (0);
268 }
269 
270 /*
271  * Destroy a master transport record.
272  * Frees the space allocated for a transport record.
273  */
274 static void
275 svc_cots_kdestroy(SVCMASTERXPRT *xprt)
276 {
277 	struct cots_master_data *cmd = (struct cots_master_data *)xprt->xp_p2;
278 
279 	ASSERT(cmd);
280 
281 	if (xprt->xp_netid)
282 		kmem_free(xprt->xp_netid, strlen(xprt->xp_netid) + 1);
283 	if (xprt->xp_addrmask.maxlen)
284 		kmem_free(xprt->xp_addrmask.buf, xprt->xp_addrmask.maxlen);
285 
286 	mutex_destroy(&xprt->xp_req_lock);
287 	mutex_destroy(&xprt->xp_thread_lock);
288 
289 	kmem_free(cmd, sizeof (*cmd) + xprt->xp_rtaddr.maxlen);
290 	kmem_free(xprt, sizeof (SVCMASTERXPRT));
291 }
292 
293 /*
294  * svc_tli_kcreate() calls this function at the end to tell
295  * rpcmod that the transport is ready to receive requests.
296  */
297 static void
298 svc_cots_kstart(SVCMASTERXPRT *xprt)
299 {
300 	struct cots_master_data *cmd = (struct cots_master_data *)xprt->xp_p2;
301 
302 	if (cmd->cmd_xprt_started == 0) {
303 		/*
304 		 * Acquire the xp_req_lock in order to use xp_wq
305 		 * safely (we don't want to qenable a queue that has
306 		 * already been closed).
307 		 */
308 		mutex_enter(&xprt->xp_req_lock);
309 		if (cmd->cmd_xprt_started == 0 &&
310 			xprt->xp_wq != NULL) {
311 			(*mir_start)(xprt->xp_wq);
312 			cmd->cmd_xprt_started = 1;
313 		}
314 		mutex_exit(&xprt->xp_req_lock);
315 	}
316 }
317 
318 /*
319  * Transport-type specific part of svc_xprt_cleanup().
320  */
321 static void
322 svc_cots_kclone_destroy(SVCXPRT *clone_xprt)
323 {
324 	cots_data_t *cd = (cots_data_t *)clone_xprt->xp_p2buf;
325 
326 	if (cd->cd_req_mp) {
327 		freemsg(cd->cd_req_mp);
328 		cd->cd_req_mp = (mblk_t *)0;
329 	}
330 	ASSERT(cd->cd_mp == NULL);
331 }
332 
333 /*
334  * Receive rpc requests.
335  * Checks if the message is intact, and deserializes the call packet.
336  */
337 static bool_t
338 svc_cots_krecv(SVCXPRT *clone_xprt, mblk_t *mp, struct rpc_msg *msg)
339 {
340 	cots_data_t *cd = (cots_data_t *)clone_xprt->xp_p2buf;
341 	XDR *xdrs = &clone_xprt->xp_xdrin;
342 	struct rpc_cots_server *stats = CLONE2STATS(clone_xprt);
343 
344 	TRACE_0(TR_FAC_KRPC, TR_SVC_COTS_KRECV_START,
345 	    "svc_cots_krecv_start:");
346 	RPCLOG(4, "svc_cots_krecv_start clone_xprt = %p:\n",
347 	    (void *)clone_xprt);
348 
349 	RSSTAT_INCR(stats, rscalls);
350 
351 	if (mp->b_datap->db_type != M_DATA) {
352 		RPCLOG(16, "svc_cots_krecv bad db_type %d\n",
353 		    mp->b_datap->db_type);
354 		goto bad;
355 	}
356 
357 	xdrmblk_init(xdrs, mp, XDR_DECODE, 0);
358 
359 	TRACE_0(TR_FAC_KRPC, TR_XDR_CALLMSG_START,
360 	    "xdr_callmsg_start:");
361 	RPCLOG0(4, "xdr_callmsg_start:\n");
362 	if (!xdr_callmsg(xdrs, msg)) {
363 		TRACE_1(TR_FAC_KRPC, TR_XDR_CALLMSG_END,
364 		    "xdr_callmsg_end:(%S)", "bad");
365 		RPCLOG0(1, "svc_cots_krecv xdr_callmsg failure\n");
366 		RSSTAT_INCR(stats, rsxdrcall);
367 		goto bad;
368 	}
369 	TRACE_1(TR_FAC_KRPC, TR_XDR_CALLMSG_END,
370 	    "xdr_callmsg_end:(%S)", "good");
371 
372 	clone_xprt->xp_xid = msg->rm_xid;
373 	cd->cd_req_mp = mp;
374 
375 	TRACE_1(TR_FAC_KRPC, TR_SVC_COTS_KRECV_END,
376 	    "svc_cots_krecv_end:(%S)", "good");
377 	RPCLOG0(4, "svc_cots_krecv_end:good\n");
378 	return (TRUE);
379 
380 bad:
381 	if (mp)
382 		freemsg(mp);
383 
384 	RSSTAT_INCR(stats, rsbadcalls);
385 	TRACE_1(TR_FAC_KRPC, TR_SVC_COTS_KRECV_END,
386 	    "svc_cots_krecv_end:(%S)", "bad");
387 	return (FALSE);
388 }
389 
390 /*
391  * Send rpc reply.
392  */
393 static bool_t
394 svc_cots_ksend(SVCXPRT *clone_xprt, struct rpc_msg *msg)
395 {
396 	/* LINTED pointer alignment */
397 	cots_data_t *cd = (cots_data_t *)clone_xprt->xp_p2buf;
398 	XDR *xdrs = &(clone_xprt->xp_xdrout);
399 	int retval = FALSE;
400 	mblk_t *mp;
401 	xdrproc_t xdr_results;
402 	caddr_t xdr_location;
403 	bool_t has_args;
404 
405 	TRACE_0(TR_FAC_KRPC, TR_SVC_COTS_KSEND_START,
406 	    "svc_cots_ksend_start:");
407 
408 	/*
409 	 * If there is a result procedure specified in the reply message,
410 	 * it will be processed in the xdr_replymsg and SVCAUTH_WRAP.
411 	 * We need to make sure it won't be processed twice, so we null
412 	 * it for xdr_replymsg here.
413 	 */
414 	has_args = FALSE;
415 	if (msg->rm_reply.rp_stat == MSG_ACCEPTED &&
416 	    msg->rm_reply.rp_acpt.ar_stat == SUCCESS) {
417 		if ((xdr_results = msg->acpted_rply.ar_results.proc) != NULL) {
418 			has_args = TRUE;
419 			xdr_location = msg->acpted_rply.ar_results.where;
420 			msg->acpted_rply.ar_results.proc = xdr_void;
421 			msg->acpted_rply.ar_results.where = NULL;
422 		}
423 	}
424 
425 	mp = cd->cd_mp;
426 	if (mp) {
427 		/*
428 		 * The program above pre-allocated an mblk and put
429 		 * the data in place.
430 		 */
431 		cd->cd_mp = (mblk_t *)NULL;
432 		if (!(xdr_replymsg_body(xdrs, msg) &&
433 		    (!has_args || SVCAUTH_WRAP(&clone_xprt->xp_auth, xdrs,
434 		    xdr_results, xdr_location)))) {
435 			RPCLOG0(1, "svc_cots_ksend: "
436 			    "xdr_replymsg_body/SVCAUTH_WRAP failed\n");
437 			freemsg(mp);
438 			goto out;
439 		}
440 	} else {
441 		int	len;
442 		int	mpsize;
443 
444 		/*
445 		 * Leave space for protocol headers.
446 		 */
447 		len = MSG_OFFSET + clone_xprt->xp_msg_size;
448 
449 		/*
450 		 * Allocate an initial mblk for the response data.
451 		 */
452 		while (!(mp = allocb(len, BPRI_LO))) {
453 			RPCLOG0(16, "svc_cots_ksend: allocb failed failed\n");
454 			if (strwaitbuf(len, BPRI_LO)) {
455 				TRACE_1(TR_FAC_KRPC, TR_SVC_COTS_KSEND_END,
456 				    "svc_cots_ksend_end:(%S)", "strwaitbuf");
457 				RPCLOG0(1,
458 				    "svc_cots_ksend: strwaitbuf failed\n");
459 				goto out;
460 			}
461 		}
462 
463 		/*
464 		 * Initialize the XDR decode stream.  Additional mblks
465 		 * will be allocated if necessary.  They will be TIDU
466 		 * sized.
467 		 */
468 		xdrmblk_init(xdrs, mp, XDR_ENCODE, clone_xprt->xp_msg_size);
469 		mpsize = MBLKSIZE(mp);
470 		ASSERT(mpsize >= len);
471 		ASSERT(mp->b_rptr == mp->b_datap->db_base);
472 
473 		/*
474 		 * If the size of mblk is not appreciably larger than what we
475 		 * asked, then resize the mblk to exactly len bytes. Reason for
476 		 * this: suppose len is 1600 bytes, the tidu is 1460 bytes
477 		 * (from TCP over ethernet), and the arguments to RPC require
478 		 * 2800 bytes. Ideally we want the protocol to render two
479 		 * ~1400 byte segments over the wire. If allocb() gives us a 2k
480 		 * mblk, and we allocate a second mblk for the rest, the
481 		 * protocol module may generate 3 segments over the wire:
482 		 * 1460 bytes for the first, 448 (2048 - 1600) for the 2nd, and
483 		 * 892 for the 3rd. If we "waste" 448 bytes in the first mblk,
484 		 * the XDR encoding will generate two ~1400 byte mblks, and the
485 		 * protocol module is more likely to produce properly sized
486 		 * segments.
487 		 */
488 		if ((mpsize >> 1) <= len) {
489 			mp->b_rptr += (mpsize - len);
490 		}
491 
492 		/*
493 		 * Adjust b_rptr to reserve space for the non-data protocol
494 		 * headers that any downstream modules might like to add, and
495 		 * for the record marking header.
496 		 */
497 		mp->b_rptr += (MSG_OFFSET + RM_HDR_SIZE);
498 
499 		XDR_SETPOS(xdrs, (uint_t)(mp->b_rptr - mp->b_datap->db_base));
500 		ASSERT(mp->b_wptr == mp->b_rptr);
501 
502 		msg->rm_xid = clone_xprt->xp_xid;
503 
504 		TRACE_0(TR_FAC_KRPC, TR_XDR_REPLYMSG_START,
505 		    "xdr_replymsg_start:");
506 		if (!(xdr_replymsg(xdrs, msg) &&
507 		    (!has_args || SVCAUTH_WRAP(&clone_xprt->xp_auth, xdrs,
508 		    xdr_results, xdr_location)))) {
509 			TRACE_1(TR_FAC_KRPC, TR_XDR_REPLYMSG_END,
510 			    "xdr_replymsg_end:(%S)", "bad");
511 			freemsg(mp);
512 			RPCLOG0(1, "svc_cots_ksend: xdr_replymsg/SVCAUTH_WRAP "
513 				"failed\n");
514 			goto out;
515 		}
516 		TRACE_1(TR_FAC_KRPC, TR_XDR_REPLYMSG_END,
517 		    "xdr_replymsg_end:(%S)", "good");
518 	}
519 
520 	put(clone_xprt->xp_wq, mp);
521 	retval = TRUE;
522 
523 out:
524 	/*
525 	 * This is completely disgusting.  If public is set it is
526 	 * a pointer to a structure whose first field is the address
527 	 * of the function to free that structure and any related
528 	 * stuff.  (see rrokfree in nfs_xdr.c).
529 	 */
530 	if (xdrs->x_public) {
531 		/* LINTED pointer alignment */
532 		(**((int (**)())xdrs->x_public))(xdrs->x_public);
533 	}
534 
535 	TRACE_1(TR_FAC_KRPC, TR_SVC_COTS_KSEND_END,
536 	    "svc_cots_ksend_end:(%S)", "done");
537 	return (retval);
538 }
539 
540 /*
541  * Deserialize arguments.
542  */
543 static bool_t
544 svc_cots_kgetargs(SVCXPRT *clone_xprt, xdrproc_t xdr_args,
545     caddr_t args_ptr)
546 {
547 	return (SVCAUTH_UNWRAP(&clone_xprt->xp_auth, &clone_xprt->xp_xdrin,
548 	    xdr_args, args_ptr));
549 }
550 
551 static bool_t
552 svc_cots_kfreeargs(SVCXPRT *clone_xprt, xdrproc_t xdr_args,
553     caddr_t args_ptr)
554 {
555 	cots_data_t *cd = (cots_data_t *)clone_xprt->xp_p2buf;
556 	mblk_t *mp;
557 	bool_t retval;
558 
559 	/*
560 	 * It is important to call the XDR routine before
561 	 * freeing the request mblk.  Structures in the
562 	 * XDR data may point into the mblk and require that
563 	 * the memory be intact during the free routine.
564 	 */
565 	if (args_ptr) {
566 		/* LINTED pointer alignment */
567 		XDR	*xdrs = &clone_xprt->xp_xdrin;
568 
569 		xdrs->x_op = XDR_FREE;
570 		retval = (*xdr_args)(xdrs, args_ptr);
571 	} else
572 		retval = TRUE;
573 
574 	if ((mp = cd->cd_req_mp) != NULL) {
575 		cd->cd_req_mp = (mblk_t *)0;
576 		freemsg(mp);
577 	}
578 
579 	return (retval);
580 }
581 
582 static int32_t *
583 svc_cots_kgetres(SVCXPRT *clone_xprt, int size)
584 {
585 	/* LINTED pointer alignment */
586 	cots_data_t *cd = (cots_data_t *)clone_xprt->xp_p2buf;
587 	XDR *xdrs = &clone_xprt->xp_xdrout;
588 	mblk_t *mp;
589 	int32_t *buf;
590 	struct rpc_msg rply;
591 	int len;
592 	int mpsize;
593 
594 	/*
595 	 * Leave space for protocol headers.
596 	 */
597 	len = MSG_OFFSET + clone_xprt->xp_msg_size;
598 
599 	/*
600 	 * Allocate an initial mblk for the response data.
601 	 */
602 	while ((mp = allocb(len, BPRI_LO)) == NULL) {
603 		if (strwaitbuf(len, BPRI_LO))
604 			return (FALSE);
605 	}
606 
607 	/*
608 	 * Initialize the XDR decode stream.  Additional mblks
609 	 * will be allocated if necessary.  They will be TIDU
610 	 * sized.
611 	 */
612 	xdrmblk_init(xdrs, mp, XDR_ENCODE, clone_xprt->xp_msg_size);
613 	mpsize = MBLKSIZE(mp);
614 	ASSERT(mpsize >= len);
615 	ASSERT(mp->b_rptr == mp->b_datap->db_base);
616 
617 	/*
618 	 * If the size of mblk is not appreciably larger than what we
619 	 * asked, then resize the mblk to exactly len bytes. Reason for
620 	 * this: suppose len is 1600 bytes, the tidu is 1460 bytes
621 	 * (from TCP over ethernet), and the arguments to RPC require
622 	 * 2800 bytes. Ideally we want the protocol to render two
623 	 * ~1400 byte segments over the wire. If allocb() gives us a 2k
624 	 * mblk, and we allocate a second mblk for the rest, the
625 	 * protocol module may generate 3 segments over the wire:
626 	 * 1460 bytes for the first, 448 (2048 - 1600) for the 2nd, and
627 	 * 892 for the 3rd. If we "waste" 448 bytes in the first mblk,
628 	 * the XDR encoding will generate two ~1400 byte mblks, and the
629 	 * protocol module is more likely to produce properly sized
630 	 * segments.
631 	 */
632 	if ((mpsize >> 1) <= len) {
633 		mp->b_rptr += (mpsize - len);
634 	}
635 
636 	/*
637 	 * Adjust b_rptr to reserve space for the non-data protocol
638 	 * headers that any downstream modules might like to add, and
639 	 * for the record marking header.
640 	 */
641 	mp->b_rptr += (MSG_OFFSET + RM_HDR_SIZE);
642 
643 	XDR_SETPOS(xdrs, (uint_t)(mp->b_rptr - mp->b_datap->db_base));
644 	ASSERT(mp->b_wptr == mp->b_rptr);
645 
646 	/*
647 	 * Assume a successful RPC since most of them are.
648 	 */
649 	rply.rm_xid = clone_xprt->xp_xid;
650 	rply.rm_direction = REPLY;
651 	rply.rm_reply.rp_stat = MSG_ACCEPTED;
652 	rply.acpted_rply.ar_verf = clone_xprt->xp_verf;
653 	rply.acpted_rply.ar_stat = SUCCESS;
654 
655 	if (!xdr_replymsg_hdr(xdrs, &rply)) {
656 		freeb(mp);
657 		return (NULL);
658 	}
659 
660 
661 	buf = XDR_INLINE(xdrs, size);
662 	if (buf == NULL) {
663 		ASSERT(cd->cd_mp == NULL);
664 		freemsg(mp);
665 	} else {
666 		cd->cd_mp = mp;
667 	}
668 	return (buf);
669 }
670 
671 static void
672 svc_cots_kfreeres(SVCXPRT *clone_xprt)
673 {
674 	cots_data_t *cd;
675 	mblk_t *mp;
676 
677 	cd = (cots_data_t *)clone_xprt->xp_p2buf;
678 	if ((mp = cd->cd_mp) != NULL) {
679 		cd->cd_mp = (mblk_t *)NULL;
680 		freemsg(mp);
681 	}
682 }
683 
684 /*
685  * the dup cacheing routines below provide a cache of non-failure
686  * transaction id's.  rpc service routines can use this to detect
687  * retransmissions and re-send a non-failure response.
688  */
689 
690 /*
691  * MAXDUPREQS is the number of cached items.  It should be adjusted
692  * to the service load so that there is likely to be a response entry
693  * when the first retransmission comes in.
694  */
695 #define	MAXDUPREQS	1024
696 
697 /*
698  * This should be appropriately scaled to MAXDUPREQS.
699  */
700 #define	DRHASHSZ	257
701 
702 #if ((DRHASHSZ & (DRHASHSZ - 1)) == 0)
703 #define	XIDHASH(xid)	((xid) & (DRHASHSZ - 1))
704 #else
705 #define	XIDHASH(xid)	((xid) % DRHASHSZ)
706 #endif
707 #define	DRHASH(dr)	XIDHASH((dr)->dr_xid)
708 #define	REQTOXID(req)	((req)->rq_xprt->xp_xid)
709 
710 static int	cotsndupreqs = 0;
711 static int	cotsmaxdupreqs = MAXDUPREQS;
712 static kmutex_t cotsdupreq_lock;
713 static struct dupreq *cotsdrhashtbl[DRHASHSZ];
714 static int	cotsdrhashstat[DRHASHSZ];
715 
716 static void unhash(struct dupreq *);
717 
718 /*
719  * cotsdrmru points to the head of a circular linked list in lru order.
720  * cotsdrmru->dr_next == drlru
721  */
722 struct dupreq *cotsdrmru;
723 
724 /*
725  * PSARC 2003/523 Contract Private Interface
726  * svc_cots_kdup
727  * Changes must be reviewed by Solaris File Sharing
728  * Changes must be communicated to contract-2003-523@sun.com
729  *
730  * svc_cots_kdup searches the request cache and returns 0 if the
731  * request is not found in the cache.  If it is found, then it
732  * returns the state of the request (in progress or done) and
733  * the status or attributes that were part of the original reply.
734  *
735  * If DUP_DONE (there is a duplicate) svc_cots_kdup copies over the
736  * value of the response. In that case, also return in *dupcachedp
737  * whether the response free routine is cached in the dupreq - in which case
738  * the caller should not be freeing it, because it will be done later
739  * in the svc_cots_kdup code when the dupreq is reused.
740  */
741 static int
742 svc_cots_kdup(struct svc_req *req, caddr_t res, int size, struct dupreq **drpp,
743 	bool_t *dupcachedp)
744 {
745 	struct rpc_cots_server *stats = CLONE2STATS(req->rq_xprt);
746 	struct dupreq *dr;
747 	uint32_t xid;
748 	uint32_t drhash;
749 	int status;
750 
751 	xid = REQTOXID(req);
752 	mutex_enter(&cotsdupreq_lock);
753 	RSSTAT_INCR(stats, rsdupchecks);
754 	/*
755 	 * Check to see whether an entry already exists in the cache.
756 	 */
757 	dr = cotsdrhashtbl[XIDHASH(xid)];
758 	while (dr != NULL) {
759 		if (dr->dr_xid == xid &&
760 		    dr->dr_proc == req->rq_proc &&
761 		    dr->dr_prog == req->rq_prog &&
762 		    dr->dr_vers == req->rq_vers &&
763 		    dr->dr_addr.len == req->rq_xprt->xp_rtaddr.len &&
764 		    bcmp((caddr_t)dr->dr_addr.buf,
765 		    (caddr_t)req->rq_xprt->xp_rtaddr.buf,
766 		    dr->dr_addr.len) == 0) {
767 			status = dr->dr_status;
768 			if (status == DUP_DONE) {
769 				bcopy(dr->dr_resp.buf, res, size);
770 				if (dupcachedp != NULL)
771 					*dupcachedp = (dr->dr_resfree != NULL);
772 				TRACE_0(TR_FAC_KRPC, TR_SVC_COTS_KDUP_DONE,
773 				    "svc_cots_kdup: DUP_DONE");
774 			} else {
775 				dr->dr_status = DUP_INPROGRESS;
776 				*drpp = dr;
777 				TRACE_0(TR_FAC_KRPC,
778 				    TR_SVC_COTS_KDUP_INPROGRESS,
779 				    "svc_cots_kdup: DUP_INPROGRESS");
780 			}
781 			RSSTAT_INCR(stats, rsdupreqs);
782 			mutex_exit(&cotsdupreq_lock);
783 			return (status);
784 		}
785 		dr = dr->dr_chain;
786 	}
787 
788 	/*
789 	 * There wasn't an entry, either allocate a new one or recycle
790 	 * an old one.
791 	 */
792 	if (cotsndupreqs < cotsmaxdupreqs) {
793 		dr = kmem_alloc(sizeof (*dr), KM_NOSLEEP);
794 		if (dr == NULL) {
795 			mutex_exit(&cotsdupreq_lock);
796 			return (DUP_ERROR);
797 		}
798 		dr->dr_resp.buf = NULL;
799 		dr->dr_resp.maxlen = 0;
800 		dr->dr_addr.buf = NULL;
801 		dr->dr_addr.maxlen = 0;
802 		if (cotsdrmru) {
803 			dr->dr_next = cotsdrmru->dr_next;
804 			cotsdrmru->dr_next = dr;
805 		} else {
806 			dr->dr_next = dr;
807 		}
808 		cotsndupreqs++;
809 	} else {
810 		dr = cotsdrmru->dr_next;
811 		while (dr->dr_status == DUP_INPROGRESS) {
812 			dr = dr->dr_next;
813 			if (dr == cotsdrmru->dr_next) {
814 				cmn_err(CE_WARN, "svc_cots_kdup no slots free");
815 				mutex_exit(&cotsdupreq_lock);
816 				return (DUP_ERROR);
817 			}
818 		}
819 		unhash(dr);
820 		if (dr->dr_resfree) {
821 			(*dr->dr_resfree)(dr->dr_resp.buf);
822 		}
823 	}
824 	dr->dr_resfree = NULL;
825 	cotsdrmru = dr;
826 
827 	dr->dr_xid = REQTOXID(req);
828 	dr->dr_prog = req->rq_prog;
829 	dr->dr_vers = req->rq_vers;
830 	dr->dr_proc = req->rq_proc;
831 	if (dr->dr_addr.maxlen < req->rq_xprt->xp_rtaddr.len) {
832 		if (dr->dr_addr.buf != NULL)
833 			kmem_free(dr->dr_addr.buf, dr->dr_addr.maxlen);
834 		dr->dr_addr.maxlen = req->rq_xprt->xp_rtaddr.len;
835 		dr->dr_addr.buf = kmem_alloc(dr->dr_addr.maxlen, KM_NOSLEEP);
836 		if (dr->dr_addr.buf == NULL) {
837 			dr->dr_addr.maxlen = 0;
838 			dr->dr_status = DUP_DROP;
839 			mutex_exit(&cotsdupreq_lock);
840 			return (DUP_ERROR);
841 		}
842 	}
843 	dr->dr_addr.len = req->rq_xprt->xp_rtaddr.len;
844 	bcopy(req->rq_xprt->xp_rtaddr.buf, dr->dr_addr.buf, dr->dr_addr.len);
845 	if (dr->dr_resp.maxlen < size) {
846 		if (dr->dr_resp.buf != NULL)
847 			kmem_free(dr->dr_resp.buf, dr->dr_resp.maxlen);
848 		dr->dr_resp.maxlen = (unsigned int)size;
849 		dr->dr_resp.buf = kmem_alloc(size, KM_NOSLEEP);
850 		if (dr->dr_resp.buf == NULL) {
851 			dr->dr_resp.maxlen = 0;
852 			dr->dr_status = DUP_DROP;
853 			mutex_exit(&cotsdupreq_lock);
854 			return (DUP_ERROR);
855 		}
856 	}
857 	dr->dr_status = DUP_INPROGRESS;
858 
859 	drhash = (uint32_t)DRHASH(dr);
860 	dr->dr_chain = cotsdrhashtbl[drhash];
861 	cotsdrhashtbl[drhash] = dr;
862 	cotsdrhashstat[drhash]++;
863 	mutex_exit(&cotsdupreq_lock);
864 	*drpp = dr;
865 	return (DUP_NEW);
866 }
867 
868 /*
869  * PSARC 2003/523 Contract Private Interface
870  * svc_cots_kdupdone
871  * Changes must be reviewed by Solaris File Sharing
872  * Changes must be communicated to contract-2003-523@sun.com
873  *
874  * svc_cots_kdupdone marks the request done (DUP_DONE or DUP_DROP)
875  * and stores the response.
876  */
877 static void
878 svc_cots_kdupdone(struct dupreq *dr, caddr_t res, void (*dis_resfree)(),
879 	int size, int status)
880 {
881 	ASSERT(dr->dr_resfree == NULL);
882 	if (status == DUP_DONE) {
883 		bcopy(res, dr->dr_resp.buf, size);
884 		dr->dr_resfree = dis_resfree;
885 	}
886 	dr->dr_status = status;
887 }
888 
889 /*
890  * This routine expects that the mutex, cotsdupreq_lock, is already held.
891  */
892 static void
893 unhash(struct dupreq *dr)
894 {
895 	struct dupreq *drt;
896 	struct dupreq *drtprev = NULL;
897 	uint32_t drhash;
898 
899 	ASSERT(MUTEX_HELD(&cotsdupreq_lock));
900 
901 	drhash = (uint32_t)DRHASH(dr);
902 	drt = cotsdrhashtbl[drhash];
903 	while (drt != NULL) {
904 		if (drt == dr) {
905 			cotsdrhashstat[drhash]--;
906 			if (drtprev == NULL) {
907 				cotsdrhashtbl[drhash] = drt->dr_chain;
908 			} else {
909 				drtprev->dr_chain = drt->dr_chain;
910 			}
911 			return;
912 		}
913 		drtprev = drt;
914 		drt = drt->dr_chain;
915 	}
916 }
917 
918 void
919 svc_cots_stats_init(zoneid_t zoneid, struct rpc_cots_server **statsp)
920 {
921 	*statsp = (struct rpc_cots_server *)rpcstat_zone_init_common(zoneid,
922 	    "unix", "rpc_cots_server", (const kstat_named_t *)&cots_rsstat_tmpl,
923 	    sizeof (cots_rsstat_tmpl));
924 }
925 
926 void
927 svc_cots_stats_fini(zoneid_t zoneid, struct rpc_cots_server **statsp)
928 {
929 	rpcstat_zone_fini_common(zoneid, "unix", "rpc_cots_server");
930 	kmem_free(*statsp, sizeof (cots_rsstat_tmpl));
931 }
932 
933 void
934 svc_cots_init(void)
935 {
936 	/*
937 	 * Check to make sure that the cots private data will fit into
938 	 * the stack buffer allocated by svc_run.  The ASSERT is a safety
939 	 * net if the cots_data_t structure ever changes.
940 	 */
941 	/*CONSTANTCONDITION*/
942 	ASSERT(sizeof (cots_data_t) <= SVC_P2LEN);
943 
944 	mutex_init(&cots_kcreate_lock, NULL, MUTEX_DEFAULT, NULL);
945 	mutex_init(&cotsdupreq_lock, NULL, MUTEX_DEFAULT, NULL);
946 }
947