lkml.org 
[lkml]   [2018]   [Feb]   [25]   [last100]   RSS Feed
Views: [wrap][no wrap]   [headers]  [forward] 
 
Messages in this thread
/
SubjectRe: [RFC PATCH v2] ptr_ring: linked list fallback
From
Date


On 2018年02月26日 09:17, Michael S. Tsirkin wrote:
> So pointer rings work fine, but they have a problem: make them too small
> and not enough entries fit. Make them too large and you start flushing
> your cache and running out of memory.
>
> This is a new idea of mine: a ring backed by a linked list. Once you run
> out of ring entries, instead of a drop you fall back on a list with a
> common lock.
>
> Should work well for the case where the ring is typically sized
> correctly, but will help address the fact that some user try to set e.g.
> tx queue length to 1000000.
>
> In other words, the idea is that if a user sets a really huge TX queue
> length, we allocate a ptr_ring which is smaller, and use the backup
> linked list when necessary to provide the requested TX queue length
> legitimately.
>
> My hope this will move us closer to direction where e.g. fw codel can
> use ptr rings without locking at all. The API is still very rough, and
> I really need to take a hard look at lock nesting.
>
> Compiled only, sending for early feedback/flames.
>
> Signed-off-by: Michael S. Tsirkin <mst@redhat.com>
> ---
>
> changes from v1:
> - added clarifications by DaveM in the commit log
> - build fixes
>
> include/linux/ptr_ring.h | 64 +++++++++++++++++++++++++++++++++++++++++++++---
> 1 file changed, 61 insertions(+), 3 deletions(-)
>
> diff --git a/include/linux/ptr_ring.h b/include/linux/ptr_ring.h
> index d72b2e7..8aa8882 100644
> --- a/include/linux/ptr_ring.h
> +++ b/include/linux/ptr_ring.h
> @@ -31,11 +31,18 @@
> #include <asm/errno.h>
> #endif
>
> +/* entries must start with the following structure */
> +struct plist {
> + struct plist *next;
> + struct plist *last; /* only valid in the 1st entry */
> +};

So I wonder whether or not it's better to do this in e.g skb_array
implementation. Then it can use its own prev/next field.

> +
> struct ptr_ring {
> int producer ____cacheline_aligned_in_smp;
> spinlock_t producer_lock;
> int consumer_head ____cacheline_aligned_in_smp; /* next valid entry */
> int consumer_tail; /* next entry to invalidate */
> + struct plist *consumer_list;
> spinlock_t consumer_lock;
> /* Shared consumer/producer data */
> /* Read-only by both the producer and the consumer */
> @@ -120,10 +127,40 @@ static inline int __ptr_ring_produce(struct ptr_ring *r, void *ptr)
> }
>
> /*
> - * Note: resize (below) nests producer lock within consumer lock, so if you
> - * consume in interrupt or BH context, you must disable interrupts/BH when
> - * calling this.
> + * Note: resize API with the _fallback should be used when calling this.
> */
> +static inline int ptr_ring_produce_fallback(struct ptr_ring *r, void *ptr)
> +{
> + int ret;
> + unsigned long flags;
> + struct plist *p = ptr;
> +
> + p->next = NULL;
> + p->last = p;
> +
> + spin_lock_irqsave(&r->producer_lock, flags);
> + ret = __ptr_ring_produce(r, ptr);
> + if (ret) {
> + spin_lock(&r->consumer_lock);
> + ret = __ptr_ring_produce(r, ptr);
> + if (ret) {
> + int producer = r->producer ? r->producer - 1 :
> + r->size - 1;
> + struct plist *first = r->queue[producer];
> +
> + BUG_ON(!first);
> +
> + first->last->next = p;
> + first->last = p;

I believe we still need a limitation on the total size of the queue.

Thanks

> + }
> + spin_unlock(&r->consumer_lock);
> + }
> +
> + spin_unlock_irqrestore(&r->producer_lock, flags);
> +
> + return ret;
> +}
> +
> static inline int ptr_ring_produce(struct ptr_ring *r, void *ptr)
> {
> int ret;
> @@ -135,6 +172,7 @@ static inline int ptr_ring_produce(struct ptr_ring *r, void *ptr)
> return ret;
> }
>
> +
> static inline int ptr_ring_produce_irq(struct ptr_ring *r, void *ptr)
> {
> int ret;
> @@ -359,6 +397,26 @@ static inline void *ptr_ring_consume_bh(struct ptr_ring *r)
> return ptr;
> }
>
> +static inline void *ptr_ring_consume_fallback(struct ptr_ring *r)
> +{
> + unsigned long flags;
> + struct plist *ptr;
> +
> + spin_lock_irqsave(&r->consumer_lock, flags);
> + if (r->consumer_list) {
> + ptr = r->consumer_list;
> + r->consumer_list = ptr->next;
> + } else {
> + ptr = __ptr_ring_consume(r);
> + if (ptr) {
> + r->consumer_list = ptr->next;
> + }
> + }
> + spin_unlock_irqrestore(&r->consumer_lock, flags);
> +
> + return ptr;
> +}
> +
> static inline int ptr_ring_consume_batched(struct ptr_ring *r,
> void **array, int n)
> {

\
 
 \ /
  Last update: 2018-02-26 04:16    [W:0.086 / U:5.040 seconds]
©2003-2020 Jasper Spaans|hosted at Digital Ocean and TransIP|Read the blog|Advertise on this site