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