rte_ring_c11_pvt.h

Go to the documentation of this file.

/* 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 unsigned int
__rte_ring_headtail_move_head_st(struct rte_ring_headtail *d,
        const struct rte_ring_headtail *s, uint32_t capacity,
        unsigned int n,
        enum rte_ring_queue_behavior behavior,
        uint32_t *old_head, uint32_t *new_head, uint32_t *entries)
{
    uint32_t stail;

    /* Single producer: only this thread writes d->head,
     * so a relaxed load is sufficient.
     */
    *old_head = rte_atomic_load_explicit(&d->head, rte_memory_order_relaxed);

    /* Acquire pairs with the consumer's release-store of tail in __rte_ring_update_tail,
     * ensuring the consumer's ring-element reads are complete before
     * we observe the updated tail.
     */
    stail = rte_atomic_load_explicit(&s->tail, rte_memory_order_acquire);

    /* Unsigned subtraction is modulo 2^32, so entries is always in
     * [0, capacity) even if old_head > stail.
     */
    *entries = capacity + stail - *old_head;

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

    if (n > 0) {
        *new_head = *old_head + n;
        rte_atomic_store_explicit(&d->head, *new_head, rte_memory_order_relaxed);
    }

    return n;
}

static __rte_always_inline unsigned int
__rte_ring_headtail_move_head_mt(struct rte_ring_headtail *d,
        const struct rte_ring_headtail *s, uint32_t capacity,
        unsigned int n,
        enum rte_ring_queue_behavior behavior,
        uint32_t *old_head, uint32_t *new_head, uint32_t *entries)
{
    uint32_t stail;
    bool success;
    unsigned int max = n;

    /*
     * A0: Establishes a synchronizing edge with R1.
     * Ensure that this thread observes same values
     * to stail observed by the thread that updated
     * d->head.
     * If not, an unsafe partial order may ensue.
     */
    *old_head = rte_atomic_load_explicit(&d->head,
            rte_memory_order_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.
         */
        stail = rte_atomic_load_explicit(&s->tail,
                    rte_memory_order_acquire);

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

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

        if (n == 0)
            return 0;

        *new_head = *old_head + n;
        /* 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 stail
         * observed by that thread on CAS failure (to retry
         * with an updated *old_head).
         */
        success = rte_atomic_compare_exchange_strong_explicit(
                    &d->head, old_head, *new_head,
                    rte_memory_order_release,
                    rte_memory_order_acquire);
    } while (unlikely(!success));
    return n;
}

#endif /* _RTE_RING_C11_PVT_H_ */