From mboxrd@z Thu Jan 1 00:00:00 1970 Return-Path: Received: from mail-pf0-f196.google.com (mail-pf0-f196.google.com [209.85.192.196]) by dpdk.org (Postfix) with ESMTP id C1E913257 for ; Mon, 27 Nov 2017 03:01:10 +0100 (CET) Received: by mail-pf0-f196.google.com with SMTP id q4so16672553pfg.13 for ; Sun, 26 Nov 2017 18:01:10 -0800 (PST) DKIM-Signature: v=1; a=rsa-sha256; c=relaxed/relaxed; d=gmail.com; s=20161025; h=from:to:cc:subject:date:message-id:in-reply-to:references; bh=Izrq30g81SWEx7YK+hRYIVslfyhsgjZWdnJ6etsytyg=; b=VWhfUDC+yP4p+hKWwd0IblYNXrMKKLEp3Unmqs3nYW3QMfBdcCjI+t7+AI60HjubP3 tBlddpLMZO58sN04rnh5yiRxm0UKshBL53kwkjUCa3S1NG0Z36fRd5llfqhh4v3OJRpK QcQkl/eYsuwnQQ36CyfSslNkgDwFLrHFAZYycV0iLqXIpMWKKmovDA6GFEC4376VfHVB RzHbwR/zXDidbb5pMBNU+AtN9jBJkdjfGciWJDmNN8T4MRMZL2DoSXGyG9aL4a5tP1sR 0N5ekma6MgUU5rUe/HYh/kWtdks+u/gKMafYEv+F9CNc48yOvIkRiGFZW0e27dOohqa7 LCww== X-Google-DKIM-Signature: v=1; a=rsa-sha256; c=relaxed/relaxed; d=1e100.net; s=20161025; h=x-gm-message-state:from:to:cc:subject:date:message-id:in-reply-to :references; bh=Izrq30g81SWEx7YK+hRYIVslfyhsgjZWdnJ6etsytyg=; b=TBDXMxLa/x8I0offEQRMAzELeGl+HeeeqyUIYMINQPybLYvkO4n4H1rFvYVR+jc7Rl SoPf0xuqVThlavjJAXDOnYYXw+LAZOJ5c3oVTMBTnG5H2mng5Xg2vmBMx4BLa28IP7Fd RIvvCLnzpeyqoOGbNT0suehg6PPj83/XctPiCyGJYlkq1HQycVea1hkmVftzSzDhCSCe 4rLhJNro+5X/Za2nYjFD0aXP6zgHiBpYYh9k2y5wIFN62A2LoLQJbGZ8b2MsQq91+rDW IUZhp+v14UAYiwc+eXRnTwiM+G0QW8wpKpgVeUXOqujiWAZBK3goXxRH4vsy/+gIPGTD pX9g== X-Gm-Message-State: AJaThX5sPRQMAYoIXYr+n0cI/nWDtE9OMBPegNnH2Rt9DgUrmDa4usH7 Vr4Nl4gQY9DNAgwMVp6+feId2w== X-Google-Smtp-Source: AGs4zMZo7IkSRXq2zmZRo9LcC7lV7/6WiAHMuosL1/lAGHLhdx7ugrs0YiVonrOlCAk3ivN1jn4Ksg== X-Received: by 10.101.93.136 with SMTP id f8mr25045326pgt.291.1511748069652; Sun, 26 Nov 2017 18:01:09 -0800 (PST) Received: from localhost.localdomain ([38.106.11.25]) by smtp.gmail.com with ESMTPSA id a87sm48926743pfg.159.2017.11.26.18.01.03 (version=TLS1_2 cipher=ECDHE-RSA-AES128-SHA bits=128/128); Sun, 26 Nov 2017 18:01:09 -0800 (PST) From: Jia He To: jerin.jacob@caviumnetworks.com, dev@dpdk.org, bruce.richardson@intel.com, konstantin.ananyev@intel.com Cc: olivier.matz@6wind.com, jianbo.liu@arm.com, hemant.agrawal@nxp.com, Jia He , Jia He Date: Sun, 26 Nov 2017 18:00:24 -0800 Message-Id: <1511748024-7245-4-git-send-email-hejianet@gmail.com> X-Mailer: git-send-email 2.7.4 In-Reply-To: <1511748024-7245-1-git-send-email-hejianet@gmail.com> References: <1510291392-33251-1-git-send-email-hejianet@gmail.com> <1511748024-7245-1-git-send-email-hejianet@gmail.com> Subject: [dpdk-dev] [PATCH V6 3/3] ring: introduce new header file to support C11 memory model X-BeenThere: dev@dpdk.org X-Mailman-Version: 2.1.15 Precedence: list List-Id: DPDK patches and discussions List-Unsubscribe: , List-Archive: List-Post: List-Help: List-Subscribe: , X-List-Received-Date: Mon, 27 Nov 2017 02:01:11 -0000 To support C11 memory model barrier, 2 options are suggested by Jerin: 1. use rte_smp_rmb 2. use load_acquire/store_release(refer to [1]). CONFIG_RTE_RING_USE_C11_MEM_MODEL is provided, and by default it is "y" only on arm64 so far. The reason why providing 2 options is due to the performance benchmark difference in different arm machines, refer to [2]. We haven't tested on ppc64. If anyone verifies it, he can add CONFIG_RTE_RING_USE_C11_MEM_MODEL=y to ppc64 config files. [1] https://github.com/freebsd/freebsd/blob/master/sys/sys/buf_ring.h#L170 [2] http://dpdk.org/ml/archives/dev/2017-October/080861.html Signed-off-by: Jia He Suggested-by: Jerin Jacob --- config/common_armv8a_linuxapp | 2 + lib/librte_ring/Makefile | 3 +- lib/librte_ring/rte_ring.h | 14 ++- lib/librte_ring/rte_ring_c11_mem.h | 186 +++++++++++++++++++++++++++++++++++++ 4 files changed, 203 insertions(+), 2 deletions(-) create mode 100644 lib/librte_ring/rte_ring_c11_mem.h diff --git a/config/common_armv8a_linuxapp b/config/common_armv8a_linuxapp index 6732d1e..5b7b2eb 100644 --- a/config/common_armv8a_linuxapp +++ b/config/common_armv8a_linuxapp @@ -49,3 +49,5 @@ CONFIG_RTE_LIBRTE_SFC_EFX_PMD=n CONFIG_RTE_LIBRTE_AVP_PMD=n CONFIG_RTE_SCHED_VECTOR=n + +CONFIG_RTE_RING_USE_C11_MEM_MODEL=y diff --git a/lib/librte_ring/Makefile b/lib/librte_ring/Makefile index c959945..a2682e7 100644 --- a/lib/librte_ring/Makefile +++ b/lib/librte_ring/Makefile @@ -46,6 +46,7 @@ SRCS-$(CONFIG_RTE_LIBRTE_RING) := rte_ring.c # install includes SYMLINK-$(CONFIG_RTE_LIBRTE_RING)-include := rte_ring.h \ - rte_ring_generic.h + rte_ring_generic.h \ + rte_ring_c11_mem.h include $(RTE_SDK)/mk/rte.lib.mk diff --git a/lib/librte_ring/rte_ring.h b/lib/librte_ring/rte_ring.h index 519614c..3343eba 100644 --- a/lib/librte_ring/rte_ring.h +++ b/lib/librte_ring/rte_ring.h @@ -356,8 +356,20 @@ void rte_ring_dump(FILE *f, const struct rte_ring *r); } \ } while (0) -/* Move common functions to generic file */ +/* Between load and load. there might be cpu reorder in weak model + * (powerpc/arm). + * There are 2 choices for the users + * 1.use rmb() memory barrier + * 2.use one-direcion load_acquire/store_release barrier,defined by + * CONFIG_RTE_RING_USE_C11_MEM_MODEL=y + * It depends on performance test results. + * By default, move common functions to rte_ring_generic.h + */ +#ifdef RTE_RING_USE_C11_MEM_MODEL +#include "rte_ring_c11_mem.h" +#else #include "rte_ring_generic.h" +#endif /** * @internal Enqueue several objects on the ring diff --git a/lib/librte_ring/rte_ring_c11_mem.h b/lib/librte_ring/rte_ring_c11_mem.h new file mode 100644 index 0000000..ca21e95 --- /dev/null +++ b/lib/librte_ring/rte_ring_c11_mem.h @@ -0,0 +1,186 @@ +/*- + * BSD LICENSE + * + * Copyright(c) 2017 hxt-semitech. All rights reserved. + * + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions + * are met: + * + * * Redistributions of source code must retain the above copyright + * notice, this list of conditions and the following disclaimer. + * * Redistributions in binary form must reproduce the above copyright + * notice, this list of conditions and the following disclaimer in + * the documentation and/or other materials provided with the + * distribution. + * * Neither the name of hxt-semitech nor the names of its + * contributors may be used to endorse or promote products derived + * from this software without specific prior written permission. + * + * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS + * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT + * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR + * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT + * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, + * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT + * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, + * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY + * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT + * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE + * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + */ + +#ifndef _RTE_RING_C11_MEM_H_ +#define _RTE_RING_C11_MEM_H_ + +static __rte_always_inline void +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) + while (unlikely(ht->tail != old_val)) + rte_pause(); + + __atomic_store_n(&ht->tail, new_val, __ATOMIC_RELEASE); +} + +/** + * @internal This function updates the producer head for enqueue + * + * @param r + * A pointer to the ring structure + * @param is_sp + * Indicates whether multi-producer path is needed or not + * @param n + * The number of elements we will want to enqueue, i.e. how far should the + * head be moved + * @param behavior + * RTE_RING_QUEUE_FIXED: Enqueue a fixed number of items from a ring + * RTE_RING_QUEUE_VARIABLE: Enqueue as many items as possible from ring + * @param old_head + * Returns head value as it was before the move, i.e. where enqueue starts + * @param new_head + * Returns the current/new head value i.e. where enqueue finishes + * @param free_entries + * Returns the amount of free space in the ring BEFORE head was moved + * @return + * Actual number of objects enqueued. + * If behavior == RTE_RING_QUEUE_FIXED, this will be 0 or n only. + */ +static __rte_always_inline unsigned int +__rte_ring_move_prod_head(struct rte_ring *r, 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 = __atomic_load_n(&r->prod.head, + __ATOMIC_ACQUIRE); + const uint32_t cons_tail = r->cons.tail; + /* + * 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 + success = __atomic_compare_exchange_n(&r->prod.head, + old_head, *new_head, + 0, __ATOMIC_ACQUIRE, + __ATOMIC_RELAXED); + } while (unlikely(success == 0)); + return n; +} + +/** + * @internal This function updates the consumer head for dequeue + * + * @param r + * A pointer to the ring structure + * @param is_sc + * Indicates whether multi-consumer path is needed or not + * @param n + * The number of elements we will want to enqueue, i.e. how far should the + * head be moved + * @param behavior + * RTE_RING_QUEUE_FIXED: Dequeue a fixed number of items from a ring + * RTE_RING_QUEUE_VARIABLE: Dequeue as many items as possible from ring + * @param old_head + * Returns head value as it was before the move, i.e. where dequeue starts + * @param new_head + * Returns the current/new head value i.e. where dequeue finishes + * @param entries + * Returns the number of entries in the ring BEFORE head was moved + * @return + * - Actual number of objects dequeued. + * If behavior == RTE_RING_QUEUE_FIXED, this will be 0 or n only. + */ +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; + int success; + + /* move cons.head atomically */ + do { + /* Restore n as it may change every loop */ + n = max; + *old_head = __atomic_load_n(&r->cons.head, + __ATOMIC_ACQUIRE); + const uint32_t prod_tail = r->prod.tail; + /* 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 + success = __atomic_compare_exchange_n(&r->cons.head, + old_head, *new_head, + 0, __ATOMIC_ACQUIRE, + __ATOMIC_RELAXED); + } while (unlikely(success == 0)); + return n; +} + +#endif /* _RTE_RING_C11_MEM_H_ */ -- 2.7.4