Блог

При выполнении реализации MPSC DPDK (с несколькими производителями и одним потребителем) API кольцевого буфера, я нашел код для перемещения заголовка производителя для вставки новых элементов в кольцевой буфер. Функция заключается в следующем :

     static __rte_always_inline unsi&ned int
    __rte_rin&_move_prod_head(struct rte_rin& *r, unsi&ned int is_sp,
            unsi&ned int n, enum rte_rin&_queue_behavior behavior,
            uint32_t *old_head, uint32_t *new_head,
            uint32_t *free_entries)
    {
        const uint32_t capacity = r-&&t;capacity;
        uint32_t cons_tail;
        unsi&ned int max = n;
        int success;
    
        *old_head = __atomic_load_n(amp;r-&&t;prod.head, __ATOMIC_RELAXED);
        do {
            /* Reset n to the initial burst count */
            n = max;
    
            /* Ensure the head is read before tail */
            __atomic_thread_fence(__ATOMIC_ACQUIRE);
    
            /* load-acquire synchronize with store-release of ht-&&t;tail
             * in update_tail.
             */
            cons_tail = __atomic_load_n(amp;r-&&t;cons.tail,
                        __ATOMIC_ACQUIRE);
    
            /* The subtraction is done between two unsi&ned 32bits value
             * (the result is always modulo 32 bits even if we have
             * *old_head &&t; 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 enou&h room in rin& */
            if (unlikely(n &&t; *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-&&t;prod.head = *new_head, success = 1;
            else
                /* on failure, *old_head is updated */
                success = __atomic_compare_exchan&e_n(amp;r-&&t;prod.head,
                        old_head, *new_head,
                        0, __ATOMIC_RELAXED,
                        __ATOMIC_RELAXED);
        } while (unlikely(success == 0));
        return n;
    }
  

Загрузка и сравнение head производителя выполняется с использованием __АТОМАРНОГО_упорядочения памяти. Разве это не проблема, когда несколько производителей из разных потоков производят в очередь. Или я что-то упускаю?

https://doc.dpdk.or&/&uides/pro&_&uide/rin&_lib.html описывает базовый механизм, который DPDK использует для реализации кольцевого буфера.