1 /*- 2 * Copyright (c) 2014 Chelsio Communications, Inc. 3 * All rights reserved. 4 * Written by: Navdeep Parhar <np@FreeBSD.org> 5 * 6 * Redistribution and use in source and binary forms, with or without 7 * modification, are permitted provided that the following conditions 8 * are met: 9 * 1. Redistributions of source code must retain the above copyright 10 * notice, this list of conditions and the following disclaimer. 11 * 2. Redistributions in binary form must reproduce the above copyright 12 * notice, this list of conditions and the following disclaimer in the 13 * documentation and/or other materials provided with the distribution. 14 * 15 * THIS SOFTWARE IS PROVIDED BY THE AUTHOR AND CONTRIBUTORS ``AS IS'' AND 16 * ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE 17 * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE 18 * ARE DISCLAIMED. IN NO EVENT SHALL THE AUTHOR OR CONTRIBUTORS BE LIABLE 19 * FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL 20 * DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS 21 * OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) 22 * HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT 23 * LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY 24 * OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF 25 * SUCH DAMAGE. 26 */ 27 28 #include <sys/cdefs.h> 29 __FBSDID("$FreeBSD$"); 30 31 #include <sys/types.h> 32 #include <sys/param.h> 33 #include <sys/systm.h> 34 #include <sys/counter.h> 35 #include <sys/lock.h> 36 #include <sys/mutex.h> 37 #include <sys/malloc.h> 38 #include <machine/cpu.h> 39 40 #if defined(__powerpc__) || defined(__mips__) 41 #define NO_64BIT_ATOMICS 42 #endif 43 44 #if defined(__i386__) 45 #define atomic_cmpset_acq_64 atomic_cmpset_64 46 #define atomic_cmpset_rel_64 atomic_cmpset_64 47 #endif 48 49 #include <net/mp_ring.h> 50 51 union ring_state { 52 struct { 53 uint16_t pidx_head; 54 uint16_t pidx_tail; 55 uint16_t cidx; 56 uint16_t flags; 57 }; 58 uint64_t state; 59 }; 60 61 enum { 62 IDLE = 0, /* consumer ran to completion, nothing more to do. */ 63 BUSY, /* consumer is running already, or will be shortly. */ 64 STALLED, /* consumer stopped due to lack of resources. */ 65 ABDICATED, /* consumer stopped even though there was work to be 66 done because it wants another thread to take over. */ 67 }; 68 69 static inline uint16_t 70 space_available(struct ifmp_ring *r, union ring_state s) 71 { 72 uint16_t x = r->size - 1; 73 74 if (s.cidx == s.pidx_head) 75 return (x); 76 else if (s.cidx > s.pidx_head) 77 return (s.cidx - s.pidx_head - 1); 78 else 79 return (x - s.pidx_head + s.cidx); 80 } 81 82 static inline uint16_t 83 increment_idx(struct ifmp_ring *r, uint16_t idx, uint16_t n) 84 { 85 int x = r->size - idx; 86 87 MPASS(x > 0); 88 return (x > n ? idx + n : n - x); 89 } 90 91 /* Consumer is about to update the ring's state to s */ 92 static inline uint16_t 93 state_to_flags(union ring_state s, int abdicate) 94 { 95 96 if (s.cidx == s.pidx_tail) 97 return (IDLE); 98 else if (abdicate && s.pidx_tail != s.pidx_head) 99 return (ABDICATED); 100 101 return (BUSY); 102 } 103 104 #ifdef NO_64BIT_ATOMICS 105 static void 106 drain_ring_locked(struct ifmp_ring *r, union ring_state os, uint16_t prev, int budget) 107 { 108 union ring_state ns; 109 int n, pending, total; 110 uint16_t cidx = os.cidx; 111 uint16_t pidx = os.pidx_tail; 112 113 MPASS(os.flags == BUSY); 114 MPASS(cidx != pidx); 115 116 if (prev == IDLE) 117 counter_u64_add(r->starts, 1); 118 pending = 0; 119 total = 0; 120 121 while (cidx != pidx) { 122 123 /* Items from cidx to pidx are available for consumption. */ 124 n = r->drain(r, cidx, pidx); 125 if (n == 0) { 126 os.state = ns.state = r->state; 127 ns.cidx = cidx; 128 ns.flags = STALLED; 129 r->state = ns.state; 130 if (prev != STALLED) 131 counter_u64_add(r->stalls, 1); 132 else if (total > 0) { 133 counter_u64_add(r->restarts, 1); 134 counter_u64_add(r->stalls, 1); 135 } 136 break; 137 } 138 cidx = increment_idx(r, cidx, n); 139 pending += n; 140 total += n; 141 142 /* 143 * We update the cidx only if we've caught up with the pidx, the 144 * real cidx is getting too far ahead of the one visible to 145 * everyone else, or we have exceeded our budget. 146 */ 147 if (cidx != pidx && pending < 64 && total < budget) 148 continue; 149 150 os.state = ns.state = r->state; 151 ns.cidx = cidx; 152 ns.flags = state_to_flags(ns, total >= budget); 153 r->state = ns.state; 154 155 if (ns.flags == ABDICATED) 156 counter_u64_add(r->abdications, 1); 157 if (ns.flags != BUSY) { 158 /* Wrong loop exit if we're going to stall. */ 159 MPASS(ns.flags != STALLED); 160 if (prev == STALLED) { 161 MPASS(total > 0); 162 counter_u64_add(r->restarts, 1); 163 } 164 break; 165 } 166 167 /* 168 * The acquire style atomic above guarantees visibility of items 169 * associated with any pidx change that we notice here. 170 */ 171 pidx = ns.pidx_tail; 172 pending = 0; 173 } 174 } 175 #else 176 /* 177 * Caller passes in a state, with a guarantee that there is work to do and that 178 * all items up to the pidx_tail in the state are visible. 179 */ 180 static void 181 drain_ring_lockless(struct ifmp_ring *r, union ring_state os, uint16_t prev, int budget) 182 { 183 union ring_state ns; 184 int n, pending, total; 185 uint16_t cidx = os.cidx; 186 uint16_t pidx = os.pidx_tail; 187 188 MPASS(os.flags == BUSY); 189 MPASS(cidx != pidx); 190 191 if (prev == IDLE) 192 counter_u64_add(r->starts, 1); 193 pending = 0; 194 total = 0; 195 196 while (cidx != pidx) { 197 198 /* Items from cidx to pidx are available for consumption. */ 199 n = r->drain(r, cidx, pidx); 200 if (n == 0) { 201 critical_enter(); 202 do { 203 os.state = ns.state = r->state; 204 ns.cidx = cidx; 205 ns.flags = STALLED; 206 } while (atomic_cmpset_64(&r->state, os.state, 207 ns.state) == 0); 208 critical_exit(); 209 if (prev != STALLED) 210 counter_u64_add(r->stalls, 1); 211 else if (total > 0) { 212 counter_u64_add(r->restarts, 1); 213 counter_u64_add(r->stalls, 1); 214 } 215 break; 216 } 217 cidx = increment_idx(r, cidx, n); 218 pending += n; 219 total += n; 220 221 /* 222 * We update the cidx only if we've caught up with the pidx, the 223 * real cidx is getting too far ahead of the one visible to 224 * everyone else, or we have exceeded our budget. 225 */ 226 if (cidx != pidx && pending < 64 && total < budget) 227 continue; 228 critical_enter(); 229 do { 230 os.state = ns.state = r->state; 231 ns.cidx = cidx; 232 ns.flags = state_to_flags(ns, total >= budget); 233 } while (atomic_cmpset_acq_64(&r->state, os.state, ns.state) == 0); 234 critical_exit(); 235 236 if (ns.flags == ABDICATED) 237 counter_u64_add(r->abdications, 1); 238 if (ns.flags != BUSY) { 239 /* Wrong loop exit if we're going to stall. */ 240 MPASS(ns.flags != STALLED); 241 if (prev == STALLED) { 242 MPASS(total > 0); 243 counter_u64_add(r->restarts, 1); 244 } 245 break; 246 } 247 248 /* 249 * The acquire style atomic above guarantees visibility of items 250 * associated with any pidx change that we notice here. 251 */ 252 pidx = ns.pidx_tail; 253 pending = 0; 254 } 255 } 256 #endif 257 258 int 259 ifmp_ring_alloc(struct ifmp_ring **pr, int size, void *cookie, mp_ring_drain_t drain, 260 mp_ring_can_drain_t can_drain, struct malloc_type *mt, int flags) 261 { 262 struct ifmp_ring *r; 263 264 /* All idx are 16b so size can be 65536 at most */ 265 if (pr == NULL || size < 2 || size > 65536 || drain == NULL || 266 can_drain == NULL) 267 return (EINVAL); 268 *pr = NULL; 269 flags &= M_NOWAIT | M_WAITOK; 270 MPASS(flags != 0); 271 272 r = malloc(__offsetof(struct ifmp_ring, items[size]), mt, flags | M_ZERO); 273 if (r == NULL) 274 return (ENOMEM); 275 r->size = size; 276 r->cookie = cookie; 277 r->mt = mt; 278 r->drain = drain; 279 r->can_drain = can_drain; 280 r->enqueues = counter_u64_alloc(flags); 281 r->drops = counter_u64_alloc(flags); 282 r->starts = counter_u64_alloc(flags); 283 r->stalls = counter_u64_alloc(flags); 284 r->restarts = counter_u64_alloc(flags); 285 r->abdications = counter_u64_alloc(flags); 286 if (r->enqueues == NULL || r->drops == NULL || r->starts == NULL || 287 r->stalls == NULL || r->restarts == NULL || 288 r->abdications == NULL) { 289 ifmp_ring_free(r); 290 return (ENOMEM); 291 } 292 293 *pr = r; 294 #ifdef NO_64BIT_ATOMICS 295 mtx_init(&r->lock, "mp_ring lock", NULL, MTX_DEF); 296 #endif 297 return (0); 298 } 299 300 void 301 ifmp_ring_free(struct ifmp_ring *r) 302 { 303 304 if (r == NULL) 305 return; 306 307 if (r->enqueues != NULL) 308 counter_u64_free(r->enqueues); 309 if (r->drops != NULL) 310 counter_u64_free(r->drops); 311 if (r->starts != NULL) 312 counter_u64_free(r->starts); 313 if (r->stalls != NULL) 314 counter_u64_free(r->stalls); 315 if (r->restarts != NULL) 316 counter_u64_free(r->restarts); 317 if (r->abdications != NULL) 318 counter_u64_free(r->abdications); 319 320 free(r, r->mt); 321 } 322 323 /* 324 * Enqueue n items and maybe drain the ring for some time. 325 * 326 * Returns an errno. 327 */ 328 #ifdef NO_64BIT_ATOMICS 329 int 330 ifmp_ring_enqueue(struct ifmp_ring *r, void **items, int n, int budget) 331 { 332 union ring_state os, ns; 333 uint16_t pidx_start, pidx_stop; 334 int i; 335 336 MPASS(items != NULL); 337 MPASS(n > 0); 338 339 mtx_lock(&r->lock); 340 /* 341 * Reserve room for the new items. Our reservation, if successful, is 342 * from 'pidx_start' to 'pidx_stop'. 343 */ 344 os.state = r->state; 345 if (n >= space_available(r, os)) { 346 counter_u64_add(r->drops, n); 347 MPASS(os.flags != IDLE); 348 if (os.flags == STALLED) 349 ifmp_ring_check_drainage(r, 0); 350 return (ENOBUFS); 351 } 352 ns.state = os.state; 353 ns.pidx_head = increment_idx(r, os.pidx_head, n); 354 r->state = ns.state; 355 pidx_start = os.pidx_head; 356 pidx_stop = ns.pidx_head; 357 358 /* 359 * Wait for other producers who got in ahead of us to enqueue their 360 * items, one producer at a time. It is our turn when the ring's 361 * pidx_tail reaches the beginning of our reservation (pidx_start). 362 */ 363 while (ns.pidx_tail != pidx_start) { 364 cpu_spinwait(); 365 ns.state = r->state; 366 } 367 368 /* Now it is our turn to fill up the area we reserved earlier. */ 369 i = pidx_start; 370 do { 371 r->items[i] = *items++; 372 if (__predict_false(++i == r->size)) 373 i = 0; 374 } while (i != pidx_stop); 375 376 /* 377 * Update the ring's pidx_tail. The release style atomic guarantees 378 * that the items are visible to any thread that sees the updated pidx. 379 */ 380 os.state = ns.state = r->state; 381 ns.pidx_tail = pidx_stop; 382 ns.flags = BUSY; 383 r->state = ns.state; 384 counter_u64_add(r->enqueues, n); 385 386 /* 387 * Turn into a consumer if some other thread isn't active as a consumer 388 * already. 389 */ 390 if (os.flags != BUSY) 391 drain_ring_locked(r, ns, os.flags, budget); 392 393 mtx_unlock(&r->lock); 394 return (0); 395 } 396 397 #else 398 int 399 ifmp_ring_enqueue(struct ifmp_ring *r, void **items, int n, int budget) 400 { 401 union ring_state os, ns; 402 uint16_t pidx_start, pidx_stop; 403 int i; 404 405 MPASS(items != NULL); 406 MPASS(n > 0); 407 408 /* 409 * Reserve room for the new items. Our reservation, if successful, is 410 * from 'pidx_start' to 'pidx_stop'. 411 */ 412 for (;;) { 413 os.state = r->state; 414 if (n >= space_available(r, os)) { 415 counter_u64_add(r->drops, n); 416 MPASS(os.flags != IDLE); 417 if (os.flags == STALLED) 418 ifmp_ring_check_drainage(r, 0); 419 return (ENOBUFS); 420 } 421 ns.state = os.state; 422 ns.pidx_head = increment_idx(r, os.pidx_head, n); 423 critical_enter(); 424 if (atomic_cmpset_64(&r->state, os.state, ns.state)) 425 break; 426 critical_exit(); 427 cpu_spinwait(); 428 } 429 pidx_start = os.pidx_head; 430 pidx_stop = ns.pidx_head; 431 432 /* 433 * Wait for other producers who got in ahead of us to enqueue their 434 * items, one producer at a time. It is our turn when the ring's 435 * pidx_tail reaches the beginning of our reservation (pidx_start). 436 */ 437 while (ns.pidx_tail != pidx_start) { 438 cpu_spinwait(); 439 ns.state = r->state; 440 } 441 442 /* Now it is our turn to fill up the area we reserved earlier. */ 443 i = pidx_start; 444 do { 445 r->items[i] = *items++; 446 if (__predict_false(++i == r->size)) 447 i = 0; 448 } while (i != pidx_stop); 449 450 /* 451 * Update the ring's pidx_tail. The release style atomic guarantees 452 * that the items are visible to any thread that sees the updated pidx. 453 */ 454 do { 455 os.state = ns.state = r->state; 456 ns.pidx_tail = pidx_stop; 457 if (os.flags == IDLE) 458 ns.flags = ABDICATED; 459 } while (atomic_cmpset_rel_64(&r->state, os.state, ns.state) == 0); 460 critical_exit(); 461 counter_u64_add(r->enqueues, n); 462 463 return (0); 464 } 465 #endif 466 467 void 468 ifmp_ring_check_drainage(struct ifmp_ring *r, int budget) 469 { 470 union ring_state os, ns; 471 472 os.state = r->state; 473 if ((os.flags != STALLED && os.flags != ABDICATED) || // Only continue in STALLED and ABDICATED 474 os.pidx_head != os.pidx_tail || // Require work to be available 475 (os.flags != ABDICATED && r->can_drain(r) == 0)) // Can either drain, or everyone left 476 return; 477 478 MPASS(os.cidx != os.pidx_tail); /* implied by STALLED */ 479 ns.state = os.state; 480 ns.flags = BUSY; 481 482 483 #ifdef NO_64BIT_ATOMICS 484 mtx_lock(&r->lock); 485 if (r->state != os.state) { 486 mtx_unlock(&r->lock); 487 return; 488 } 489 r->state = ns.state; 490 drain_ring_locked(r, ns, os.flags, budget); 491 mtx_unlock(&r->lock); 492 #else 493 /* 494 * The acquire style atomic guarantees visibility of items associated 495 * with the pidx that we read here. 496 */ 497 if (!atomic_cmpset_acq_64(&r->state, os.state, ns.state)) 498 return; 499 500 501 drain_ring_lockless(r, ns, os.flags, budget); 502 #endif 503 } 504 505 void 506 ifmp_ring_reset_stats(struct ifmp_ring *r) 507 { 508 509 counter_u64_zero(r->enqueues); 510 counter_u64_zero(r->drops); 511 counter_u64_zero(r->starts); 512 counter_u64_zero(r->stalls); 513 counter_u64_zero(r->restarts); 514 counter_u64_zero(r->abdications); 515 } 516 517 int 518 ifmp_ring_is_idle(struct ifmp_ring *r) 519 { 520 union ring_state s; 521 522 s.state = r->state; 523 if (s.pidx_head == s.pidx_tail && s.pidx_tail == s.cidx && 524 s.flags == IDLE) 525 return (1); 526 527 return (0); 528 } 529 530 int 531 ifmp_ring_is_stalled(struct ifmp_ring *r) 532 { 533 union ring_state s; 534 535 s.state = r->state; 536 if (s.pidx_head == s.pidx_tail && s.flags == STALLED) 537 return (1); 538 539 return (0); 540 } 541