dpdk ring 多/单生产者、多/单消费者

static __rte_always_inline unsigned
rte_ring_sp_enqueue_burst(struct rte_ring *r, void * const *obj_table,
                         unsigned int n, unsigned int *free_space)
{
        return __rte_ring_do_enqueue(r, obj_table, n,
                        RTE_RING_QUEUE_VARIABLE, __IS_SP, free_space);
}

static __rte_always_inline unsigned
rte_ring_mp_enqueue_burst(struct rte_ring *r, void * const *obj_table,
                         unsigned int n, unsigned int *free_space)
{
        return __rte_ring_do_enqueue(r, obj_table, n,
                        RTE_RING_QUEUE_VARIABLE, __IS_MP, free_space);
}

static __rte_always_inline unsigned int
__rte_ring_move_prod_head(struct rte_ring *r, unsigned int is_sp,
                unsigned int n, enum rte_ring_queue_behavior behavior,
                uint32_t *old_head, uint32_t *new_head,
                uint32_t *free_entries)
{
        const uint32_t capacity = r->capacity;
        unsigned int max = n;
        int success;

        do {
                /* Reset n to the initial burst count */
                n = max;

                *old_head = r->prod.head;

                /* add rmb barrier to avoid load/load reorder in weak
                 * memory model. It is noop on x86
                 */
                rte_smp_rmb();

                /*
                 *  The subtraction is done between two unsigned 32bits value
                 * (the result is always modulo 32 bits even if we have
                 * *old_head > cons_tail). So ‘free_entries‘ is always between 0
                 * and capacity (which is < size).
                 */
                *free_entries = (capacity + r->cons.tail - *old_head);

                /* check that we have enough room in ring */

                 * and capacity (which is < size).
                 */
                *free_entries = (capacity + r->cons.tail - *old_head);

                /* check that we have enough room in ring */
                if (unlikely(n > *free_entries))
                        n = (behavior == RTE_RING_QUEUE_FIXED) ?
                                        0 : *free_entries;

                if (n == 0)
                        return 0;

                *new_head = *old_head + n;
                if (is_sp)
                        r->prod.head = *new_head, success = 1;
                else
                        success = rte_atomic32_cmpset(&r->prod.head,
                                        *old_head, *new_head);
        } while (unlikely(success == 0));
        return n;
}