From mboxrd@z Thu Jan 1 00:00:00 1970 Return-Path: Received: from mail-pl0-f66.google.com (mail-pl0-f66.google.com [209.85.160.66]) by dpdk.org (Postfix) with ESMTP id 8F2A11B29A for ; Wed, 17 Jan 2018 05:04:33 +0100 (CET) Received: by mail-pl0-f66.google.com with SMTP id p5so7836971plo.12 for ; Tue, 16 Jan 2018 20:04:33 -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=qiPIdR1xS8iSAQhZ+/CkrnYXEab68bsereXXPDH7R2s=; b=q4nOIr1S/vBGeIJysp+kjht3XY36wqGZVdZmzP7EfN3p4a52n8rjeiDzxemisX2nFL Cn2JgjUwRVpgTbTdxxx8DT/+TlpdzEy5gpwAliqXGLrMdsjJDobMQSqweF1yveccIOl8 1E7cDf741cU4D4sQJQ/HXsEyzl62wXYX8UV5JO00vAzU/SUttJ4lN68vCTVYlW6Slwct iNmiI4KuM96nGFIQJ19n02evnVsiukt0zg5Q9F61ifl9egcYFfGrXvWfRr6n4KXu4nQr MfypRSIaxgSmudYe4EW7NsCXP3NWHRXj57/qhT97IlbqR7vayoGxASWF2ei38G2lk+Qe FQVQ== 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=qiPIdR1xS8iSAQhZ+/CkrnYXEab68bsereXXPDH7R2s=; b=S6YMR31zUtetNGq1QWt5Kh28WOXkKKY7yrIxA40ON4w8Ivv0lRux9CdQ8w6DXuTReC Y0Xt/lFozrmtcDXdU9i6Zxe652nHSfpHgB1msFILRvM8KcQgqnMqQZ7d2V81cpNxZWVe EEw/Vmc1CRsxHsj5z5BxRjGTHp6rDsQJXAsXHJyRmYzA3/fhQo9llvqEqLYNZ01lf6wp Yy702nOMooyBNuPe+OofbqeDuMN5NQnc3MHFEs7H/SX8UrVwa7VGSKNjDFKGi5iI3tpl +Iy9JUejC7cE7bL1oi6QEzy+oEaV5T95byd1iQIetmB4Nk8qVuF8ONA5zBTA2wADtt8Q 0C7Q== X-Gm-Message-State: AKwxytddlLCzdIIZdnMdeVtE0CXLV2a5LWxi7qGGED+TM64qRjahEzl7 yEUgFVwM2qYdytb8sitzg7S75Q== X-Google-Smtp-Source: ACJfBosd64aCj9WswPNL6XtM6WZGHGg24e9BEDB7bYJyNCehjt16l4vmaDp+1Rnx6xTFz4vu/GfHeg== X-Received: by 10.84.211.3 with SMTP id b3mr9135013pli.24.1516161872568; Tue, 16 Jan 2018 20:04:32 -0800 (PST) Received: from localhost.localdomain ([38.106.11.25]) by smtp.gmail.com with ESMTPSA id b68sm3977759pfg.159.2018.01.16.20.04.25 (version=TLS1_2 cipher=ECDHE-RSA-AES128-SHA bits=128/128); Tue, 16 Jan 2018 20:04:32 -0800 (PST) From: Jia He To: dev@dpdk.org, Thomas Monjalon Cc: Jerin Jacob , Jianbo Liu , Jan Viktorin , Olivier Matz , konstantin.ananyev@intel.com, hemant.agrawal@nxp.com, Jia He , Jia He Date: Tue, 16 Jan 2018 20:03:51 -0800 Message-Id: <1516161831-28719-4-git-send-email-hejianet@gmail.com> X-Mailer: git-send-email 2.7.4 In-Reply-To: <1516161831-28719-1-git-send-email-hejianet@gmail.com> References: <1516161831-28719-1-git-send-email-hejianet@gmail.com> Subject: [dpdk-dev] [PATCH v8 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: Wed, 17 Jan 2018 04:04:33 -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 "n" on any architectures 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 Acked-by: Jerin Jacob Acked-by: Olivier Matz Acked-by: Jianbo Liu --- config/common_linuxapp | 2 + lib/librte_ring/Makefile | 3 +- lib/librte_ring/rte_ring.h | 14 ++- lib/librte_ring/rte_ring_c11_mem.h | 193 +++++++++++++++++++++++++++++++++++++ 4 files changed, 210 insertions(+), 2 deletions(-) create mode 100644 lib/librte_ring/rte_ring_c11_mem.h diff --git a/config/common_linuxapp b/config/common_linuxapp index 74c7d64..999fbaa 100644 --- a/config/common_linuxapp +++ b/config/common_linuxapp @@ -50,3 +50,5 @@ CONFIG_RTE_LIBRTE_AVP_PMD=y CONFIG_RTE_LIBRTE_NFP_PMD=y CONFIG_RTE_LIBRTE_POWER=y CONFIG_RTE_VIRTIO_USER=y + +CONFIG_RTE_RING_USE_C11_MEM_MODEL=n diff --git a/lib/librte_ring/Makefile b/lib/librte_ring/Makefile index 281ffb7..bde8907 100644 --- a/lib/librte_ring/Makefile +++ b/lib/librte_ring/Makefile @@ -18,6 +18,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..20b7ff6 --- /dev/null +++ b/lib/librte_ring/rte_ring_c11_mem.h @@ -0,0 +1,193 @@ +/*- + * Copyright(c) 2017 Intel Corporation. All rights reserved. + * All rights reserved. + * + * SPDX-License-Identifier: BSD-3-Clause + */ + +/* + * Derived from FreeBSD's bufring.h + * + ************************************************************************** + * + * Copyright (c) 2007-2009 Kip Macy kmacy@freebsd.org + * All rights reserved. + * + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions are met: + * + * 1. Redistributions of source code must retain the above copyright notice, + * this list of conditions and the following disclaimer. + * + * 2. The name of Kip Macy nor the names of other + * 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