Quelle  ptr_ring.h   Sprache: C

/* SPDX-License-Identifier: GPL-2.0-or-later */
/*
 * Definitions for the 'struct ptr_ring' datastructure.
 *
 * Author:
 *  Michael S. Tsirkin <mst@redhat.com>
 *
 * Copyright (C) 2016 Red Hat, Inc.
 *
 * This is a limited-size FIFO maintaining pointers in FIFO order, with
 * one CPU producing entries and another consuming entries from a FIFO.
 *
 * This implementation tries to minimize cache-contention when there is a
 * single producer and a single consumer CPU.
 */

#ifndef _LINUX_PTR_RING_H
#define _LINUX_PTR_RING_H 1

#ifdef __KERNEL__
#include <linux/spinlock.h>
#include <linux/cache.h>
#include <linux/types.h>
#include <linux/compiler.h>
#include <linux/slab.h>
#include <linux/mm.h>
#include <asm/errno.h>
#endif

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 */
 spinlock_t consumer_lock;
 /* Shared consumer/producer data */
 /* Read-only by both the producer and the consumer */
 int size ____cacheline_aligned_in_smp; /* max entries in queue */
 int batch; /* number of entries to consume in a batch */
 void **queue;
};

/* Note: callers invoking this in a loop must use a compiler barrier,
 * for example cpu_relax().
 *
 * NB: this is unlike __ptr_ring_empty in that callers must hold producer_lock:
 * see e.g. ptr_ring_full.
 */
static inline bool __ptr_ring_full(struct ptr_ring *r)
{
 return r->queue[r->producer];
}

static inline bool ptr_ring_full(struct ptr_ring *r)
{
 bool ret;

 spin_lock(&r->producer_lock);
 ret = __ptr_ring_full(r);
 spin_unlock(&r->producer_lock);

 return ret;
}

static inline bool ptr_ring_full_irq(struct ptr_ring *r)
{
 bool ret;

 spin_lock_irq(&r->producer_lock);
 ret = __ptr_ring_full(r);
 spin_unlock_irq(&r->producer_lock);

 return ret;
}

static inline bool ptr_ring_full_any(struct ptr_ring *r)
{
 unsigned long flags;
 bool ret;

 spin_lock_irqsave(&r->producer_lock, flags);
 ret = __ptr_ring_full(r);
 spin_unlock_irqrestore(&r->producer_lock, flags);

 return ret;
}

static inline bool ptr_ring_full_bh(struct ptr_ring *r)
{
 bool ret;

 spin_lock_bh(&r->producer_lock);
 ret = __ptr_ring_full(r);
 spin_unlock_bh(&r->producer_lock);

 return ret;
}

/* Note: callers invoking this in a loop must use a compiler barrier,
 * for example cpu_relax(). Callers must hold producer_lock.
 * Callers are responsible for making sure pointer that is being queued
 * points to a valid data.
 */
static inline int __ptr_ring_produce(struct ptr_ring *r, void *ptr)
{
 if (unlikely(!r->size) || r->queue[r->producer])
  return -ENOSPC;

 /* Make sure the pointer we are storing points to a valid data. */
 /* Pairs with the dependency ordering in __ptr_ring_consume. */
 smp_wmb();

 WRITE_ONCE(r->queue[r->producer++], ptr);
 if (unlikely(r->producer >= r->size))
  r->producer = 0;
 return 0;
}

/*
 * 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.
 */
static inline int ptr_ring_produce(struct ptr_ring *r, void *ptr)
{
 int ret;

 spin_lock(&r->producer_lock);
 ret = __ptr_ring_produce(r, ptr);
 spin_unlock(&r->producer_lock);

 return ret;
}

static inline int ptr_ring_produce_irq(struct ptr_ring *r, void *ptr)
{
 int ret;

 spin_lock_irq(&r->producer_lock);
 ret = __ptr_ring_produce(r, ptr);
 spin_unlock_irq(&r->producer_lock);

 return ret;
}

static inline int ptr_ring_produce_any(struct ptr_ring *r, void *ptr)
{
 unsigned long flags;
 int ret;

 spin_lock_irqsave(&r->producer_lock, flags);
 ret = __ptr_ring_produce(r, ptr);
 spin_unlock_irqrestore(&r->producer_lock, flags);

 return ret;
}

static inline int ptr_ring_produce_bh(struct ptr_ring *r, void *ptr)
{
 int ret;

 spin_lock_bh(&r->producer_lock);
 ret = __ptr_ring_produce(r, ptr);
 spin_unlock_bh(&r->producer_lock);

 return ret;
}

static inline void *__ptr_ring_peek(struct ptr_ring *r)
{
 if (likely(r->size))
  return READ_ONCE(r->queue[r->consumer_head]);
 return NULL;
}

/*
 * Test ring empty status without taking any locks.
 *
 * NB: This is only safe to call if ring is never resized.
 *
 * However, if some other CPU consumes ring entries at the same time, the value
 * returned is not guaranteed to be correct.
 *
 * In this case - to avoid incorrectly detecting the ring
 * as empty - the CPU consuming the ring entries is responsible
 * for either consuming all ring entries until the ring is empty,
 * or synchronizing with some other CPU and causing it to
 * re-test __ptr_ring_empty and/or consume the ring enteries
 * after the synchronization point.
 *
 * Note: callers invoking this in a loop must use a compiler barrier,
 * for example cpu_relax().
 */
static inline bool __ptr_ring_empty(struct ptr_ring *r)
{
 if (likely(r->size))
  return !r->queue[READ_ONCE(r->consumer_head)];
 return true;
}

static inline bool ptr_ring_empty(struct ptr_ring *r)
{
 bool ret;

 spin_lock(&r->consumer_lock);
 ret = __ptr_ring_empty(r);
 spin_unlock(&r->consumer_lock);

 return ret;
}

static inline bool ptr_ring_empty_irq(struct ptr_ring *r)
{
 bool ret;

 spin_lock_irq(&r->consumer_lock);
 ret = __ptr_ring_empty(r);
 spin_unlock_irq(&r->consumer_lock);

 return ret;
}

static inline bool ptr_ring_empty_any(struct ptr_ring *r)
{
 unsigned long flags;
 bool ret;

 spin_lock_irqsave(&r->consumer_lock, flags);
 ret = __ptr_ring_empty(r);
 spin_unlock_irqrestore(&r->consumer_lock, flags);

 return ret;
}

static inline bool ptr_ring_empty_bh(struct ptr_ring *r)
{
 bool ret;

 spin_lock_bh(&r->consumer_lock);
 ret = __ptr_ring_empty(r);
 spin_unlock_bh(&r->consumer_lock);

 return ret;
}

/* Must only be called after __ptr_ring_peek returned !NULL */
static inline void __ptr_ring_discard_one(struct ptr_ring *r)
{
 /* Fundamentally, what we want to do is update consumer
  * index and zero out the entry so producer can reuse it.
  * Doing it naively at each consume would be as simple as:
  *       consumer = r->consumer;
  *       r->queue[consumer++] = NULL;
  *       if (unlikely(consumer >= r->size))
  *               consumer = 0;
  *       r->consumer = consumer;
  * but that is suboptimal when the ring is full as producer is writing
  * out new entries in the same cache line.  Defer these updates until a
  * batch of entries has been consumed.
 */
 /* Note: we must keep consumer_head valid at all times for __ptr_ring_empty
  * to work correctly.
 */
 int consumer_head = r->consumer_head;
 int head = consumer_head++;

 /* Once we have processed enough entries invalidate them in
  * the ring all at once so producer can reuse their space in the ring.
  * We also do this when we reach end of the ring - not mandatory
  * but helps keep the implementation simple.
 */
 if (unlikely(consumer_head - r->consumer_tail >= r->batch ||
       consumer_head >= r->size)) {
  /* Zero out entries in the reverse order: this way we touch the
   * cache line that producer might currently be reading the last;
   * producer won't make progress and touch other cache lines
   * besides the first one until we write out all entries.
 */
  while (likely(head >= r->consumer_tail))
   r->queue[head--] = NULL;
  r->consumer_tail = consumer_head;
 }
 if (unlikely(consumer_head >= r->size)) {
  consumer_head = 0;
  r->consumer_tail = 0;
 }
 /* matching READ_ONCE in __ptr_ring_empty for lockless tests */
 WRITE_ONCE(r->consumer_head, consumer_head);
}

static inline void *__ptr_ring_consume(struct ptr_ring *r)
{
 void *ptr;

 /* The READ_ONCE in __ptr_ring_peek guarantees that anyone
  * accessing data through the pointer is up to date. Pairs
  * with smp_wmb in __ptr_ring_produce.
 */
 ptr = __ptr_ring_peek(r);
 if (ptr)
  __ptr_ring_discard_one(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
 * producing.
 */
static inline void *ptr_ring_consume(struct ptr_ring *r)
{
 void *ptr;

 spin_lock(&r->consumer_lock);
 ptr = __ptr_ring_consume(r);
 spin_unlock(&r->consumer_lock);

 return ptr;
}

static inline void *ptr_ring_consume_irq(struct ptr_ring *r)
{
 void *ptr;

 spin_lock_irq(&r->consumer_lock);
 ptr = __ptr_ring_consume(r);
 spin_unlock_irq(&r->consumer_lock);

 return ptr;
}

static inline void *ptr_ring_consume_any(struct ptr_ring *r)
{
 unsigned long flags;
 void *ptr;

 spin_lock_irqsave(&r->consumer_lock, flags);
 ptr = __ptr_ring_consume(r);
 spin_unlock_irqrestore(&r->consumer_lock, flags);

 return ptr;
}

static inline void *ptr_ring_consume_bh(struct ptr_ring *r)
{
 void *ptr;

 spin_lock_bh(&r->consumer_lock);
 ptr = __ptr_ring_consume(r);
 spin_unlock_bh(&r->consumer_lock);

 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.
 */
#define __PTR_RING_PEEK_CALL(r, f) ((f)(__ptr_ring_peek(r)))

#define PTR_RING_PEEK_CALL(r, f) ({ \
 typeof((f)(NULL)) __PTR_RING_PEEK_CALL_v; \
 \
 spin_lock(&(r)->consumer_lock); \
 __PTR_RING_PEEK_CALL_v = __PTR_RING_PEEK_CALL(r, f); \
 spin_unlock(&(r)->consumer_lock); \
 __PTR_RING_PEEK_CALL_v; \
})

#define PTR_RING_PEEK_CALL_IRQ(r, f) ({ \
 typeof((f)(NULL)) __PTR_RING_PEEK_CALL_v; \
 \
 spin_lock_irq(&(r)->consumer_lock); \
 __PTR_RING_PEEK_CALL_v = __PTR_RING_PEEK_CALL(r, f); \
 spin_unlock_irq(&(r)->consumer_lock); \
 __PTR_RING_PEEK_CALL_v; \
})

#define PTR_RING_PEEK_CALL_BH(r, f) ({ \
 typeof((f)(NULL)) __PTR_RING_PEEK_CALL_v; \
 \
 spin_lock_bh(&(r)->consumer_lock); \
 __PTR_RING_PEEK_CALL_v = __PTR_RING_PEEK_CALL(r, f); \
 spin_unlock_bh(&(r)->consumer_lock); \
 __PTR_RING_PEEK_CALL_v; \
})

#define PTR_RING_PEEK_CALL_ANY(r, f) ({ \
 typeof((f)(NULL)) __PTR_RING_PEEK_CALL_v; \
 unsigned long __PTR_RING_PEEK_CALL_f;\
 \
 spin_lock_irqsave(&(r)->consumer_lock, __PTR_RING_PEEK_CALL_f); \
 __PTR_RING_PEEK_CALL_v = __PTR_RING_PEEK_CALL(r, f); \
 spin_unlock_irqrestore(&(r)->consumer_lock, __PTR_RING_PEEK_CALL_f); \
 __PTR_RING_PEEK_CALL_v; \
})

/* Not all gfp_t flags (besides GFP_KERNEL) are allowed. See
 * documentation for vmalloc for which of them are legal.
 */
static inline void **__ptr_ring_init_queue_alloc_noprof(unsigned int size, gfp_t gfp)
{
 if (size > KMALLOC_MAX_SIZE / sizeof(void *))
  return NULL;
 return kvmalloc_array_noprof(size, sizeof(void *), gfp | __GFP_ZERO);
}

static inline void __ptr_ring_set_size(struct ptr_ring *r, int size)
{
 r->size = size;
 r->batch = SMP_CACHE_BYTES * 2 / sizeof(*(r->queue));
 /* We need to set batch at least to 1 to make logic
  * in __ptr_ring_discard_one work correctly.
  * Batching too much (because ring is small) would cause a lot of
  * burstiness. Needs tuning, for now disable batching.
 */
 if (r->batch > r->size / 2 || !r->batch)
  r->batch = 1;
}

static inline int ptr_ring_init_noprof(struct ptr_ring *r, int size, gfp_t gfp)
{
 r->queue = __ptr_ring_init_queue_alloc_noprof(size, gfp);
 if (!r->queue)
  return -ENOMEM;

 __ptr_ring_set_size(r, size);
 r->producer = r->consumer_head = r->consumer_tail = 0;
 spin_lock_init(&r->producer_lock);
 spin_lock_init(&r->consumer_lock);

 return 0;
}
#define ptr_ring_init(...) alloc_hooks(ptr_ring_init_noprof(__VA_ARGS__))

/*
 * Return entries into ring. Destroy entries that don't fit.
 *
 * Note: this is expected to be a rare slow path operation.
 *
 * Note: producer lock is nested within consumer lock, so if you
 * resize you must make sure all uses nest correctly.
 * In particular if you consume ring in interrupt or BH context, you must
 * disable interrupts/BH when doing so.
 */
static inline void ptr_ring_unconsume(struct ptr_ring *r, void **batch, int n,
          void (*destroy)(void *))
{
 unsigned long flags;
 int head;

 spin_lock_irqsave(&r->consumer_lock, flags);
 spin_lock(&r->producer_lock);

 if (!r->size)
  goto done;

 /*
  * Clean out buffered entries (for simplicity). This way following code
  * can test entries for NULL and if not assume they are valid.
 */
 head = r->consumer_head - 1;
 while (likely(head >= r->consumer_tail))
  r->queue[head--] = NULL;
 r->consumer_tail = r->consumer_head;

 /*
  * Go over entries in batch, start moving head back and copy entries.
  * Stop when we run into previously unconsumed entries.
 */
 while (n) {
  head = r->consumer_head - 1;
  if (head < 0)
   head = r->size - 1;
  if (r->queue[head]) {
   /* This batch entry will have to be destroyed. */
   goto done;
  }
  r->queue[head] = batch[--n];
  r->consumer_tail = head;
  /* matching READ_ONCE in __ptr_ring_empty for lockless tests */
  WRITE_ONCE(r->consumer_head, head);
 }

done:
 /* Destroy all entries left in the batch. */
 while (n)
  destroy(batch[--n]);
 spin_unlock(&r->producer_lock);
 spin_unlock_irqrestore(&r->consumer_lock, flags);
}

static inline void **__ptr_ring_swap_queue(struct ptr_ring *r, void **queue,
        int size, gfp_t gfp,
        void (*destroy)(void *))
{
 int producer = 0;
 void **old;
 void *ptr;

 while ((ptr = __ptr_ring_consume(r)))
  if (producer < size)
   queue[producer++] = ptr;
  else if (destroy)
   destroy(ptr);

 if (producer >= size)
  producer = 0;
 __ptr_ring_set_size(r, size);
 r->producer = producer;
 r->consumer_head = 0;
 r->consumer_tail = 0;
 old = r->queue;
 r->queue = queue;

 return old;
}

/*
 * Note: producer lock is nested within consumer lock, so if you
 * resize you must make sure all uses nest correctly.
 * In particular if you consume ring in interrupt or BH context, you must
 * disable interrupts/BH when doing so.
 */
static inline int ptr_ring_resize_noprof(struct ptr_ring *r, int size, gfp_t gfp,
      void (*destroy)(void *))
{
 unsigned long flags;
 void **queue = __ptr_ring_init_queue_alloc_noprof(size, gfp);
 void **old;

 if (!queue)
  return -ENOMEM;

 spin_lock_irqsave(&(r)->consumer_lock, flags);
 spin_lock(&(r)->producer_lock);

 old = __ptr_ring_swap_queue(r, queue, size, gfp, destroy);

 spin_unlock(&(r)->producer_lock);
 spin_unlock_irqrestore(&(r)->consumer_lock, flags);

 kvfree(old);

 return 0;
}
#define ptr_ring_resize(...) alloc_hooks(ptr_ring_resize_noprof(__VA_ARGS__))

/*
 * Note: producer lock is nested within consumer lock, so if you
 * resize you must make sure all uses nest correctly.
 * In particular if you consume ring in BH context, you must
 * disable BH when doing so.
 */
static inline int ptr_ring_resize_multiple_bh_noprof(struct ptr_ring **rings,
           unsigned int nrings,
           int size, gfp_t gfp,
           void (*destroy)(void *))
{
 void ***queues;
 int i;

 queues = kmalloc_array_noprof(nrings, sizeof(*queues), gfp);
 if (!queues)
  goto noqueues;

 for (i = 0; i < nrings; ++i) {
  queues[i] = __ptr_ring_init_queue_alloc_noprof(size, gfp);
  if (!queues[i])
   goto nomem;
 }

 for (i = 0; i < nrings; ++i) {
  spin_lock_bh(&(rings[i])->consumer_lock);
  spin_lock(&(rings[i])->producer_lock);
  queues[i] = __ptr_ring_swap_queue(rings[i], queues[i],
        size, gfp, destroy);
  spin_unlock(&(rings[i])->producer_lock);
  spin_unlock_bh(&(rings[i])->consumer_lock);
 }

 for (i = 0; i < nrings; ++i)
  kvfree(queues[i]);

 kfree(queues);

 return 0;

nomem:
 while (--i >= 0)
  kvfree(queues[i]);

 kfree(queues);

noqueues:
 return -ENOMEM;
}
#define ptr_ring_resize_multiple_bh(...) \
  alloc_hooks(ptr_ring_resize_multiple_bh_noprof(__VA_ARGS__))

static inline void ptr_ring_cleanup(struct ptr_ring *r, void (*destroy)(void *))
{
 void *ptr;

 if (destroy)
  while ((ptr = ptr_ring_consume(r)))
   destroy(ptr);
 kvfree(r->queue);
}

#endif /* _LINUX_PTR_RING_H  */