#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 pad1 ____cacheline_aligned_in_smp;
u32 consumer ____cacheline_aligned_in_smp;
u32 pad2 ____cacheline_aligned_in_smp;
u32 flags;
u32 pad3 ____cacheline_aligned_in_smp;
};
struct xdp_rxtx_ring {
struct xdp_ring ptrs;
struct xdp_desc desc[] ____cacheline_aligned_in_smp;
};
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;
};
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 xsk_buff_pool *pool)
{
if (!xp_validate_desc(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 xsk_buff_pool *pool)
{
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, pool))
return true;
q->cached_cons++;
}
return false;
}
static inline u32 xskq_cons_read_desc_batch(struct xsk_queue *q,
struct xdp_desc *descs,
struct xsk_buff_pool *pool, u32 max)
{
u32 cached_cons = q->cached_cons, nb_entries = 0;
while (cached_cons != q->cached_prod && nb_entries < max) {
struct xdp_rxtx_ring *ring = (struct xdp_rxtx_ring *)q->ring;
u32 idx = cached_cons & q->ring_mask;
descs[nb_entries] = ring->desc[idx];
if (unlikely(!xskq_cons_is_valid_desc(q, &descs[nb_entries], pool))) {
cached_cons++;
continue;
}
nb_entries++;
cached_cons++;
}
return nb_entries;
}
static inline void __xskq_cons_release(struct xsk_queue *q)
{
smp_mb();
WRITE_ONCE(q->ring->consumer, q->cached_cons);
}
static inline void __xskq_cons_peek(struct xsk_queue *q)
{
q->cached_prod = READ_ONCE(q->ring->producer);
smp_rmb();
}
static inline void xskq_cons_get_entries(struct xsk_queue *q)
{
__xskq_cons_release(q);
__xskq_cons_peek(q);
}
static inline u32 xskq_cons_nb_entries(struct xsk_queue *q, u32 max)
{
u32 entries = q->cached_prod - q->cached_cons;
if (entries >= max)
return max;
__xskq_cons_peek(q);
entries = q->cached_prod - q->cached_cons;
return entries >= max ? max : entries;
}
static inline bool xskq_cons_has_entries(struct xsk_queue *q, u32 cnt)
{
return xskq_cons_nb_entries(q, cnt) >= cnt ? true : false;
}
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 xsk_buff_pool *pool)
{
if (q->cached_prod == q->cached_cons)
xskq_cons_get_entries(q);
return xskq_cons_read_desc(q, desc, pool);
}
static inline u32 xskq_cons_peek_desc_batch(struct xsk_queue *q, struct xdp_desc *descs,
struct xsk_buff_pool *pool, u32 max)
{
u32 entries = xskq_cons_nb_entries(q, max);
return xskq_cons_read_desc_batch(q, descs, pool, entries);
}
static inline void xskq_cons_release(struct xsk_queue *q)
{
q->cached_cons++;
}
static inline void xskq_cons_release_n(struct xsk_queue *q, u32 cnt)
{
q->cached_cons += cnt;
}
static inline bool xskq_cons_is_full(struct xsk_queue *q)
{
return READ_ONCE(q->ring->producer) - READ_ONCE(q->ring->consumer) ==
q->nentries;
}
static inline u32 xskq_cons_present_entries(struct xsk_queue *q)
{
return READ_ONCE(q->ring->producer) - READ_ONCE(q->ring->consumer);
}
static inline u32 xskq_prod_nb_free(struct xsk_queue *q, u32 max)
{
u32 free_entries = q->nentries - (q->cached_prod - q->cached_cons);
if (free_entries >= max)
return max;
q->cached_cons = READ_ONCE(q->ring->consumer);
free_entries = q->nentries - (q->cached_prod - q->cached_cons);
return free_entries >= max ? max : free_entries;
}
static inline bool xskq_prod_is_full(struct xsk_queue *q)
{
return xskq_prod_nb_free(q, 1) ? false : true;
}
static inline void xskq_prod_cancel(struct xsk_queue *q)
{
q->cached_prod--;
}
static inline int xskq_prod_reserve(struct xsk_queue *q)
{
if (xskq_prod_is_full(q))
return -ENOSPC;
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;
ring->desc[q->cached_prod++ & q->ring_mask] = addr;
return 0;
}
static inline u32 xskq_prod_reserve_addr_batch(struct xsk_queue *q, struct xdp_desc *descs,
u32 max)
{
struct xdp_umem_ring *ring = (struct xdp_umem_ring *)q->ring;
u32 nb_entries, i, cached_prod;
nb_entries = xskq_prod_nb_free(q, max);
cached_prod = q->cached_prod;
for (i = 0; i < nb_entries; i++)
ring->desc[cached_prod++ & q->ring_mask] = descs[i].addr;
q->cached_prod = cached_prod;
return nb_entries;
}
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;
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();
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)
{
return READ_ONCE(q->ring->consumer) == READ_ONCE(q->ring->producer);
}
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 */