[mirror_ubuntu-bionic-kernel.git] / include / linux / ptr_ring.h

/*
 *	Definitions for the 'struct ptr_ring' datastructure.
 *
 *	Author:
 *		Michael S. Tsirkin <mst@redhat.com>
 *
 *	Copyright (C) 2016 Red Hat, Inc.
 *
 *	This program is free software; you can redistribute it and/or modify it
 *	under the terms of the GNU General Public License as published by the
 *	Free Software Foundation; either version 2 of the License, or (at your
 *	option) any later version.
 *
 *	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/cache.h>
#include <linux/slab.h>
#include <asm/errno.h>
#endif

struct ptr_ring {
	int producer ____cacheline_aligned_in_smp;
	spinlock_t producer_lock;
	int consumer ____cacheline_aligned_in_smp;
	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 */
	void **queue;
};

/* Note: callers invoking this in a loop must use a compiler barrier,
 * for example cpu_relax().  If ring is ever resized, callers must hold
 * producer_lock - see e.g. ptr_ring_full.  Otherwise, if callers don't hold
 * producer_lock, the next call to __ptr_ring_produce may fail.
 */
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.
 */
static inline int __ptr_ring_produce(struct ptr_ring *r, void *ptr)
{
	if (unlikely(!r->size) || r->queue[r->producer])
		return -ENOSPC;

	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;
}

/* Note: callers invoking this in a loop must use a compiler barrier,
 * for example cpu_relax(). Callers must take consumer_lock
 * if they dereference the pointer - see e.g. PTR_RING_PEEK_CALL.
 * If ring is never resized, and if the pointer is merely
 * tested, there's no need to take the lock - see e.g.  __ptr_ring_empty.
 */
static inline void *__ptr_ring_peek(struct ptr_ring *r)
{
	if (likely(r->size))
		return r->queue[r->consumer];
	return NULL;
}

/* Note: callers invoking this in a loop must use a compiler barrier,
 * for example cpu_relax(). Callers must take consumer_lock
 * if the ring is ever resized - see e.g. ptr_ring_empty.
 */
static inline bool __ptr_ring_empty(struct ptr_ring *r)
{
	return !__ptr_ring_peek(r);
}

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)
{
	r->queue[r->consumer++] = NULL;
	if (unlikely(r->consumer >= r->size))
		r->consumer = 0;
}

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

	ptr = __ptr_ring_peek(r);
	if (ptr)
		__ptr_ring_discard_one(r);

	return ptr;
}

/*
 * 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;
}

/* 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; \
})

static inline void **__ptr_ring_init_queue_alloc(int size, gfp_t gfp)
{
	return kzalloc(ALIGN(size * sizeof(void *), SMP_CACHE_BYTES), gfp);
}

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

	r->size = size;
	r->producer = r->consumer = 0;
	spin_lock_init(&r->producer_lock);
	spin_lock_init(&r->consumer_lock);

	return 0;
}

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);

	r->size = size;
	r->producer = producer;
	r->consumer = 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(struct ptr_ring *r, int size, gfp_t gfp,
				  void (*destroy)(void *))
{
	unsigned long flags;
	void **queue = __ptr_ring_init_queue_alloc(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);

	kfree(old);

	return 0;
}

/*
 * 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_multiple(struct ptr_ring **rings, int nrings,
					   int size,
					   gfp_t gfp, void (*destroy)(void *))
{
	unsigned long flags;
	void ***queues;
	int i;

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

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

	for (i = 0; i < nrings; ++i) {
		spin_lock_irqsave(&(rings[i])->consumer_lock, flags);
		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_irqrestore(&(rings[i])->consumer_lock, flags);
	}

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

	kfree(queues);

	return 0;

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

	kfree(queues);

noqueues:
	return -ENOMEM;
}

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);
	kfree(r->queue);
}

#endif /* _LINUX_PTR_RING_H  */
Commit	Line	Data
2e0ab8ca MT	1	/*
	2	* Definitions for the 'struct ptr_ring' datastructure.
	3	*
	4	* Author:
	5	* Michael S. Tsirkin <mst@redhat.com>
	6	*
	7	* Copyright (C) 2016 Red Hat, Inc.
	8	*
	9	* This program is free software; you can redistribute it and/or modify it
	10	* under the terms of the GNU General Public License as published by the
	11	* Free Software Foundation; either version 2 of the License, or (at your
	12	* option) any later version.
	13	*
	14	* This is a limited-size FIFO maintaining pointers in FIFO order, with
	15	* one CPU producing entries and another consuming entries from a FIFO.
	16	*
	17	* This implementation tries to minimize cache-contention when there is a
	18	* single producer and a single consumer CPU.
	19	*/
	20
	21	#ifndef _LINUX_PTR_RING_H
	22	#define _LINUX_PTR_RING_H 1
	23
	24	#ifdef __KERNEL__
	25	#include <linux/spinlock.h>
	26	#include <linux/cache.h>
	27	#include <linux/types.h>
	28	#include <linux/compiler.h>
	29	#include <linux/cache.h>
	30	#include <linux/slab.h>
	31	#include <asm/errno.h>
	32	#endif
	33
	34	struct ptr_ring {
	35	int producer ____cacheline_aligned_in_smp;
	36	spinlock_t producer_lock;
	37	int consumer ____cacheline_aligned_in_smp;
	38	spinlock_t consumer_lock;
	39	/* Shared consumer/producer data */
	40	/* Read-only by both the producer and the consumer */
	41	int size ____cacheline_aligned_in_smp; /* max entries in queue */
	42	void **queue;
	43	};
	44
	45	/* Note: callers invoking this in a loop must use a compiler barrier,
5d49de53 MT	46	* for example cpu_relax(). If ring is ever resized, callers must hold
	47	* producer_lock - see e.g. ptr_ring_full. Otherwise, if callers don't hold
	48	* producer_lock, the next call to __ptr_ring_produce may fail.
2e0ab8ca MT	49	*/
	50	static inline bool __ptr_ring_full(struct ptr_ring *r)
	51	{
	52	return r->queue[r->producer];
	53	}
	54
	55	static inline bool ptr_ring_full(struct ptr_ring *r)
	56	{
5d49de53 MT	57	bool ret;
	58
	59	spin_lock(&r->producer_lock);
	60	ret = __ptr_ring_full(r);
	61	spin_unlock(&r->producer_lock);
	62
	63	return ret;
	64	}
	65
	66	static inline bool ptr_ring_full_irq(struct ptr_ring *r)
	67	{
	68	bool ret;
	69
	70	spin_lock_irq(&r->producer_lock);
	71	ret = __ptr_ring_full(r);
	72	spin_unlock_irq(&r->producer_lock);
	73
	74	return ret;
	75	}
	76
	77	static inline bool ptr_ring_full_any(struct ptr_ring *r)
	78	{
	79	unsigned long flags;
	80	bool ret;
	81
	82	spin_lock_irqsave(&r->producer_lock, flags);
	83	ret = __ptr_ring_full(r);
	84	spin_unlock_irqrestore(&r->producer_lock, flags);
	85
	86	return ret;
	87	}
	88
	89	static inline bool ptr_ring_full_bh(struct ptr_ring *r)
	90	{
	91	bool ret;
	92
	93	spin_lock_bh(&r->producer_lock);
	94	ret = __ptr_ring_full(r);
	95	spin_unlock_bh(&r->producer_lock);
	96
	97	return ret;
2e0ab8ca MT	98	}
	99
	100	/* Note: callers invoking this in a loop must use a compiler barrier,
5d49de53	101	* for example cpu_relax(). Callers must hold producer_lock.
2e0ab8ca MT	102	*/
	103	static inline int __ptr_ring_produce(struct ptr_ring r, void ptr)
	104	{
982fb490	105	if (unlikely(!r->size) \|\| r->queue[r->producer])
2e0ab8ca MT	106	return -ENOSPC;
	107
	108	r->queue[r->producer++] = ptr;
	109	if (unlikely(r->producer >= r->size))
	110	r->producer = 0;
	111	return 0;
	112	}
	113
e7169530 MT	114	/*
	115	* Note: resize (below) nests producer lock within consumer lock, so if you
	116	* consume in interrupt or BH context, you must disable interrupts/BH when
	117	* calling this.
	118	*/
2e0ab8ca MT	119	static inline int ptr_ring_produce(struct ptr_ring r, void ptr)
	120	{
	121	int ret;
	122
	123	spin_lock(&r->producer_lock);
	124	ret = __ptr_ring_produce(r, ptr);
	125	spin_unlock(&r->producer_lock);
	126
	127	return ret;
	128	}
	129
	130	static inline int ptr_ring_produce_irq(struct ptr_ring r, void ptr)
	131	{
	132	int ret;
	133
	134	spin_lock_irq(&r->producer_lock);
	135	ret = __ptr_ring_produce(r, ptr);
	136	spin_unlock_irq(&r->producer_lock);
	137
	138	return ret;
	139	}
	140
	141	static inline int ptr_ring_produce_any(struct ptr_ring r, void ptr)
	142	{
	143	unsigned long flags;
	144	int ret;
	145
	146	spin_lock_irqsave(&r->producer_lock, flags);
	147	ret = __ptr_ring_produce(r, ptr);
	148	spin_unlock_irqrestore(&r->producer_lock, flags);
	149
	150	return ret;
	151	}
	152
	153	static inline int ptr_ring_produce_bh(struct ptr_ring r, void ptr)
	154	{
	155	int ret;
	156
	157	spin_lock_bh(&r->producer_lock);
	158	ret = __ptr_ring_produce(r, ptr);
	159	spin_unlock_bh(&r->producer_lock);
	160
	161	return ret;
	162	}
	163
	164	/* Note: callers invoking this in a loop must use a compiler barrier,
	165	* for example cpu_relax(). Callers must take consumer_lock
	166	* if they dereference the pointer - see e.g. PTR_RING_PEEK_CALL.
5d49de53 MT	167	* If ring is never resized, and if the pointer is merely
5d49de53 MT	168	* tested, there's no need to take the lock - see e.g. __ptr_ring_empty.
2e0ab8ca MT	169	*/
	170	static inline void __ptr_ring_peek(struct ptr_ring r)
	171	{
982fb490 JW	172	if (likely(r->size))
	173	return r->queue[r->consumer];
	174	return NULL;
2e0ab8ca MT	175	}
2e0ab8ca MT	176
5d49de53 MT	177	/* Note: callers invoking this in a loop must use a compiler barrier,
	178	* for example cpu_relax(). Callers must take consumer_lock
	179	* if the ring is ever resized - see e.g. ptr_ring_empty.
	180	*/
	181	static inline bool __ptr_ring_empty(struct ptr_ring *r)
2e0ab8ca	182	{
2e0ab8ca MT	183	return !__ptr_ring_peek(r);
	184	}
	185
5d49de53 MT	186	static inline bool ptr_ring_empty(struct ptr_ring *r)
	187	{
	188	bool ret;
	189
	190	spin_lock(&r->consumer_lock);
	191	ret = __ptr_ring_empty(r);
	192	spin_unlock(&r->consumer_lock);
	193
	194	return ret;
	195	}
	196
	197	static inline bool ptr_ring_empty_irq(struct ptr_ring *r)
	198	{
	199	bool ret;
	200
	201	spin_lock_irq(&r->consumer_lock);
	202	ret = __ptr_ring_empty(r);
	203	spin_unlock_irq(&r->consumer_lock);
	204
	205	return ret;
	206	}
	207
	208	static inline bool ptr_ring_empty_any(struct ptr_ring *r)
	209	{
	210	unsigned long flags;
	211	bool ret;
	212
	213	spin_lock_irqsave(&r->consumer_lock, flags);
	214	ret = __ptr_ring_empty(r);
	215	spin_unlock_irqrestore(&r->consumer_lock, flags);
	216
	217	return ret;
	218	}
	219
	220	static inline bool ptr_ring_empty_bh(struct ptr_ring *r)
	221	{
	222	bool ret;
	223
	224	spin_lock_bh(&r->consumer_lock);
	225	ret = __ptr_ring_empty(r);
	226	spin_unlock_bh(&r->consumer_lock);
	227
	228	return ret;
	229	}
	230
2e0ab8ca MT	231	/* Must only be called after __ptr_ring_peek returned !NULL */
	232	static inline void __ptr_ring_discard_one(struct ptr_ring *r)
	233	{
	234	r->queue[r->consumer++] = NULL;
	235	if (unlikely(r->consumer >= r->size))
	236	r->consumer = 0;
	237	}
	238
	239	static inline void __ptr_ring_consume(struct ptr_ring r)
	240	{
	241	void *ptr;
	242
	243	ptr = __ptr_ring_peek(r);
	244	if (ptr)
	245	__ptr_ring_discard_one(r);
	246
	247	return ptr;
	248	}
	249
e7169530 MT	250	/*
	251	* Note: resize (below) nests producer lock within consumer lock, so if you
	252	* call this in interrupt or BH context, you must disable interrupts/BH when
	253	* producing.
	254	*/
2e0ab8ca MT	255	static inline void ptr_ring_consume(struct ptr_ring r)
	256	{
	257	void *ptr;
	258
	259	spin_lock(&r->consumer_lock);
	260	ptr = __ptr_ring_consume(r);
	261	spin_unlock(&r->consumer_lock);
	262
	263	return ptr;
	264	}
	265
	266	static inline void ptr_ring_consume_irq(struct ptr_ring r)
	267	{
	268	void *ptr;
	269
	270	spin_lock_irq(&r->consumer_lock);
	271	ptr = __ptr_ring_consume(r);
	272	spin_unlock_irq(&r->consumer_lock);
	273
	274	return ptr;
	275	}
	276
	277	static inline void ptr_ring_consume_any(struct ptr_ring r)
	278	{
	279	unsigned long flags;
	280	void *ptr;
	281
	282	spin_lock_irqsave(&r->consumer_lock, flags);
	283	ptr = __ptr_ring_consume(r);
	284	spin_unlock_irqrestore(&r->consumer_lock, flags);
	285
	286	return ptr;
	287	}
	288
	289	static inline void ptr_ring_consume_bh(struct ptr_ring r)
	290	{
	291	void *ptr;
	292
	293	spin_lock_bh(&r->consumer_lock);
	294	ptr = __ptr_ring_consume(r);
	295	spin_unlock_bh(&r->consumer_lock);
	296
	297	return ptr;
	298	}
	299
	300	/* Cast to structure type and call a function without discarding from FIFO.
	301	* Function must return a value.
	302	* Callers must take consumer_lock.
	303	*/
	304	#define __PTR_RING_PEEK_CALL(r, f) ((f)(__ptr_ring_peek(r)))
	305
	306	#define PTR_RING_PEEK_CALL(r, f) ({ \
	307	typeof((f)(NULL)) __PTR_RING_PEEK_CALL_v; \
	308	\
	309	spin_lock(&(r)->consumer_lock); \
	310	__PTR_RING_PEEK_CALL_v = __PTR_RING_PEEK_CALL(r, f); \
	311	spin_unlock(&(r)->consumer_lock); \
	312	__PTR_RING_PEEK_CALL_v; \
	313	})
	314
	315	#define PTR_RING_PEEK_CALL_IRQ(r, f) ({ \
	316	typeof((f)(NULL)) __PTR_RING_PEEK_CALL_v; \
	317	\
	318	spin_lock_irq(&(r)->consumer_lock); \
319	__PTR_RING_PEEK_CALL_v = __PTR_RING_PEEK_CALL(r, f); \
320	spin_unlock_irq(&(r)->consumer_lock); \
321	__PTR_RING_PEEK_CALL_v; \
322	})
323
324	#define PTR_RING_PEEK_CALL_BH(r, f) ({ \
325	typeof((f)(NULL)) __PTR_RING_PEEK_CALL_v; \
326	\
327	spin_lock_bh(&(r)->consumer_lock); \
328	__PTR_RING_PEEK_CALL_v = __PTR_RING_PEEK_CALL(r, f); \
329	spin_unlock_bh(&(r)->consumer_lock); \
330	__PTR_RING_PEEK_CALL_v; \
331	})
332
333	#define PTR_RING_PEEK_CALL_ANY(r, f) ({ \
334	typeof((f)(NULL)) __PTR_RING_PEEK_CALL_v; \
335	unsigned long __PTR_RING_PEEK_CALL_f;\
336	\
337	spin_lock_irqsave(&(r)->consumer_lock, __PTR_RING_PEEK_CALL_f); \
338	__PTR_RING_PEEK_CALL_v = __PTR_RING_PEEK_CALL(r, f); \
339	spin_unlock_irqrestore(&(r)->consumer_lock, __PTR_RING_PEEK_CALL_f); \
340	__PTR_RING_PEEK_CALL_v; \
341	})
342
5d49de53 MT	343	static inline void **__ptr_ring_init_queue_alloc(int size, gfp_t gfp)
	344	{
	345	return kzalloc(ALIGN(size * sizeof(void *), SMP_CACHE_BYTES), gfp);
	346	}
	347
2e0ab8ca MT	348	static inline int ptr_ring_init(struct ptr_ring *r, int size, gfp_t gfp)
2e0ab8ca MT	349	{
5d49de53	350	r->queue = __ptr_ring_init_queue_alloc(size, gfp);
2e0ab8ca MT	351	if (!r->queue)
	352	return -ENOMEM;
	353
	354	r->size = size;
	355	r->producer = r->consumer = 0;
	356	spin_lock_init(&r->producer_lock);
	357	spin_lock_init(&r->consumer_lock);
	358
	359	return 0;
	360	}
	361
59e6ae53 MT	362	static inline void *__ptr_ring_swap_queue(struct ptr_ring r, void **queue,
	363	int size, gfp_t gfp,
	364	void (destroy)(void ))
5d49de53	365	{
5d49de53	366	int producer = 0;
5d49de53 MT	367	void **old;
	368	void *ptr;
	369
e7169530	370	while ((ptr = __ptr_ring_consume(r)))
5d49de53 MT	371	if (producer < size)
	372	queue[producer++] = ptr;
	373	else if (destroy)
	374	destroy(ptr);
	375
	376	r->size = size;
	377	r->producer = producer;
	378	r->consumer = 0;
	379	old = r->queue;
	380	r->queue = queue;
	381
59e6ae53 MT	382	return old;
	383	}
	384
e7169530 MT	385	/*
	386	* Note: producer lock is nested within consumer lock, so if you
	387	* resize you must make sure all uses nest correctly.
	388	* In particular if you consume ring in interrupt or BH context, you must
	389	* disable interrupts/BH when doing so.
	390	*/
59e6ae53 MT	391	static inline int ptr_ring_resize(struct ptr_ring *r, int size, gfp_t gfp,
	392	void (destroy)(void ))
	393	{
	394	unsigned long flags;
	395	void **queue = __ptr_ring_init_queue_alloc(size, gfp);
	396	void **old;
	397
	398	if (!queue)
	399	return -ENOMEM;
	400
e7169530 MT	401	spin_lock_irqsave(&(r)->consumer_lock, flags);
e7169530 MT	402	spin_lock(&(r)->producer_lock);
59e6ae53 MT	403
	404	old = __ptr_ring_swap_queue(r, queue, size, gfp, destroy);
	405
e7169530 MT	406	spin_unlock(&(r)->producer_lock);
e7169530 MT	407	spin_unlock_irqrestore(&(r)->consumer_lock, flags);
5d49de53 MT	408
	409	kfree(old);
	410
	411	return 0;
	412	}
	413
e7169530 MT	414	/*
	415	* Note: producer lock is nested within consumer lock, so if you
	416	* resize you must make sure all uses nest correctly.
	417	* In particular if you consume ring in interrupt or BH context, you must
	418	* disable interrupts/BH when doing so.
	419	*/
59e6ae53 MT	420	static inline int ptr_ring_resize_multiple(struct ptr_ring **rings, int nrings,
	421	int size,
	422	gfp_t gfp, void (destroy)(void ))
	423	{
	424	unsigned long flags;
	425	void ***queues;
	426	int i;
	427
	428	queues = kmalloc(nrings * sizeof *queues, gfp);
	429	if (!queues)
	430	goto noqueues;
	431
	432	for (i = 0; i < nrings; ++i) {
	433	queues[i] = __ptr_ring_init_queue_alloc(size, gfp);
	434	if (!queues[i])
	435	goto nomem;
	436	}
	437
	438	for (i = 0; i < nrings; ++i) {
e7169530 MT	439	spin_lock_irqsave(&(rings[i])->consumer_lock, flags);
e7169530 MT	440	spin_lock(&(rings[i])->producer_lock);
59e6ae53 MT	441	queues[i] = __ptr_ring_swap_queue(rings[i], queues[i],
59e6ae53 MT	442	size, gfp, destroy);
e7169530 MT	443	spin_unlock(&(rings[i])->producer_lock);
e7169530 MT	444	spin_unlock_irqrestore(&(rings[i])->consumer_lock, flags);
59e6ae53 MT	445	}
	446
	447	for (i = 0; i < nrings; ++i)
	448	kfree(queues[i]);
	449
	450	kfree(queues);
	451
	452	return 0;
	453
	454	nomem:
	455	while (--i >= 0)
	456	kfree(queues[i]);
	457
	458	kfree(queues);
	459
	460	noqueues:
	461	return -ENOMEM;
	462	}
	463
5d49de53	464	static inline void ptr_ring_cleanup(struct ptr_ring r, void (destroy)(void *))
2e0ab8ca	465	{
5d49de53 MT	466	void *ptr;
	467
	468	if (destroy)
	469	while ((ptr = ptr_ring_consume(r)))
	470	destroy(ptr);
2e0ab8ca MT	471	kfree(r->queue);
	472	}
	473
	474	#endif /* _LINUX_PTR_RING_H */