xref: /titanic_52/usr/src/uts/common/inet/tcp/tcp_fusion.c (revision 9e39c5ba00a55fa05777cc94b148296af305e135)
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 2009 Sun Microsystems, Inc.  All rights reserved.
23  * Use is subject to license terms.
24  */
25 
26 #include <sys/types.h>
27 #include <sys/stream.h>
28 #include <sys/strsun.h>
29 #include <sys/strsubr.h>
30 #include <sys/debug.h>
31 #include <sys/sdt.h>
32 #include <sys/cmn_err.h>
33 #include <sys/tihdr.h>
34 
35 #include <inet/common.h>
36 #include <inet/optcom.h>
37 #include <inet/ip.h>
38 #include <inet/ip_if.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 #include <inet/ip_if.h>
46 
47 /*
48  * This file implements TCP fusion - a protocol-less data path for TCP
49  * loopback connections.  The fusion of two local TCP endpoints occurs
50  * at connection establishment time.  Various conditions (see details
51  * in tcp_fuse()) need to be met for fusion to be successful.  If it
52  * fails, we fall back to the regular TCP data path; if it succeeds,
53  * both endpoints proceed to use tcp_fuse_output() as the transmit path.
54  * tcp_fuse_output() enqueues application data directly onto the peer's
55  * receive queue; no protocol processing is involved.  After enqueueing
56  * the data, the sender can either push (putnext) data up the receiver's
57  * read queue; or the sender can simply return and let the receiver
58  * retrieve the enqueued data via the synchronous streams entry point
59  * tcp_fuse_rrw().  The latter path is taken if synchronous streams is
60  * enabled (the default).  It is disabled if sockfs no longer resides
61  * directly on top of tcp module due to a module insertion or removal.
62  * It also needs to be temporarily disabled when sending urgent data
63  * because the tcp_fuse_rrw() path bypasses the M_PROTO processing done
64  * by strsock_proto() hook.
65  *
66  * Sychronization is handled by squeue and the mutex tcp_non_sq_lock.
67  * One of the requirements for fusion to succeed is that both endpoints
68  * need to be using the same squeue.  This ensures that neither side
69  * can disappear while the other side is still sending data.  By itself,
70  * squeue is not sufficient for guaranteeing safety when synchronous
71  * streams is enabled.  The reason is that tcp_fuse_rrw() doesn't enter
72  * the squeue and its access to tcp_rcv_list and other fusion-related
73  * fields needs to be sychronized with the sender.  tcp_non_sq_lock is
74  * used for this purpose.  When there is urgent data, the sender needs
75  * to push the data up the receiver's streams read queue.  In order to
76  * avoid holding the tcp_non_sq_lock across putnext(), the sender sets
77  * the peer tcp's tcp_fuse_syncstr_plugged bit and releases tcp_non_sq_lock
78  * (see macro TCP_FUSE_SYNCSTR_PLUG_DRAIN()).  If tcp_fuse_rrw() enters
79  * after this point, it will see that synchronous streams is plugged and
80  * will wait on tcp_fuse_plugcv.  After the sender has finished pushing up
81  * all urgent data, it will clear the tcp_fuse_syncstr_plugged bit using
82  * TCP_FUSE_SYNCSTR_UNPLUG_DRAIN().  This will cause any threads waiting
83  * on tcp_fuse_plugcv to return EBUSY, and in turn cause strget() to call
84  * getq_noenab() to dequeue data from the stream head instead.  Once the
85  * data on the stream head has been consumed, tcp_fuse_rrw() may again
86  * be used to process tcp_rcv_list.  However, if TCP_FUSE_SYNCSTR_STOP()
87  * has been called, all future calls to tcp_fuse_rrw() will return EBUSY,
88  * effectively disabling synchronous streams.
89  *
90  * The following note applies only to the synchronous streams mode.
91  *
92  * Flow control is done by checking the size of receive buffer and
93  * the number of data blocks, both set to different limits.  This is
94  * different than regular streams flow control where cumulative size
95  * check dominates block count check -- streams queue high water mark
96  * typically represents bytes.  Each enqueue triggers notifications
97  * to the receiving process; a build up of data blocks indicates a
98  * slow receiver and the sender should be blocked or informed at the
99  * earliest moment instead of further wasting system resources.  In
100  * effect, this is equivalent to limiting the number of outstanding
101  * segments in flight.
102  */
103 
104 /*
105  * Setting this to false means we disable fusion altogether and
106  * loopback connections would go through the protocol paths.
107  */
108 boolean_t do_tcp_fusion = B_TRUE;
109 
110 /*
111  * Enabling this flag allows sockfs to retrieve data directly
112  * from a fused tcp endpoint using synchronous streams interface.
113  */
114 boolean_t do_tcp_direct_sockfs = B_FALSE;
115 
116 /*
117  * This is the minimum amount of outstanding writes allowed on
118  * a synchronous streams-enabled receiving endpoint before the
119  * sender gets flow-controlled.  Setting this value to 0 means
120  * that the data block limit is equivalent to the byte count
121  * limit, which essentially disables the check.
122  */
123 #define	TCP_FUSION_RCV_UNREAD_MIN	8
124 uint_t tcp_fusion_rcv_unread_min = TCP_FUSION_RCV_UNREAD_MIN;
125 
126 static void		tcp_fuse_syncstr_enable(tcp_t *);
127 static void		tcp_fuse_syncstr_disable(tcp_t *);
128 static boolean_t	strrput_sig(queue_t *, boolean_t);
129 
130 /*
131  * Return true if this connection needs some IP functionality
132  */
133 static boolean_t
134 tcp_loopback_needs_ip(tcp_t *tcp, netstack_t *ns)
135 {
136 	ipsec_stack_t	*ipss = ns->netstack_ipsec;
137 
138 	/*
139 	 * If ire is not cached, do not use fusion
140 	 */
141 	if (tcp->tcp_connp->conn_ire_cache == NULL) {
142 		/*
143 		 * There is no need to hold conn_lock here because when called
144 		 * from tcp_fuse() there can be no window where conn_ire_cache
145 		 * can change. This is not true when called from
146 		 * tcp_fuse_output() as conn_ire_cache can become null just
147 		 * after the check. It will be necessary to recheck for a NULL
148 		 * conn_ire_cache in tcp_fuse_output() to avoid passing a
149 		 * stale ill pointer to FW_HOOKS.
150 		 */
151 		return (B_TRUE);
152 	}
153 	if (tcp->tcp_ipversion == IPV4_VERSION) {
154 		if (tcp->tcp_ip_hdr_len != IP_SIMPLE_HDR_LENGTH)
155 			return (B_TRUE);
156 		if (CONN_OUTBOUND_POLICY_PRESENT(tcp->tcp_connp, ipss))
157 			return (B_TRUE);
158 		if (CONN_INBOUND_POLICY_PRESENT(tcp->tcp_connp, ipss))
159 			return (B_TRUE);
160 	} else {
161 		if (tcp->tcp_ip_hdr_len != IPV6_HDR_LEN)
162 			return (B_TRUE);
163 		if (CONN_OUTBOUND_POLICY_PRESENT_V6(tcp->tcp_connp, ipss))
164 			return (B_TRUE);
165 		if (CONN_INBOUND_POLICY_PRESENT_V6(tcp->tcp_connp, ipss))
166 			return (B_TRUE);
167 	}
168 	if (!CONN_IS_LSO_MD_FASTPATH(tcp->tcp_connp))
169 		return (B_TRUE);
170 	return (B_FALSE);
171 }
172 
173 
174 /*
175  * This routine gets called by the eager tcp upon changing state from
176  * SYN_RCVD to ESTABLISHED.  It fuses a direct path between itself
177  * and the active connect tcp such that the regular tcp processings
178  * may be bypassed under allowable circumstances.  Because the fusion
179  * requires both endpoints to be in the same squeue, it does not work
180  * for simultaneous active connects because there is no easy way to
181  * switch from one squeue to another once the connection is created.
182  * This is different from the eager tcp case where we assign it the
183  * same squeue as the one given to the active connect tcp during open.
184  */
185 void
186 tcp_fuse(tcp_t *tcp, uchar_t *iphdr, tcph_t *tcph)
187 {
188 	conn_t *peer_connp, *connp = tcp->tcp_connp;
189 	tcp_t *peer_tcp;
190 	tcp_stack_t	*tcps = tcp->tcp_tcps;
191 	netstack_t	*ns;
192 	ip_stack_t	*ipst = tcps->tcps_netstack->netstack_ip;
193 
194 	ASSERT(!tcp->tcp_fused);
195 	ASSERT(tcp->tcp_loopback);
196 	ASSERT(tcp->tcp_loopback_peer == NULL);
197 	/*
198 	 * We need to inherit q_hiwat of the listener tcp, but we can't
199 	 * really use tcp_listener since we get here after sending up
200 	 * T_CONN_IND and tcp_wput_accept() may be called independently,
201 	 * at which point tcp_listener is cleared; this is why we use
202 	 * tcp_saved_listener.  The listener itself is guaranteed to be
203 	 * around until tcp_accept_finish() is called on this eager --
204 	 * this won't happen until we're done since we're inside the
205 	 * eager's perimeter now.
206 	 *
207 	 * We can also get called in the case were a connection needs
208 	 * to be re-fused. In this case tcp_saved_listener will be
209 	 * NULL but tcp_refuse will be true.
210 	 */
211 	ASSERT(tcp->tcp_saved_listener != NULL || tcp->tcp_refuse);
212 	/*
213 	 * Lookup peer endpoint; search for the remote endpoint having
214 	 * the reversed address-port quadruplet in ESTABLISHED state,
215 	 * which is guaranteed to be unique in the system.  Zone check
216 	 * is applied accordingly for loopback address, but not for
217 	 * local address since we want fusion to happen across Zones.
218 	 */
219 	if (tcp->tcp_ipversion == IPV4_VERSION) {
220 		peer_connp = ipcl_conn_tcp_lookup_reversed_ipv4(connp,
221 		    (ipha_t *)iphdr, tcph, ipst);
222 	} else {
223 		peer_connp = ipcl_conn_tcp_lookup_reversed_ipv6(connp,
224 		    (ip6_t *)iphdr, tcph, ipst);
225 	}
226 
227 	/*
228 	 * We can only proceed if peer exists, resides in the same squeue
229 	 * as our conn and is not raw-socket.  The squeue assignment of
230 	 * this eager tcp was done earlier at the time of SYN processing
231 	 * in ip_fanout_tcp{_v6}.  Note that similar squeues by itself
232 	 * doesn't guarantee a safe condition to fuse, hence we perform
233 	 * additional tests below.
234 	 */
235 	ASSERT(peer_connp == NULL || peer_connp != connp);
236 	if (peer_connp == NULL || peer_connp->conn_sqp != connp->conn_sqp ||
237 	    !IPCL_IS_TCP(peer_connp)) {
238 		if (peer_connp != NULL) {
239 			TCP_STAT(tcps, tcp_fusion_unqualified);
240 			CONN_DEC_REF(peer_connp);
241 		}
242 		return;
243 	}
244 	peer_tcp = peer_connp->conn_tcp;	/* active connect tcp */
245 
246 	ASSERT(peer_tcp != NULL && peer_tcp != tcp && !peer_tcp->tcp_fused);
247 	ASSERT(peer_tcp->tcp_loopback && peer_tcp->tcp_loopback_peer == NULL);
248 	ASSERT(peer_connp->conn_sqp == connp->conn_sqp);
249 
250 	/*
251 	 * Fuse the endpoints; we perform further checks against both
252 	 * tcp endpoints to ensure that a fusion is allowed to happen.
253 	 * In particular we bail out for non-simple TCP/IP or if IPsec/
254 	 * IPQoS policy/kernel SSL exists. We also need to check if
255 	 * the connection is quiescent to cover the case when we are
256 	 * trying to re-enable fusion after IPobservability is turned off.
257 	 */
258 	ns = tcps->tcps_netstack;
259 	ipst = ns->netstack_ip;
260 
261 	if (!tcp->tcp_unfusable && !peer_tcp->tcp_unfusable &&
262 	    !tcp_loopback_needs_ip(tcp, ns) &&
263 	    !tcp_loopback_needs_ip(peer_tcp, ns) &&
264 	    tcp->tcp_kssl_ent == NULL &&
265 	    tcp->tcp_xmit_head == NULL && peer_tcp->tcp_xmit_head == NULL &&
266 	    !IPP_ENABLED(IPP_LOCAL_OUT|IPP_LOCAL_IN, ipst)) {
267 		mblk_t *mp;
268 		queue_t *peer_rq = peer_tcp->tcp_rq;
269 
270 		ASSERT(!TCP_IS_DETACHED(peer_tcp));
271 		ASSERT(tcp->tcp_fused_sigurg_mp == NULL);
272 		ASSERT(peer_tcp->tcp_fused_sigurg_mp == NULL);
273 		ASSERT(tcp->tcp_kssl_ctx == NULL);
274 
275 		/*
276 		 * We need to drain data on both endpoints during unfuse.
277 		 * If we need to send up SIGURG at the time of draining,
278 		 * we want to be sure that an mblk is readily available.
279 		 * This is why we pre-allocate the M_PCSIG mblks for both
280 		 * endpoints which will only be used during/after unfuse.
281 		 */
282 		if (!IPCL_IS_NONSTR(tcp->tcp_connp)) {
283 			if ((mp = allocb(1, BPRI_HI)) == NULL)
284 				goto failed;
285 
286 			tcp->tcp_fused_sigurg_mp = mp;
287 		}
288 
289 		if (!IPCL_IS_NONSTR(peer_tcp->tcp_connp)) {
290 			if ((mp = allocb(1, BPRI_HI)) == NULL)
291 				goto failed;
292 
293 			peer_tcp->tcp_fused_sigurg_mp = mp;
294 		}
295 
296 		if (!IPCL_IS_NONSTR(peer_tcp->tcp_connp) &&
297 		    (mp = allocb(sizeof (struct stroptions),
298 		    BPRI_HI)) == NULL) {
299 			goto failed;
300 		}
301 
302 		/* Fuse both endpoints */
303 		peer_tcp->tcp_loopback_peer = tcp;
304 		tcp->tcp_loopback_peer = peer_tcp;
305 		peer_tcp->tcp_fused = tcp->tcp_fused = B_TRUE;
306 
307 		/*
308 		 * We never use regular tcp paths in fusion and should
309 		 * therefore clear tcp_unsent on both endpoints.  Having
310 		 * them set to non-zero values means asking for trouble
311 		 * especially after unfuse, where we may end up sending
312 		 * through regular tcp paths which expect xmit_list and
313 		 * friends to be correctly setup.
314 		 */
315 		peer_tcp->tcp_unsent = tcp->tcp_unsent = 0;
316 
317 		tcp_timers_stop(tcp);
318 		tcp_timers_stop(peer_tcp);
319 
320 		/*
321 		 * At this point we are a detached eager tcp and therefore
322 		 * don't have a queue assigned to us until accept happens.
323 		 * In the mean time the peer endpoint may immediately send
324 		 * us data as soon as fusion is finished, and we need to be
325 		 * able to flow control it in case it sends down huge amount
326 		 * of data while we're still detached.  To prevent that we
327 		 * inherit the listener's recv_hiwater value; this is temporary
328 		 * since we'll repeat the process intcp_accept_finish().
329 		 */
330 		if (!tcp->tcp_refuse) {
331 			(void) tcp_fuse_set_rcv_hiwat(tcp,
332 			    tcp->tcp_saved_listener->tcp_recv_hiwater);
333 
334 			/*
335 			 * Set the stream head's write offset value to zero
336 			 * since we won't be needing any room for TCP/IP
337 			 * headers; tell it to not break up the writes (this
338 			 * would reduce the amount of work done by kmem); and
339 			 * configure our receive buffer. Note that we can only
340 			 * do this for the active connect tcp since our eager is
341 			 * still detached; it will be dealt with later in
342 			 * tcp_accept_finish().
343 			 */
344 			if (!IPCL_IS_NONSTR(peer_tcp->tcp_connp)) {
345 				struct stroptions *stropt;
346 
347 				DB_TYPE(mp) = M_SETOPTS;
348 				mp->b_wptr += sizeof (*stropt);
349 
350 				stropt = (struct stroptions *)mp->b_rptr;
351 				stropt->so_flags = SO_MAXBLK|SO_WROFF|SO_HIWAT;
352 				stropt->so_maxblk = tcp_maxpsz_set(peer_tcp,
353 				    B_FALSE);
354 				stropt->so_wroff = 0;
355 
356 				/*
357 				 * Record the stream head's high water mark for
358 				 * peer endpoint; this is used for flow-control
359 				 * purposes in tcp_fuse_output().
360 				 */
361 				stropt->so_hiwat = tcp_fuse_set_rcv_hiwat(
362 				    peer_tcp, peer_rq->q_hiwat);
363 
364 				tcp->tcp_refuse = B_FALSE;
365 				peer_tcp->tcp_refuse = B_FALSE;
366 				/* Send the options up */
367 				putnext(peer_rq, mp);
368 			} else {
369 				struct sock_proto_props sopp;
370 
371 				/* The peer is a non-STREAMS end point */
372 				ASSERT(IPCL_IS_TCP(peer_connp));
373 
374 				(void) tcp_fuse_set_rcv_hiwat(tcp,
375 				    tcp->tcp_saved_listener->tcp_recv_hiwater);
376 
377 				sopp.sopp_flags = SOCKOPT_MAXBLK |
378 				    SOCKOPT_WROFF | SOCKOPT_RCVHIWAT;
379 				sopp.sopp_maxblk = tcp_maxpsz_set(peer_tcp,
380 				    B_FALSE);
381 				sopp.sopp_wroff = 0;
382 				sopp.sopp_rxhiwat = tcp_fuse_set_rcv_hiwat(
383 				    peer_tcp, peer_tcp->tcp_recv_hiwater);
384 				(*peer_connp->conn_upcalls->su_set_proto_props)
385 				    (peer_connp->conn_upper_handle, &sopp);
386 			}
387 		}
388 		tcp->tcp_refuse = B_FALSE;
389 		peer_tcp->tcp_refuse = B_FALSE;
390 	} else {
391 		TCP_STAT(tcps, tcp_fusion_unqualified);
392 	}
393 	CONN_DEC_REF(peer_connp);
394 	return;
395 
396 failed:
397 	if (tcp->tcp_fused_sigurg_mp != NULL) {
398 		freeb(tcp->tcp_fused_sigurg_mp);
399 		tcp->tcp_fused_sigurg_mp = NULL;
400 	}
401 	if (peer_tcp->tcp_fused_sigurg_mp != NULL) {
402 		freeb(peer_tcp->tcp_fused_sigurg_mp);
403 		peer_tcp->tcp_fused_sigurg_mp = NULL;
404 	}
405 	CONN_DEC_REF(peer_connp);
406 }
407 
408 /*
409  * Unfuse a previously-fused pair of tcp loopback endpoints.
410  */
411 void
412 tcp_unfuse(tcp_t *tcp)
413 {
414 	tcp_t *peer_tcp = tcp->tcp_loopback_peer;
415 
416 	ASSERT(tcp->tcp_fused && peer_tcp != NULL);
417 	ASSERT(peer_tcp->tcp_fused && peer_tcp->tcp_loopback_peer == tcp);
418 	ASSERT(tcp->tcp_connp->conn_sqp == peer_tcp->tcp_connp->conn_sqp);
419 	ASSERT(tcp->tcp_unsent == 0 && peer_tcp->tcp_unsent == 0);
420 
421 	/*
422 	 * We disable synchronous streams, drain any queued data and
423 	 * clear tcp_direct_sockfs.  The synchronous streams entry
424 	 * points will become no-ops after this point.
425 	 */
426 	tcp_fuse_disable_pair(tcp, B_TRUE);
427 
428 	/*
429 	 * Update th_seq and th_ack in the header template
430 	 */
431 	U32_TO_ABE32(tcp->tcp_snxt, tcp->tcp_tcph->th_seq);
432 	U32_TO_ABE32(tcp->tcp_rnxt, tcp->tcp_tcph->th_ack);
433 	U32_TO_ABE32(peer_tcp->tcp_snxt, peer_tcp->tcp_tcph->th_seq);
434 	U32_TO_ABE32(peer_tcp->tcp_rnxt, peer_tcp->tcp_tcph->th_ack);
435 
436 	/* Unfuse the endpoints */
437 	peer_tcp->tcp_fused = tcp->tcp_fused = B_FALSE;
438 	peer_tcp->tcp_loopback_peer = tcp->tcp_loopback_peer = NULL;
439 	if (!IPCL_IS_NONSTR(peer_tcp->tcp_connp)) {
440 		ASSERT(peer_tcp->tcp_fused_sigurg_mp != NULL);
441 		freeb(peer_tcp->tcp_fused_sigurg_mp);
442 		peer_tcp->tcp_fused_sigurg_mp = NULL;
443 	}
444 	if (!IPCL_IS_NONSTR(tcp->tcp_connp)) {
445 		ASSERT(tcp->tcp_fused_sigurg_mp != NULL);
446 		freeb(tcp->tcp_fused_sigurg_mp);
447 		tcp->tcp_fused_sigurg_mp = NULL;
448 	}
449 }
450 
451 /*
452  * Fusion output routine for urgent data.  This routine is called by
453  * tcp_fuse_output() for handling non-M_DATA mblks.
454  */
455 void
456 tcp_fuse_output_urg(tcp_t *tcp, mblk_t *mp)
457 {
458 	mblk_t *mp1;
459 	struct T_exdata_ind *tei;
460 	tcp_t *peer_tcp = tcp->tcp_loopback_peer;
461 	mblk_t *head, *prev_head = NULL;
462 	tcp_stack_t	*tcps = tcp->tcp_tcps;
463 
464 	ASSERT(tcp->tcp_fused);
465 	ASSERT(peer_tcp != NULL && peer_tcp->tcp_loopback_peer == tcp);
466 	ASSERT(DB_TYPE(mp) == M_PROTO || DB_TYPE(mp) == M_PCPROTO);
467 	ASSERT(mp->b_cont != NULL && DB_TYPE(mp->b_cont) == M_DATA);
468 	ASSERT(MBLKL(mp) >= sizeof (*tei) && MBLKL(mp->b_cont) > 0);
469 
470 	/*
471 	 * Urgent data arrives in the form of T_EXDATA_REQ from above.
472 	 * Each occurence denotes a new urgent pointer.  For each new
473 	 * urgent pointer we signal (SIGURG) the receiving app to indicate
474 	 * that it needs to go into urgent mode.  This is similar to the
475 	 * urgent data handling in the regular tcp.  We don't need to keep
476 	 * track of where the urgent pointer is, because each T_EXDATA_REQ
477 	 * "advances" the urgent pointer for us.
478 	 *
479 	 * The actual urgent data carried by T_EXDATA_REQ is then prepended
480 	 * by a T_EXDATA_IND before being enqueued behind any existing data
481 	 * destined for the receiving app.  There is only a single urgent
482 	 * pointer (out-of-band mark) for a given tcp.  If the new urgent
483 	 * data arrives before the receiving app reads some existing urgent
484 	 * data, the previous marker is lost.  This behavior is emulated
485 	 * accordingly below, by removing any existing T_EXDATA_IND messages
486 	 * and essentially converting old urgent data into non-urgent.
487 	 */
488 	ASSERT(tcp->tcp_valid_bits & TCP_URG_VALID);
489 	/* Let sender get out of urgent mode */
490 	tcp->tcp_valid_bits &= ~TCP_URG_VALID;
491 
492 	/*
493 	 * This flag indicates that a signal needs to be sent up.
494 	 * This flag will only get cleared once SIGURG is delivered and
495 	 * is not affected by the tcp_fused flag -- delivery will still
496 	 * happen even after an endpoint is unfused, to handle the case
497 	 * where the sending endpoint immediately closes/unfuses after
498 	 * sending urgent data and the accept is not yet finished.
499 	 */
500 	peer_tcp->tcp_fused_sigurg = B_TRUE;
501 
502 	/* Reuse T_EXDATA_REQ mblk for T_EXDATA_IND */
503 	DB_TYPE(mp) = M_PROTO;
504 	tei = (struct T_exdata_ind *)mp->b_rptr;
505 	tei->PRIM_type = T_EXDATA_IND;
506 	tei->MORE_flag = 0;
507 	mp->b_wptr = (uchar_t *)&tei[1];
508 
509 	TCP_STAT(tcps, tcp_fusion_urg);
510 	BUMP_MIB(&tcps->tcps_mib, tcpOutUrg);
511 
512 	head = peer_tcp->tcp_rcv_list;
513 	while (head != NULL) {
514 		/*
515 		 * Remove existing T_EXDATA_IND, keep the data which follows
516 		 * it and relink our list.  Note that we don't modify the
517 		 * tcp_rcv_last_tail since it never points to T_EXDATA_IND.
518 		 */
519 		if (DB_TYPE(head) != M_DATA) {
520 			mp1 = head;
521 
522 			ASSERT(DB_TYPE(mp1->b_cont) == M_DATA);
523 			head = mp1->b_cont;
524 			mp1->b_cont = NULL;
525 			head->b_next = mp1->b_next;
526 			mp1->b_next = NULL;
527 			if (prev_head != NULL)
528 				prev_head->b_next = head;
529 			if (peer_tcp->tcp_rcv_list == mp1)
530 				peer_tcp->tcp_rcv_list = head;
531 			if (peer_tcp->tcp_rcv_last_head == mp1)
532 				peer_tcp->tcp_rcv_last_head = head;
533 			freeb(mp1);
534 		}
535 		prev_head = head;
536 		head = head->b_next;
537 	}
538 }
539 
540 /*
541  * Fusion output routine, called by tcp_output() and tcp_wput_proto().
542  * If we are modifying any member that can be changed outside the squeue,
543  * like tcp_flow_stopped, we need to take tcp_non_sq_lock.
544  */
545 boolean_t
546 tcp_fuse_output(tcp_t *tcp, mblk_t *mp, uint32_t send_size)
547 {
548 	tcp_t *peer_tcp = tcp->tcp_loopback_peer;
549 	uint_t max_unread;
550 	boolean_t flow_stopped, peer_data_queued = B_FALSE;
551 	boolean_t urgent = (DB_TYPE(mp) != M_DATA);
552 	boolean_t push = B_TRUE;
553 	mblk_t *mp1 = mp;
554 	ill_t *ilp, *olp;
555 	ipif_t *iifp, *oifp;
556 	ipha_t *ipha;
557 	ip6_t *ip6h;
558 	tcph_t *tcph;
559 	uint_t ip_hdr_len;
560 	uint32_t seq;
561 	uint32_t recv_size = send_size;
562 	tcp_stack_t	*tcps = tcp->tcp_tcps;
563 	netstack_t	*ns = tcps->tcps_netstack;
564 	ip_stack_t	*ipst = ns->netstack_ip;
565 
566 	ASSERT(tcp->tcp_fused);
567 	ASSERT(peer_tcp != NULL && peer_tcp->tcp_loopback_peer == tcp);
568 	ASSERT(tcp->tcp_connp->conn_sqp == peer_tcp->tcp_connp->conn_sqp);
569 	ASSERT(DB_TYPE(mp) == M_DATA || DB_TYPE(mp) == M_PROTO ||
570 	    DB_TYPE(mp) == M_PCPROTO);
571 
572 	/* If this connection requires IP, unfuse and use regular path */
573 	if (tcp_loopback_needs_ip(tcp, ns) ||
574 	    tcp_loopback_needs_ip(peer_tcp, ns) ||
575 	    IPP_ENABLED(IPP_LOCAL_OUT|IPP_LOCAL_IN, ipst) ||
576 	    list_head(&ipst->ips_ipobs_cb_list) != NULL) {
577 		TCP_STAT(tcps, tcp_fusion_aborted);
578 		tcp->tcp_refuse = B_TRUE;
579 		peer_tcp->tcp_refuse = B_TRUE;
580 
581 		bcopy(peer_tcp->tcp_tcph, &tcp->tcp_saved_tcph,
582 		    sizeof (tcph_t));
583 		bcopy(tcp->tcp_tcph, &peer_tcp->tcp_saved_tcph,
584 		    sizeof (tcph_t));
585 		if (tcp->tcp_ipversion == IPV4_VERSION) {
586 			bcopy(peer_tcp->tcp_ipha, &tcp->tcp_saved_ipha,
587 			    sizeof (ipha_t));
588 			bcopy(tcp->tcp_ipha, &peer_tcp->tcp_saved_ipha,
589 			    sizeof (ipha_t));
590 		} else {
591 			bcopy(peer_tcp->tcp_ip6h, &tcp->tcp_saved_ip6h,
592 			    sizeof (ip6_t));
593 			bcopy(tcp->tcp_ip6h, &peer_tcp->tcp_saved_ip6h,
594 			    sizeof (ip6_t));
595 		}
596 		goto unfuse;
597 	}
598 
599 	if (send_size == 0) {
600 		freemsg(mp);
601 		return (B_TRUE);
602 	}
603 	max_unread = peer_tcp->tcp_fuse_rcv_unread_hiwater;
604 
605 	/*
606 	 * Handle urgent data; we either send up SIGURG to the peer now
607 	 * or do it later when we drain, in case the peer is detached
608 	 * or if we're short of memory for M_PCSIG mblk.
609 	 */
610 	if (urgent) {
611 		/*
612 		 * We stop synchronous streams when we have urgent data
613 		 * queued to prevent tcp_fuse_rrw() from pulling it.  If
614 		 * for some reasons the urgent data can't be delivered
615 		 * below, synchronous streams will remain stopped until
616 		 * someone drains the tcp_rcv_list.
617 		 */
618 		TCP_FUSE_SYNCSTR_PLUG_DRAIN(peer_tcp);
619 		tcp_fuse_output_urg(tcp, mp);
620 
621 		mp1 = mp->b_cont;
622 	}
623 
624 	if (tcp->tcp_ipversion == IPV4_VERSION &&
625 	    (HOOKS4_INTERESTED_LOOPBACK_IN(ipst) ||
626 	    HOOKS4_INTERESTED_LOOPBACK_OUT(ipst)) ||
627 	    tcp->tcp_ipversion == IPV6_VERSION &&
628 	    (HOOKS6_INTERESTED_LOOPBACK_IN(ipst) ||
629 	    HOOKS6_INTERESTED_LOOPBACK_OUT(ipst))) {
630 		/*
631 		 * Build ip and tcp header to satisfy FW_HOOKS.
632 		 * We only build it when any hook is present.
633 		 */
634 		if ((mp1 = tcp_xmit_mp(tcp, mp1, tcp->tcp_mss, NULL, NULL,
635 		    tcp->tcp_snxt, B_TRUE, NULL, B_FALSE)) == NULL)
636 			/* If tcp_xmit_mp fails, use regular path */
637 			goto unfuse;
638 
639 		/*
640 		 * The ipif and ill can be safely referenced under the
641 		 * protection of conn_lock - see head of function comment for
642 		 * conn_get_held_ipif(). It is necessary to check that both
643 		 * the ipif and ill can be looked up (i.e. not condemned). If
644 		 * not, bail out and unfuse this connection.
645 		 */
646 		mutex_enter(&peer_tcp->tcp_connp->conn_lock);
647 		if ((peer_tcp->tcp_connp->conn_ire_cache == NULL) ||
648 		    (peer_tcp->tcp_connp->conn_ire_cache->ire_marks &
649 		    IRE_MARK_CONDEMNED) ||
650 		    ((oifp = peer_tcp->tcp_connp->conn_ire_cache->ire_ipif)
651 		    == NULL) ||
652 		    (!IPIF_CAN_LOOKUP(oifp)) ||
653 		    ((olp = oifp->ipif_ill) == NULL) ||
654 		    (ill_check_and_refhold(olp) != 0)) {
655 			mutex_exit(&peer_tcp->tcp_connp->conn_lock);
656 			goto unfuse;
657 		}
658 		mutex_exit(&peer_tcp->tcp_connp->conn_lock);
659 
660 		/* PFHooks: LOOPBACK_OUT */
661 		if (tcp->tcp_ipversion == IPV4_VERSION) {
662 			ipha = (ipha_t *)mp1->b_rptr;
663 
664 			DTRACE_PROBE4(ip4__loopback__out__start,
665 			    ill_t *, NULL, ill_t *, olp,
666 			    ipha_t *, ipha, mblk_t *, mp1);
667 			FW_HOOKS(ipst->ips_ip4_loopback_out_event,
668 			    ipst->ips_ipv4firewall_loopback_out,
669 			    NULL, olp, ipha, mp1, mp1, 0, ipst);
670 			DTRACE_PROBE1(ip4__loopback__out__end, mblk_t *, mp1);
671 		} else {
672 			ip6h = (ip6_t *)mp1->b_rptr;
673 
674 			DTRACE_PROBE4(ip6__loopback__out__start,
675 			    ill_t *, NULL, ill_t *, olp,
676 			    ip6_t *, ip6h, mblk_t *, mp1);
677 			FW_HOOKS6(ipst->ips_ip6_loopback_out_event,
678 			    ipst->ips_ipv6firewall_loopback_out,
679 			    NULL, olp, ip6h, mp1, mp1, 0, ipst);
680 			DTRACE_PROBE1(ip6__loopback__out__end, mblk_t *, mp1);
681 		}
682 		ill_refrele(olp);
683 
684 		if (mp1 == NULL)
685 			goto unfuse;
686 
687 		/*
688 		 * The ipif and ill can be safely referenced under the
689 		 * protection of conn_lock - see head of function comment for
690 		 * conn_get_held_ipif(). It is necessary to check that both
691 		 * the ipif and ill can be looked up (i.e. not condemned). If
692 		 * not, bail out and unfuse this connection.
693 		 */
694 		mutex_enter(&tcp->tcp_connp->conn_lock);
695 		if ((tcp->tcp_connp->conn_ire_cache == NULL) ||
696 		    (tcp->tcp_connp->conn_ire_cache->ire_marks &
697 		    IRE_MARK_CONDEMNED) ||
698 		    ((iifp = tcp->tcp_connp->conn_ire_cache->ire_ipif)
699 		    == NULL) ||
700 		    (!IPIF_CAN_LOOKUP(iifp)) ||
701 		    ((ilp = iifp->ipif_ill) == NULL) ||
702 		    (ill_check_and_refhold(ilp) != 0)) {
703 			mutex_exit(&tcp->tcp_connp->conn_lock);
704 			goto unfuse;
705 		}
706 		mutex_exit(&tcp->tcp_connp->conn_lock);
707 
708 		/* PFHooks: LOOPBACK_IN */
709 		if (tcp->tcp_ipversion == IPV4_VERSION) {
710 			DTRACE_PROBE4(ip4__loopback__in__start,
711 			    ill_t *, ilp, ill_t *, NULL,
712 			    ipha_t *, ipha, mblk_t *, mp1);
713 			FW_HOOKS(ipst->ips_ip4_loopback_in_event,
714 			    ipst->ips_ipv4firewall_loopback_in,
715 			    ilp, NULL, ipha, mp1, mp1, 0, ipst);
716 			DTRACE_PROBE1(ip4__loopback__in__end, mblk_t *, mp1);
717 			ill_refrele(ilp);
718 			if (mp1 == NULL)
719 				goto unfuse;
720 
721 			ip_hdr_len = IPH_HDR_LENGTH(ipha);
722 		} else {
723 			DTRACE_PROBE4(ip6__loopback__in__start,
724 			    ill_t *, ilp, ill_t *, NULL,
725 			    ip6_t *, ip6h, mblk_t *, mp1);
726 			FW_HOOKS6(ipst->ips_ip6_loopback_in_event,
727 			    ipst->ips_ipv6firewall_loopback_in,
728 			    ilp, NULL, ip6h, mp1, mp1, 0, ipst);
729 			DTRACE_PROBE1(ip6__loopback__in__end, mblk_t *, mp1);
730 			ill_refrele(ilp);
731 			if (mp1 == NULL)
732 				goto unfuse;
733 
734 			ip_hdr_len = ip_hdr_length_v6(mp1, ip6h);
735 		}
736 
737 		/* Data length might be changed by FW_HOOKS */
738 		tcph = (tcph_t *)&mp1->b_rptr[ip_hdr_len];
739 		seq = ABE32_TO_U32(tcph->th_seq);
740 		recv_size += seq - tcp->tcp_snxt;
741 
742 		/*
743 		 * The message duplicated by tcp_xmit_mp is freed.
744 		 * Note: the original message passed in remains unchanged.
745 		 */
746 		freemsg(mp1);
747 	}
748 
749 	mutex_enter(&peer_tcp->tcp_non_sq_lock);
750 	/*
751 	 * Wake up and signal the peer; it is okay to do this before
752 	 * enqueueing because we are holding the lock.  One of the
753 	 * advantages of synchronous streams is the ability for us to
754 	 * find out when the application performs a read on the socket,
755 	 * by way of tcp_fuse_rrw() entry point being called.  Every
756 	 * data that gets enqueued onto the receiver is treated as if
757 	 * it has arrived at the receiving endpoint, thus generating
758 	 * SIGPOLL/SIGIO for asynchronous socket just as in the strrput()
759 	 * case.  However, we only wake up the application when necessary,
760 	 * i.e. during the first enqueue.  When tcp_fuse_rrw() is called
761 	 * it will send everything upstream.
762 	 */
763 	if (peer_tcp->tcp_direct_sockfs && !urgent &&
764 	    !TCP_IS_DETACHED(peer_tcp)) {
765 		/* Update poll events and send SIGPOLL/SIGIO if necessary */
766 		STR_WAKEUP_SENDSIG(STREAM(peer_tcp->tcp_rq),
767 		    peer_tcp->tcp_rcv_list);
768 	}
769 
770 	/*
771 	 * Enqueue data into the peer's receive list; we may or may not
772 	 * drain the contents depending on the conditions below.
773 	 *
774 	 * tcp_hard_binding indicates that accept has not yet completed,
775 	 * in which case we use tcp_rcv_enqueue() instead of calling
776 	 * su_recv directly. Queued data will be drained when the accept
777 	 * completes (in tcp_accept_finish()).
778 	 */
779 	if (IPCL_IS_NONSTR(peer_tcp->tcp_connp) &&
780 	    !peer_tcp->tcp_hard_binding) {
781 		int error;
782 		int flags = 0;
783 
784 		if ((tcp->tcp_valid_bits & TCP_URG_VALID) &&
785 		    (tcp->tcp_urg == tcp->tcp_snxt)) {
786 			flags = MSG_OOB;
787 			(*peer_tcp->tcp_connp->conn_upcalls->su_signal_oob)
788 			    (peer_tcp->tcp_connp->conn_upper_handle, 0);
789 			tcp->tcp_valid_bits &= ~TCP_URG_VALID;
790 		}
791 		(*peer_tcp->tcp_connp->conn_upcalls->su_recv)(
792 		    peer_tcp->tcp_connp->conn_upper_handle, mp, recv_size,
793 		    flags, &error, &push);
794 		ASSERT(error != EOPNOTSUPP);
795 	} else {
796 		if (IPCL_IS_NONSTR(peer_tcp->tcp_connp) &&
797 		    (tcp->tcp_valid_bits & TCP_URG_VALID) &&
798 		    (tcp->tcp_urg == tcp->tcp_snxt)) {
799 			/*
800 			 * Can not deal with urgent pointers
801 			 * that arrive before the connection has been
802 			 * accept()ed.
803 			 */
804 			tcp->tcp_valid_bits &= ~TCP_URG_VALID;
805 			freemsg(mp);
806 			mutex_exit(&peer_tcp->tcp_non_sq_lock);
807 			return (B_TRUE);
808 		}
809 
810 		tcp_rcv_enqueue(peer_tcp, mp, recv_size);
811 	}
812 
813 	/* In case it wrapped around and also to keep it constant */
814 	peer_tcp->tcp_rwnd += recv_size;
815 	/*
816 	 * We increase the peer's unread message count here whilst still
817 	 * holding it's tcp_non_sq_lock. This ensures that the increment
818 	 * occurs in the same lock acquisition perimeter as the enqueue.
819 	 * Depending on lock hierarchy, we can release these locks which
820 	 * creates a window in which we can race with tcp_fuse_rrw()
821 	 */
822 	peer_tcp->tcp_fuse_rcv_unread_cnt++;
823 
824 	/*
825 	 * Exercise flow-control when needed; we will get back-enabled
826 	 * in either tcp_accept_finish(), tcp_unfuse(), or tcp_fuse_rrw().
827 	 * If tcp_direct_sockfs is on or if the peer endpoint is detached,
828 	 * we emulate streams flow control by checking the peer's queue
829 	 * size and high water mark; otherwise we simply use canputnext()
830 	 * to decide if we need to stop our flow.
831 	 *
832 	 * The outstanding unread data block check does not apply for a
833 	 * detached receiver; this is to avoid unnecessary blocking of the
834 	 * sender while the accept is currently in progress and is quite
835 	 * similar to the regular tcp.
836 	 */
837 	if (TCP_IS_DETACHED(peer_tcp) || max_unread == 0)
838 		max_unread = UINT_MAX;
839 
840 	/*
841 	 * Since we are accessing our tcp_flow_stopped and might modify it,
842 	 * we need to take tcp->tcp_non_sq_lock. The lock for the highest
843 	 * address is held first. Dropping peer_tcp->tcp_non_sq_lock should
844 	 * not be an issue here since we are within the squeue and the peer
845 	 * won't disappear.
846 	 */
847 	if (tcp > peer_tcp) {
848 		mutex_exit(&peer_tcp->tcp_non_sq_lock);
849 		mutex_enter(&tcp->tcp_non_sq_lock);
850 		mutex_enter(&peer_tcp->tcp_non_sq_lock);
851 	} else {
852 		mutex_enter(&tcp->tcp_non_sq_lock);
853 	}
854 	flow_stopped = tcp->tcp_flow_stopped;
855 	if (((peer_tcp->tcp_direct_sockfs || TCP_IS_DETACHED(peer_tcp)) &&
856 	    (peer_tcp->tcp_rcv_cnt >= peer_tcp->tcp_fuse_rcv_hiwater ||
857 	    peer_tcp->tcp_fuse_rcv_unread_cnt >= max_unread)) ||
858 	    (!peer_tcp->tcp_direct_sockfs && !TCP_IS_DETACHED(peer_tcp) &&
859 	    !IPCL_IS_NONSTR(peer_tcp->tcp_connp) &&
860 	    !canputnext(peer_tcp->tcp_rq))) {
861 		peer_data_queued = B_TRUE;
862 	}
863 
864 	if (!flow_stopped && (peer_data_queued ||
865 	    (TCP_UNSENT_BYTES(tcp) >= tcp->tcp_xmit_hiwater))) {
866 		tcp_setqfull(tcp);
867 		flow_stopped = B_TRUE;
868 		TCP_STAT(tcps, tcp_fusion_flowctl);
869 		DTRACE_PROBE4(tcp__fuse__output__flowctl, tcp_t *, tcp,
870 		    uint_t, send_size, uint_t, peer_tcp->tcp_rcv_cnt,
871 		    uint_t, peer_tcp->tcp_fuse_rcv_unread_cnt);
872 	} else if (flow_stopped && !peer_data_queued &&
873 	    (TCP_UNSENT_BYTES(tcp) <= tcp->tcp_xmit_lowater)) {
874 		tcp_clrqfull(tcp);
875 		TCP_STAT(tcps, tcp_fusion_backenabled);
876 		flow_stopped = B_FALSE;
877 	}
878 	mutex_exit(&tcp->tcp_non_sq_lock);
879 
880 	/*
881 	 * If we are in synchronous streams mode and the peer read queue is
882 	 * not full then schedule a push timer if one is not scheduled
883 	 * already. This is needed for applications which use MSG_PEEK to
884 	 * determine the number of bytes available before issuing a 'real'
885 	 * read. It also makes flow control more deterministic, particularly
886 	 * for smaller message sizes.
887 	 */
888 	if (!urgent && peer_tcp->tcp_direct_sockfs &&
889 	    peer_tcp->tcp_push_tid == 0 && !TCP_IS_DETACHED(peer_tcp) &&
890 	    canputnext(peer_tcp->tcp_rq)) {
891 		peer_tcp->tcp_push_tid = TCP_TIMER(peer_tcp, tcp_push_timer,
892 		    MSEC_TO_TICK(tcps->tcps_push_timer_interval));
893 	}
894 	mutex_exit(&peer_tcp->tcp_non_sq_lock);
895 	ipst->ips_loopback_packets++;
896 	tcp->tcp_last_sent_len = send_size;
897 
898 	/* Need to adjust the following SNMP MIB-related variables */
899 	tcp->tcp_snxt += send_size;
900 	tcp->tcp_suna = tcp->tcp_snxt;
901 	peer_tcp->tcp_rnxt += recv_size;
902 	peer_tcp->tcp_rack = peer_tcp->tcp_rnxt;
903 
904 	BUMP_MIB(&tcps->tcps_mib, tcpOutDataSegs);
905 	UPDATE_MIB(&tcps->tcps_mib, tcpOutDataBytes, send_size);
906 
907 	BUMP_MIB(&tcps->tcps_mib, tcpInSegs);
908 	BUMP_MIB(&tcps->tcps_mib, tcpInDataInorderSegs);
909 	UPDATE_MIB(&tcps->tcps_mib, tcpInDataInorderBytes, send_size);
910 
911 	BUMP_LOCAL(tcp->tcp_obsegs);
912 	BUMP_LOCAL(peer_tcp->tcp_ibsegs);
913 
914 	DTRACE_PROBE2(tcp__fuse__output, tcp_t *, tcp, uint_t, send_size);
915 
916 	if (!TCP_IS_DETACHED(peer_tcp)) {
917 		/*
918 		 * Drain the peer's receive queue it has urgent data or if
919 		 * we're not flow-controlled.  There is no need for draining
920 		 * normal data when tcp_direct_sockfs is on because the peer
921 		 * will pull the data via tcp_fuse_rrw().
922 		 */
923 		if (urgent || (!flow_stopped && !peer_tcp->tcp_direct_sockfs)) {
924 			ASSERT(IPCL_IS_NONSTR(peer_tcp->tcp_connp) ||
925 			    peer_tcp->tcp_rcv_list != NULL);
926 			/*
927 			 * For TLI-based streams, a thread in tcp_accept_swap()
928 			 * can race with us.  That thread will ensure that the
929 			 * correct peer_tcp->tcp_rq is globally visible before
930 			 * peer_tcp->tcp_detached is visible as clear, but we
931 			 * must also ensure that the load of tcp_rq cannot be
932 			 * reordered to be before the tcp_detached check.
933 			 */
934 			membar_consumer();
935 			(void) tcp_fuse_rcv_drain(peer_tcp->tcp_rq, peer_tcp,
936 			    NULL);
937 			/*
938 			 * If synchronous streams was stopped above due
939 			 * to the presence of urgent data, re-enable it.
940 			 */
941 			if (urgent)
942 				TCP_FUSE_SYNCSTR_UNPLUG_DRAIN(peer_tcp);
943 		}
944 	}
945 	return (B_TRUE);
946 unfuse:
947 	tcp_unfuse(tcp);
948 	return (B_FALSE);
949 }
950 
951 /*
952  * This routine gets called to deliver data upstream on a fused or
953  * previously fused tcp loopback endpoint; the latter happens only
954  * when there is a pending SIGURG signal plus urgent data that can't
955  * be sent upstream in the past.
956  */
957 boolean_t
958 tcp_fuse_rcv_drain(queue_t *q, tcp_t *tcp, mblk_t **sigurg_mpp)
959 {
960 	mblk_t *mp;
961 	conn_t	*connp = tcp->tcp_connp;
962 
963 #ifdef DEBUG
964 	uint_t cnt = 0;
965 #endif
966 	tcp_stack_t	*tcps = tcp->tcp_tcps;
967 	tcp_t		*peer_tcp = tcp->tcp_loopback_peer;
968 	boolean_t	sd_rd_eof = B_FALSE;
969 
970 	ASSERT(tcp->tcp_loopback);
971 	ASSERT(tcp->tcp_fused || tcp->tcp_fused_sigurg);
972 	ASSERT(!tcp->tcp_fused || tcp->tcp_loopback_peer != NULL);
973 	ASSERT(IPCL_IS_NONSTR(connp) || sigurg_mpp != NULL || tcp->tcp_fused);
974 
975 	/* No need for the push timer now, in case it was scheduled */
976 	if (tcp->tcp_push_tid != 0) {
977 		(void) TCP_TIMER_CANCEL(tcp, tcp->tcp_push_tid);
978 		tcp->tcp_push_tid = 0;
979 	}
980 	/*
981 	 * If there's urgent data sitting in receive list and we didn't
982 	 * get a chance to send up a SIGURG signal, make sure we send
983 	 * it first before draining in order to ensure that SIOCATMARK
984 	 * works properly.
985 	 */
986 	if (tcp->tcp_fused_sigurg) {
987 		tcp->tcp_fused_sigurg = B_FALSE;
988 		if (IPCL_IS_NONSTR(connp)) {
989 			(*connp->conn_upcalls->su_signal_oob)
990 			    (connp->conn_upper_handle, 0);
991 		} else {
992 			/*
993 			 * sigurg_mpp is normally NULL, i.e. when we're still
994 			 * fused and didn't get here because of tcp_unfuse().
995 			 * In this case try hard to allocate the M_PCSIG mblk.
996 			 */
997 			if (sigurg_mpp == NULL &&
998 			    (mp = allocb(1, BPRI_HI)) == NULL &&
999 			    (mp = allocb_tryhard(1)) == NULL) {
1000 				/* Alloc failed; try again next time */
1001 				tcp->tcp_push_tid = TCP_TIMER(tcp,
1002 				    tcp_push_timer,
1003 				    MSEC_TO_TICK(
1004 				    tcps->tcps_push_timer_interval));
1005 				return (B_TRUE);
1006 			} else if (sigurg_mpp != NULL) {
1007 				/*
1008 				 * Use the supplied M_PCSIG mblk; it means we're
1009 				 * either unfused or in the process of unfusing,
1010 				 * and the drain must happen now.
1011 				 */
1012 				mp = *sigurg_mpp;
1013 				*sigurg_mpp = NULL;
1014 			}
1015 			ASSERT(mp != NULL);
1016 
1017 			/* Send up the signal */
1018 			DB_TYPE(mp) = M_PCSIG;
1019 			*mp->b_wptr++ = (uchar_t)SIGURG;
1020 			putnext(q, mp);
1021 		}
1022 		/*
1023 		 * Let the regular tcp_rcv_drain() path handle
1024 		 * draining the data if we're no longer fused.
1025 		 */
1026 		if (!tcp->tcp_fused)
1027 			return (B_FALSE);
1028 	}
1029 
1030 	/*
1031 	 * In the synchronous streams case, we generate SIGPOLL/SIGIO for
1032 	 * each M_DATA that gets enqueued onto the receiver.  At this point
1033 	 * we are about to drain any queued data via putnext().  In order
1034 	 * to avoid extraneous signal generation from strrput(), we set
1035 	 * STRGETINPROG flag at the stream head prior to the draining and
1036 	 * restore it afterwards.  This masks out signal generation only
1037 	 * for M_DATA messages and does not affect urgent data. We only do
1038 	 * this if the STREOF flag is not set which can happen if the
1039 	 * application shuts down the read side of a stream. In this case
1040 	 * we simply free these messages to approximate the flushq behavior
1041 	 * which normally occurs when STREOF is on the stream head read queue.
1042 	 */
1043 	if (tcp->tcp_direct_sockfs)
1044 		sd_rd_eof = strrput_sig(q, B_FALSE);
1045 
1046 	/* Drain the data */
1047 	while ((mp = tcp->tcp_rcv_list) != NULL) {
1048 		tcp->tcp_rcv_list = mp->b_next;
1049 		mp->b_next = NULL;
1050 #ifdef DEBUG
1051 		cnt += msgdsize(mp);
1052 #endif
1053 		ASSERT(!IPCL_IS_NONSTR(connp));
1054 		if (sd_rd_eof) {
1055 			freemsg(mp);
1056 		} else {
1057 			putnext(q, mp);
1058 			TCP_STAT(tcps, tcp_fusion_putnext);
1059 		}
1060 	}
1061 
1062 	if (tcp->tcp_direct_sockfs && !sd_rd_eof)
1063 		(void) strrput_sig(q, B_TRUE);
1064 
1065 #ifdef DEBUG
1066 	ASSERT(cnt == tcp->tcp_rcv_cnt);
1067 #endif
1068 	tcp->tcp_rcv_last_head = NULL;
1069 	tcp->tcp_rcv_last_tail = NULL;
1070 	tcp->tcp_rcv_cnt = 0;
1071 	tcp->tcp_fuse_rcv_unread_cnt = 0;
1072 	tcp->tcp_rwnd = tcp->tcp_recv_hiwater;
1073 
1074 	if (peer_tcp->tcp_flow_stopped && (TCP_UNSENT_BYTES(peer_tcp) <=
1075 	    peer_tcp->tcp_xmit_lowater)) {
1076 		tcp_clrqfull(peer_tcp);
1077 		TCP_STAT(tcps, tcp_fusion_backenabled);
1078 	}
1079 
1080 	return (B_TRUE);
1081 }
1082 
1083 /*
1084  * Synchronous stream entry point for sockfs to retrieve
1085  * data directly from tcp_rcv_list.
1086  * tcp_fuse_rrw() might end up modifying the peer's tcp_flow_stopped,
1087  * for which it  must take the tcp_non_sq_lock of the peer as well
1088  * making any change. The order of taking the locks is based on
1089  * the TCP pointer itself. Before we get the peer we need to take
1090  * our tcp_non_sq_lock so that the peer doesn't disappear. However,
1091  * we cannot drop the lock if we have to grab the peer's lock (because
1092  * of ordering), since the peer might disappear in the interim. So,
1093  * we take our tcp_non_sq_lock, get the peer, increment the ref on the
1094  * peer's conn, drop all the locks and then take the tcp_non_sq_lock in the
1095  * desired order. Incrementing the conn ref on the peer means that the
1096  * peer won't disappear when we drop our tcp_non_sq_lock.
1097  */
1098 int
1099 tcp_fuse_rrw(queue_t *q, struiod_t *dp)
1100 {
1101 	tcp_t *tcp = Q_TO_CONN(q)->conn_tcp;
1102 	mblk_t *mp;
1103 	tcp_t *peer_tcp;
1104 	tcp_stack_t	*tcps = tcp->tcp_tcps;
1105 
1106 	mutex_enter(&tcp->tcp_non_sq_lock);
1107 
1108 	/*
1109 	 * If tcp_fuse_syncstr_plugged is set, then another thread is moving
1110 	 * the underlying data to the stream head.  We need to wait until it's
1111 	 * done, then return EBUSY so that strget() will dequeue data from the
1112 	 * stream head to ensure data is drained in-order.
1113 	 */
1114 plugged:
1115 	if (tcp->tcp_fuse_syncstr_plugged) {
1116 		do {
1117 			cv_wait(&tcp->tcp_fuse_plugcv, &tcp->tcp_non_sq_lock);
1118 		} while (tcp->tcp_fuse_syncstr_plugged);
1119 
1120 		mutex_exit(&tcp->tcp_non_sq_lock);
1121 		TCP_STAT(tcps, tcp_fusion_rrw_plugged);
1122 		TCP_STAT(tcps, tcp_fusion_rrw_busy);
1123 		return (EBUSY);
1124 	}
1125 
1126 	peer_tcp = tcp->tcp_loopback_peer;
1127 
1128 	/*
1129 	 * If someone had turned off tcp_direct_sockfs or if synchronous
1130 	 * streams is stopped, we return EBUSY.  This causes strget() to
1131 	 * dequeue data from the stream head instead.
1132 	 */
1133 	if (!tcp->tcp_direct_sockfs || tcp->tcp_fuse_syncstr_stopped) {
1134 		mutex_exit(&tcp->tcp_non_sq_lock);
1135 		TCP_STAT(tcps, tcp_fusion_rrw_busy);
1136 		return (EBUSY);
1137 	}
1138 
1139 	/*
1140 	 * Grab lock in order. The highest addressed tcp is locked first.
1141 	 * We don't do this within the tcp_rcv_list check since if we
1142 	 * have to drop the lock, for ordering, then the tcp_rcv_list
1143 	 * could change.
1144 	 */
1145 	if (peer_tcp > tcp) {
1146 		CONN_INC_REF(peer_tcp->tcp_connp);
1147 		mutex_exit(&tcp->tcp_non_sq_lock);
1148 		mutex_enter(&peer_tcp->tcp_non_sq_lock);
1149 		mutex_enter(&tcp->tcp_non_sq_lock);
1150 		/*
1151 		 * This might have changed in the interim
1152 		 * Once read-side tcp_non_sq_lock is dropped above
1153 		 * anything can happen, we need to check all
1154 		 * known conditions again once we reaquire
1155 		 * read-side tcp_non_sq_lock.
1156 		 */
1157 		if (tcp->tcp_fuse_syncstr_plugged) {
1158 			mutex_exit(&peer_tcp->tcp_non_sq_lock);
1159 			CONN_DEC_REF(peer_tcp->tcp_connp);
1160 			goto plugged;
1161 		}
1162 		if (!tcp->tcp_direct_sockfs || tcp->tcp_fuse_syncstr_stopped) {
1163 			mutex_exit(&tcp->tcp_non_sq_lock);
1164 			mutex_exit(&peer_tcp->tcp_non_sq_lock);
1165 			CONN_DEC_REF(peer_tcp->tcp_connp);
1166 			TCP_STAT(tcps, tcp_fusion_rrw_busy);
1167 			return (EBUSY);
1168 		}
1169 		CONN_DEC_REF(peer_tcp->tcp_connp);
1170 	} else {
1171 		mutex_enter(&peer_tcp->tcp_non_sq_lock);
1172 	}
1173 
1174 	if ((mp = tcp->tcp_rcv_list) != NULL) {
1175 
1176 		DTRACE_PROBE3(tcp__fuse__rrw, tcp_t *, tcp,
1177 		    uint32_t, tcp->tcp_rcv_cnt, ssize_t, dp->d_uio.uio_resid);
1178 
1179 		tcp->tcp_rcv_list = NULL;
1180 		TCP_STAT(tcps, tcp_fusion_rrw_msgcnt);
1181 
1182 		/*
1183 		 * At this point nothing should be left in tcp_rcv_list.
1184 		 * The only possible case where we would have a chain of
1185 		 * b_next-linked messages is urgent data, but we wouldn't
1186 		 * be here if that's true since urgent data is delivered
1187 		 * via putnext() and synchronous streams is stopped until
1188 		 * tcp_fuse_rcv_drain() is finished.
1189 		 */
1190 		ASSERT(DB_TYPE(mp) == M_DATA && mp->b_next == NULL);
1191 
1192 		tcp->tcp_rcv_last_head = NULL;
1193 		tcp->tcp_rcv_last_tail = NULL;
1194 		tcp->tcp_rcv_cnt = 0;
1195 		tcp->tcp_fuse_rcv_unread_cnt = 0;
1196 
1197 		if (peer_tcp->tcp_flow_stopped &&
1198 		    (TCP_UNSENT_BYTES(peer_tcp) <=
1199 		    peer_tcp->tcp_xmit_lowater)) {
1200 			tcp_clrqfull(peer_tcp);
1201 			TCP_STAT(tcps, tcp_fusion_backenabled);
1202 		}
1203 	}
1204 	mutex_exit(&peer_tcp->tcp_non_sq_lock);
1205 	/*
1206 	 * Either we just dequeued everything or we get here from sockfs
1207 	 * and have nothing to return; in this case clear RSLEEP.
1208 	 */
1209 	ASSERT(tcp->tcp_rcv_last_head == NULL);
1210 	ASSERT(tcp->tcp_rcv_last_tail == NULL);
1211 	ASSERT(tcp->tcp_rcv_cnt == 0);
1212 	ASSERT(tcp->tcp_fuse_rcv_unread_cnt == 0);
1213 	STR_WAKEUP_CLEAR(STREAM(q));
1214 
1215 	mutex_exit(&tcp->tcp_non_sq_lock);
1216 	dp->d_mp = mp;
1217 	return (0);
1218 }
1219 
1220 /*
1221  * Synchronous stream entry point used by certain ioctls to retrieve
1222  * information about or peek into the tcp_rcv_list.
1223  */
1224 int
1225 tcp_fuse_rinfop(queue_t *q, infod_t *dp)
1226 {
1227 	tcp_t	*tcp = Q_TO_CONN(q)->conn_tcp;
1228 	mblk_t	*mp;
1229 	uint_t	cmd = dp->d_cmd;
1230 	int	res = 0;
1231 	int	error = 0;
1232 	struct stdata *stp = STREAM(q);
1233 
1234 	mutex_enter(&tcp->tcp_non_sq_lock);
1235 	/* If shutdown on read has happened, return nothing */
1236 	mutex_enter(&stp->sd_lock);
1237 	if (stp->sd_flag & STREOF) {
1238 		mutex_exit(&stp->sd_lock);
1239 		goto done;
1240 	}
1241 	mutex_exit(&stp->sd_lock);
1242 
1243 	/*
1244 	 * It is OK not to return an answer if tcp_rcv_list is
1245 	 * currently not accessible.
1246 	 */
1247 	if (!tcp->tcp_direct_sockfs || tcp->tcp_fuse_syncstr_stopped ||
1248 	    tcp->tcp_fuse_syncstr_plugged || (mp = tcp->tcp_rcv_list) == NULL)
1249 		goto done;
1250 
1251 	if (cmd & INFOD_COUNT) {
1252 		/*
1253 		 * We have at least one message and
1254 		 * could return only one at a time.
1255 		 */
1256 		dp->d_count++;
1257 		res |= INFOD_COUNT;
1258 	}
1259 	if (cmd & INFOD_BYTES) {
1260 		/*
1261 		 * Return size of all data messages.
1262 		 */
1263 		dp->d_bytes += tcp->tcp_rcv_cnt;
1264 		res |= INFOD_BYTES;
1265 	}
1266 	if (cmd & INFOD_FIRSTBYTES) {
1267 		/*
1268 		 * Return size of first data message.
1269 		 */
1270 		dp->d_bytes = msgdsize(mp);
1271 		res |= INFOD_FIRSTBYTES;
1272 		dp->d_cmd &= ~INFOD_FIRSTBYTES;
1273 	}
1274 	if (cmd & INFOD_COPYOUT) {
1275 		mblk_t *mp1;
1276 		int n;
1277 
1278 		if (DB_TYPE(mp) == M_DATA) {
1279 			mp1 = mp;
1280 		} else {
1281 			mp1 = mp->b_cont;
1282 			ASSERT(mp1 != NULL);
1283 		}
1284 
1285 		/*
1286 		 * Return data contents of first message.
1287 		 */
1288 		ASSERT(DB_TYPE(mp1) == M_DATA);
1289 		while (mp1 != NULL && dp->d_uiop->uio_resid > 0) {
1290 			n = MIN(dp->d_uiop->uio_resid, MBLKL(mp1));
1291 			if (n != 0 && (error = uiomove((char *)mp1->b_rptr, n,
1292 			    UIO_READ, dp->d_uiop)) != 0) {
1293 				goto done;
1294 			}
1295 			mp1 = mp1->b_cont;
1296 		}
1297 		res |= INFOD_COPYOUT;
1298 		dp->d_cmd &= ~INFOD_COPYOUT;
1299 	}
1300 done:
1301 	mutex_exit(&tcp->tcp_non_sq_lock);
1302 
1303 	dp->d_res |= res;
1304 
1305 	return (error);
1306 }
1307 
1308 /*
1309  * Enable synchronous streams on a fused tcp loopback endpoint.
1310  */
1311 static void
1312 tcp_fuse_syncstr_enable(tcp_t *tcp)
1313 {
1314 	queue_t *rq = tcp->tcp_rq;
1315 	struct stdata *stp = STREAM(rq);
1316 
1317 	/* We can only enable synchronous streams for sockfs mode */
1318 	tcp->tcp_direct_sockfs = tcp->tcp_issocket && do_tcp_direct_sockfs;
1319 
1320 	if (!tcp->tcp_direct_sockfs)
1321 		return;
1322 
1323 	mutex_enter(&stp->sd_lock);
1324 	mutex_enter(QLOCK(rq));
1325 
1326 	/*
1327 	 * We replace our q_qinfo with one that has the qi_rwp entry point.
1328 	 * Clear SR_SIGALLDATA because we generate the equivalent signal(s)
1329 	 * for every enqueued data in tcp_fuse_output().
1330 	 */
1331 	rq->q_qinfo = &tcp_loopback_rinit;
1332 	rq->q_struiot = tcp_loopback_rinit.qi_struiot;
1333 	stp->sd_struiordq = rq;
1334 	stp->sd_rput_opt &= ~SR_SIGALLDATA;
1335 
1336 	mutex_exit(QLOCK(rq));
1337 	mutex_exit(&stp->sd_lock);
1338 }
1339 
1340 /*
1341  * Disable synchronous streams on a fused tcp loopback endpoint.
1342  */
1343 static void
1344 tcp_fuse_syncstr_disable(tcp_t *tcp)
1345 {
1346 	queue_t *rq = tcp->tcp_rq;
1347 	struct stdata *stp = STREAM(rq);
1348 
1349 	if (!tcp->tcp_direct_sockfs)
1350 		return;
1351 
1352 	mutex_enter(&stp->sd_lock);
1353 	mutex_enter(QLOCK(rq));
1354 
1355 	/*
1356 	 * Reset q_qinfo to point to the default tcp entry points.
1357 	 * Also restore SR_SIGALLDATA so that strrput() can generate
1358 	 * the signals again for future M_DATA messages.
1359 	 */
1360 	rq->q_qinfo = &tcp_rinitv4;	/* No open - same as rinitv6 */
1361 	rq->q_struiot = tcp_rinitv4.qi_struiot;
1362 	stp->sd_struiordq = NULL;
1363 	stp->sd_rput_opt |= SR_SIGALLDATA;
1364 	tcp->tcp_direct_sockfs = B_FALSE;
1365 
1366 	mutex_exit(QLOCK(rq));
1367 	mutex_exit(&stp->sd_lock);
1368 }
1369 
1370 /*
1371  * Enable synchronous streams on a pair of fused tcp endpoints.
1372  */
1373 void
1374 tcp_fuse_syncstr_enable_pair(tcp_t *tcp)
1375 {
1376 	tcp_t *peer_tcp = tcp->tcp_loopback_peer;
1377 
1378 	ASSERT(tcp->tcp_fused);
1379 	ASSERT(peer_tcp != NULL);
1380 
1381 	tcp_fuse_syncstr_enable(tcp);
1382 	tcp_fuse_syncstr_enable(peer_tcp);
1383 }
1384 
1385 /*
1386  * Used to enable/disable signal generation at the stream head. We already
1387  * generated the signal(s) for these messages when they were enqueued on the
1388  * receiver. We also check if STREOF is set here. If it is, we return false
1389  * and let the caller decide what to do.
1390  */
1391 static boolean_t
1392 strrput_sig(queue_t *q, boolean_t on)
1393 {
1394 	struct stdata *stp = STREAM(q);
1395 
1396 	mutex_enter(&stp->sd_lock);
1397 	if (stp->sd_flag == STREOF) {
1398 		mutex_exit(&stp->sd_lock);
1399 		return (B_TRUE);
1400 	}
1401 	if (on)
1402 		stp->sd_flag &= ~STRGETINPROG;
1403 	else
1404 		stp->sd_flag |= STRGETINPROG;
1405 	mutex_exit(&stp->sd_lock);
1406 
1407 	return (B_FALSE);
1408 }
1409 
1410 /*
1411  * Disable synchronous streams on a pair of fused tcp endpoints and drain
1412  * any queued data; called either during unfuse or upon transitioning from
1413  * a socket to a stream endpoint due to _SIOCSOCKFALLBACK.
1414  */
1415 void
1416 tcp_fuse_disable_pair(tcp_t *tcp, boolean_t unfusing)
1417 {
1418 	tcp_t *peer_tcp = tcp->tcp_loopback_peer;
1419 	tcp_stack_t	*tcps = tcp->tcp_tcps;
1420 
1421 	ASSERT(tcp->tcp_fused);
1422 	ASSERT(peer_tcp != NULL);
1423 
1424 	/*
1425 	 * Force any tcp_fuse_rrw() calls to block until we've moved the data
1426 	 * onto the stream head.
1427 	 */
1428 	TCP_FUSE_SYNCSTR_PLUG_DRAIN(tcp);
1429 	TCP_FUSE_SYNCSTR_PLUG_DRAIN(peer_tcp);
1430 
1431 	/*
1432 	 * Cancel any pending push timers.
1433 	 */
1434 	if (tcp->tcp_push_tid != 0) {
1435 		(void) TCP_TIMER_CANCEL(tcp, tcp->tcp_push_tid);
1436 		tcp->tcp_push_tid = 0;
1437 	}
1438 	if (peer_tcp->tcp_push_tid != 0) {
1439 		(void) TCP_TIMER_CANCEL(peer_tcp, peer_tcp->tcp_push_tid);
1440 		peer_tcp->tcp_push_tid = 0;
1441 	}
1442 
1443 	/*
1444 	 * Drain any pending data; the detached check is needed because
1445 	 * we may be called as a result of a tcp_unfuse() triggered by
1446 	 * tcp_fuse_output().  Note that in case of a detached tcp, the
1447 	 * draining will happen later after the tcp is unfused.  For non-
1448 	 * urgent data, this can be handled by the regular tcp_rcv_drain().
1449 	 * If we have urgent data sitting in the receive list, we will
1450 	 * need to send up a SIGURG signal first before draining the data.
1451 	 * All of these will be handled by the code in tcp_fuse_rcv_drain()
1452 	 * when called from tcp_rcv_drain().
1453 	 */
1454 	if (!TCP_IS_DETACHED(tcp)) {
1455 		(void) tcp_fuse_rcv_drain(tcp->tcp_rq, tcp,
1456 		    (unfusing ? &tcp->tcp_fused_sigurg_mp : NULL));
1457 	}
1458 	if (!TCP_IS_DETACHED(peer_tcp)) {
1459 		(void) tcp_fuse_rcv_drain(peer_tcp->tcp_rq, peer_tcp,
1460 		    (unfusing ? &peer_tcp->tcp_fused_sigurg_mp : NULL));
1461 	}
1462 
1463 	/*
1464 	 * Make all current and future tcp_fuse_rrw() calls fail with EBUSY.
1465 	 * To ensure threads don't sneak past the checks in tcp_fuse_rrw(),
1466 	 * a given stream must be stopped prior to being unplugged (but the
1467 	 * ordering of operations between the streams is unimportant).
1468 	 */
1469 	TCP_FUSE_SYNCSTR_STOP(tcp);
1470 	TCP_FUSE_SYNCSTR_STOP(peer_tcp);
1471 	TCP_FUSE_SYNCSTR_UNPLUG_DRAIN(tcp);
1472 	TCP_FUSE_SYNCSTR_UNPLUG_DRAIN(peer_tcp);
1473 
1474 	/* Lift up any flow-control conditions */
1475 	if (tcp->tcp_flow_stopped) {
1476 		tcp_clrqfull(tcp);
1477 		TCP_STAT(tcps, tcp_fusion_backenabled);
1478 	}
1479 	if (peer_tcp->tcp_flow_stopped) {
1480 		tcp_clrqfull(peer_tcp);
1481 		TCP_STAT(tcps, tcp_fusion_backenabled);
1482 	}
1483 
1484 	/* Disable synchronous streams */
1485 	if (!IPCL_IS_NONSTR(tcp->tcp_connp))
1486 		tcp_fuse_syncstr_disable(tcp);
1487 	if (!IPCL_IS_NONSTR(peer_tcp->tcp_connp))
1488 		tcp_fuse_syncstr_disable(peer_tcp);
1489 }
1490 
1491 /*
1492  * Calculate the size of receive buffer for a fused tcp endpoint.
1493  */
1494 size_t
1495 tcp_fuse_set_rcv_hiwat(tcp_t *tcp, size_t rwnd)
1496 {
1497 	tcp_stack_t	*tcps = tcp->tcp_tcps;
1498 
1499 	ASSERT(tcp->tcp_fused);
1500 
1501 	/* Ensure that value is within the maximum upper bound */
1502 	if (rwnd > tcps->tcps_max_buf)
1503 		rwnd = tcps->tcps_max_buf;
1504 
1505 	/* Obey the absolute minimum tcp receive high water mark */
1506 	if (rwnd < tcps->tcps_sth_rcv_hiwat)
1507 		rwnd = tcps->tcps_sth_rcv_hiwat;
1508 
1509 	/*
1510 	 * Round up to system page size in case SO_RCVBUF is modified
1511 	 * after SO_SNDBUF; the latter is also similarly rounded up.
1512 	 */
1513 	rwnd = P2ROUNDUP_TYPED(rwnd, PAGESIZE, size_t);
1514 	tcp->tcp_fuse_rcv_hiwater = rwnd;
1515 	return (rwnd);
1516 }
1517 
1518 /*
1519  * Calculate the maximum outstanding unread data block for a fused tcp endpoint.
1520  */
1521 int
1522 tcp_fuse_maxpsz_set(tcp_t *tcp)
1523 {
1524 	tcp_t *peer_tcp = tcp->tcp_loopback_peer;
1525 	uint_t sndbuf = tcp->tcp_xmit_hiwater;
1526 	uint_t maxpsz = sndbuf;
1527 
1528 	ASSERT(tcp->tcp_fused);
1529 	ASSERT(peer_tcp != NULL);
1530 	ASSERT(peer_tcp->tcp_fuse_rcv_hiwater != 0);
1531 	/*
1532 	 * In the fused loopback case, we want the stream head to split
1533 	 * up larger writes into smaller chunks for a more accurate flow-
1534 	 * control accounting.  Our maxpsz is half of the sender's send
1535 	 * buffer or the receiver's receive buffer, whichever is smaller.
1536 	 * We round up the buffer to system page size due to the lack of
1537 	 * TCP MSS concept in Fusion.
1538 	 */
1539 	if (maxpsz > peer_tcp->tcp_fuse_rcv_hiwater)
1540 		maxpsz = peer_tcp->tcp_fuse_rcv_hiwater;
1541 	maxpsz = P2ROUNDUP_TYPED(maxpsz, PAGESIZE, uint_t) >> 1;
1542 
1543 	/*
1544 	 * Calculate the peer's limit for the number of outstanding unread
1545 	 * data block.  This is the amount of data blocks that are allowed
1546 	 * to reside in the receiver's queue before the sender gets flow
1547 	 * controlled.  It is used only in the synchronous streams mode as
1548 	 * a way to throttle the sender when it performs consecutive writes
1549 	 * faster than can be read.  The value is derived from SO_SNDBUF in
1550 	 * order to give the sender some control; we divide it with a large
1551 	 * value (16KB) to produce a fairly low initial limit.
1552 	 */
1553 	if (tcp_fusion_rcv_unread_min == 0) {
1554 		/* A value of 0 means that we disable the check */
1555 		peer_tcp->tcp_fuse_rcv_unread_hiwater = 0;
1556 	} else {
1557 		peer_tcp->tcp_fuse_rcv_unread_hiwater =
1558 		    MAX(sndbuf >> 14, tcp_fusion_rcv_unread_min);
1559 	}
1560 	return (maxpsz);
1561 }
1562