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