xref: /titanic_51/usr/src/uts/common/inet/tcp/tcp_fusion.c (revision 8f176e333efb8f4f331f3fe919478983c4d5ee2e)
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 (the "License").
6  * You may not use this file except in compliance with the License.
7  *
8  * You can obtain a copy of the license at usr/src/OPENSOLARIS.LICENSE
9  * or http://www.opensolaris.org/os/licensing.
10  * See the License for the specific language governing permissions
11  * and limitations under the License.
12  *
13  * When distributing Covered Code, include this CDDL HEADER in each
14  * file and include the License file at usr/src/OPENSOLARIS.LICENSE.
15  * If applicable, add the following below this CDDL HEADER, with the
16  * fields enclosed by brackets "[]" replaced with your own identifying
17  * information: Portions Copyright [yyyy] [name of copyright owner]
18  *
19  * CDDL HEADER END
20  */
21 /*
22  * Copyright 2006 Sun Microsystems, Inc.  All rights reserved.
23  * Use is subject to license terms.
24  */
25 
26 #pragma ident	"%Z%%M%	%I%	%E% SMI"
27 
28 #include <sys/types.h>
29 #include <sys/stream.h>
30 #include <sys/strsun.h>
31 #include <sys/strsubr.h>
32 #include <sys/debug.h>
33 #include <sys/sdt.h>
34 #include <sys/cmn_err.h>
35 #include <sys/tihdr.h>
36 
37 #include <inet/common.h>
38 #include <inet/ip.h>
39 #include <inet/ip_impl.h>
40 #include <inet/tcp.h>
41 #include <inet/tcp_impl.h>
42 #include <inet/ipsec_impl.h>
43 #include <inet/ipclassifier.h>
44 #include <inet/ipp_common.h>
45 
46 /*
47  * This file implements TCP fusion - a protocol-less data path for TCP
48  * loopback connections.  The fusion of two local TCP endpoints occurs
49  * at connection establishment time.  Various conditions (see details
50  * in tcp_fuse()) need to be met for fusion to be successful.  If it
51  * fails, we fall back to the regular TCP data path; if it succeeds,
52  * both endpoints proceed to use tcp_fuse_output() as the transmit path.
53  * tcp_fuse_output() enqueues application data directly onto the peer's
54  * receive queue; no protocol processing is involved.  After enqueueing
55  * the data, the sender can either push (putnext) data up the receiver's
56  * read queue; or the sender can simply return and let the receiver
57  * retrieve the enqueued data via the synchronous streams entry point
58  * tcp_fuse_rrw().  The latter path is taken if synchronous streams is
59  * enabled (the default).  It is disabled if sockfs no longer resides
60  * directly on top of tcp module due to a module insertion or removal.
61  * It also needs to be temporarily disabled when sending urgent data
62  * because the tcp_fuse_rrw() path bypasses the M_PROTO processing done
63  * by strsock_proto() hook.
64  *
65  * Sychronization is handled by squeue and the mutex tcp_fuse_lock.
66  * One of the requirements for fusion to succeed is that both endpoints
67  * need to be using the same squeue.  This ensures that neither side
68  * can disappear while the other side is still sending data.  By itself,
69  * squeue is not sufficient for guaranteeing safety when synchronous
70  * streams is enabled.  The reason is that tcp_fuse_rrw() doesn't enter
71  * the squeue and its access to tcp_rcv_list and other fusion-related
72  * fields needs to be sychronized with the sender.  tcp_fuse_lock is
73  * used for this purpose.  When there is urgent data, the sender needs
74  * to push the data up the receiver's streams read queue.  In order to
75  * avoid holding the tcp_fuse_lock across putnext(), the sender sets
76  * the peer tcp's tcp_fuse_syncstr_plugged bit and releases tcp_fuse_lock
77  * (see macro TCP_FUSE_SYNCSTR_PLUG_DRAIN()).  If tcp_fuse_rrw() enters
78  * after this point, it will see that synchronous streams is plugged and
79  * will wait on tcp_fuse_plugcv.  After the sender has finished pushing up
80  * all urgent data, it will clear the tcp_fuse_syncstr_plugged bit using
81  * TCP_FUSE_SYNCSTR_UNPLUG_DRAIN().  This will cause any threads waiting
82  * on tcp_fuse_plugcv to return EBUSY, and in turn cause strget() to call
83  * getq_noenab() to dequeue data from the stream head instead.  Once the
84  * data on the stream head has been consumed, tcp_fuse_rrw() may again
85  * be used to process tcp_rcv_list.  However, if TCP_FUSE_SYNCSTR_STOP()
86  * has been called, all future calls to tcp_fuse_rrw() will return EBUSY,
87  * effectively disabling synchronous streams.
88  *
89  * The following note applies only to the synchronous streams mode.
90  *
91  * Flow control is done by checking the size of receive buffer and
92  * the number of data blocks, both set to different limits.  This is
93  * different than regular streams flow control where cumulative size
94  * check dominates block count check -- streams queue high water mark
95  * typically represents bytes.  Each enqueue triggers notifications
96  * to the receiving process; a build up of data blocks indicates a
97  * slow receiver and the sender should be blocked or informed at the
98  * earliest moment instead of further wasting system resources.  In
99  * effect, this is equivalent to limiting the number of outstanding
100  * segments in flight.
101  */
102 
103 /*
104  * Macros that determine whether or not IP processing is needed for TCP.
105  */
106 #define	TCP_IPOPT_POLICY_V4(tcp)					\
107 	((tcp)->tcp_ipversion == IPV4_VERSION &&			\
108 	((tcp)->tcp_ip_hdr_len != IP_SIMPLE_HDR_LENGTH ||		\
109 	CONN_OUTBOUND_POLICY_PRESENT((tcp)->tcp_connp) ||		\
110 	CONN_INBOUND_POLICY_PRESENT((tcp)->tcp_connp)))
111 
112 #define	TCP_IPOPT_POLICY_V6(tcp)					\
113 	((tcp)->tcp_ipversion == IPV6_VERSION &&			\
114 	((tcp)->tcp_ip_hdr_len != IPV6_HDR_LEN ||			\
115 	CONN_OUTBOUND_POLICY_PRESENT_V6((tcp)->tcp_connp) ||		\
116 	CONN_INBOUND_POLICY_PRESENT_V6((tcp)->tcp_connp)))
117 
118 #define	TCP_LOOPBACK_IP(tcp)						\
119 	(TCP_IPOPT_POLICY_V4(tcp) || TCP_IPOPT_POLICY_V6(tcp) ||	\
120 	!CONN_IS_LSO_MD_FASTPATH((tcp)->tcp_connp))
121 
122 /*
123  * Setting this to false means we disable fusion altogether and
124  * loopback connections would go through the protocol paths.
125  */
126 boolean_t do_tcp_fusion = B_TRUE;
127 
128 /*
129  * Enabling this flag allows sockfs to retrieve data directly
130  * from a fused tcp endpoint using synchronous streams interface.
131  */
132 boolean_t do_tcp_direct_sockfs = B_TRUE;
133 
134 /*
135  * This is the minimum amount of outstanding writes allowed on
136  * a synchronous streams-enabled receiving endpoint before the
137  * sender gets flow-controlled.  Setting this value to 0 means
138  * that the data block limit is equivalent to the byte count
139  * limit, which essentially disables the check.
140  */
141 #define	TCP_FUSION_RCV_UNREAD_MIN	8
142 uint_t tcp_fusion_rcv_unread_min = TCP_FUSION_RCV_UNREAD_MIN;
143 
144 static void	tcp_fuse_syncstr_enable(tcp_t *);
145 static void	tcp_fuse_syncstr_disable(tcp_t *);
146 static void	strrput_sig(queue_t *, boolean_t);
147 
148 /*
149  * This routine gets called by the eager tcp upon changing state from
150  * SYN_RCVD to ESTABLISHED.  It fuses a direct path between itself
151  * and the active connect tcp such that the regular tcp processings
152  * may be bypassed under allowable circumstances.  Because the fusion
153  * requires both endpoints to be in the same squeue, it does not work
154  * for simultaneous active connects because there is no easy way to
155  * switch from one squeue to another once the connection is created.
156  * This is different from the eager tcp case where we assign it the
157  * same squeue as the one given to the active connect tcp during open.
158  */
159 void
160 tcp_fuse(tcp_t *tcp, uchar_t *iphdr, tcph_t *tcph)
161 {
162 	conn_t *peer_connp, *connp = tcp->tcp_connp;
163 	tcp_t *peer_tcp;
164 
165 	ASSERT(!tcp->tcp_fused);
166 	ASSERT(tcp->tcp_loopback);
167 	ASSERT(tcp->tcp_loopback_peer == NULL);
168 	/*
169 	 * We need to inherit q_hiwat of the listener tcp, but we can't
170 	 * really use tcp_listener since we get here after sending up
171 	 * T_CONN_IND and tcp_wput_accept() may be called independently,
172 	 * at which point tcp_listener is cleared; this is why we use
173 	 * tcp_saved_listener.  The listener itself is guaranteed to be
174 	 * around until tcp_accept_finish() is called on this eager --
175 	 * this won't happen until we're done since we're inside the
176 	 * eager's perimeter now.
177 	 */
178 	ASSERT(tcp->tcp_saved_listener != NULL);
179 
180 	/*
181 	 * Lookup peer endpoint; search for the remote endpoint having
182 	 * the reversed address-port quadruplet in ESTABLISHED state,
183 	 * which is guaranteed to be unique in the system.  Zone check
184 	 * is applied accordingly for loopback address, but not for
185 	 * local address since we want fusion to happen across Zones.
186 	 */
187 	if (tcp->tcp_ipversion == IPV4_VERSION) {
188 		peer_connp = ipcl_conn_tcp_lookup_reversed_ipv4(connp,
189 		    (ipha_t *)iphdr, tcph);
190 	} else {
191 		peer_connp = ipcl_conn_tcp_lookup_reversed_ipv6(connp,
192 		    (ip6_t *)iphdr, tcph);
193 	}
194 
195 	/*
196 	 * We can only proceed if peer exists, resides in the same squeue
197 	 * as our conn and is not raw-socket.  The squeue assignment of
198 	 * this eager tcp was done earlier at the time of SYN processing
199 	 * in ip_fanout_tcp{_v6}.  Note that similar squeues by itself
200 	 * doesn't guarantee a safe condition to fuse, hence we perform
201 	 * additional tests below.
202 	 */
203 	ASSERT(peer_connp == NULL || peer_connp != connp);
204 	if (peer_connp == NULL || peer_connp->conn_sqp != connp->conn_sqp ||
205 	    !IPCL_IS_TCP(peer_connp)) {
206 		if (peer_connp != NULL) {
207 			TCP_STAT(tcp_fusion_unqualified);
208 			CONN_DEC_REF(peer_connp);
209 		}
210 		return;
211 	}
212 	peer_tcp = peer_connp->conn_tcp;	/* active connect tcp */
213 
214 	ASSERT(peer_tcp != NULL && peer_tcp != tcp && !peer_tcp->tcp_fused);
215 	ASSERT(peer_tcp->tcp_loopback && peer_tcp->tcp_loopback_peer == NULL);
216 	ASSERT(peer_connp->conn_sqp == connp->conn_sqp);
217 
218 	/*
219 	 * Fuse the endpoints; we perform further checks against both
220 	 * tcp endpoints to ensure that a fusion is allowed to happen.
221 	 * In particular we bail out for non-simple TCP/IP or if IPsec/
222 	 * IPQoS policy/kernel SSL exists.
223 	 */
224 	if (!tcp->tcp_unfusable && !peer_tcp->tcp_unfusable &&
225 	    !TCP_LOOPBACK_IP(tcp) && !TCP_LOOPBACK_IP(peer_tcp) &&
226 	    tcp->tcp_kssl_ent == NULL &&
227 	    !IPP_ENABLED(IPP_LOCAL_OUT|IPP_LOCAL_IN)) {
228 		mblk_t *mp;
229 		struct stroptions *stropt;
230 		queue_t *peer_rq = peer_tcp->tcp_rq;
231 
232 		ASSERT(!TCP_IS_DETACHED(peer_tcp) && peer_rq != NULL);
233 		ASSERT(tcp->tcp_fused_sigurg_mp == NULL);
234 		ASSERT(peer_tcp->tcp_fused_sigurg_mp == NULL);
235 		ASSERT(tcp->tcp_kssl_ctx == NULL);
236 
237 		/*
238 		 * We need to drain data on both endpoints during unfuse.
239 		 * If we need to send up SIGURG at the time of draining,
240 		 * we want to be sure that an mblk is readily available.
241 		 * This is why we pre-allocate the M_PCSIG mblks for both
242 		 * endpoints which will only be used during/after unfuse.
243 		 */
244 		if ((mp = allocb(1, BPRI_HI)) == NULL)
245 			goto failed;
246 
247 		tcp->tcp_fused_sigurg_mp = mp;
248 
249 		if ((mp = allocb(1, BPRI_HI)) == NULL)
250 			goto failed;
251 
252 		peer_tcp->tcp_fused_sigurg_mp = mp;
253 
254 		/* Allocate M_SETOPTS mblk */
255 		if ((mp = allocb(sizeof (*stropt), BPRI_HI)) == NULL)
256 			goto failed;
257 
258 		/* Fuse both endpoints */
259 		peer_tcp->tcp_loopback_peer = tcp;
260 		tcp->tcp_loopback_peer = peer_tcp;
261 		peer_tcp->tcp_fused = tcp->tcp_fused = B_TRUE;
262 
263 		/*
264 		 * We never use regular tcp paths in fusion and should
265 		 * therefore clear tcp_unsent on both endpoints.  Having
266 		 * them set to non-zero values means asking for trouble
267 		 * especially after unfuse, where we may end up sending
268 		 * through regular tcp paths which expect xmit_list and
269 		 * friends to be correctly setup.
270 		 */
271 		peer_tcp->tcp_unsent = tcp->tcp_unsent = 0;
272 
273 		tcp_timers_stop(tcp);
274 		tcp_timers_stop(peer_tcp);
275 
276 		/*
277 		 * At this point we are a detached eager tcp and therefore
278 		 * don't have a queue assigned to us until accept happens.
279 		 * In the mean time the peer endpoint may immediately send
280 		 * us data as soon as fusion is finished, and we need to be
281 		 * able to flow control it in case it sends down huge amount
282 		 * of data while we're still detached.  To prevent that we
283 		 * inherit the listener's q_hiwat value; this is temporary
284 		 * since we'll repeat the process in tcp_accept_finish().
285 		 */
286 		(void) tcp_fuse_set_rcv_hiwat(tcp,
287 		    tcp->tcp_saved_listener->tcp_rq->q_hiwat);
288 
289 		/*
290 		 * Set the stream head's write offset value to zero since we
291 		 * won't be needing any room for TCP/IP headers; tell it to
292 		 * not break up the writes (this would reduce the amount of
293 		 * work done by kmem); and configure our receive buffer.
294 		 * Note that we can only do this for the active connect tcp
295 		 * since our eager is still detached; it will be dealt with
296 		 * later in tcp_accept_finish().
297 		 */
298 		DB_TYPE(mp) = M_SETOPTS;
299 		mp->b_wptr += sizeof (*stropt);
300 
301 		stropt = (struct stroptions *)mp->b_rptr;
302 		stropt->so_flags = SO_MAXBLK | SO_WROFF | SO_HIWAT;
303 		stropt->so_maxblk = tcp_maxpsz_set(peer_tcp, B_FALSE);
304 		stropt->so_wroff = 0;
305 
306 		/*
307 		 * Record the stream head's high water mark for
308 		 * peer endpoint; this is used for flow-control
309 		 * purposes in tcp_fuse_output().
310 		 */
311 		stropt->so_hiwat = tcp_fuse_set_rcv_hiwat(peer_tcp,
312 		    peer_rq->q_hiwat);
313 
314 		/* Send the options up */
315 		putnext(peer_rq, mp);
316 	} else {
317 		TCP_STAT(tcp_fusion_unqualified);
318 	}
319 	CONN_DEC_REF(peer_connp);
320 	return;
321 
322 failed:
323 	if (tcp->tcp_fused_sigurg_mp != NULL) {
324 		freeb(tcp->tcp_fused_sigurg_mp);
325 		tcp->tcp_fused_sigurg_mp = NULL;
326 	}
327 	if (peer_tcp->tcp_fused_sigurg_mp != NULL) {
328 		freeb(peer_tcp->tcp_fused_sigurg_mp);
329 		peer_tcp->tcp_fused_sigurg_mp = NULL;
330 	}
331 	CONN_DEC_REF(peer_connp);
332 }
333 
334 /*
335  * Unfuse a previously-fused pair of tcp loopback endpoints.
336  */
337 void
338 tcp_unfuse(tcp_t *tcp)
339 {
340 	tcp_t *peer_tcp = tcp->tcp_loopback_peer;
341 
342 	ASSERT(tcp->tcp_fused && peer_tcp != NULL);
343 	ASSERT(peer_tcp->tcp_fused && peer_tcp->tcp_loopback_peer == tcp);
344 	ASSERT(tcp->tcp_connp->conn_sqp == peer_tcp->tcp_connp->conn_sqp);
345 	ASSERT(tcp->tcp_unsent == 0 && peer_tcp->tcp_unsent == 0);
346 	ASSERT(tcp->tcp_fused_sigurg_mp != NULL);
347 	ASSERT(peer_tcp->tcp_fused_sigurg_mp != NULL);
348 
349 	/*
350 	 * We disable synchronous streams, drain any queued data and
351 	 * clear tcp_direct_sockfs.  The synchronous streams entry
352 	 * points will become no-ops after this point.
353 	 */
354 	tcp_fuse_disable_pair(tcp, B_TRUE);
355 
356 	/*
357 	 * Update th_seq and th_ack in the header template
358 	 */
359 	U32_TO_ABE32(tcp->tcp_snxt, tcp->tcp_tcph->th_seq);
360 	U32_TO_ABE32(tcp->tcp_rnxt, tcp->tcp_tcph->th_ack);
361 	U32_TO_ABE32(peer_tcp->tcp_snxt, peer_tcp->tcp_tcph->th_seq);
362 	U32_TO_ABE32(peer_tcp->tcp_rnxt, peer_tcp->tcp_tcph->th_ack);
363 
364 	/* Unfuse the endpoints */
365 	peer_tcp->tcp_fused = tcp->tcp_fused = B_FALSE;
366 	peer_tcp->tcp_loopback_peer = tcp->tcp_loopback_peer = NULL;
367 }
368 
369 /*
370  * Fusion output routine for urgent data.  This routine is called by
371  * tcp_fuse_output() for handling non-M_DATA mblks.
372  */
373 void
374 tcp_fuse_output_urg(tcp_t *tcp, mblk_t *mp)
375 {
376 	mblk_t *mp1;
377 	struct T_exdata_ind *tei;
378 	tcp_t *peer_tcp = tcp->tcp_loopback_peer;
379 	mblk_t *head, *prev_head = NULL;
380 
381 	ASSERT(tcp->tcp_fused);
382 	ASSERT(peer_tcp != NULL && peer_tcp->tcp_loopback_peer == tcp);
383 	ASSERT(DB_TYPE(mp) == M_PROTO || DB_TYPE(mp) == M_PCPROTO);
384 	ASSERT(mp->b_cont != NULL && DB_TYPE(mp->b_cont) == M_DATA);
385 	ASSERT(MBLKL(mp) >= sizeof (*tei) && MBLKL(mp->b_cont) > 0);
386 
387 	/*
388 	 * Urgent data arrives in the form of T_EXDATA_REQ from above.
389 	 * Each occurence denotes a new urgent pointer.  For each new
390 	 * urgent pointer we signal (SIGURG) the receiving app to indicate
391 	 * that it needs to go into urgent mode.  This is similar to the
392 	 * urgent data handling in the regular tcp.  We don't need to keep
393 	 * track of where the urgent pointer is, because each T_EXDATA_REQ
394 	 * "advances" the urgent pointer for us.
395 	 *
396 	 * The actual urgent data carried by T_EXDATA_REQ is then prepended
397 	 * by a T_EXDATA_IND before being enqueued behind any existing data
398 	 * destined for the receiving app.  There is only a single urgent
399 	 * pointer (out-of-band mark) for a given tcp.  If the new urgent
400 	 * data arrives before the receiving app reads some existing urgent
401 	 * data, the previous marker is lost.  This behavior is emulated
402 	 * accordingly below, by removing any existing T_EXDATA_IND messages
403 	 * and essentially converting old urgent data into non-urgent.
404 	 */
405 	ASSERT(tcp->tcp_valid_bits & TCP_URG_VALID);
406 	/* Let sender get out of urgent mode */
407 	tcp->tcp_valid_bits &= ~TCP_URG_VALID;
408 
409 	/*
410 	 * This flag indicates that a signal needs to be sent up.
411 	 * This flag will only get cleared once SIGURG is delivered and
412 	 * is not affected by the tcp_fused flag -- delivery will still
413 	 * happen even after an endpoint is unfused, to handle the case
414 	 * where the sending endpoint immediately closes/unfuses after
415 	 * sending urgent data and the accept is not yet finished.
416 	 */
417 	peer_tcp->tcp_fused_sigurg = B_TRUE;
418 
419 	/* Reuse T_EXDATA_REQ mblk for T_EXDATA_IND */
420 	DB_TYPE(mp) = M_PROTO;
421 	tei = (struct T_exdata_ind *)mp->b_rptr;
422 	tei->PRIM_type = T_EXDATA_IND;
423 	tei->MORE_flag = 0;
424 	mp->b_wptr = (uchar_t *)&tei[1];
425 
426 	TCP_STAT(tcp_fusion_urg);
427 	BUMP_MIB(&tcp_mib, tcpOutUrg);
428 
429 	head = peer_tcp->tcp_rcv_list;
430 	while (head != NULL) {
431 		/*
432 		 * Remove existing T_EXDATA_IND, keep the data which follows
433 		 * it and relink our list.  Note that we don't modify the
434 		 * tcp_rcv_last_tail since it never points to T_EXDATA_IND.
435 		 */
436 		if (DB_TYPE(head) != M_DATA) {
437 			mp1 = head;
438 
439 			ASSERT(DB_TYPE(mp1->b_cont) == M_DATA);
440 			head = mp1->b_cont;
441 			mp1->b_cont = NULL;
442 			head->b_next = mp1->b_next;
443 			mp1->b_next = NULL;
444 			if (prev_head != NULL)
445 				prev_head->b_next = head;
446 			if (peer_tcp->tcp_rcv_list == mp1)
447 				peer_tcp->tcp_rcv_list = head;
448 			if (peer_tcp->tcp_rcv_last_head == mp1)
449 				peer_tcp->tcp_rcv_last_head = head;
450 			freeb(mp1);
451 		}
452 		prev_head = head;
453 		head = head->b_next;
454 	}
455 }
456 
457 /*
458  * Fusion output routine, called by tcp_output() and tcp_wput_proto().
459  */
460 boolean_t
461 tcp_fuse_output(tcp_t *tcp, mblk_t *mp, uint32_t send_size)
462 {
463 	tcp_t *peer_tcp = tcp->tcp_loopback_peer;
464 	uint_t max_unread;
465 	boolean_t flow_stopped;
466 	boolean_t urgent = (DB_TYPE(mp) != M_DATA);
467 	mblk_t *mp1 = mp;
468 	ill_t *ilp, *olp;
469 	ipha_t *ipha;
470 	ip6_t *ip6h;
471 	tcph_t *tcph;
472 	uint_t ip_hdr_len;
473 	uint32_t seq;
474 	uint32_t recv_size = send_size;
475 
476 	ASSERT(tcp->tcp_fused);
477 	ASSERT(peer_tcp != NULL && peer_tcp->tcp_loopback_peer == tcp);
478 	ASSERT(tcp->tcp_connp->conn_sqp == peer_tcp->tcp_connp->conn_sqp);
479 	ASSERT(DB_TYPE(mp) == M_DATA || DB_TYPE(mp) == M_PROTO ||
480 	    DB_TYPE(mp) == M_PCPROTO);
481 
482 	max_unread = peer_tcp->tcp_fuse_rcv_unread_hiwater;
483 
484 	/* If this connection requires IP, unfuse and use regular path */
485 	if (TCP_LOOPBACK_IP(tcp) || TCP_LOOPBACK_IP(peer_tcp) ||
486 	    IPP_ENABLED(IPP_LOCAL_OUT|IPP_LOCAL_IN)) {
487 		TCP_STAT(tcp_fusion_aborted);
488 		goto unfuse;
489 	}
490 
491 	if (send_size == 0) {
492 		freemsg(mp);
493 		return (B_TRUE);
494 	}
495 
496 	/*
497 	 * Handle urgent data; we either send up SIGURG to the peer now
498 	 * or do it later when we drain, in case the peer is detached
499 	 * or if we're short of memory for M_PCSIG mblk.
500 	 */
501 	if (urgent) {
502 		/*
503 		 * We stop synchronous streams when we have urgent data
504 		 * queued to prevent tcp_fuse_rrw() from pulling it.  If
505 		 * for some reasons the urgent data can't be delivered
506 		 * below, synchronous streams will remain stopped until
507 		 * someone drains the tcp_rcv_list.
508 		 */
509 		TCP_FUSE_SYNCSTR_PLUG_DRAIN(peer_tcp);
510 		tcp_fuse_output_urg(tcp, mp);
511 
512 		mp1 = mp->b_cont;
513 	}
514 
515 	if (tcp->tcp_ipversion == IPV4_VERSION &&
516 	    (HOOKS4_INTERESTED_LOOPBACK_IN ||
517 	    HOOKS4_INTERESTED_LOOPBACK_OUT) ||
518 	    tcp->tcp_ipversion == IPV6_VERSION &&
519 	    (HOOKS6_INTERESTED_LOOPBACK_IN ||
520 	    HOOKS6_INTERESTED_LOOPBACK_OUT)) {
521 		/*
522 		 * Build ip and tcp header to satisfy FW_HOOKS.
523 		 * We only build it when any hook is present.
524 		 */
525 		if ((mp1 = tcp_xmit_mp(tcp, mp1, tcp->tcp_mss, NULL, NULL,
526 		    tcp->tcp_snxt, B_TRUE, NULL, B_FALSE)) == NULL)
527 			/* If tcp_xmit_mp fails, use regular path */
528 			goto unfuse;
529 
530 		ASSERT(peer_tcp->tcp_connp->conn_ire_cache->ire_ipif != NULL);
531 		olp = peer_tcp->tcp_connp->conn_ire_cache->ire_ipif->ipif_ill;
532 		/* PFHooks: LOOPBACK_OUT */
533 		if (tcp->tcp_ipversion == IPV4_VERSION) {
534 			ipha = (ipha_t *)mp1->b_rptr;
535 
536 			DTRACE_PROBE4(ip4__loopback__out__start,
537 			    ill_t *, NULL, ill_t *, olp,
538 			    ipha_t *, ipha, mblk_t *, mp1);
539 			FW_HOOKS(ip4_loopback_out_event,
540 			    ipv4firewall_loopback_out,
541 			    NULL, olp, ipha, mp1, mp1);
542 			DTRACE_PROBE1(ip4__loopback__out__end, mblk_t *, mp1);
543 		} else {
544 			ip6h = (ip6_t *)mp1->b_rptr;
545 
546 			DTRACE_PROBE4(ip6__loopback__out__start,
547 			    ill_t *, NULL, ill_t *, olp,
548 			    ip6_t *, ip6h, mblk_t *, mp1);
549 			FW_HOOKS6(ip6_loopback_out_event,
550 			    ipv6firewall_loopback_out,
551 			    NULL, olp, ip6h, mp1, mp1);
552 			DTRACE_PROBE1(ip6__loopback__out__end, mblk_t *, mp1);
553 		}
554 		if (mp1 == NULL)
555 			goto unfuse;
556 
557 
558 		/* PFHooks: LOOPBACK_IN */
559 		ASSERT(tcp->tcp_connp->conn_ire_cache->ire_ipif != NULL);
560 		ilp = tcp->tcp_connp->conn_ire_cache->ire_ipif->ipif_ill;
561 
562 		if (tcp->tcp_ipversion == IPV4_VERSION) {
563 			DTRACE_PROBE4(ip4__loopback__in__start,
564 			    ill_t *, ilp, ill_t *, NULL,
565 			    ipha_t *, ipha, mblk_t *, mp1);
566 			FW_HOOKS(ip4_loopback_in_event,
567 			    ipv4firewall_loopback_in,
568 			    ilp, NULL, ipha, mp1, mp1);
569 			DTRACE_PROBE1(ip4__loopback__in__end, mblk_t *, mp1);
570 			if (mp1 == NULL)
571 				goto unfuse;
572 
573 			ip_hdr_len = IPH_HDR_LENGTH(ipha);
574 		} else {
575 			DTRACE_PROBE4(ip6__loopback__in__start,
576 			    ill_t *, ilp, ill_t *, NULL,
577 			    ip6_t *, ip6h, mblk_t *, mp1);
578 			FW_HOOKS6(ip6_loopback_in_event,
579 			    ipv6firewall_loopback_in,
580 			    ilp, NULL, ip6h, mp1, mp1);
581 			DTRACE_PROBE1(ip6__loopback__in__end, mblk_t *, mp1);
582 			if (mp1 == NULL)
583 				goto unfuse;
584 
585 			ip_hdr_len = ip_hdr_length_v6(mp1, ip6h);
586 		}
587 
588 		/* Data length might be changed by FW_HOOKS */
589 		tcph = (tcph_t *)&mp1->b_rptr[ip_hdr_len];
590 		seq = ABE32_TO_U32(tcph->th_seq);
591 		recv_size += seq - tcp->tcp_snxt;
592 
593 		/*
594 		 * The message duplicated by tcp_xmit_mp is freed.
595 		 * Note: the original message passed in remains unchanged.
596 		 */
597 		freemsg(mp1);
598 	}
599 
600 	mutex_enter(&peer_tcp->tcp_fuse_lock);
601 	/*
602 	 * Wake up and signal the peer; it is okay to do this before
603 	 * enqueueing because we are holding the lock.  One of the
604 	 * advantages of synchronous streams is the ability for us to
605 	 * find out when the application performs a read on the socket,
606 	 * by way of tcp_fuse_rrw() entry point being called.  Every
607 	 * data that gets enqueued onto the receiver is treated as if
608 	 * it has arrived at the receiving endpoint, thus generating
609 	 * SIGPOLL/SIGIO for asynchronous socket just as in the strrput()
610 	 * case.  However, we only wake up the application when necessary,
611 	 * i.e. during the first enqueue.  When tcp_fuse_rrw() is called
612 	 * it will send everything upstream.
613 	 */
614 	if (peer_tcp->tcp_direct_sockfs && !urgent &&
615 	    !TCP_IS_DETACHED(peer_tcp)) {
616 		if (peer_tcp->tcp_rcv_list == NULL)
617 			STR_WAKEUP_SET(STREAM(peer_tcp->tcp_rq));
618 		/* Update poll events and send SIGPOLL/SIGIO if necessary */
619 		STR_SENDSIG(STREAM(peer_tcp->tcp_rq));
620 	}
621 
622 	/*
623 	 * Enqueue data into the peer's receive list; we may or may not
624 	 * drain the contents depending on the conditions below.
625 	 */
626 	tcp_rcv_enqueue(peer_tcp, mp, recv_size);
627 
628 	/* In case it wrapped around and also to keep it constant */
629 	peer_tcp->tcp_rwnd += recv_size;
630 
631 	/*
632 	 * Exercise flow-control when needed; we will get back-enabled
633 	 * in either tcp_accept_finish(), tcp_unfuse(), or tcp_fuse_rrw().
634 	 * If tcp_direct_sockfs is on or if the peer endpoint is detached,
635 	 * we emulate streams flow control by checking the peer's queue
636 	 * size and high water mark; otherwise we simply use canputnext()
637 	 * to decide if we need to stop our flow.
638 	 *
639 	 * The outstanding unread data block check does not apply for a
640 	 * detached receiver; this is to avoid unnecessary blocking of the
641 	 * sender while the accept is currently in progress and is quite
642 	 * similar to the regular tcp.
643 	 */
644 	if (TCP_IS_DETACHED(peer_tcp) || max_unread == 0)
645 		max_unread = UINT_MAX;
646 
647 	flow_stopped = tcp->tcp_flow_stopped;
648 	if (!flow_stopped &&
649 	    (((peer_tcp->tcp_direct_sockfs || TCP_IS_DETACHED(peer_tcp)) &&
650 	    (peer_tcp->tcp_rcv_cnt >= peer_tcp->tcp_fuse_rcv_hiwater ||
651 	    ++peer_tcp->tcp_fuse_rcv_unread_cnt >= max_unread)) ||
652 	    (!peer_tcp->tcp_direct_sockfs &&
653 	    !TCP_IS_DETACHED(peer_tcp) && !canputnext(peer_tcp->tcp_rq)))) {
654 		tcp_setqfull(tcp);
655 		flow_stopped = B_TRUE;
656 		TCP_STAT(tcp_fusion_flowctl);
657 		DTRACE_PROBE4(tcp__fuse__output__flowctl, tcp_t *, tcp,
658 		    uint_t, send_size, uint_t, peer_tcp->tcp_rcv_cnt,
659 		    uint_t, peer_tcp->tcp_fuse_rcv_unread_cnt);
660 	} else if (flow_stopped &&
661 	    TCP_UNSENT_BYTES(tcp) <= tcp->tcp_xmit_lowater) {
662 		tcp_clrqfull(tcp);
663 		flow_stopped = B_FALSE;
664 	}
665 
666 	loopback_packets++;
667 	tcp->tcp_last_sent_len = send_size;
668 
669 	/* Need to adjust the following SNMP MIB-related variables */
670 	tcp->tcp_snxt += send_size;
671 	tcp->tcp_suna = tcp->tcp_snxt;
672 	peer_tcp->tcp_rnxt += recv_size;
673 	peer_tcp->tcp_rack = peer_tcp->tcp_rnxt;
674 
675 	BUMP_MIB(&tcp_mib, tcpOutDataSegs);
676 	UPDATE_MIB(&tcp_mib, tcpOutDataBytes, send_size);
677 
678 	BUMP_MIB(&tcp_mib, tcpInSegs);
679 	BUMP_MIB(&tcp_mib, tcpInDataInorderSegs);
680 	UPDATE_MIB(&tcp_mib, tcpInDataInorderBytes, send_size);
681 
682 	BUMP_LOCAL(tcp->tcp_obsegs);
683 	BUMP_LOCAL(peer_tcp->tcp_ibsegs);
684 
685 	mutex_exit(&peer_tcp->tcp_fuse_lock);
686 
687 	DTRACE_PROBE2(tcp__fuse__output, tcp_t *, tcp, uint_t, send_size);
688 
689 	if (!TCP_IS_DETACHED(peer_tcp)) {
690 		/*
691 		 * Drain the peer's receive queue it has urgent data or if
692 		 * we're not flow-controlled.  There is no need for draining
693 		 * normal data when tcp_direct_sockfs is on because the peer
694 		 * will pull the data via tcp_fuse_rrw().
695 		 */
696 		if (urgent || (!flow_stopped && !peer_tcp->tcp_direct_sockfs)) {
697 			ASSERT(peer_tcp->tcp_rcv_list != NULL);
698 			/*
699 			 * For TLI-based streams, a thread in tcp_accept_swap()
700 			 * can race with us.  That thread will ensure that the
701 			 * correct peer_tcp->tcp_rq is globally visible before
702 			 * peer_tcp->tcp_detached is visible as clear, but we
703 			 * must also ensure that the load of tcp_rq cannot be
704 			 * reordered to be before the tcp_detached check.
705 			 */
706 			membar_consumer();
707 			(void) tcp_fuse_rcv_drain(peer_tcp->tcp_rq, peer_tcp,
708 			    NULL);
709 			/*
710 			 * If synchronous streams was stopped above due
711 			 * to the presence of urgent data, re-enable it.
712 			 */
713 			if (urgent)
714 				TCP_FUSE_SYNCSTR_UNPLUG_DRAIN(peer_tcp);
715 		}
716 	}
717 	return (B_TRUE);
718 unfuse:
719 	tcp_unfuse(tcp);
720 	return (B_FALSE);
721 }
722 
723 /*
724  * This routine gets called to deliver data upstream on a fused or
725  * previously fused tcp loopback endpoint; the latter happens only
726  * when there is a pending SIGURG signal plus urgent data that can't
727  * be sent upstream in the past.
728  */
729 boolean_t
730 tcp_fuse_rcv_drain(queue_t *q, tcp_t *tcp, mblk_t **sigurg_mpp)
731 {
732 	mblk_t *mp;
733 #ifdef DEBUG
734 	uint_t cnt = 0;
735 #endif
736 
737 	ASSERT(tcp->tcp_loopback);
738 	ASSERT(tcp->tcp_fused || tcp->tcp_fused_sigurg);
739 	ASSERT(!tcp->tcp_fused || tcp->tcp_loopback_peer != NULL);
740 	ASSERT(sigurg_mpp != NULL || tcp->tcp_fused);
741 
742 	/* No need for the push timer now, in case it was scheduled */
743 	if (tcp->tcp_push_tid != 0) {
744 		(void) TCP_TIMER_CANCEL(tcp, tcp->tcp_push_tid);
745 		tcp->tcp_push_tid = 0;
746 	}
747 	/*
748 	 * If there's urgent data sitting in receive list and we didn't
749 	 * get a chance to send up a SIGURG signal, make sure we send
750 	 * it first before draining in order to ensure that SIOCATMARK
751 	 * works properly.
752 	 */
753 	if (tcp->tcp_fused_sigurg) {
754 		/*
755 		 * sigurg_mpp is normally NULL, i.e. when we're still
756 		 * fused and didn't get here because of tcp_unfuse().
757 		 * In this case try hard to allocate the M_PCSIG mblk.
758 		 */
759 		if (sigurg_mpp == NULL &&
760 		    (mp = allocb(1, BPRI_HI)) == NULL &&
761 		    (mp = allocb_tryhard(1)) == NULL) {
762 			/* Alloc failed; try again next time */
763 			tcp->tcp_push_tid = TCP_TIMER(tcp, tcp_push_timer,
764 			    MSEC_TO_TICK(tcp_push_timer_interval));
765 			return (B_TRUE);
766 		} else if (sigurg_mpp != NULL) {
767 			/*
768 			 * Use the supplied M_PCSIG mblk; it means we're
769 			 * either unfused or in the process of unfusing,
770 			 * and the drain must happen now.
771 			 */
772 			mp = *sigurg_mpp;
773 			*sigurg_mpp = NULL;
774 		}
775 		ASSERT(mp != NULL);
776 
777 		tcp->tcp_fused_sigurg = B_FALSE;
778 		/* Send up the signal */
779 		DB_TYPE(mp) = M_PCSIG;
780 		*mp->b_wptr++ = (uchar_t)SIGURG;
781 		putnext(q, mp);
782 		/*
783 		 * Let the regular tcp_rcv_drain() path handle
784 		 * draining the data if we're no longer fused.
785 		 */
786 		if (!tcp->tcp_fused)
787 			return (B_FALSE);
788 	}
789 
790 	/*
791 	 * In the synchronous streams case, we generate SIGPOLL/SIGIO for
792 	 * each M_DATA that gets enqueued onto the receiver.  At this point
793 	 * we are about to drain any queued data via putnext().  In order
794 	 * to avoid extraneous signal generation from strrput(), we set
795 	 * STRGETINPROG flag at the stream head prior to the draining and
796 	 * restore it afterwards.  This masks out signal generation only
797 	 * for M_DATA messages and does not affect urgent data.
798 	 */
799 	if (tcp->tcp_direct_sockfs)
800 		strrput_sig(q, B_FALSE);
801 
802 	/* Drain the data */
803 	while ((mp = tcp->tcp_rcv_list) != NULL) {
804 		tcp->tcp_rcv_list = mp->b_next;
805 		mp->b_next = NULL;
806 #ifdef DEBUG
807 		cnt += msgdsize(mp);
808 #endif
809 		putnext(q, mp);
810 		TCP_STAT(tcp_fusion_putnext);
811 	}
812 
813 	if (tcp->tcp_direct_sockfs)
814 		strrput_sig(q, B_TRUE);
815 
816 	ASSERT(cnt == tcp->tcp_rcv_cnt);
817 	tcp->tcp_rcv_last_head = NULL;
818 	tcp->tcp_rcv_last_tail = NULL;
819 	tcp->tcp_rcv_cnt = 0;
820 	tcp->tcp_fuse_rcv_unread_cnt = 0;
821 	tcp->tcp_rwnd = q->q_hiwat;
822 
823 	return (B_TRUE);
824 }
825 
826 /*
827  * Synchronous stream entry point for sockfs to retrieve
828  * data directly from tcp_rcv_list.
829  */
830 int
831 tcp_fuse_rrw(queue_t *q, struiod_t *dp)
832 {
833 	tcp_t *tcp = Q_TO_CONN(q)->conn_tcp;
834 	mblk_t *mp;
835 
836 	mutex_enter(&tcp->tcp_fuse_lock);
837 
838 	/*
839 	 * If tcp_fuse_syncstr_plugged is set, then another thread is moving
840 	 * the underlying data to the stream head.  We need to wait until it's
841 	 * done, then return EBUSY so that strget() will dequeue data from the
842 	 * stream head to ensure data is drained in-order.
843 	 */
844 	if (tcp->tcp_fuse_syncstr_plugged) {
845 		do {
846 			cv_wait(&tcp->tcp_fuse_plugcv, &tcp->tcp_fuse_lock);
847 		} while (tcp->tcp_fuse_syncstr_plugged);
848 
849 		mutex_exit(&tcp->tcp_fuse_lock);
850 		TCP_STAT(tcp_fusion_rrw_plugged);
851 		TCP_STAT(tcp_fusion_rrw_busy);
852 		return (EBUSY);
853 	}
854 
855 	/*
856 	 * If someone had turned off tcp_direct_sockfs or if synchronous
857 	 * streams is stopped, we return EBUSY.  This causes strget() to
858 	 * dequeue data from the stream head instead.
859 	 */
860 	if (!tcp->tcp_direct_sockfs || tcp->tcp_fuse_syncstr_stopped) {
861 		mutex_exit(&tcp->tcp_fuse_lock);
862 		TCP_STAT(tcp_fusion_rrw_busy);
863 		return (EBUSY);
864 	}
865 
866 	if ((mp = tcp->tcp_rcv_list) != NULL) {
867 		tcp_t *peer_tcp = tcp->tcp_loopback_peer;
868 
869 		DTRACE_PROBE3(tcp__fuse__rrw, tcp_t *, tcp,
870 		    uint32_t, tcp->tcp_rcv_cnt, ssize_t, dp->d_uio.uio_resid);
871 
872 		tcp->tcp_rcv_list = NULL;
873 		TCP_STAT(tcp_fusion_rrw_msgcnt);
874 
875 		/*
876 		 * At this point nothing should be left in tcp_rcv_list.
877 		 * The only possible case where we would have a chain of
878 		 * b_next-linked messages is urgent data, but we wouldn't
879 		 * be here if that's true since urgent data is delivered
880 		 * via putnext() and synchronous streams is stopped until
881 		 * tcp_fuse_rcv_drain() is finished.
882 		 */
883 		ASSERT(DB_TYPE(mp) == M_DATA && mp->b_next == NULL);
884 
885 		tcp->tcp_rcv_last_head = NULL;
886 		tcp->tcp_rcv_last_tail = NULL;
887 		tcp->tcp_rcv_cnt = 0;
888 		tcp->tcp_fuse_rcv_unread_cnt = 0;
889 
890 		if (peer_tcp->tcp_flow_stopped) {
891 			tcp_clrqfull(peer_tcp);
892 			TCP_STAT(tcp_fusion_backenabled);
893 		}
894 	}
895 
896 	/*
897 	 * Either we just dequeued everything or we get here from sockfs
898 	 * and have nothing to return; in this case clear RSLEEP.
899 	 */
900 	ASSERT(tcp->tcp_rcv_last_head == NULL);
901 	ASSERT(tcp->tcp_rcv_last_tail == NULL);
902 	ASSERT(tcp->tcp_rcv_cnt == 0);
903 	ASSERT(tcp->tcp_fuse_rcv_unread_cnt == 0);
904 	STR_WAKEUP_CLEAR(STREAM(q));
905 
906 	mutex_exit(&tcp->tcp_fuse_lock);
907 	dp->d_mp = mp;
908 	return (0);
909 }
910 
911 /*
912  * Synchronous stream entry point used by certain ioctls to retrieve
913  * information about or peek into the tcp_rcv_list.
914  */
915 int
916 tcp_fuse_rinfop(queue_t *q, infod_t *dp)
917 {
918 	tcp_t	*tcp = Q_TO_CONN(q)->conn_tcp;
919 	mblk_t	*mp;
920 	uint_t	cmd = dp->d_cmd;
921 	int	res = 0;
922 	int	error = 0;
923 	struct stdata *stp = STREAM(q);
924 
925 	mutex_enter(&tcp->tcp_fuse_lock);
926 	/* If shutdown on read has happened, return nothing */
927 	mutex_enter(&stp->sd_lock);
928 	if (stp->sd_flag & STREOF) {
929 		mutex_exit(&stp->sd_lock);
930 		goto done;
931 	}
932 	mutex_exit(&stp->sd_lock);
933 
934 	/*
935 	 * It is OK not to return an answer if tcp_rcv_list is
936 	 * currently not accessible.
937 	 */
938 	if (!tcp->tcp_direct_sockfs || tcp->tcp_fuse_syncstr_stopped ||
939 	    tcp->tcp_fuse_syncstr_plugged || (mp = tcp->tcp_rcv_list) == NULL)
940 		goto done;
941 
942 	if (cmd & INFOD_COUNT) {
943 		/*
944 		 * We have at least one message and
945 		 * could return only one at a time.
946 		 */
947 		dp->d_count++;
948 		res |= INFOD_COUNT;
949 	}
950 	if (cmd & INFOD_BYTES) {
951 		/*
952 		 * Return size of all data messages.
953 		 */
954 		dp->d_bytes += tcp->tcp_rcv_cnt;
955 		res |= INFOD_BYTES;
956 	}
957 	if (cmd & INFOD_FIRSTBYTES) {
958 		/*
959 		 * Return size of first data message.
960 		 */
961 		dp->d_bytes = msgdsize(mp);
962 		res |= INFOD_FIRSTBYTES;
963 		dp->d_cmd &= ~INFOD_FIRSTBYTES;
964 	}
965 	if (cmd & INFOD_COPYOUT) {
966 		mblk_t *mp1;
967 		int n;
968 
969 		if (DB_TYPE(mp) == M_DATA) {
970 			mp1 = mp;
971 		} else {
972 			mp1 = mp->b_cont;
973 			ASSERT(mp1 != NULL);
974 		}
975 
976 		/*
977 		 * Return data contents of first message.
978 		 */
979 		ASSERT(DB_TYPE(mp1) == M_DATA);
980 		while (mp1 != NULL && dp->d_uiop->uio_resid > 0) {
981 			n = MIN(dp->d_uiop->uio_resid, MBLKL(mp1));
982 			if (n != 0 && (error = uiomove((char *)mp1->b_rptr, n,
983 			    UIO_READ, dp->d_uiop)) != 0) {
984 				goto done;
985 			}
986 			mp1 = mp1->b_cont;
987 		}
988 		res |= INFOD_COPYOUT;
989 		dp->d_cmd &= ~INFOD_COPYOUT;
990 	}
991 done:
992 	mutex_exit(&tcp->tcp_fuse_lock);
993 
994 	dp->d_res |= res;
995 
996 	return (error);
997 }
998 
999 /*
1000  * Enable synchronous streams on a fused tcp loopback endpoint.
1001  */
1002 static void
1003 tcp_fuse_syncstr_enable(tcp_t *tcp)
1004 {
1005 	queue_t *rq = tcp->tcp_rq;
1006 	struct stdata *stp = STREAM(rq);
1007 
1008 	/* We can only enable synchronous streams for sockfs mode */
1009 	tcp->tcp_direct_sockfs = tcp->tcp_issocket && do_tcp_direct_sockfs;
1010 
1011 	if (!tcp->tcp_direct_sockfs)
1012 		return;
1013 
1014 	mutex_enter(&stp->sd_lock);
1015 	mutex_enter(QLOCK(rq));
1016 
1017 	/*
1018 	 * We replace our q_qinfo with one that has the qi_rwp entry point.
1019 	 * Clear SR_SIGALLDATA because we generate the equivalent signal(s)
1020 	 * for every enqueued data in tcp_fuse_output().
1021 	 */
1022 	rq->q_qinfo = &tcp_loopback_rinit;
1023 	rq->q_struiot = tcp_loopback_rinit.qi_struiot;
1024 	stp->sd_struiordq = rq;
1025 	stp->sd_rput_opt &= ~SR_SIGALLDATA;
1026 
1027 	mutex_exit(QLOCK(rq));
1028 	mutex_exit(&stp->sd_lock);
1029 }
1030 
1031 /*
1032  * Disable synchronous streams on a fused tcp loopback endpoint.
1033  */
1034 static void
1035 tcp_fuse_syncstr_disable(tcp_t *tcp)
1036 {
1037 	queue_t *rq = tcp->tcp_rq;
1038 	struct stdata *stp = STREAM(rq);
1039 
1040 	if (!tcp->tcp_direct_sockfs)
1041 		return;
1042 
1043 	mutex_enter(&stp->sd_lock);
1044 	mutex_enter(QLOCK(rq));
1045 
1046 	/*
1047 	 * Reset q_qinfo to point to the default tcp entry points.
1048 	 * Also restore SR_SIGALLDATA so that strrput() can generate
1049 	 * the signals again for future M_DATA messages.
1050 	 */
1051 	rq->q_qinfo = &tcp_rinit;
1052 	rq->q_struiot = tcp_rinit.qi_struiot;
1053 	stp->sd_struiordq = NULL;
1054 	stp->sd_rput_opt |= SR_SIGALLDATA;
1055 	tcp->tcp_direct_sockfs = B_FALSE;
1056 
1057 	mutex_exit(QLOCK(rq));
1058 	mutex_exit(&stp->sd_lock);
1059 }
1060 
1061 /*
1062  * Enable synchronous streams on a pair of fused tcp endpoints.
1063  */
1064 void
1065 tcp_fuse_syncstr_enable_pair(tcp_t *tcp)
1066 {
1067 	tcp_t *peer_tcp = tcp->tcp_loopback_peer;
1068 
1069 	ASSERT(tcp->tcp_fused);
1070 	ASSERT(peer_tcp != NULL);
1071 
1072 	tcp_fuse_syncstr_enable(tcp);
1073 	tcp_fuse_syncstr_enable(peer_tcp);
1074 }
1075 
1076 /*
1077  * Allow or disallow signals to be generated by strrput().
1078  */
1079 static void
1080 strrput_sig(queue_t *q, boolean_t on)
1081 {
1082 	struct stdata *stp = STREAM(q);
1083 
1084 	mutex_enter(&stp->sd_lock);
1085 	if (on)
1086 		stp->sd_flag &= ~STRGETINPROG;
1087 	else
1088 		stp->sd_flag |= STRGETINPROG;
1089 	mutex_exit(&stp->sd_lock);
1090 }
1091 
1092 /*
1093  * Disable synchronous streams on a pair of fused tcp endpoints and drain
1094  * any queued data; called either during unfuse or upon transitioning from
1095  * a socket to a stream endpoint due to _SIOCSOCKFALLBACK.
1096  */
1097 void
1098 tcp_fuse_disable_pair(tcp_t *tcp, boolean_t unfusing)
1099 {
1100 	tcp_t *peer_tcp = tcp->tcp_loopback_peer;
1101 
1102 	ASSERT(tcp->tcp_fused);
1103 	ASSERT(peer_tcp != NULL);
1104 
1105 	/*
1106 	 * Force any tcp_fuse_rrw() calls to block until we've moved the data
1107 	 * onto the stream head.
1108 	 */
1109 	TCP_FUSE_SYNCSTR_PLUG_DRAIN(tcp);
1110 	TCP_FUSE_SYNCSTR_PLUG_DRAIN(peer_tcp);
1111 
1112 	/*
1113 	 * Drain any pending data; the detached check is needed because
1114 	 * we may be called as a result of a tcp_unfuse() triggered by
1115 	 * tcp_fuse_output().  Note that in case of a detached tcp, the
1116 	 * draining will happen later after the tcp is unfused.  For non-
1117 	 * urgent data, this can be handled by the regular tcp_rcv_drain().
1118 	 * If we have urgent data sitting in the receive list, we will
1119 	 * need to send up a SIGURG signal first before draining the data.
1120 	 * All of these will be handled by the code in tcp_fuse_rcv_drain()
1121 	 * when called from tcp_rcv_drain().
1122 	 */
1123 	if (!TCP_IS_DETACHED(tcp)) {
1124 		(void) tcp_fuse_rcv_drain(tcp->tcp_rq, tcp,
1125 		    (unfusing ? &tcp->tcp_fused_sigurg_mp : NULL));
1126 	}
1127 	if (!TCP_IS_DETACHED(peer_tcp)) {
1128 		(void) tcp_fuse_rcv_drain(peer_tcp->tcp_rq, peer_tcp,
1129 		    (unfusing ? &peer_tcp->tcp_fused_sigurg_mp : NULL));
1130 	}
1131 
1132 	/*
1133 	 * Make all current and future tcp_fuse_rrw() calls fail with EBUSY.
1134 	 * To ensure threads don't sneak past the checks in tcp_fuse_rrw(),
1135 	 * a given stream must be stopped prior to being unplugged (but the
1136 	 * ordering of operations between the streams is unimportant).
1137 	 */
1138 	TCP_FUSE_SYNCSTR_STOP(tcp);
1139 	TCP_FUSE_SYNCSTR_STOP(peer_tcp);
1140 	TCP_FUSE_SYNCSTR_UNPLUG_DRAIN(tcp);
1141 	TCP_FUSE_SYNCSTR_UNPLUG_DRAIN(peer_tcp);
1142 
1143 	/* Lift up any flow-control conditions */
1144 	if (tcp->tcp_flow_stopped) {
1145 		tcp_clrqfull(tcp);
1146 		TCP_STAT(tcp_fusion_backenabled);
1147 	}
1148 	if (peer_tcp->tcp_flow_stopped) {
1149 		tcp_clrqfull(peer_tcp);
1150 		TCP_STAT(tcp_fusion_backenabled);
1151 	}
1152 
1153 	/* Disable synchronous streams */
1154 	tcp_fuse_syncstr_disable(tcp);
1155 	tcp_fuse_syncstr_disable(peer_tcp);
1156 }
1157 
1158 /*
1159  * Calculate the size of receive buffer for a fused tcp endpoint.
1160  */
1161 size_t
1162 tcp_fuse_set_rcv_hiwat(tcp_t *tcp, size_t rwnd)
1163 {
1164 	ASSERT(tcp->tcp_fused);
1165 
1166 	/* Ensure that value is within the maximum upper bound */
1167 	if (rwnd > tcp_max_buf)
1168 		rwnd = tcp_max_buf;
1169 
1170 	/* Obey the absolute minimum tcp receive high water mark */
1171 	if (rwnd < tcp_sth_rcv_hiwat)
1172 		rwnd = tcp_sth_rcv_hiwat;
1173 
1174 	/*
1175 	 * Round up to system page size in case SO_RCVBUF is modified
1176 	 * after SO_SNDBUF; the latter is also similarly rounded up.
1177 	 */
1178 	rwnd = P2ROUNDUP_TYPED(rwnd, PAGESIZE, size_t);
1179 	tcp->tcp_fuse_rcv_hiwater = rwnd;
1180 	return (rwnd);
1181 }
1182 
1183 /*
1184  * Calculate the maximum outstanding unread data block for a fused tcp endpoint.
1185  */
1186 int
1187 tcp_fuse_maxpsz_set(tcp_t *tcp)
1188 {
1189 	tcp_t *peer_tcp = tcp->tcp_loopback_peer;
1190 	uint_t sndbuf = tcp->tcp_xmit_hiwater;
1191 	uint_t maxpsz = sndbuf;
1192 
1193 	ASSERT(tcp->tcp_fused);
1194 	ASSERT(peer_tcp != NULL);
1195 	ASSERT(peer_tcp->tcp_fuse_rcv_hiwater != 0);
1196 	/*
1197 	 * In the fused loopback case, we want the stream head to split
1198 	 * up larger writes into smaller chunks for a more accurate flow-
1199 	 * control accounting.  Our maxpsz is half of the sender's send
1200 	 * buffer or the receiver's receive buffer, whichever is smaller.
1201 	 * We round up the buffer to system page size due to the lack of
1202 	 * TCP MSS concept in Fusion.
1203 	 */
1204 	if (maxpsz > peer_tcp->tcp_fuse_rcv_hiwater)
1205 		maxpsz = peer_tcp->tcp_fuse_rcv_hiwater;
1206 	maxpsz = P2ROUNDUP_TYPED(maxpsz, PAGESIZE, uint_t) >> 1;
1207 
1208 	/*
1209 	 * Calculate the peer's limit for the number of outstanding unread
1210 	 * data block.  This is the amount of data blocks that are allowed
1211 	 * to reside in the receiver's queue before the sender gets flow
1212 	 * controlled.  It is used only in the synchronous streams mode as
1213 	 * a way to throttle the sender when it performs consecutive writes
1214 	 * faster than can be read.  The value is derived from SO_SNDBUF in
1215 	 * order to give the sender some control; we divide it with a large
1216 	 * value (16KB) to produce a fairly low initial limit.
1217 	 */
1218 	if (tcp_fusion_rcv_unread_min == 0) {
1219 		/* A value of 0 means that we disable the check */
1220 		peer_tcp->tcp_fuse_rcv_unread_hiwater = 0;
1221 	} else {
1222 		peer_tcp->tcp_fuse_rcv_unread_hiwater =
1223 		    MAX(sndbuf >> 14, tcp_fusion_rcv_unread_min);
1224 	}
1225 	return (maxpsz);
1226 }
1227