/*- * Copyright (c) 2014 Chelsio Communications, Inc. * All rights reserved. * Written by: Navdeep Parhar * * Redistribution and use in source and binary forms, with or without * modification, are permitted provided that the following conditions * are met: * 1. Redistributions of source code must retain the above copyright * notice, this list of conditions and the following disclaimer. * 2. Redistributions in binary form must reproduce the above copyright * notice, this list of conditions and the following disclaimer in the * documentation and/or other materials provided with the distribution. * * THIS SOFTWARE IS PROVIDED BY THE AUTHOR AND CONTRIBUTORS ``AS IS'' AND * ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE * ARE DISCLAIMED. IN NO EVENT SHALL THE AUTHOR OR CONTRIBUTORS BE LIABLE * FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL * DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS * OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) * HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT * LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY * OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF * SUCH DAMAGE. */ #include #include #include #include #include #include #include #include #include #include #include "t4_mp_ring.h" #if defined(__i386__) #define atomic_cmpset_acq_64 atomic_cmpset_64 #define atomic_cmpset_rel_64 atomic_cmpset_64 #endif /* * mp_ring handles multiple threads (producers) enqueueing data to a tx queue. * The thread that is writing the hardware descriptors is the consumer and it * runs with the consumer lock held. A producer becomes the consumer if there * isn't one already. The consumer runs with the flags sets to BUSY and * consumes everything (IDLE or COALESCING) or gets STALLED. If it is running * over its budget it sets flags to TOO_BUSY. A producer that observes a * TOO_BUSY consumer will become the new consumer by setting flags to * TAKING_OVER. The original consumer stops and sets the flags back to BUSY for * the new consumer. * * COALESCING is the same as IDLE except there are items being held in the hope * that they can be coalesced with items that follow. The driver must arrange * for a tx update or some other event that transmits all the held items in a * timely manner if nothing else is enqueued. */ union ring_state { struct { uint16_t pidx_head; uint16_t pidx_tail; uint16_t cidx; uint16_t flags; }; uint64_t state; }; enum { IDLE = 0, /* tx is all caught up, nothing to do. */ COALESCING, /* IDLE, but tx frames are being held for coalescing */ BUSY, /* consumer is running already, or will be shortly. */ TOO_BUSY, /* consumer is running and is beyond its budget */ TAKING_OVER, /* new consumer taking over from a TOO_BUSY consumer */ STALLED, /* consumer stopped due to lack of resources. */ }; enum { C_FAST = 0, C_2, C_3, C_TAKEOVER, }; static inline uint16_t space_available(struct mp_ring *r, union ring_state s) { uint16_t x = r->size - 1; if (s.cidx == s.pidx_head) return (x); else if (s.cidx > s.pidx_head) return (s.cidx - s.pidx_head - 1); else return (x - s.pidx_head + s.cidx); } static inline uint16_t increment_idx(struct mp_ring *r, uint16_t idx, uint16_t n) { int x = r->size - idx; MPASS(x > 0); return (x > n ? idx + n : n - x); } /* * Consumer. Called with the consumer lock held and a guarantee that there is * work to do. */ static void drain_ring(struct mp_ring *r, int budget) { union ring_state os, ns; int n, pending, total; uint16_t cidx; uint16_t pidx; bool coalescing; mtx_assert(r->cons_lock, MA_OWNED); os.state = atomic_load_acq_64(&r->state); MPASS(os.flags == BUSY); cidx = os.cidx; pidx = os.pidx_tail; MPASS(cidx != pidx); pending = 0; total = 0; while (cidx != pidx) { /* Items from cidx to pidx are available for consumption. */ n = r->drain(r, cidx, pidx, &coalescing); if (n == 0) { critical_enter(); os.state = atomic_load_64(&r->state); do { ns.state = os.state; ns.cidx = cidx; MPASS(os.flags == BUSY || os.flags == TOO_BUSY || os.flags == TAKING_OVER); if (os.flags == TAKING_OVER) ns.flags = BUSY; else ns.flags = STALLED; } while (atomic_fcmpset_64(&r->state, &os.state, ns.state) == 0); critical_exit(); if (os.flags == TAKING_OVER) counter_u64_add(r->abdications, 1); else if (ns.flags == STALLED) counter_u64_add(r->stalls, 1); break; } cidx = increment_idx(r, cidx, n); pending += n; total += n; counter_u64_add(r->consumed, n); os.state = atomic_load_64(&r->state); do { MPASS(os.flags == BUSY || os.flags == TOO_BUSY || os.flags == TAKING_OVER); ns.state = os.state; ns.cidx = cidx; if (__predict_false(os.flags == TAKING_OVER)) { MPASS(total >= budget); ns.flags = BUSY; continue; } if (cidx == os.pidx_tail) { ns.flags = coalescing ? COALESCING : IDLE; continue; } if (total >= budget) { ns.flags = TOO_BUSY; continue; } MPASS(os.flags == BUSY); if (pending < 32) break; } while (atomic_fcmpset_acq_64(&r->state, &os.state, ns.state) == 0); if (__predict_false(os.flags == TAKING_OVER)) { MPASS(ns.flags == BUSY); counter_u64_add(r->abdications, 1); break; } if (ns.flags == IDLE || ns.flags == COALESCING) { MPASS(ns.pidx_tail == cidx); if (ns.pidx_head != ns.pidx_tail) counter_u64_add(r->cons_idle2, 1); else counter_u64_add(r->cons_idle, 1); break; } /* * The acquire style atomic above guarantees visibility of items * associated with any pidx change that we notice here. */ pidx = ns.pidx_tail; pending = 0; } #ifdef INVARIANTS if (os.flags == TAKING_OVER) MPASS(ns.flags == BUSY); else { MPASS(ns.flags == IDLE || ns.flags == COALESCING || ns.flags == STALLED); } #endif } static void drain_txpkts(struct mp_ring *r, union ring_state os, int budget) { union ring_state ns; uint16_t cidx = os.cidx; uint16_t pidx = os.pidx_tail; bool coalescing; mtx_assert(r->cons_lock, MA_OWNED); MPASS(os.flags == BUSY); MPASS(cidx == pidx); r->drain(r, cidx, pidx, &coalescing); MPASS(coalescing == false); critical_enter(); os.state = atomic_load_64(&r->state); do { ns.state = os.state; MPASS(os.flags == BUSY); MPASS(os.cidx == cidx); if (ns.cidx == ns.pidx_tail) ns.flags = IDLE; else ns.flags = BUSY; } while (atomic_fcmpset_acq_64(&r->state, &os.state, ns.state) == 0); critical_exit(); if (ns.flags == BUSY) drain_ring(r, budget); } int mp_ring_alloc(struct mp_ring **pr, int size, void *cookie, ring_drain_t drain, ring_can_drain_t can_drain, struct malloc_type *mt, struct mtx *lck, int flags) { struct mp_ring *r; int i; /* All idx are 16b so size can be 65536 at most */ if (pr == NULL || size < 2 || size > 65536 || drain == NULL || can_drain == NULL) return (EINVAL); *pr = NULL; flags &= M_NOWAIT | M_WAITOK; MPASS(flags != 0); r = malloc(__offsetof(struct mp_ring, items[size]), mt, flags | M_ZERO); if (r == NULL) return (ENOMEM); r->size = size; r->cookie = cookie; r->mt = mt; r->drain = drain; r->can_drain = can_drain; r->cons_lock = lck; if ((r->dropped = counter_u64_alloc(flags)) == NULL) goto failed; for (i = 0; i < nitems(r->consumer); i++) { if ((r->consumer[i] = counter_u64_alloc(flags)) == NULL) goto failed; } if ((r->not_consumer = counter_u64_alloc(flags)) == NULL) goto failed; if ((r->abdications = counter_u64_alloc(flags)) == NULL) goto failed; if ((r->stalls = counter_u64_alloc(flags)) == NULL) goto failed; if ((r->consumed = counter_u64_alloc(flags)) == NULL) goto failed; if ((r->cons_idle = counter_u64_alloc(flags)) == NULL) goto failed; if ((r->cons_idle2 = counter_u64_alloc(flags)) == NULL) goto failed; *pr = r; return (0); failed: mp_ring_free(r); return (ENOMEM); } void mp_ring_free(struct mp_ring *r) { int i; if (r == NULL) return; if (r->dropped != NULL) counter_u64_free(r->dropped); for (i = 0; i < nitems(r->consumer); i++) { if (r->consumer[i] != NULL) counter_u64_free(r->consumer[i]); } if (r->not_consumer != NULL) counter_u64_free(r->not_consumer); if (r->abdications != NULL) counter_u64_free(r->abdications); if (r->stalls != NULL) counter_u64_free(r->stalls); if (r->consumed != NULL) counter_u64_free(r->consumed); if (r->cons_idle != NULL) counter_u64_free(r->cons_idle); if (r->cons_idle2 != NULL) counter_u64_free(r->cons_idle2); free(r, r->mt); } /* * Enqueue n items and maybe drain the ring for some time. * * Returns an errno. */ int mp_ring_enqueue(struct mp_ring *r, void **items, int n, int budget) { union ring_state os, ns; uint16_t pidx_start, pidx_stop; int i, nospc, cons; bool consumer; MPASS(items != NULL); MPASS(n > 0); /* * Reserve room for the new items. Our reservation, if successful, is * from 'pidx_start' to 'pidx_stop'. */ nospc = 0; os.state = atomic_load_64(&r->state); for (;;) { for (;;) { if (__predict_true(space_available(r, os) >= n)) break; /* Not enough room in the ring. */ MPASS(os.flags != IDLE); MPASS(os.flags != COALESCING); if (__predict_false(++nospc > 100)) { counter_u64_add(r->dropped, n); return (ENOBUFS); } if (os.flags == STALLED) mp_ring_check_drainage(r, 64); else cpu_spinwait(); os.state = atomic_load_64(&r->state); } /* There is room in the ring. */ cons = -1; ns.state = os.state; ns.pidx_head = increment_idx(r, os.pidx_head, n); if (os.flags == IDLE || os.flags == COALESCING) { MPASS(os.pidx_tail == os.cidx); if (os.pidx_head == os.pidx_tail) { cons = C_FAST; ns.pidx_tail = increment_idx(r, os.pidx_tail, n); } else cons = C_2; ns.flags = BUSY; } else if (os.flags == TOO_BUSY) { cons = C_TAKEOVER; ns.flags = TAKING_OVER; } critical_enter(); if (atomic_fcmpset_64(&r->state, &os.state, ns.state)) break; critical_exit(); cpu_spinwait(); }; pidx_start = os.pidx_head; pidx_stop = ns.pidx_head; if (cons == C_FAST) { i = pidx_start; do { r->items[i] = *items++; if (__predict_false(++i == r->size)) i = 0; } while (i != pidx_stop); critical_exit(); counter_u64_add(r->consumer[C_FAST], 1); mtx_lock(r->cons_lock); drain_ring(r, budget); mtx_unlock(r->cons_lock); return (0); } /* * Wait for other producers who got in ahead of us to enqueue their * items, one producer at a time. It is our turn when the ring's * pidx_tail reaches the beginning of our reservation (pidx_start). */ while (ns.pidx_tail != pidx_start) { cpu_spinwait(); ns.state = atomic_load_64(&r->state); } /* Now it is our turn to fill up the area we reserved earlier. */ i = pidx_start; do { r->items[i] = *items++; if (__predict_false(++i == r->size)) i = 0; } while (i != pidx_stop); /* * Update the ring's pidx_tail. The release style atomic guarantees * that the items are visible to any thread that sees the updated pidx. */ os.state = atomic_load_64(&r->state); do { consumer = false; ns.state = os.state; ns.pidx_tail = pidx_stop; if (os.flags == IDLE || os.flags == COALESCING || (os.flags == STALLED && r->can_drain(r))) { MPASS(cons == -1); consumer = true; ns.flags = BUSY; } } while (atomic_fcmpset_rel_64(&r->state, &os.state, ns.state) == 0); critical_exit(); if (cons == -1) { if (consumer) cons = C_3; else { counter_u64_add(r->not_consumer, 1); return (0); } } MPASS(cons > C_FAST && cons < nitems(r->consumer)); counter_u64_add(r->consumer[cons], 1); mtx_lock(r->cons_lock); drain_ring(r, budget); mtx_unlock(r->cons_lock); return (0); } void mp_ring_check_drainage(struct mp_ring *r, int budget) { union ring_state os, ns; os.state = atomic_load_64(&r->state); if (os.flags == STALLED && r->can_drain(r)) { MPASS(os.cidx != os.pidx_tail); /* implied by STALLED */ ns.state = os.state; ns.flags = BUSY; if (atomic_cmpset_acq_64(&r->state, os.state, ns.state)) { mtx_lock(r->cons_lock); drain_ring(r, budget); mtx_unlock(r->cons_lock); } } else if (os.flags == COALESCING) { MPASS(os.cidx == os.pidx_tail); ns.state = os.state; ns.flags = BUSY; if (atomic_cmpset_acq_64(&r->state, os.state, ns.state)) { mtx_lock(r->cons_lock); drain_txpkts(r, ns, budget); mtx_unlock(r->cons_lock); } } } void mp_ring_reset_stats(struct mp_ring *r) { int i; counter_u64_zero(r->dropped); for (i = 0; i < nitems(r->consumer); i++) counter_u64_zero(r->consumer[i]); counter_u64_zero(r->not_consumer); counter_u64_zero(r->abdications); counter_u64_zero(r->stalls); counter_u64_zero(r->consumed); counter_u64_zero(r->cons_idle); counter_u64_zero(r->cons_idle2); } bool mp_ring_is_idle(struct mp_ring *r) { union ring_state s; s.state = atomic_load_64(&r->state); if (s.pidx_head == s.pidx_tail && s.pidx_tail == s.cidx && s.flags == IDLE) return (true); return (false); } void mp_ring_sysctls(struct mp_ring *r, struct sysctl_ctx_list *ctx, struct sysctl_oid_list *children) { struct sysctl_oid *oid; oid = SYSCTL_ADD_NODE(ctx, children, OID_AUTO, "mp_ring", CTLFLAG_RD | CTLFLAG_MPSAFE, NULL, "mp_ring statistics"); children = SYSCTL_CHILDREN(oid); SYSCTL_ADD_U64(ctx, children, OID_AUTO, "state", CTLFLAG_RD, __DEVOLATILE(uint64_t *, &r->state), 0, "ring state"); SYSCTL_ADD_COUNTER_U64(ctx, children, OID_AUTO, "dropped", CTLFLAG_RD, &r->dropped, "# of items dropped"); SYSCTL_ADD_COUNTER_U64(ctx, children, OID_AUTO, "consumed", CTLFLAG_RD, &r->consumed, "# of items consumed"); SYSCTL_ADD_COUNTER_U64(ctx, children, OID_AUTO, "fast_consumer", CTLFLAG_RD, &r->consumer[C_FAST], "# of times producer became consumer (fast)"); SYSCTL_ADD_COUNTER_U64(ctx, children, OID_AUTO, "consumer2", CTLFLAG_RD, &r->consumer[C_2], "# of times producer became consumer (2)"); SYSCTL_ADD_COUNTER_U64(ctx, children, OID_AUTO, "consumer3", CTLFLAG_RD, &r->consumer[C_3], "# of times producer became consumer (3)"); SYSCTL_ADD_COUNTER_U64(ctx, children, OID_AUTO, "takeovers", CTLFLAG_RD, &r->consumer[C_TAKEOVER], "# of times producer took over from another consumer."); SYSCTL_ADD_COUNTER_U64(ctx, children, OID_AUTO, "not_consumer", CTLFLAG_RD, &r->not_consumer, "# of times producer did not become consumer"); SYSCTL_ADD_COUNTER_U64(ctx, children, OID_AUTO, "abdications", CTLFLAG_RD, &r->abdications, "# of consumer abdications"); SYSCTL_ADD_COUNTER_U64(ctx, children, OID_AUTO, "stalls", CTLFLAG_RD, &r->stalls, "# of consumer stalls"); SYSCTL_ADD_COUNTER_U64(ctx, children, OID_AUTO, "cons_idle", CTLFLAG_RD, &r->cons_idle, "# of times consumer ran fully to completion"); SYSCTL_ADD_COUNTER_U64(ctx, children, OID_AUTO, "cons_idle2", CTLFLAG_RD, &r->cons_idle2, "# of times consumer idled when another enqueue was in progress"); }