rte_ring_c11_pvt.h

/* SPDX-License-Identifier: BSD-3-Clause
 *
 * Copyright (c) 2017,2018 HXT-semitech Corporation.
 * Copyright (c) 2007-2009 Kip Macy kmacy@freebsd.org
 * Copyright (c) 2021 Arm Limited
 * All rights reserved.
 * Derived from FreeBSD's bufring.h
 * Used as BSD-3 Licensed with permission from Kip Macy.
 */

#ifndef _RTE_RING_C11_PVT_H_
#define _RTE_RING_C11_PVT_H_

static __rte_always_inline void
__rte_ring_update_tail(struct rte_ring_headtail *ht, uint32_t old_val,
        uint32_t new_val, uint32_t single, uint32_t enqueue)
{
    RTE_SET_USED(enqueue);

    /*
     * If there are other enqueues/dequeues in progress that preceded us,
     * we need to wait for them to complete
     */
    if (!single)
        rte_wait_until_equal_32(&ht->tail, old_val, __ATOMIC_RELAXED);

    /*
     * R0: Establishes a synchronizing edge with load-acquire of
     * cons_tail at A1 or prod_tail at A4.
     * Ensures that memory effects by this thread on ring elements array
     * is observed by a different thread of the other type.
     */
    __atomic_store_n(&ht->tail, new_val, __ATOMIC_RELEASE);
}

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;
    uint32_t cons_tail;
    unsigned int max = n;
    int success;

    /*
     * A0: Establishes a synchronizing edge with R1.
     * Ensure that this thread observes same values
     * to cons_tail observed by the thread that
     * updated r->prod.head.
     * If not, an unsafe partial order may ensue.
     */
    *old_head = __atomic_load_n(&r->prod.head, __ATOMIC_ACQUIRE);
    do {
        /* Reset n to the initial burst count */
        n = max;

        /*
         * A1: Establishes a synchronizing edge with R0.
         * Ensures that other thread's memory effects on
         * ring elements array is observed by the time
         * this thread observes its tail update.
         */
        cons_tail = __atomic_load_n(&r->cons.tail, __ATOMIC_ACQUIRE);

        /* 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 + 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
            /* on failure, *old_head is updated */
            /*
             * R1/A2.
             * R1: Establishes a synchronizing edge with A0 of a
             * different thread.
             * A2: Establishes a synchronizing edge with R1 of a
             * different thread to observe same value for
             * cons_tail observed by that thread on CAS failure
             * (to retry with an updated *old_head).
             */
            success = __atomic_compare_exchange_n(&r->prod.head,
                    old_head, *new_head,
                    0, __ATOMIC_RELEASE,
                    __ATOMIC_ACQUIRE);
    } while (unlikely(success == 0));
    return n;
}

static __rte_always_inline unsigned int
__rte_ring_move_cons_head(struct rte_ring *r, int is_sc,
        unsigned int n, enum rte_ring_queue_behavior behavior,
        uint32_t *old_head, uint32_t *new_head,
        uint32_t *entries)
{
    unsigned int max = n;
    uint32_t prod_tail;
    int success;

    /*
     * A3: Establishes a synchronizing edge with R2.
     * Ensure that this thread observes same values
     * to prod_tail observed by the thread that
     * updated r->cons.head.
     * If not, an unsafe partial order may ensue.
     */
    *old_head = __atomic_load_n(&r->cons.head, __ATOMIC_ACQUIRE);
    do {
        /* Restore n as it may change every loop */
        n = max;

        /*
         * A4: Establishes a synchronizing edge with R0.
         * Ensures that other thread's memory effects on
         * ring elements array is observed by the time
         * this thread observes its tail update.
         */
        prod_tail = __atomic_load_n(&r->prod.tail, __ATOMIC_ACQUIRE);

        /* The subtraction is done between two unsigned 32bits value
         * (the result is always modulo 32 bits even if we have
         * cons_head > prod_tail). So 'entries' is always between 0
         * and size(ring)-1.
         */
        *entries = (prod_tail - *old_head);

        /* Set the actual entries for dequeue */
        if (n > *entries)
            n = (behavior == RTE_RING_QUEUE_FIXED) ? 0 : *entries;

        if (unlikely(n == 0))
            return 0;

        *new_head = *old_head + n;
        if (is_sc)
            r->cons.head = *new_head, success = 1;
        else
            /* on failure, *old_head will be updated */
            /*
             * R2/A5.
             * R2: Establishes a synchronizing edge with A3 of a
             * different thread.
             * A5: Establishes a synchronizing edge with R2 of a
             * different thread to observe same value for
             * prod_tail observed by that thread on CAS failure
             * (to retry with an updated *old_head).
             */
            success = __atomic_compare_exchange_n(&r->cons.head,
                            old_head, *new_head,
                            0, __ATOMIC_RELEASE,
                            __ATOMIC_ACQUIRE);
    } while (unlikely(success == 0));
    return n;
}

#endif /* _RTE_RING_C11_PVT_H_ */