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