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