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#include <net/xsk_buff_pool.h> 13 14#include "xsk.h" 15 16struct xdp_ring { 17 u32 producer ____cacheline_aligned_in_smp; 18 /* Hinder the adjacent cache prefetcher to prefetch the consumer 19 * pointer if the producer pointer is touched and vice versa. 20 */ 21 u32 pad1 ____cacheline_aligned_in_smp; 22 u32 consumer ____cacheline_aligned_in_smp; 23 u32 pad2 ____cacheline_aligned_in_smp; 24 u32 flags; 25 u32 pad3 ____cacheline_aligned_in_smp; 26}; 27 28/* Used for the RX and TX queues for packets */ 29struct xdp_rxtx_ring { 30 struct xdp_ring ptrs; 31 struct xdp_desc desc[] ____cacheline_aligned_in_smp; 32}; 33 34/* Used for the fill and completion queues for buffers */ 35struct xdp_umem_ring { 36 struct xdp_ring ptrs; 37 u64 desc[] ____cacheline_aligned_in_smp; 38}; 39 40struct xsk_queue { 41 u32 ring_mask; 42 u32 nentries; 43 u32 cached_prod; 44 u32 cached_cons; 45 struct xdp_ring *ring; 46 u64 invalid_descs; 47 u64 queue_empty_descs; 48 size_t ring_vmalloc_size; 49}; 50 51struct parsed_desc { 52 u32 mb; 53 u32 valid; 54}; 55 56/* The structure of the shared state of the rings are a simple 57 * circular buffer, as outlined in 58 * Documentation/core-api/circular-buffers.rst. For the Rx and 59 * completion ring, the kernel is the producer and user space is the 60 * consumer. For the Tx and fill rings, the kernel is the consumer and 61 * user space is the producer. 62 * 63 * producer consumer 64 * 65 * if (LOAD ->consumer) { (A) LOAD.acq ->producer (C) 66 * STORE $data LOAD $data 67 * STORE.rel ->producer (B) STORE.rel ->consumer (D) 68 * } 69 * 70 * (A) pairs with (D), and (B) pairs with (C). 71 * 72 * Starting with (B), it protects the data from being written after 73 * the producer pointer. If this barrier was missing, the consumer 74 * could observe the producer pointer being set and thus load the data 75 * before the producer has written the new data. The consumer would in 76 * this case load the old data. 77 * 78 * (C) protects the consumer from speculatively loading the data before 79 * the producer pointer actually has been read. If we do not have this 80 * barrier, some architectures could load old data as speculative loads 81 * are not discarded as the CPU does not know there is a dependency 82 * between ->producer and data. 83 * 84 * (A) is a control dependency that separates the load of ->consumer 85 * from the stores of $data. In case ->consumer indicates there is no 86 * room in the buffer to store $data we do not. The dependency will 87 * order both of the stores after the loads. So no barrier is needed. 88 * 89 * (D) protects the load of the data to be observed to happen after the 90 * store of the consumer pointer. If we did not have this memory 91 * barrier, the producer could observe the consumer pointer being set 92 * and overwrite the data with a new value before the consumer got the 93 * chance to read the old value. The consumer would thus miss reading 94 * the old entry and very likely read the new entry twice, once right 95 * now and again after circling through the ring. 96 */ 97 98/* The operations on the rings are the following: 99 * 100 * producer consumer 101 * 102 * RESERVE entries PEEK in the ring for entries 103 * WRITE data into the ring READ data from the ring 104 * SUBMIT entries RELEASE entries 105 * 106 * The producer reserves one or more entries in the ring. It can then 107 * fill in these entries and finally submit them so that they can be 108 * seen and read by the consumer. 109 * 110 * The consumer peeks into the ring to see if the producer has written 111 * any new entries. If so, the consumer can then read these entries 112 * and when it is done reading them release them back to the producer 113 * so that the producer can use these slots to fill in new entries. 114 * 115 * The function names below reflect these operations. 116 */ 117 118/* Functions that read and validate content from consumer rings. */ 119 120static inline void __xskq_cons_read_addr_unchecked(struct xsk_queue *q, u32 cached_cons, u64 *addr) 121{ 122 struct xdp_umem_ring *ring = (struct xdp_umem_ring *)q->ring; 123 u32 idx = cached_cons & q->ring_mask; 124 125 *addr = ring->desc[idx]; 126} 127 128static inline bool xskq_cons_read_addr_unchecked(struct xsk_queue *q, u64 *addr) 129{ 130 if (q->cached_cons != q->cached_prod) { 131 __xskq_cons_read_addr_unchecked(q, q->cached_cons, addr); 132 return true; 133 } 134 135 return false; 136} 137 138static inline bool xp_unused_options_set(u32 options) 139{ 140 return options & ~(XDP_PKT_CONTD | XDP_TX_METADATA); 141} 142 143static inline bool xp_aligned_validate_desc(struct xsk_buff_pool *pool, 144 struct xdp_desc *desc) 145{ 146 u64 addr = desc->addr - pool->tx_metadata_len; 147 u64 len = desc->len + pool->tx_metadata_len; 148 u64 offset = addr & (pool->chunk_size - 1); 149 150 if (!desc->len) 151 return false; 152 153 if (offset + len > pool->chunk_size) 154 return false; 155 156 if (addr >= pool->addrs_cnt) 157 return false; 158 159 if (xp_unused_options_set(desc->options)) 160 return false; 161 return true; 162} 163 164static inline bool xp_unaligned_validate_desc(struct xsk_buff_pool *pool, 165 struct xdp_desc *desc) 166{ 167 u64 addr = xp_unaligned_add_offset_to_addr(desc->addr) - pool->tx_metadata_len; 168 u64 len = desc->len + pool->tx_metadata_len; 169 170 if (!desc->len) 171 return false; 172 173 if (len > pool->chunk_size) 174 return false; 175 176 if (addr >= pool->addrs_cnt || addr + len > pool->addrs_cnt || 177 xp_desc_crosses_non_contig_pg(pool, addr, len)) 178 return false; 179 180 if (xp_unused_options_set(desc->options)) 181 return false; 182 return true; 183} 184 185static inline bool xp_validate_desc(struct xsk_buff_pool *pool, 186 struct xdp_desc *desc) 187{ 188 return pool->unaligned ? xp_unaligned_validate_desc(pool, desc) : 189 xp_aligned_validate_desc(pool, desc); 190} 191 192static inline bool xskq_has_descs(struct xsk_queue *q) 193{ 194 return q->cached_cons != q->cached_prod; 195} 196 197static inline bool xskq_cons_is_valid_desc(struct xsk_queue *q, 198 struct xdp_desc *d, 199 struct xsk_buff_pool *pool) 200{ 201 if (!xp_validate_desc(pool, d)) { 202 q->invalid_descs++; 203 return false; 204 } 205 return true; 206} 207 208static inline bool xskq_cons_read_desc(struct xsk_queue *q, 209 struct xdp_desc *desc, 210 struct xsk_buff_pool *pool) 211{ 212 if (q->cached_cons != q->cached_prod) { 213 struct xdp_rxtx_ring *ring = (struct xdp_rxtx_ring *)q->ring; 214 u32 idx = q->cached_cons & q->ring_mask; 215 216 *desc = ring->desc[idx]; 217 return xskq_cons_is_valid_desc(q, desc, pool); 218 } 219 220 q->queue_empty_descs++; 221 return false; 222} 223 224static inline void xskq_cons_release_n(struct xsk_queue *q, u32 cnt) 225{ 226 q->cached_cons += cnt; 227} 228 229static inline void parse_desc(struct xsk_queue *q, struct xsk_buff_pool *pool, 230 struct xdp_desc *desc, struct parsed_desc *parsed) 231{ 232 parsed->valid = xskq_cons_is_valid_desc(q, desc, pool); 233 parsed->mb = xp_mb_desc(desc); 234} 235 236static inline 237u32 xskq_cons_read_desc_batch(struct xsk_queue *q, struct xsk_buff_pool *pool, 238 u32 max) 239{ 240 u32 cached_cons = q->cached_cons, nb_entries = 0; 241 struct xdp_desc *descs = pool->tx_descs; 242 u32 total_descs = 0, nr_frags = 0; 243 244 /* track first entry, if stumble upon *any* invalid descriptor, rewind 245 * current packet that consists of frags and stop the processing 246 */ 247 while (cached_cons != q->cached_prod && nb_entries < max) { 248 struct xdp_rxtx_ring *ring = (struct xdp_rxtx_ring *)q->ring; 249 u32 idx = cached_cons & q->ring_mask; 250 struct parsed_desc parsed; 251 252 descs[nb_entries] = ring->desc[idx]; 253 cached_cons++; 254 parse_desc(q, pool, &descs[nb_entries], &parsed); 255 if (unlikely(!parsed.valid)) 256 break; 257 258 if (likely(!parsed.mb)) { 259 total_descs += (nr_frags + 1); 260 nr_frags = 0; 261 } else { 262 nr_frags++; 263 if (nr_frags == pool->netdev->xdp_zc_max_segs) { 264 nr_frags = 0; 265 break; 266 } 267 } 268 nb_entries++; 269 } 270 271 cached_cons -= nr_frags; 272 /* Release valid plus any invalid entries */ 273 xskq_cons_release_n(q, cached_cons - q->cached_cons); 274 return total_descs; 275} 276 277/* Functions for consumers */ 278 279static inline void __xskq_cons_release(struct xsk_queue *q) 280{ 281 smp_store_release(&q->ring->consumer, q->cached_cons); /* D, matchees A */ 282} 283 284static inline void __xskq_cons_peek(struct xsk_queue *q) 285{ 286 /* Refresh the local pointer */ 287 q->cached_prod = smp_load_acquire(&q->ring->producer); /* C, matches B */ 288} 289 290static inline void xskq_cons_get_entries(struct xsk_queue *q) 291{ 292 __xskq_cons_release(q); 293 __xskq_cons_peek(q); 294} 295 296static inline u32 xskq_cons_nb_entries(struct xsk_queue *q, u32 max) 297{ 298 u32 entries = q->cached_prod - q->cached_cons; 299 300 if (entries >= max) 301 return max; 302 303 __xskq_cons_peek(q); 304 entries = q->cached_prod - q->cached_cons; 305 306 return entries >= max ? max : entries; 307} 308 309static inline bool xskq_cons_has_entries(struct xsk_queue *q, u32 cnt) 310{ 311 return xskq_cons_nb_entries(q, cnt) >= cnt; 312} 313 314static inline bool xskq_cons_peek_addr_unchecked(struct xsk_queue *q, u64 *addr) 315{ 316 if (q->cached_prod == q->cached_cons) 317 xskq_cons_get_entries(q); 318 return xskq_cons_read_addr_unchecked(q, addr); 319} 320 321static inline bool xskq_cons_peek_desc(struct xsk_queue *q, 322 struct xdp_desc *desc, 323 struct xsk_buff_pool *pool) 324{ 325 if (q->cached_prod == q->cached_cons) 326 xskq_cons_get_entries(q); 327 return xskq_cons_read_desc(q, desc, pool); 328} 329 330/* To improve performance in the xskq_cons_release functions, only update local state here. 331 * Reflect this to global state when we get new entries from the ring in 332 * xskq_cons_get_entries() and whenever Rx or Tx processing are completed in the NAPI loop. 333 */ 334static inline void xskq_cons_release(struct xsk_queue *q) 335{ 336 q->cached_cons++; 337} 338 339static inline void xskq_cons_cancel_n(struct xsk_queue *q, u32 cnt) 340{ 341 q->cached_cons -= cnt; 342} 343 344static inline u32 xskq_cons_present_entries(struct xsk_queue *q) 345{ 346 /* No barriers needed since data is not accessed */ 347 return READ_ONCE(q->ring->producer) - READ_ONCE(q->ring->consumer); 348} 349 350/* Functions for producers */ 351 352static inline u32 xskq_prod_nb_free(struct xsk_queue *q, u32 max) 353{ 354 u32 free_entries = q->nentries - (q->cached_prod - q->cached_cons); 355 356 if (free_entries >= max) 357 return max; 358 359 /* Refresh the local tail pointer */ 360 q->cached_cons = READ_ONCE(q->ring->consumer); 361 free_entries = q->nentries - (q->cached_prod - q->cached_cons); 362 363 return free_entries >= max ? max : free_entries; 364} 365 366static inline bool xskq_prod_is_full(struct xsk_queue *q) 367{ 368 return xskq_prod_nb_free(q, 1) ? false : true; 369} 370 371static inline void xskq_prod_cancel_n(struct xsk_queue *q, u32 cnt) 372{ 373 q->cached_prod -= cnt; 374} 375 376static inline int xskq_prod_reserve(struct xsk_queue *q) 377{ 378 if (xskq_prod_is_full(q)) 379 return -ENOSPC; 380 381 /* A, matches D */ 382 q->cached_prod++; 383 return 0; 384} 385 386static inline int xskq_prod_reserve_addr(struct xsk_queue *q, u64 addr) 387{ 388 struct xdp_umem_ring *ring = (struct xdp_umem_ring *)q->ring; 389 390 if (xskq_prod_is_full(q)) 391 return -ENOSPC; 392 393 /* A, matches D */ 394 ring->desc[q->cached_prod++ & q->ring_mask] = addr; 395 return 0; 396} 397 398static inline void xskq_prod_write_addr_batch(struct xsk_queue *q, struct xdp_desc *descs, 399 u32 nb_entries) 400{ 401 struct xdp_umem_ring *ring = (struct xdp_umem_ring *)q->ring; 402 u32 i, cached_prod; 403 404 /* A, matches D */ 405 cached_prod = q->cached_prod; 406 for (i = 0; i < nb_entries; i++) 407 ring->desc[cached_prod++ & q->ring_mask] = descs[i].addr; 408 q->cached_prod = cached_prod; 409} 410 411static inline int xskq_prod_reserve_desc(struct xsk_queue *q, 412 u64 addr, u32 len, u32 flags) 413{ 414 struct xdp_rxtx_ring *ring = (struct xdp_rxtx_ring *)q->ring; 415 u32 idx; 416 417 if (xskq_prod_is_full(q)) 418 return -ENOBUFS; 419 420 /* A, matches D */ 421 idx = q->cached_prod++ & q->ring_mask; 422 ring->desc[idx].addr = addr; 423 ring->desc[idx].len = len; 424 ring->desc[idx].options = flags; 425 426 return 0; 427} 428 429static inline void __xskq_prod_submit(struct xsk_queue *q, u32 idx) 430{ 431 smp_store_release(&q->ring->producer, idx); /* B, matches C */ 432} 433 434static inline void xskq_prod_submit(struct xsk_queue *q) 435{ 436 __xskq_prod_submit(q, q->cached_prod); 437} 438 439static inline void xskq_prod_submit_n(struct xsk_queue *q, u32 nb_entries) 440{ 441 __xskq_prod_submit(q, q->ring->producer + nb_entries); 442} 443 444static inline bool xskq_prod_is_empty(struct xsk_queue *q) 445{ 446 /* No barriers needed since data is not accessed */ 447 return READ_ONCE(q->ring->consumer) == READ_ONCE(q->ring->producer); 448} 449 450/* For both producers and consumers */ 451 452static inline u64 xskq_nb_invalid_descs(struct xsk_queue *q) 453{ 454 return q ? q->invalid_descs : 0; 455} 456 457static inline u64 xskq_nb_queue_empty_descs(struct xsk_queue *q) 458{ 459 return q ? q->queue_empty_descs : 0; 460} 461 462struct xsk_queue *xskq_create(u32 nentries, bool umem_queue); 463void xskq_destroy(struct xsk_queue *q_ops); 464 465#endif /* _LINUX_XSK_QUEUE_H */ 466