ptr_ring: introduce batch dequeuing

This patch introduce a batched version of consuming, consumer can
dequeue more than one pointers from the ring at a time. We don't care
about the reorder of reading here so no need for compiler barrier.

Signed-off-by: Jason Wang <jasowang@redhat.com>
Signed-off-by: David S. Miller <davem@davemloft.net>

diff --git a/include/linux/ptr_ring.h b/include/linux/ptr_ring.h
index 796b90f6d4e90028783a7b085456f48f1150ae21..d8c97ec8a8e66d865f516f708a5b3f8fe567da29 100644 (file)
--- a/include/linux/ptr_ring.h
+++ b/include/linux/ptr_ring.h
@@ -278,6 +278,22 @@ static inline void *__ptr_ring_consume(struct ptr_ring *r)
        return ptr;
 }
 
+static inline int __ptr_ring_consume_batched(struct ptr_ring *r,
+                                            void **array, int n)
+{
+       void *ptr;
+       int i;
+
+       for (i = 0; i < n; i++) {
+               ptr = __ptr_ring_consume(r);
+               if (!ptr)
+                       break;
+               array[i] = ptr;
+       }
+
+       return i;
+}
+
 /*
  * Note: resize (below) nests producer lock within consumer lock, so if you
  * call this in interrupt or BH context, you must disable interrupts/BH when
@@ -328,6 +344,55 @@ static inline void *ptr_ring_consume_bh(struct ptr_ring *r)
        return ptr;
 }
 
+static inline int ptr_ring_consume_batched(struct ptr_ring *r,
+                                          void **array, int n)
+{
+       int ret;
+
+       spin_lock(&r->consumer_lock);
+       ret = __ptr_ring_consume_batched(r, array, n);
+       spin_unlock(&r->consumer_lock);
+
+       return ret;
+}
+
+static inline int ptr_ring_consume_batched_irq(struct ptr_ring *r,
+                                              void **array, int n)
+{
+       int ret;
+
+       spin_lock_irq(&r->consumer_lock);
+       ret = __ptr_ring_consume_batched(r, array, n);
+       spin_unlock_irq(&r->consumer_lock);
+
+       return ret;
+}
+
+static inline int ptr_ring_consume_batched_any(struct ptr_ring *r,
+                                              void **array, int n)
+{
+       unsigned long flags;
+       int ret;
+
+       spin_lock_irqsave(&r->consumer_lock, flags);
+       ret = __ptr_ring_consume_batched(r, array, n);
+       spin_unlock_irqrestore(&r->consumer_lock, flags);
+
+       return ret;
+}
+
+static inline int ptr_ring_consume_batched_bh(struct ptr_ring *r,
+                                             void **array, int n)
+{
+       int ret;
+
+       spin_lock_bh(&r->consumer_lock);
+       ret = __ptr_ring_consume_batched(r, array, n);
+       spin_unlock_bh(&r->consumer_lock);
+
+       return ret;
+}
+
 /* Cast to structure type and call a function without discarding from FIFO.
  * Function must return a value.
  * Callers must take consumer_lock.