[mirror_ubuntu-jammy-kernel.git] / net / xdp / xsk_queue.h

/* SPDX-License-Identifier: GPL-2.0 */
/* XDP user-space ring structure
 * Copyright(c) 2018 Intel Corporation.
 */

#ifndef _LINUX_XSK_QUEUE_H
#define _LINUX_XSK_QUEUE_H

#include <linux/types.h>
#include <linux/if_xdp.h>
#include <net/xdp_sock.h>
#include <net/xsk_buff_pool.h>

#include "xsk.h"

struct xdp_ring {
	u32 producer ____cacheline_aligned_in_smp;
	u32 consumer ____cacheline_aligned_in_smp;
	u32 flags;
};

/* Used for the RX and TX queues for packets */
struct xdp_rxtx_ring {
	struct xdp_ring ptrs;
	struct xdp_desc desc[] ____cacheline_aligned_in_smp;
};

/* Used for the fill and completion queues for buffers */
struct xdp_umem_ring {
	struct xdp_ring ptrs;
	u64 desc[] ____cacheline_aligned_in_smp;
};

struct xsk_queue {
	u32 ring_mask;
	u32 nentries;
	u32 cached_prod;
	u32 cached_cons;
	struct xdp_ring *ring;
	u64 invalid_descs;
	u64 queue_empty_descs;
};

/* The structure of the shared state of the rings are the same as the
 * ring buffer in kernel/events/ring_buffer.c. For the Rx and completion
 * ring, the kernel is the producer and user space is the consumer. For
 * the Tx and fill rings, the kernel is the consumer and user space is
 * the producer.
 *
 * producer                         consumer
 *
 * if (LOAD ->consumer) {           LOAD ->producer
 *                    (A)           smp_rmb()       (C)
 *    STORE $data                   LOAD $data
 *    smp_wmb()       (B)           smp_mb()        (D)
 *    STORE ->producer              STORE ->consumer
 * }
 *
 * (A) pairs with (D), and (B) pairs with (C).
 *
 * Starting with (B), it protects the data from being written after
 * the producer pointer. If this barrier was missing, the consumer
 * could observe the producer pointer being set and thus load the data
 * before the producer has written the new data. The consumer would in
 * this case load the old data.
 *
 * (C) protects the consumer from speculatively loading the data before
 * the producer pointer actually has been read. If we do not have this
 * barrier, some architectures could load old data as speculative loads
 * are not discarded as the CPU does not know there is a dependency
 * between ->producer and data.
 *
 * (A) is a control dependency that separates the load of ->consumer
 * from the stores of $data. In case ->consumer indicates there is no
 * room in the buffer to store $data we do not. So no barrier is needed.
 *
 * (D) protects the load of the data to be observed to happen after the
 * store of the consumer pointer. If we did not have this memory
 * barrier, the producer could observe the consumer pointer being set
 * and overwrite the data with a new value before the consumer got the
 * chance to read the old value. The consumer would thus miss reading
 * the old entry and very likely read the new entry twice, once right
 * now and again after circling through the ring.
 */

/* The operations on the rings are the following:
 *
 * producer                           consumer
 *
 * RESERVE entries                    PEEK in the ring for entries
 * WRITE data into the ring           READ data from the ring
 * SUBMIT entries                     RELEASE entries
 *
 * The producer reserves one or more entries in the ring. It can then
 * fill in these entries and finally submit them so that they can be
 * seen and read by the consumer.
 *
 * The consumer peeks into the ring to see if the producer has written
 * any new entries. If so, the producer can then read these entries
 * and when it is done reading them release them back to the producer
 * so that the producer can use these slots to fill in new entries.
 *
 * The function names below reflect these operations.
 */

/* Functions that read and validate content from consumer rings. */

static inline bool xskq_cons_read_addr_unchecked(struct xsk_queue *q, u64 *addr)
{
	struct xdp_umem_ring *ring = (struct xdp_umem_ring *)q->ring;

	if (q->cached_cons != q->cached_prod) {
		u32 idx = q->cached_cons & q->ring_mask;

		*addr = ring->desc[idx];
		return true;
	}

	return false;
}

static inline bool xp_aligned_validate_desc(struct xsk_buff_pool *pool,
					    struct xdp_desc *desc)
{
	u64 chunk, chunk_end;

	chunk = xp_aligned_extract_addr(pool, desc->addr);
	chunk_end = xp_aligned_extract_addr(pool, desc->addr + desc->len);
	if (chunk != chunk_end)
		return false;

	if (chunk >= pool->addrs_cnt)
		return false;

	if (desc->options)
		return false;
	return true;
}

static inline bool xp_unaligned_validate_desc(struct xsk_buff_pool *pool,
					      struct xdp_desc *desc)
{
	u64 addr, base_addr;

	base_addr = xp_unaligned_extract_addr(desc->addr);
	addr = xp_unaligned_add_offset_to_addr(desc->addr);

	if (desc->len > pool->chunk_size)
		return false;

	if (base_addr >= pool->addrs_cnt || addr >= pool->addrs_cnt ||
	    xp_desc_crosses_non_contig_pg(pool, addr, desc->len))
		return false;

	if (desc->options)
		return false;
	return true;
}

static inline bool xp_validate_desc(struct xsk_buff_pool *pool,
				    struct xdp_desc *desc)
{
	return pool->unaligned ? xp_unaligned_validate_desc(pool, desc) :
		xp_aligned_validate_desc(pool, desc);
}

static inline bool xskq_cons_is_valid_desc(struct xsk_queue *q,
					   struct xdp_desc *d,
					   struct xdp_umem *umem)
{
	if (!xp_validate_desc(umem->pool, d)) {
		q->invalid_descs++;
		return false;
	}
	return true;
}

static inline bool xskq_cons_read_desc(struct xsk_queue *q,
				       struct xdp_desc *desc,
				       struct xdp_umem *umem)
{
	while (q->cached_cons != q->cached_prod) {
		struct xdp_rxtx_ring *ring = (struct xdp_rxtx_ring *)q->ring;
		u32 idx = q->cached_cons & q->ring_mask;

		*desc = ring->desc[idx];
		if (xskq_cons_is_valid_desc(q, desc, umem))
			return true;

		q->cached_cons++;
	}

	return false;
}

/* Functions for consumers */

static inline void __xskq_cons_release(struct xsk_queue *q)
{
	smp_mb(); /* D, matches A */
	WRITE_ONCE(q->ring->consumer, q->cached_cons);
}

static inline void __xskq_cons_peek(struct xsk_queue *q)
{
	/* Refresh the local pointer */
	q->cached_prod = READ_ONCE(q->ring->producer);
	smp_rmb(); /* C, matches B */
}

static inline void xskq_cons_get_entries(struct xsk_queue *q)
{
	__xskq_cons_release(q);
	__xskq_cons_peek(q);
}

static inline bool xskq_cons_has_entries(struct xsk_queue *q, u32 cnt)
{
	u32 entries = q->cached_prod - q->cached_cons;

	if (entries >= cnt)
		return true;

	__xskq_cons_peek(q);
	entries = q->cached_prod - q->cached_cons;

	return entries >= cnt;
}

static inline bool xskq_cons_peek_addr_unchecked(struct xsk_queue *q, u64 *addr)
{
	if (q->cached_prod == q->cached_cons)
		xskq_cons_get_entries(q);
	return xskq_cons_read_addr_unchecked(q, addr);
}

static inline bool xskq_cons_peek_desc(struct xsk_queue *q,
				       struct xdp_desc *desc,
				       struct xdp_umem *umem)
{
	if (q->cached_prod == q->cached_cons)
		xskq_cons_get_entries(q);
	return xskq_cons_read_desc(q, desc, umem);
}

static inline void xskq_cons_release(struct xsk_queue *q)
{
	/* To improve performance, only update local state here.
	 * Reflect this to global state when we get new entries
	 * from the ring in xskq_cons_get_entries() and whenever
	 * Rx or Tx processing are completed in the NAPI loop.
	 */
	q->cached_cons++;
}

static inline bool xskq_cons_is_full(struct xsk_queue *q)
{
	/* No barriers needed since data is not accessed */
	return READ_ONCE(q->ring->producer) - READ_ONCE(q->ring->consumer) ==
		q->nentries;
}

/* Functions for producers */

static inline bool xskq_prod_is_full(struct xsk_queue *q)
{
	u32 free_entries = q->nentries - (q->cached_prod - q->cached_cons);

	if (free_entries)
		return false;

	/* Refresh the local tail pointer */
	q->cached_cons = READ_ONCE(q->ring->consumer);
	free_entries = q->nentries - (q->cached_prod - q->cached_cons);

	return !free_entries;
}

static inline int xskq_prod_reserve(struct xsk_queue *q)
{
	if (xskq_prod_is_full(q))
		return -ENOSPC;

	/* A, matches D */
	q->cached_prod++;
	return 0;
}

static inline int xskq_prod_reserve_addr(struct xsk_queue *q, u64 addr)
{
	struct xdp_umem_ring *ring = (struct xdp_umem_ring *)q->ring;

	if (xskq_prod_is_full(q))
		return -ENOSPC;

	/* A, matches D */
	ring->desc[q->cached_prod++ & q->ring_mask] = addr;
	return 0;
}

static inline int xskq_prod_reserve_desc(struct xsk_queue *q,
					 u64 addr, u32 len)
{
	struct xdp_rxtx_ring *ring = (struct xdp_rxtx_ring *)q->ring;
	u32 idx;

	if (xskq_prod_is_full(q))
		return -ENOSPC;

	/* A, matches D */
	idx = q->cached_prod++ & q->ring_mask;
	ring->desc[idx].addr = addr;
	ring->desc[idx].len = len;

	return 0;
}

static inline void __xskq_prod_submit(struct xsk_queue *q, u32 idx)
{
	smp_wmb(); /* B, matches C */

	WRITE_ONCE(q->ring->producer, idx);
}

static inline void xskq_prod_submit(struct xsk_queue *q)
{
	__xskq_prod_submit(q, q->cached_prod);
}

static inline void xskq_prod_submit_addr(struct xsk_queue *q, u64 addr)
{
	struct xdp_umem_ring *ring = (struct xdp_umem_ring *)q->ring;
	u32 idx = q->ring->producer;

	ring->desc[idx++ & q->ring_mask] = addr;

	__xskq_prod_submit(q, idx);
}

static inline void xskq_prod_submit_n(struct xsk_queue *q, u32 nb_entries)
{
	__xskq_prod_submit(q, q->ring->producer + nb_entries);
}

static inline bool xskq_prod_is_empty(struct xsk_queue *q)
{
	/* No barriers needed since data is not accessed */
	return READ_ONCE(q->ring->consumer) == READ_ONCE(q->ring->producer);
}

/* For both producers and consumers */

static inline u64 xskq_nb_invalid_descs(struct xsk_queue *q)
{
	return q ? q->invalid_descs : 0;
}

static inline u64 xskq_nb_queue_empty_descs(struct xsk_queue *q)
{
	return q ? q->queue_empty_descs : 0;
}

struct xsk_queue *xskq_create(u32 nentries, bool umem_queue);
void xskq_destroy(struct xsk_queue *q_ops);

#endif /* _LINUX_XSK_QUEUE_H */
Commit	Line	Data
dac09149 BT	1	/* SPDX-License-Identifier: GPL-2.0 */
dac09149 BT	2	/* XDP user-space ring structure
423f3832	3	* Copyright(c) 2018 Intel Corporation.
423f3832 MK	4	*/
	5
	6	#ifndef _LINUX_XSK_QUEUE_H
	7	#define _LINUX_XSK_QUEUE_H
	8
	9	#include <linux/types.h>
	10	#include <linux/if_xdp.h>
e61e62b9	11	#include <net/xdp_sock.h>
2b43470a	12	#include <net/xsk_buff_pool.h>
423f3832	13
89e4a376 BT	14	#include "xsk.h"
89e4a376 BT	15
b3a9e0be BT	16	struct xdp_ring {
	17	u32 producer ____cacheline_aligned_in_smp;
	18	u32 consumer ____cacheline_aligned_in_smp;
77cd0d7b	19	u32 flags;
b3a9e0be BT	20	};
	21
	22	/* Used for the RX and TX queues for packets */
	23	struct xdp_rxtx_ring {
	24	struct xdp_ring ptrs;
95e486f5	25	struct xdp_desc desc[] ____cacheline_aligned_in_smp;
b3a9e0be BT	26	};
	27
	28	/* Used for the fill and completion queues for buffers */
	29	struct xdp_umem_ring {
	30	struct xdp_ring ptrs;
95e486f5	31	u64 desc[] ____cacheline_aligned_in_smp;
b3a9e0be BT	32	};
b3a9e0be BT	33
423f3832	34	struct xsk_queue {
423f3832 MK	35	u32 ring_mask;
423f3832 MK	36	u32 nentries;
d7012f05	37	u32 cached_prod;
c5ed924b	38	u32 cached_cons;
423f3832 MK	39	struct xdp_ring *ring;
423f3832 MK	40	u64 invalid_descs;
8aa5a335	41	u64 queue_empty_descs;
423f3832 MK	42	};
423f3832 MK	43
f63666de MK	44	/* The structure of the shared state of the rings are the same as the
	45	* ring buffer in kernel/events/ring_buffer.c. For the Rx and completion
	46	* ring, the kernel is the producer and user space is the consumer. For
	47	* the Tx and fill rings, the kernel is the consumer and user space is
	48	* the producer.
	49	*
	50	* producer consumer
	51	*
	52	* if (LOAD ->consumer) { LOAD ->producer
	53	* (A) smp_rmb() (C)
	54	* STORE $data LOAD $data
	55	* smp_wmb() (B) smp_mb() (D)
	56	* STORE ->producer STORE ->consumer
	57	* }
	58	*
	59	* (A) pairs with (D), and (B) pairs with (C).
	60	*
	61	* Starting with (B), it protects the data from being written after
	62	* the producer pointer. If this barrier was missing, the consumer
	63	* could observe the producer pointer being set and thus load the data
	64	* before the producer has written the new data. The consumer would in
	65	* this case load the old data.
	66	*
	67	* (C) protects the consumer from speculatively loading the data before
	68	* the producer pointer actually has been read. If we do not have this
	69	* barrier, some architectures could load old data as speculative loads
	70	* are not discarded as the CPU does not know there is a dependency
	71	* between ->producer and data.
	72	*
	73	* (A) is a control dependency that separates the load of ->consumer
	74	* from the stores of $data. In case ->consumer indicates there is no
	75	* room in the buffer to store $data we do not. So no barrier is needed.
	76	*
	77	* (D) protects the load of the data to be observed to happen after the
	78	* store of the consumer pointer. If we did not have this memory
	79	* barrier, the producer could observe the consumer pointer being set
	80	* and overwrite the data with a new value before the consumer got the
	81	* chance to read the old value. The consumer would thus miss reading
	82	* the old entry and very likely read the new entry twice, once right
	83	* now and again after circling through the ring.
	84	*/
	85
15d8c916 MK	86	/* The operations on the rings are the following:
	87	*
	88	* producer consumer
	89	*
	90	* RESERVE entries PEEK in the ring for entries
	91	* WRITE data into the ring READ data from the ring
	92	* SUBMIT entries RELEASE entries
	93	*
	94	* The producer reserves one or more entries in the ring. It can then
	95	* fill in these entries and finally submit them so that they can be
	96	* seen and read by the consumer.
	97	*
	98	* The consumer peeks into the ring to see if the producer has written
	99	* any new entries. If so, the producer can then read these entries
	100	* and when it is done reading them release them back to the producer
	101	* so that the producer can use these slots to fill in new entries.
	102	*
	103	* The function names below reflect these operations.
	104	*/
d57d7642	105
15d8c916	106	/* Functions that read and validate content from consumer rings. */
c497176c	107
2b43470a BT	108	static inline bool xskq_cons_read_addr_unchecked(struct xsk_queue q, u64 addr)
	109	{
	110	struct xdp_umem_ring ring = (struct xdp_umem_ring )q->ring;
	111
	112	if (q->cached_cons != q->cached_prod) {
	113	u32 idx = q->cached_cons & q->ring_mask;
	114
	115	*addr = ring->desc[idx];
c05cd364 KL	116	return true;
	117	}
	118
2b43470a BT	119	return false;
2b43470a BT	120	}
35fcde7f	121
26062b18 BT	122	static inline bool xp_aligned_validate_desc(struct xsk_buff_pool *pool,
	123	struct xdp_desc *desc)
	124	{
	125	u64 chunk, chunk_end;
	126
	127	chunk = xp_aligned_extract_addr(pool, desc->addr);
	128	chunk_end = xp_aligned_extract_addr(pool, desc->addr + desc->len);
	129	if (chunk != chunk_end)
	130	return false;
	131
	132	if (chunk >= pool->addrs_cnt)
	133	return false;
	134
	135	if (desc->options)
	136	return false;
	137	return true;
	138	}
	139
	140	static inline bool xp_unaligned_validate_desc(struct xsk_buff_pool *pool,
	141	struct xdp_desc *desc)
	142	{
	143	u64 addr, base_addr;
	144
	145	base_addr = xp_unaligned_extract_addr(desc->addr);
	146	addr = xp_unaligned_add_offset_to_addr(desc->addr);
	147
	148	if (desc->len > pool->chunk_size)
	149	return false;
	150
	151	if (base_addr >= pool->addrs_cnt \|\| addr >= pool->addrs_cnt \|\|
	152	xp_desc_crosses_non_contig_pg(pool, addr, desc->len))
	153	return false;
	154
	155	if (desc->options)
	156	return false;
	157	return true;
	158	}
	159
	160	static inline bool xp_validate_desc(struct xsk_buff_pool *pool,
	161	struct xdp_desc *desc)
	162	{
	163	return pool->unaligned ? xp_unaligned_validate_desc(pool, desc) :
	164	xp_aligned_validate_desc(pool, desc);
	165	}
	166
2b43470a BT	167	static inline bool xskq_cons_is_valid_desc(struct xsk_queue *q,
	168	struct xdp_desc *d,
	169	struct xdp_umem *umem)
	170	{
	171	if (!xp_validate_desc(umem->pool, d)) {
35fcde7f MK	172	q->invalid_descs++;
	173	return false;
	174	}
35fcde7f MK	175	return true;
	176	}
	177
03896ef1 MK	178	static inline bool xskq_cons_read_desc(struct xsk_queue *q,
	179	struct xdp_desc *desc,
	180	struct xdp_umem *umem)
35fcde7f	181	{
c5ed924b	182	while (q->cached_cons != q->cached_prod) {
35fcde7f	183	struct xdp_rxtx_ring ring = (struct xdp_rxtx_ring )q->ring;
c5ed924b	184	u32 idx = q->cached_cons & q->ring_mask;
35fcde7f	185
c34787fc	186	*desc = ring->desc[idx];
03896ef1 MK	187	if (xskq_cons_is_valid_desc(q, desc, umem))
03896ef1 MK	188	return true;
35fcde7f	189
c5ed924b	190	q->cached_cons++;
35fcde7f MK	191	}
35fcde7f MK	192
03896ef1	193	return false;
35fcde7f MK	194	}
35fcde7f MK	195
15d8c916 MK	196	/* Functions for consumers */
	197
	198	static inline void __xskq_cons_release(struct xsk_queue *q)
	199	{
	200	smp_mb(); /* D, matches A */
	201	WRITE_ONCE(q->ring->consumer, q->cached_cons);
	202	}
	203
	204	static inline void __xskq_cons_peek(struct xsk_queue *q)
	205	{
	206	/* Refresh the local pointer */
	207	q->cached_prod = READ_ONCE(q->ring->producer);
	208	smp_rmb(); /* C, matches B */
	209	}
	210
	211	static inline void xskq_cons_get_entries(struct xsk_queue *q)
	212	{
	213	__xskq_cons_release(q);
	214	__xskq_cons_peek(q);
	215	}
	216
	217	static inline bool xskq_cons_has_entries(struct xsk_queue *q, u32 cnt)
	218	{
	219	u32 entries = q->cached_prod - q->cached_cons;
	220
	221	if (entries >= cnt)
	222	return true;
	223
	224	__xskq_cons_peek(q);
	225	entries = q->cached_prod - q->cached_cons;
	226
	227	return entries >= cnt;
	228	}
	229
2b43470a BT	230	static inline bool xskq_cons_peek_addr_unchecked(struct xsk_queue q, u64 addr)
	231	{
	232	if (q->cached_prod == q->cached_cons)
	233	xskq_cons_get_entries(q);
	234	return xskq_cons_read_addr_unchecked(q, addr);
	235	}
	236
03896ef1 MK	237	static inline bool xskq_cons_peek_desc(struct xsk_queue *q,
	238	struct xdp_desc *desc,
	239	struct xdp_umem *umem)
35fcde7f	240	{
c5ed924b MK	241	if (q->cached_prod == q->cached_cons)
c5ed924b MK	242	xskq_cons_get_entries(q);
03896ef1	243	return xskq_cons_read_desc(q, desc, umem);
35fcde7f MK	244	}
35fcde7f MK	245
15d8c916 MK	246	static inline void xskq_cons_release(struct xsk_queue *q)
	247	{
	248	/* To improve performance, only update local state here.
	249	* Reflect this to global state when we get new entries
30744a68 MK	250	* from the ring in xskq_cons_get_entries() and whenever
30744a68 MK	251	* Rx or Tx processing are completed in the NAPI loop.
15d8c916 MK	252	*/
	253	q->cached_cons++;
	254	}
	255
	256	static inline bool xskq_cons_is_full(struct xsk_queue *q)
	257	{
	258	/* No barriers needed since data is not accessed */
	259	return READ_ONCE(q->ring->producer) - READ_ONCE(q->ring->consumer) ==
	260	q->nentries;
	261	}
	262
	263	/* Functions for producers */
	264
	265	static inline bool xskq_prod_is_full(struct xsk_queue *q)
	266	{
	267	u32 free_entries = q->nentries - (q->cached_prod - q->cached_cons);
	268
	269	if (free_entries)
	270	return false;
	271
	272	/* Refresh the local tail pointer */
	273	q->cached_cons = READ_ONCE(q->ring->consumer);
	274	free_entries = q->nentries - (q->cached_prod - q->cached_cons);
	275
	276	return !free_entries;
	277	}
	278
	279	static inline int xskq_prod_reserve(struct xsk_queue *q)
	280	{
	281	if (xskq_prod_is_full(q))
	282	return -ENOSPC;
	283
	284	/* A, matches D */
	285	q->cached_prod++;
	286	return 0;
	287	}
	288
	289	static inline int xskq_prod_reserve_addr(struct xsk_queue *q, u64 addr)
	290	{
	291	struct xdp_umem_ring ring = (struct xdp_umem_ring )q->ring;
	292
	293	if (xskq_prod_is_full(q))
	294	return -ENOSPC;
	295
	296	/* A, matches D */
	297	ring->desc[q->cached_prod++ & q->ring_mask] = addr;
	298	return 0;
	299	}
	300
59e35e55 MK	301	static inline int xskq_prod_reserve_desc(struct xsk_queue *q,
59e35e55 MK	302	u64 addr, u32 len)
c497176c BT	303	{
c497176c BT	304	struct xdp_rxtx_ring ring = (struct xdp_rxtx_ring )q->ring;
59e35e55	305	u32 idx;
c497176c	306
df0ae6f7	307	if (xskq_prod_is_full(q))
c497176c BT	308	return -ENOSPC;
c497176c BT	309
f63666de	310	/* A, matches D */
d7012f05	311	idx = q->cached_prod++ & q->ring_mask;
bbff2f32	312	ring->desc[idx].addr = addr;
c497176c	313	ring->desc[idx].len = len;
c497176c BT	314
	315	return 0;
	316	}
	317
15d8c916	318	static inline void __xskq_prod_submit(struct xsk_queue *q, u32 idx)
35fcde7f	319	{
15d8c916 MK	320	smp_wmb(); /* B, matches C */
	321
	322	WRITE_ONCE(q->ring->producer, idx);
	323	}
	324
	325	static inline void xskq_prod_submit(struct xsk_queue *q)
	326	{
	327	__xskq_prod_submit(q, q->cached_prod);
	328	}
	329
	330	static inline void xskq_prod_submit_addr(struct xsk_queue *q, u64 addr)
	331	{
	332	struct xdp_umem_ring ring = (struct xdp_umem_ring )q->ring;
	333	u32 idx = q->ring->producer;
	334
	335	ring->desc[idx++ & q->ring_mask] = addr;
	336
	337	__xskq_prod_submit(q, idx);
	338	}
	339
	340	static inline void xskq_prod_submit_n(struct xsk_queue *q, u32 nb_entries)
	341	{
	342	__xskq_prod_submit(q, q->ring->producer + nb_entries);
35fcde7f MK	343	}
35fcde7f MK	344
59e35e55	345	static inline bool xskq_prod_is_empty(struct xsk_queue *q)
c497176c	346	{
11cc2d21 MK	347	/* No barriers needed since data is not accessed */
11cc2d21 MK	348	return READ_ONCE(q->ring->consumer) == READ_ONCE(q->ring->producer);
c497176c BT	349	}
c497176c BT	350
15d8c916 MK	351	/* For both producers and consumers */
	352
	353	static inline u64 xskq_nb_invalid_descs(struct xsk_queue *q)
	354	{
	355	return q ? q->invalid_descs : 0;
	356	}
	357
8aa5a335 CL	358	static inline u64 xskq_nb_queue_empty_descs(struct xsk_queue *q)
	359	{
	360	return q ? q->queue_empty_descs : 0;
	361	}
	362
b9b6b68e	363	struct xsk_queue *xskq_create(u32 nentries, bool umem_queue);
c497176c	364	void xskq_destroy(struct xsk_queue *q_ops);
423f3832 MK	365
423f3832 MK	366	#endif /* _LINUX_XSK_QUEUE_H */