From mboxrd@z Thu Jan 1 00:00:00 1970 Return-Path: Received: from mail-wm0-f45.google.com (mail-wm0-f45.google.com [74.125.82.45]) by dpdk.org (Postfix) with ESMTP id 1124137A0 for ; Fri, 30 Jun 2017 11:40:51 +0200 (CEST) Received: by mail-wm0-f45.google.com with SMTP id 70so4799097wmo.1 for ; Fri, 30 Jun 2017 02:40:51 -0700 (PDT) DKIM-Signature: v=1; a=rsa-sha256; c=relaxed/relaxed; d=6wind-com.20150623.gappssmtp.com; s=20150623; h=date:from:to:cc:subject:message-id:in-reply-to:references :mime-version:content-transfer-encoding; bh=IqDcKitdxq2PqiiMUUzBv7oYw320JtOkRJ6UeEgMqhU=; b=JJBnUn69tOxRu7tdobrdi6KuXufkfQhR2+B/bABFKMXC993mocEQzfme0MD8vEeCIt 2mYIENvOCrBTII7jDGy50a0lKN63I44JPg9BgywgSt7A47RoxjdM7BIYhzSM2Y1CLChG BK6sG7cdXsoxrW2npVBWDHuPIV8xd3b+vIwZpTS0Lv3Xgcn1WLlGQ4+ZwS256LACOkcf rR4zYL5fd6mB6iO1Q1W0tNuMeo9X2Ov3TZHIDd+YRn/mJoLq14tt6HuB3/0hqTmyr+u/ yDGczJqi5z0ZYnHo9K/APh7FRD39YAQlqQ7oXyzcBBRM/sJbhUDYC2Oy/N092i2OL3uv +KIw== X-Google-DKIM-Signature: v=1; a=rsa-sha256; c=relaxed/relaxed; d=1e100.net; s=20161025; h=x-gm-message-state:date:from:to:cc:subject:message-id:in-reply-to :references:mime-version:content-transfer-encoding; bh=IqDcKitdxq2PqiiMUUzBv7oYw320JtOkRJ6UeEgMqhU=; b=UnCVqx8fMS3LjnNxyjQd0Uji4s8Lc8Z2+oPLaKCgdADaY+lMQLq7k4M0j8GF8ybKVx 42Rf4xuL/w3BM7MC/O/Djku59/qcVS9cSzeXhSiHp4zRBsFZbEfJd7Zlwo9jr153nkiy Yks5fwTe/VvdGOmNLTBcu5Qm1uUUS67gfDUImDSIh6RT/TraY0qBSM/546nQPkr3yKQL twRd6pm54K8f7xFn1TbXVrV4PmkxBghmS/hNn4XOwgMwUK4JWxa85S0trbQ+A9zD4IcT RECbL35ciPi+S0Gr+dJ/bx4jDKqqhNxd5BNazZu5j6qL42vL1b3yFwsObQKTOnoaqDsC QZWw== X-Gm-Message-State: AKS2vOwz2m+2DVY4yKmeQIEGBeKMF+uZlNxa8QzC/zhjOtbvbsyc4bmN 8JwwKDryXP5bv1V2 X-Received: by 10.28.141.72 with SMTP id p69mr5329495wmd.12.1498815651343; Fri, 30 Jun 2017 02:40:51 -0700 (PDT) Received: from platinum (2a01cb0c03c651000226b0fffeed02fc.ipv6.abo.wanadoo.fr. [2a01:cb0c:3c6:5100:226:b0ff:feed:2fc]) by smtp.gmail.com with ESMTPSA id w197sm12289858wme.20.2017.06.30.02.40.50 (version=TLS1_2 cipher=ECDHE-RSA-CHACHA20-POLY1305 bits=256/256); Fri, 30 Jun 2017 02:40:51 -0700 (PDT) Date: Fri, 30 Jun 2017 11:40:46 +0200 From: Olivier Matz To: Bruce Richardson Cc: dev@dpdk.org, jerin.jacob@caviumnetworks.com Message-ID: <20170630114046.5c6eb8bb@platinum> In-Reply-To: <20170607133620.275801-2-bruce.richardson@intel.com> References: <20170607133620.275801-1-bruce.richardson@intel.com> <20170607133620.275801-2-bruce.richardson@intel.com> X-Mailer: Claws Mail 3.14.1 (GTK+ 2.24.31; x86_64-pc-linux-gnu) MIME-Version: 1.0 Content-Type: text/plain; charset=US-ASCII Content-Transfer-Encoding: 7bit Subject: Re: [dpdk-dev] [PATCH 1/5] ring: allow rings with non power-of-2 sizes 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: Fri, 30 Jun 2017 09:40:52 -0000 Hi Bruce, Few comments inline. On Wed, 7 Jun 2017 14:36:16 +0100, Bruce Richardson wrote: > The rte_rings traditionally have only supported having ring sizes as powers > of 2, with the actual usable space being the size - 1. In some cases, for > example, with an eventdev where we want to precisely control queue depths > for latency, we need to allow ring sizes which are not powers of two so we > add in an additional ring capacity value to allow that. For existing rings, > this value will be size-1, i.e. the same as the mask, but if the new > EXACT_SZ flag is passed on ring creation, the ring will have exactly the > usable space requested, although the underlying memory size may be bigger. > > Signed-off-by: Bruce Richardson Specifying the exact wanted size looks to be a better API than the current one, which is to give the power of two above the wanted value, knowing there will be only size-1 elements available. What would you think about deprecating ring init/creation without this flag in a few versions? Then, later, we could remove this flag and the new behavior would become the default. > --- > lib/librte_ring/rte_ring.c | 26 ++++++++++++-- > lib/librte_ring/rte_ring.h | 88 +++++++++++++++++++++++++++++----------------- > 2 files changed, 78 insertions(+), 36 deletions(-) > > diff --git a/lib/librte_ring/rte_ring.c b/lib/librte_ring/rte_ring.c > index 5f98c33..b8047ee 100644 > --- a/lib/librte_ring/rte_ring.c > +++ b/lib/librte_ring/rte_ring.c > @@ -140,8 +140,22 @@ rte_ring_init(struct rte_ring *r, const char *name, unsigned count, > r->flags = flags; > r->prod.single = (flags & RING_F_SP_ENQ) ? __IS_SP : __IS_MP; > r->cons.single = (flags & RING_F_SC_DEQ) ? __IS_SC : __IS_MC; > - r->size = count; > - r->mask = count - 1; > + > + if (flags & RING_F_EXACT_SZ) { > + r->size = rte_align32pow2(count + 1); > + r->mask = r->size - 1; > + r->capacity = count; > + } else { > + if ((!POWEROF2(count)) || (count > RTE_RING_SZ_MASK)) { > + RTE_LOG(ERR, RING, > + "Requested size is invalid, must be power of 2, and not exceed the size limit %u\n", > + RTE_RING_SZ_MASK); > + return -EINVAL; > + } > + r->size = count; > + r->mask = count - 1; > + r->capacity = r->mask; > + } > r->prod.head = r->cons.head = 0; > r->prod.tail = r->cons.tail = 0; > > @@ -160,10 +174,15 @@ rte_ring_create(const char *name, unsigned count, int socket_id, > ssize_t ring_size; > int mz_flags = 0; > struct rte_ring_list* ring_list = NULL; > + const unsigned int requested_count = count; > int ret; > > ring_list = RTE_TAILQ_CAST(rte_ring_tailq.head, rte_ring_list); > > + /* for an exact size ring, round up from count to a power of two */ > + if (flags & RING_F_EXACT_SZ) > + count = rte_align32pow2(count + 1); > + > ring_size = rte_ring_get_memsize(count); > if (ring_size < 0) { > rte_errno = ring_size; > @@ -194,7 +213,7 @@ rte_ring_create(const char *name, unsigned count, int socket_id, > r = mz->addr; > /* no need to check return value here, we already checked the > * arguments above */ > - rte_ring_init(r, name, count, flags); > + rte_ring_init(r, name, requested_count, flags); > > te->data = (void *) r; > r->memzone = mz; > @@ -262,6 +281,7 @@ rte_ring_dump(FILE *f, const struct rte_ring *r) > fprintf(f, "ring <%s>@%p\n", r->name, r); > fprintf(f, " flags=%x\n", r->flags); > fprintf(f, " size=%"PRIu32"\n", r->size); > + fprintf(f, " capacity=%"PRIu32"\n", r->capacity); > fprintf(f, " ct=%"PRIu32"\n", r->cons.tail); > fprintf(f, " ch=%"PRIu32"\n", r->cons.head); > fprintf(f, " pt=%"PRIu32"\n", r->prod.tail); > diff --git a/lib/librte_ring/rte_ring.h b/lib/librte_ring/rte_ring.h > index 97f025a..494d31f 100644 > --- a/lib/librte_ring/rte_ring.h > +++ b/lib/librte_ring/rte_ring.h > @@ -153,6 +153,7 @@ struct rte_ring { > /**< Memzone, if any, containing the rte_ring */ > uint32_t size; /**< Size of ring. */ > uint32_t mask; /**< Mask (size-1) of ring. */ > + uint32_t capacity; /**< Usable size of ring */ > > /** Ring producer status. */ > struct rte_ring_headtail prod __rte_aligned(PROD_ALIGN); > @@ -163,6 +164,15 @@ struct rte_ring { > > #define RING_F_SP_ENQ 0x0001 /**< The default enqueue is "single-producer". */ > #define RING_F_SC_DEQ 0x0002 /**< The default dequeue is "single-consumer". */ > +/** > + * Ring is to hold exactly requested number of entries. > + * Without this flag set, the ring size requested must be a power of 2, and the > + * usable space will be that size - 1. With the flag, the requested size will > + * be rounded up to the next power of two, but the usable space will be exactly > + * that requested. Worst case, if a power-of-2 size is requested, half the > + * ring space will be wasted. > + */ > +#define RING_F_EXACT_SZ 0x0004 > #define RTE_RING_SZ_MASK (unsigned)(0x0fffffff) /**< Ring size mask */ > > /* @internal defines for passing to the enqueue dequeue worker functions */ > @@ -389,7 +399,7 @@ __rte_ring_move_prod_head(struct rte_ring *r, int is_sp, > uint32_t *old_head, uint32_t *new_head, > uint32_t *free_entries) > { > - const uint32_t mask = r->mask; > + const uint32_t capacity = r->capacity; > unsigned int max = n; > int success; > > @@ -399,11 +409,13 @@ __rte_ring_move_prod_head(struct rte_ring *r, int is_sp, > > *old_head = r->prod.head; > const uint32_t cons_tail = r->cons.tail; > - /* The subtraction is done between two unsigned 32bits value > + /* > + * 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 size(ring)-1. */ > - *free_entries = (mask + cons_tail - *old_head); > + * 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)) > @@ -845,69 +857,63 @@ rte_ring_dequeue(struct rte_ring *r, void **obj_p) > } > > /** > - * Test if a ring is full. > + * Return the number of entries in a ring. > * > * @param r > * A pointer to the ring structure. > * @return > - * - 1: The ring is full. > - * - 0: The ring is not full. > + * The number of entries in the ring. > */ > -static inline int > -rte_ring_full(const struct rte_ring *r) > +static inline unsigned > +rte_ring_count(const struct rte_ring *r) > { > uint32_t prod_tail = r->prod.tail; > uint32_t cons_tail = r->cons.tail; > - return ((cons_tail - prod_tail - 1) & r->mask) == 0; > + return (prod_tail - cons_tail) & r->mask; > } When used on a mc/mp ring, this function can now return a value which is higher than the ring capacity. Even if this function is not really accurate when used while the ring is use, I think we should maximize the result with r->capacity. This will also avoid rte_ring_free_count() to return a overflowed value. > > /** > - * Test if a ring is empty. > + * Return the number of free entries in a ring. > * > * @param r > * A pointer to the ring structure. > * @return > - * - 1: The ring is empty. > - * - 0: The ring is not empty. > + * The number of free entries in the ring. > */ > -static inline int > -rte_ring_empty(const struct rte_ring *r) > +static inline unsigned > +rte_ring_free_count(const struct rte_ring *r) > { > - uint32_t prod_tail = r->prod.tail; > - uint32_t cons_tail = r->cons.tail; > - return !!(cons_tail == prod_tail); > + return r->capacity - rte_ring_count(r); > } > > /** > - * Return the number of entries in a ring. > + * Test if a ring is full. > * > * @param r > * A pointer to the ring structure. > * @return > - * The number of entries in the ring. > + * - 1: The ring is full. > + * - 0: The ring is not full. > */ > -static inline unsigned > -rte_ring_count(const struct rte_ring *r) > +static inline int > +rte_ring_full(const struct rte_ring *r) > { > - uint32_t prod_tail = r->prod.tail; > - uint32_t cons_tail = r->cons.tail; > - return (prod_tail - cons_tail) & r->mask; > + return rte_ring_free_count(r) == 0; > } > > /** > - * Return the number of free entries in a ring. > + * Test if a ring is empty. > * > * @param r > * A pointer to the ring structure. > * @return > - * The number of free entries in the ring. > + * - 1: The ring is empty. > + * - 0: The ring is not empty. > */ > -static inline unsigned > -rte_ring_free_count(const struct rte_ring *r) > +static inline int > +rte_ring_empty(const struct rte_ring *r) > { > - uint32_t prod_tail = r->prod.tail; > - uint32_t cons_tail = r->cons.tail; > - return (cons_tail - prod_tail - 1) & r->mask; > + return rte_ring_count(r) == 0; > } > > /** > @@ -916,7 +922,9 @@ rte_ring_free_count(const struct rte_ring *r) > * @param r > * A pointer to the ring structure. > * @return > - * The number of elements which can be stored in the ring. > + * The size of the data store used by the ring. > + * NOTE: this is not the same as the usable space in the ring. To query that > + * use ``rte_ring_get_capacity()``. > */ > static inline unsigned int > rte_ring_get_size(const struct rte_ring *r) > @@ -925,6 +933,20 @@ rte_ring_get_size(const struct rte_ring *r) > } > > /** > + * Return the number of elements which can be stored in the ring. > + * > + * @param r > + * A pointer to the ring structure. > + * @return > + * The usable size of the ring. > + */ > +static inline unsigned int > +rte_ring_get_capacity(const struct rte_ring *r) > +{ > + return r->capacity; > +} > + > +/** > * Dump the status of all rings on the console > * > * @param f I think the users of rte_ring_get_size() use this API to get the number of elements that can fit in the ring. The patch changes the semantic of this function (even if the result is the same when EXACT_SZ is not passed). This is the same for ring->size. Wouldn't it be better to rename all current occurences of "size" as "storage_size", and introduce a new "size", which is what you call "capacity". I'm asking the question but I'm not that convinced myself... nevertheless I find storage_size/size less confusing than size/capacity. Olivier