net/xdp/xsk_queue.h

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