1 /* SPDX-License-Identifier: GPL-2.0 */ 2 /* XDP user-space ring structure 3 * Copyright(c) 2018 Intel Corporation. 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> 11 #include <net/xdp_sock.h> 12 13 #define RX_BATCH_SIZE 16 14 #define LAZY_UPDATE_THRESHOLD 128 15 16 struct xdp_ring { 17 u32 producer ____cacheline_aligned_in_smp; 18 u32 consumer ____cacheline_aligned_in_smp; 19 }; 20 21 /* Used for the RX and TX queues for packets */ 22 struct xdp_rxtx_ring { 23 struct xdp_ring ptrs; 24 struct xdp_desc desc[0] ____cacheline_aligned_in_smp; 25 }; 26 27 /* Used for the fill and completion queues for buffers */ 28 struct xdp_umem_ring { 29 struct xdp_ring ptrs; 30 u64 desc[0] ____cacheline_aligned_in_smp; 31 }; 32 33 struct xsk_queue { 34 u64 chunk_mask; 35 u64 size; 36 u32 ring_mask; 37 u32 nentries; 38 u32 prod_head; 39 u32 prod_tail; 40 u32 cons_head; 41 u32 cons_tail; 42 struct xdp_ring *ring; 43 u64 invalid_descs; 44 }; 45 46 /* Common functions operating for both RXTX and umem queues */ 47 48 static inline u64 xskq_nb_invalid_descs(struct xsk_queue *q) 49 { 50 return q ? q->invalid_descs : 0; 51 } 52 53 static inline u32 xskq_nb_avail(struct xsk_queue *q, u32 dcnt) 54 { 55 u32 entries = q->prod_tail - q->cons_tail; 56 57 if (entries == 0) { 58 /* Refresh the local pointer */ 59 q->prod_tail = READ_ONCE(q->ring->producer); 60 entries = q->prod_tail - q->cons_tail; 61 } 62 63 return (entries > dcnt) ? dcnt : entries; 64 } 65 66 static inline u32 xskq_nb_free(struct xsk_queue *q, u32 producer, u32 dcnt) 67 { 68 u32 free_entries = q->nentries - (producer - q->cons_tail); 69 70 if (free_entries >= dcnt) 71 return free_entries; 72 73 /* Refresh the local tail pointer */ 74 q->cons_tail = READ_ONCE(q->ring->consumer); 75 return q->nentries - (producer - q->cons_tail); 76 } 77 78 /* UMEM queue */ 79 80 static inline bool xskq_is_valid_addr(struct xsk_queue *q, u64 addr) 81 { 82 if (addr >= q->size) { 83 q->invalid_descs++; 84 return false; 85 } 86 87 return true; 88 } 89 90 static inline u64 *xskq_validate_addr(struct xsk_queue *q, u64 *addr) 91 { 92 while (q->cons_tail != q->cons_head) { 93 struct xdp_umem_ring *ring = (struct xdp_umem_ring *)q->ring; 94 unsigned int idx = q->cons_tail & q->ring_mask; 95 96 *addr = READ_ONCE(ring->desc[idx]) & q->chunk_mask; 97 if (xskq_is_valid_addr(q, *addr)) 98 return addr; 99 100 q->cons_tail++; 101 } 102 103 return NULL; 104 } 105 106 static inline u64 *xskq_peek_addr(struct xsk_queue *q, u64 *addr) 107 { 108 if (q->cons_tail == q->cons_head) { 109 WRITE_ONCE(q->ring->consumer, q->cons_tail); 110 q->cons_head = q->cons_tail + xskq_nb_avail(q, RX_BATCH_SIZE); 111 112 /* Order consumer and data */ 113 smp_rmb(); 114 } 115 116 return xskq_validate_addr(q, addr); 117 } 118 119 static inline void xskq_discard_addr(struct xsk_queue *q) 120 { 121 q->cons_tail++; 122 } 123 124 static inline int xskq_produce_addr(struct xsk_queue *q, u64 addr) 125 { 126 struct xdp_umem_ring *ring = (struct xdp_umem_ring *)q->ring; 127 128 if (xskq_nb_free(q, q->prod_tail, 1) == 0) 129 return -ENOSPC; 130 131 ring->desc[q->prod_tail++ & q->ring_mask] = addr; 132 133 /* Order producer and data */ 134 smp_wmb(); 135 136 WRITE_ONCE(q->ring->producer, q->prod_tail); 137 return 0; 138 } 139 140 static inline int xskq_produce_addr_lazy(struct xsk_queue *q, u64 addr) 141 { 142 struct xdp_umem_ring *ring = (struct xdp_umem_ring *)q->ring; 143 144 if (xskq_nb_free(q, q->prod_head, LAZY_UPDATE_THRESHOLD) == 0) 145 return -ENOSPC; 146 147 ring->desc[q->prod_head++ & q->ring_mask] = addr; 148 return 0; 149 } 150 151 static inline void xskq_produce_flush_addr_n(struct xsk_queue *q, 152 u32 nb_entries) 153 { 154 /* Order producer and data */ 155 smp_wmb(); 156 157 q->prod_tail += nb_entries; 158 WRITE_ONCE(q->ring->producer, q->prod_tail); 159 } 160 161 static inline int xskq_reserve_addr(struct xsk_queue *q) 162 { 163 if (xskq_nb_free(q, q->prod_head, 1) == 0) 164 return -ENOSPC; 165 166 q->prod_head++; 167 return 0; 168 } 169 170 /* Rx/Tx queue */ 171 172 static inline bool xskq_is_valid_desc(struct xsk_queue *q, struct xdp_desc *d) 173 { 174 if (!xskq_is_valid_addr(q, d->addr)) 175 return false; 176 177 if (((d->addr + d->len) & q->chunk_mask) != 178 (d->addr & q->chunk_mask)) { 179 q->invalid_descs++; 180 return false; 181 } 182 183 return true; 184 } 185 186 static inline struct xdp_desc *xskq_validate_desc(struct xsk_queue *q, 187 struct xdp_desc *desc) 188 { 189 while (q->cons_tail != q->cons_head) { 190 struct xdp_rxtx_ring *ring = (struct xdp_rxtx_ring *)q->ring; 191 unsigned int idx = q->cons_tail & q->ring_mask; 192 193 *desc = READ_ONCE(ring->desc[idx]); 194 if (xskq_is_valid_desc(q, desc)) 195 return desc; 196 197 q->cons_tail++; 198 } 199 200 return NULL; 201 } 202 203 static inline struct xdp_desc *xskq_peek_desc(struct xsk_queue *q, 204 struct xdp_desc *desc) 205 { 206 if (q->cons_tail == q->cons_head) { 207 WRITE_ONCE(q->ring->consumer, q->cons_tail); 208 q->cons_head = q->cons_tail + xskq_nb_avail(q, RX_BATCH_SIZE); 209 210 /* Order consumer and data */ 211 smp_rmb(); 212 } 213 214 return xskq_validate_desc(q, desc); 215 } 216 217 static inline void xskq_discard_desc(struct xsk_queue *q) 218 { 219 q->cons_tail++; 220 } 221 222 static inline int xskq_produce_batch_desc(struct xsk_queue *q, 223 u64 addr, u32 len) 224 { 225 struct xdp_rxtx_ring *ring = (struct xdp_rxtx_ring *)q->ring; 226 unsigned int idx; 227 228 if (xskq_nb_free(q, q->prod_head, 1) == 0) 229 return -ENOSPC; 230 231 idx = (q->prod_head++) & q->ring_mask; 232 ring->desc[idx].addr = addr; 233 ring->desc[idx].len = len; 234 235 return 0; 236 } 237 238 static inline void xskq_produce_flush_desc(struct xsk_queue *q) 239 { 240 /* Order producer and data */ 241 smp_wmb(); 242 243 q->prod_tail = q->prod_head, 244 WRITE_ONCE(q->ring->producer, q->prod_tail); 245 } 246 247 static inline bool xskq_full_desc(struct xsk_queue *q) 248 { 249 return xskq_nb_avail(q, q->nentries) == q->nentries; 250 } 251 252 static inline bool xskq_empty_desc(struct xsk_queue *q) 253 { 254 return xskq_nb_free(q, q->prod_tail, q->nentries) == q->nentries; 255 } 256 257 void xskq_set_umem(struct xsk_queue *q, u64 size, u64 chunk_mask); 258 struct xsk_queue *xskq_create(u32 nentries, bool umem_queue); 259 void xskq_destroy(struct xsk_queue *q_ops); 260 261 /* Executed by the core when the entire UMEM gets freed */ 262 void xsk_reuseq_destroy(struct xdp_umem *umem); 263 264 #endif /* _LINUX_XSK_QUEUE_H */ 265