X-Git-Url: http://git.droids-corp.org/?a=blobdiff_plain;f=lib%2Flibrte_ring%2Frte_ring.h;h=97f025a1fb61968f5805483cf71c2d44e92053b0;hb=c3b047be7328c70e0542d40db15f847181208a22;hp=fa66638055a0c7022a24caf50955bcf4bb9c696a;hpb=0dfc98c507b15bc32671fa047504e2dd7107e94c;p=dpdk.git diff --git a/lib/librte_ring/rte_ring.h b/lib/librte_ring/rte_ring.h index fa66638055..97f025a1fb 100644 --- a/lib/librte_ring/rte_ring.h +++ b/lib/librte_ring/rte_ring.h @@ -147,7 +147,7 @@ struct rte_ring { * compatibility requirements, it could be changed to RTE_RING_NAMESIZE * next time the ABI changes */ - char name[RTE_MEMZONE_NAMESIZE]; /**< Name of the ring. */ + char name[RTE_MEMZONE_NAMESIZE] __rte_cache_aligned; /**< Name of the ring. */ int flags; /**< Flags supplied at creation. */ const struct rte_memzone *memzone; /**< Memzone, if any, containing the rte_ring */ @@ -159,10 +159,6 @@ struct rte_ring { /** Ring consumer status. */ struct rte_ring_headtail cons __rte_aligned(CONS_ALIGN); - - void *ring[] __rte_cache_aligned; /**< Memory space of ring starts here. - * not volatile so need to be careful - * about compiler re-ordering */ }; #define RING_F_SP_ENQ 0x0001 /**< The default enqueue is "single-producer". */ @@ -290,57 +286,80 @@ void rte_ring_dump(FILE *f, const struct rte_ring *r); /* the actual enqueue of pointers on the ring. * Placed here since identical code needed in both * single and multi producer enqueue functions */ -#define ENQUEUE_PTRS() do { \ +#define ENQUEUE_PTRS(r, ring_start, prod_head, obj_table, n, obj_type) do { \ unsigned int i; \ - const uint32_t size = r->size; \ - uint32_t idx = prod_head & r->mask; \ + const uint32_t size = (r)->size; \ + uint32_t idx = prod_head & (r)->mask; \ + obj_type *ring = (obj_type *)ring_start; \ if (likely(idx + n < size)) { \ for (i = 0; i < (n & ((~(unsigned)0x3))); i+=4, idx+=4) { \ - r->ring[idx] = obj_table[i]; \ - r->ring[idx+1] = obj_table[i+1]; \ - r->ring[idx+2] = obj_table[i+2]; \ - r->ring[idx+3] = obj_table[i+3]; \ + ring[idx] = obj_table[i]; \ + ring[idx+1] = obj_table[i+1]; \ + ring[idx+2] = obj_table[i+2]; \ + ring[idx+3] = obj_table[i+3]; \ } \ switch (n & 0x3) { \ - case 3: r->ring[idx++] = obj_table[i++]; \ - case 2: r->ring[idx++] = obj_table[i++]; \ - case 1: r->ring[idx++] = obj_table[i++]; \ + case 3: \ + ring[idx++] = obj_table[i++]; /* fallthrough */ \ + case 2: \ + ring[idx++] = obj_table[i++]; /* fallthrough */ \ + case 1: \ + ring[idx++] = obj_table[i++]; \ } \ } else { \ for (i = 0; idx < size; i++, idx++)\ - r->ring[idx] = obj_table[i]; \ + ring[idx] = obj_table[i]; \ for (idx = 0; i < n; i++, idx++) \ - r->ring[idx] = obj_table[i]; \ + ring[idx] = obj_table[i]; \ } \ -} while(0) +} while (0) /* the actual copy of pointers on the ring to obj_table. * Placed here since identical code needed in both * single and multi consumer dequeue functions */ -#define DEQUEUE_PTRS() do { \ +#define DEQUEUE_PTRS(r, ring_start, cons_head, obj_table, n, obj_type) do { \ unsigned int i; \ - uint32_t idx = cons_head & r->mask; \ - const uint32_t size = r->size; \ + uint32_t idx = cons_head & (r)->mask; \ + const uint32_t size = (r)->size; \ + obj_type *ring = (obj_type *)ring_start; \ if (likely(idx + n < size)) { \ for (i = 0; i < (n & (~(unsigned)0x3)); i+=4, idx+=4) {\ - obj_table[i] = r->ring[idx]; \ - obj_table[i+1] = r->ring[idx+1]; \ - obj_table[i+2] = r->ring[idx+2]; \ - obj_table[i+3] = r->ring[idx+3]; \ + obj_table[i] = ring[idx]; \ + obj_table[i+1] = ring[idx+1]; \ + obj_table[i+2] = ring[idx+2]; \ + obj_table[i+3] = ring[idx+3]; \ } \ switch (n & 0x3) { \ - case 3: obj_table[i++] = r->ring[idx++]; \ - case 2: obj_table[i++] = r->ring[idx++]; \ - case 1: obj_table[i++] = r->ring[idx++]; \ + case 3: \ + obj_table[i++] = ring[idx++]; /* fallthrough */ \ + case 2: \ + obj_table[i++] = ring[idx++]; /* fallthrough */ \ + case 1: \ + obj_table[i++] = ring[idx++]; \ } \ } else { \ for (i = 0; idx < size; i++, idx++) \ - obj_table[i] = r->ring[idx]; \ + obj_table[i] = ring[idx]; \ for (idx = 0; i < n; i++, idx++) \ - obj_table[i] = r->ring[idx]; \ + obj_table[i] = ring[idx]; \ } \ } while (0) +static inline __attribute__((always_inline)) void +update_tail(struct rte_ring_headtail *ht, uint32_t old_val, uint32_t new_val, + uint32_t single) +{ + /* + * If there are other enqueues/dequeues in progress that preceded us, + * we need to wait for them to complete + */ + if (!single) + while (unlikely(ht->tail != old_val)) + rte_pause(); + + ht->tail = new_val; +} + /** * @internal This function updates the producer head for enqueue * @@ -437,18 +456,10 @@ __rte_ring_do_enqueue(struct rte_ring *r, void * const *obj_table, if (n == 0) goto end; - ENQUEUE_PTRS(); + ENQUEUE_PTRS(r, &r[1], prod_head, obj_table, n, void *); rte_smp_wmb(); - /* - * If there are other enqueues in progress that preceded us, - * we need to wait for them to complete - */ - while (unlikely(r->prod.tail != prod_head)) - rte_pause(); - - r->prod.tail = prod_next; - + update_tail(&r->prod, prod_head, prod_next, is_sp); end: if (free_space != NULL) *free_space = free_entries - n; @@ -550,17 +561,10 @@ __rte_ring_do_dequeue(struct rte_ring *r, void **obj_table, if (n == 0) goto end; - DEQUEUE_PTRS(); + DEQUEUE_PTRS(r, &r[1], cons_head, obj_table, n, void *); rte_smp_rmb(); - /* - * If there are other enqueues in progress that preceded us, - * we need to wait for them to complete - */ - while (unlikely(r->cons.tail != cons_head)) - rte_pause(); - - r->cons.tail = cons_next; + update_tail(&r->cons, cons_head, cons_next, is_sc); end: if (available != NULL) @@ -837,7 +841,7 @@ rte_ring_sc_dequeue(struct rte_ring *r, void **obj_p) static inline int __attribute__((always_inline)) rte_ring_dequeue(struct rte_ring *r, void **obj_p) { - return rte_ring_dequeue_bulk(r, obj_p, 1, NULL) ? 0 : -ENOBUFS; + return rte_ring_dequeue_bulk(r, obj_p, 1, NULL) ? 0 : -ENOENT; } /**