kernel/bpf/ringbuf.c

   1 #include <linux/bpf.h>
   2 #include <linux/btf.h>
   3 #include <linux/err.h>
   4 #include <linux/irq_work.h>
   5 #include <linux/slab.h>
   6 #include <linux/filter.h>
   7 #include <linux/mm.h>
   8 #include <linux/vmalloc.h>
   9 #include <linux/wait.h>
  10 #include <linux/poll.h>
  11 #include <linux/kmemleak.h>
  12 #include <uapi/linux/btf.h>
  13 #include <linux/btf_ids.h>
  14
  15 #define RINGBUF_CREATE_FLAG_MASK (BPF_F_NUMA_NODE)
  16
  17 /* non-mmap()'able part of bpf_ringbuf (everything up to consumer page) */
  18 #define RINGBUF_PGOFF \
  19         (offsetof(struct bpf_ringbuf, consumer_pos) >> PAGE_SHIFT)
  20 /* consumer page and producer page */
  21 #define RINGBUF_POS_PAGES 2
  22
  23 #define RINGBUF_MAX_RECORD_SZ (UINT_MAX/4)
  24
  25 /* Maximum size of ring buffer area is limited by 32-bit page offset within
  26  * record header, counted in pages. Reserve 8 bits for extensibility, and take
  27  * into account few extra pages for consumer/producer pages and
  28  * non-mmap()'able parts. This gives 64GB limit, which seems plenty for single
  29  * ring buffer.
  30  */
  31 #define RINGBUF_MAX_DATA_SZ \
  32         (((1ULL << 24) - RINGBUF_POS_PAGES - RINGBUF_PGOFF) * PAGE_SIZE)
  33
  34 struct bpf_ringbuf {
  35         wait_queue_head_t waitq;
  36         struct irq_work work;
  37         u64 mask;
  38         struct page **pages;
  39         int nr_pages;
  40         spinlock_t spinlock ____cacheline_aligned_in_smp;
  41         /* For user-space producer ring buffers, an atomic_t busy bit is used
  42          * to synchronize access to the ring buffers in the kernel, rather than
  43          * the spinlock that is used for kernel-producer ring buffers. This is
  44          * done because the ring buffer must hold a lock across a BPF program's
  45          * callback:
  46          *
  47          *    __bpf_user_ringbuf_peek() // lock acquired
  48          * -> program callback_fn()
  49          * -> __bpf_user_ringbuf_sample_release() // lock released
  50          *
  51          * It is unsafe and incorrect to hold an IRQ spinlock across what could
  52          * be a long execution window, so we instead simply disallow concurrent
  53          * access to the ring buffer by kernel consumers, and return -EBUSY from
  54          * __bpf_user_ringbuf_peek() if the busy bit is held by another task.
  55          */
  56         atomic_t busy ____cacheline_aligned_in_smp;
  57         /* Consumer and producer counters are put into separate pages to
  58          * allow each position to be mapped with different permissions.
  59          * This prevents a user-space application from modifying the
  60          * position and ruining in-kernel tracking. The permissions of the
  61          * pages depend on who is producing samples: user-space or the
  62          * kernel.
  63          *
  64          * Kernel-producer
  65          * ---------------
  66          * The producer position and data pages are mapped as r/o in
  67          * userspace. For this approach, bits in the header of samples are
  68          * used to signal to user-space, and to other producers, whether a
  69          * sample is currently being written.
  70          *
  71          * User-space producer
  72          * -------------------
  73          * Only the page containing the consumer position is mapped r/o in
  74          * user-space. User-space producers also use bits of the header to
  75          * communicate to the kernel, but the kernel must carefully check and
  76          * validate each sample to ensure that they're correctly formatted, and
  77          * fully contained within the ring buffer.
  78          */
  79         unsigned long consumer_pos __aligned(PAGE_SIZE);
  80         unsigned long producer_pos __aligned(PAGE_SIZE);
  81         char data[] __aligned(PAGE_SIZE);
  82 };
  83
  84 struct bpf_ringbuf_map {
  85         struct bpf_map map;
  86         struct bpf_ringbuf *rb;
  87 };
  88
  89 /* 8-byte ring buffer record header structure */
  90 struct bpf_ringbuf_hdr {
  91         u32 len;
  92         u32 pg_off;
  93 };
  94
  95 static struct bpf_ringbuf *bpf_ringbuf_area_alloc(size_t data_sz, int numa_node)
  96 {
  97         const gfp_t flags = GFP_KERNEL_ACCOUNT | __GFP_RETRY_MAYFAIL |
  98                             __GFP_NOWARN | __GFP_ZERO;
  99         int nr_meta_pages = RINGBUF_PGOFF + RINGBUF_POS_PAGES;
 100         int nr_data_pages = data_sz >> PAGE_SHIFT;
 101         int nr_pages = nr_meta_pages + nr_data_pages;
 102         struct page **pages, *page;
 103         struct bpf_ringbuf *rb;
 104         size_t array_size;
 105         int i;
 106
 107         /* Each data page is mapped twice to allow "virtual"
 108          * continuous read of samples wrapping around the end of ring
 109          * buffer area:
 110          * ------------------------------------------------------
 111          * | meta pages |  real data pages  |  same data pages  |
 112          * ------------------------------------------------------
 113          * |            | 1 2 3 4 5 6 7 8 9 | 1 2 3 4 5 6 7 8 9 |
 114          * ------------------------------------------------------
 115          * |            | TA             DA | TA             DA |
 116          * ------------------------------------------------------
 117          *                               ^^^^^^^
 118          *                                  |
 119          * Here, no need to worry about special handling of wrapped-around
 120          * data due to double-mapped data pages. This works both in kernel and
 121          * when mmap()'ed in user-space, simplifying both kernel and
 122          * user-space implementations significantly.
 123          */
 124         array_size = (nr_meta_pages + 2 * nr_data_pages) * sizeof(*pages);
 125         pages = bpf_map_area_alloc(array_size, numa_node);
 126         if (!pages)
 127                 return NULL;
 128
 129         for (i = 0; i < nr_pages; i++) {
 130                 page = alloc_pages_node(numa_node, flags, 0);
 131                 if (!page) {
 132                         nr_pages = i;
 133                         goto err_free_pages;
 134                 }
 135                 pages[i] = page;
 136                 if (i >= nr_meta_pages)
 137                         pages[nr_data_pages + i] = page;
 138         }
 139
 140         rb = vmap(pages, nr_meta_pages + 2 * nr_data_pages,
 141                   VM_MAP | VM_USERMAP, PAGE_KERNEL);
 142         if (rb) {
 143                 kmemleak_not_leak(pages);
 144                 rb->pages = pages;
 145                 rb->nr_pages = nr_pages;
 146                 return rb;
 147         }
 148
 149 err_free_pages:
 150         for (i = 0; i < nr_pages; i++)
 151                 __free_page(pages[i]);
 152         bpf_map_area_free(pages);
 153         return NULL;
 154 }
 155
 156 static void bpf_ringbuf_notify(struct irq_work *work)
 157 {
 158         struct bpf_ringbuf *rb = container_of(work, struct bpf_ringbuf, work);
 159
 160         wake_up_all(&rb->waitq);
 161 }
 162
 163 static struct bpf_ringbuf *bpf_ringbuf_alloc(size_t data_sz, int numa_node)
 164 {
 165         struct bpf_ringbuf *rb;
 166
 167         rb = bpf_ringbuf_area_alloc(data_sz, numa_node);
 168         if (!rb)
 169                 return NULL;
 170
 171         spin_lock_init(&rb->spinlock);
 172         atomic_set(&rb->busy, 0);
 173         init_waitqueue_head(&rb->waitq);
 174         init_irq_work(&rb->work, bpf_ringbuf_notify);
 175
 176         rb->mask = data_sz - 1;
 177         rb->consumer_pos = 0;
 178         rb->producer_pos = 0;
 179
 180         return rb;
 181 }
 182
 183 static struct bpf_map *ringbuf_map_alloc(union bpf_attr *attr)
 184 {
 185         struct bpf_ringbuf_map *rb_map;
 186
 187         if (attr->map_flags & ~RINGBUF_CREATE_FLAG_MASK)
 188                 return ERR_PTR(-EINVAL);
 189
 190         if (attr->key_size || attr->value_size ||
 191             !is_power_of_2(attr->max_entries) ||
 192             !PAGE_ALIGNED(attr->max_entries))
 193                 return ERR_PTR(-EINVAL);
 194
 195 #ifdef CONFIG_64BIT
 196         /* on 32-bit arch, it's impossible to overflow record's hdr->pgoff */
 197         if (attr->max_entries > RINGBUF_MAX_DATA_SZ)
 198                 return ERR_PTR(-E2BIG);
 199 #endif
 200
 201         rb_map = bpf_map_area_alloc(sizeof(*rb_map), NUMA_NO_NODE);
 202         if (!rb_map)
 203                 return ERR_PTR(-ENOMEM);
 204
 205         bpf_map_init_from_attr(&rb_map->map, attr);
 206
 207         rb_map->rb = bpf_ringbuf_alloc(attr->max_entries, rb_map->map.numa_node);
 208         if (!rb_map->rb) {
 209                 bpf_map_area_free(rb_map);
 210                 return ERR_PTR(-ENOMEM);
 211         }
 212
 213         return &rb_map->map;
 214 }
 215
 216 static void bpf_ringbuf_free(struct bpf_ringbuf *rb)
 217 {
 218         /* copy pages pointer and nr_pages to local variable, as we are going
 219          * to unmap rb itself with vunmap() below
 220          */
 221         struct page **pages = rb->pages;
 222         int i, nr_pages = rb->nr_pages;
 223
 224         vunmap(rb);
 225         for (i = 0; i < nr_pages; i++)
 226                 __free_page(pages[i]);
 227         bpf_map_area_free(pages);
 228 }
 229
 230 static void ringbuf_map_free(struct bpf_map *map)
 231 {
 232         struct bpf_ringbuf_map *rb_map;
 233
 234         rb_map = container_of(map, struct bpf_ringbuf_map, map);
 235         bpf_ringbuf_free(rb_map->rb);
 236         bpf_map_area_free(rb_map);
 237 }
 238
 239 static void *ringbuf_map_lookup_elem(struct bpf_map *map, void *key)
 240 {
 241         return ERR_PTR(-ENOTSUPP);
 242 }
 243
 244 static int ringbuf_map_update_elem(struct bpf_map *map, void *key, void *value,
 245                                    u64 flags)
 246 {
 247         return -ENOTSUPP;
 248 }
 249
 250 static int ringbuf_map_delete_elem(struct bpf_map *map, void *key)
 251 {
 252         return -ENOTSUPP;
 253 }
 254
 255 static int ringbuf_map_get_next_key(struct bpf_map *map, void *key,
 256                                     void *next_key)
 257 {
 258         return -ENOTSUPP;
 259 }
 260
 261 static int ringbuf_map_mmap_kern(struct bpf_map *map, struct vm_area_struct *vma)
 262 {
 263         struct bpf_ringbuf_map *rb_map;
 264
 265         rb_map = container_of(map, struct bpf_ringbuf_map, map);
 266
 267         if (vma->vm_flags & VM_WRITE) {
 268                 /* allow writable mapping for the consumer_pos only */
 269                 if (vma->vm_pgoff != 0 || vma->vm_end - vma->vm_start != PAGE_SIZE)
 270                         return -EPERM;
 271         } else {
 272                 vma->vm_flags &= ~VM_MAYWRITE;
 273         }
 274         /* remap_vmalloc_range() checks size and offset constraints */
 275         return remap_vmalloc_range(vma, rb_map->rb,
 276                                    vma->vm_pgoff + RINGBUF_PGOFF);
 277 }
 278
 279 static int ringbuf_map_mmap_user(struct bpf_map *map, struct vm_area_struct *vma)
 280 {
 281         struct bpf_ringbuf_map *rb_map;
 282
 283         rb_map = container_of(map, struct bpf_ringbuf_map, map);
 284
 285         if (vma->vm_flags & VM_WRITE) {
 286                 if (vma->vm_pgoff == 0)
 287                         /* Disallow writable mappings to the consumer pointer,
 288                          * and allow writable mappings to both the producer
 289                          * position, and the ring buffer data itself.
 290                          */
 291                         return -EPERM;
 292         } else {
 293                 vma->vm_flags &= ~VM_MAYWRITE;
 294         }
 295         /* remap_vmalloc_range() checks size and offset constraints */
 296         return remap_vmalloc_range(vma, rb_map->rb, vma->vm_pgoff + RINGBUF_PGOFF);
 297 }
 298
 299 static unsigned long ringbuf_avail_data_sz(struct bpf_ringbuf *rb)
 300 {
 301         unsigned long cons_pos, prod_pos;
 302
 303         cons_pos = smp_load_acquire(&rb->consumer_pos);
 304         prod_pos = smp_load_acquire(&rb->producer_pos);
 305         return prod_pos - cons_pos;
 306 }
 307
 308 static u32 ringbuf_total_data_sz(const struct bpf_ringbuf *rb)
 309 {
 310         return rb->mask + 1;
 311 }
 312
 313 static __poll_t ringbuf_map_poll_kern(struct bpf_map *map, struct file *filp,
 314                                       struct poll_table_struct *pts)
 315 {
 316         struct bpf_ringbuf_map *rb_map;
 317
 318         rb_map = container_of(map, struct bpf_ringbuf_map, map);
 319         poll_wait(filp, &rb_map->rb->waitq, pts);
 320
 321         if (ringbuf_avail_data_sz(rb_map->rb))
 322                 return EPOLLIN | EPOLLRDNORM;
 323         return 0;
 324 }
 325
 326 static __poll_t ringbuf_map_poll_user(struct bpf_map *map, struct file *filp,
 327                                       struct poll_table_struct *pts)
 328 {
 329         struct bpf_ringbuf_map *rb_map;
 330
 331         rb_map = container_of(map, struct bpf_ringbuf_map, map);
 332         poll_wait(filp, &rb_map->rb->waitq, pts);
 333
 334         if (ringbuf_avail_data_sz(rb_map->rb) < ringbuf_total_data_sz(rb_map->rb))
 335                 return EPOLLOUT | EPOLLWRNORM;
 336         return 0;
 337 }
 338
 339 BTF_ID_LIST_SINGLE(ringbuf_map_btf_ids, struct, bpf_ringbuf_map)
 340 const struct bpf_map_ops ringbuf_map_ops = {
 341         .map_meta_equal = bpf_map_meta_equal,
 342         .map_alloc = ringbuf_map_alloc,
 343         .map_free = ringbuf_map_free,
 344         .map_mmap = ringbuf_map_mmap_kern,
 345         .map_poll = ringbuf_map_poll_kern,
 346         .map_lookup_elem = ringbuf_map_lookup_elem,
 347         .map_update_elem = ringbuf_map_update_elem,
 348         .map_delete_elem = ringbuf_map_delete_elem,
 349         .map_get_next_key = ringbuf_map_get_next_key,
 350         .map_btf_id = &ringbuf_map_btf_ids[0],
 351 };
 352
 353 BTF_ID_LIST_SINGLE(user_ringbuf_map_btf_ids, struct, bpf_ringbuf_map)
 354 const struct bpf_map_ops user_ringbuf_map_ops = {
 355         .map_meta_equal = bpf_map_meta_equal,
 356         .map_alloc = ringbuf_map_alloc,
 357         .map_free = ringbuf_map_free,
 358         .map_mmap = ringbuf_map_mmap_user,
 359         .map_poll = ringbuf_map_poll_user,
 360         .map_lookup_elem = ringbuf_map_lookup_elem,
 361         .map_update_elem = ringbuf_map_update_elem,
 362         .map_delete_elem = ringbuf_map_delete_elem,
 363         .map_get_next_key = ringbuf_map_get_next_key,
 364         .map_btf_id = &user_ringbuf_map_btf_ids[0],
 365 };
 366
 367 /* Given pointer to ring buffer record metadata and struct bpf_ringbuf itself,
 368  * calculate offset from record metadata to ring buffer in pages, rounded
 369  * down. This page offset is stored as part of record metadata and allows to
 370  * restore struct bpf_ringbuf * from record pointer. This page offset is
 371  * stored at offset 4 of record metadata header.
 372  */
 373 static size_t bpf_ringbuf_rec_pg_off(struct bpf_ringbuf *rb,
 374                                      struct bpf_ringbuf_hdr *hdr)
 375 {
 376         return ((void *)hdr - (void *)rb) >> PAGE_SHIFT;
 377 }
 378
 379 /* Given pointer to ring buffer record header, restore pointer to struct
 380  * bpf_ringbuf itself by using page offset stored at offset 4
 381  */
 382 static struct bpf_ringbuf *
 383 bpf_ringbuf_restore_from_rec(struct bpf_ringbuf_hdr *hdr)
 384 {
 385         unsigned long addr = (unsigned long)(void *)hdr;
 386         unsigned long off = (unsigned long)hdr->pg_off << PAGE_SHIFT;
 387
 388         return (void*)((addr & PAGE_MASK) - off);
 389 }
 390
 391 static void *__bpf_ringbuf_reserve(struct bpf_ringbuf *rb, u64 size)
 392 {
 393         unsigned long cons_pos, prod_pos, new_prod_pos, flags;
 394         u32 len, pg_off;
 395         struct bpf_ringbuf_hdr *hdr;
 396
 397         if (unlikely(size > RINGBUF_MAX_RECORD_SZ))
 398                 return NULL;
 399
 400         len = round_up(size + BPF_RINGBUF_HDR_SZ, 8);
 401         if (len > ringbuf_total_data_sz(rb))
 402                 return NULL;
 403
 404         cons_pos = smp_load_acquire(&rb->consumer_pos);
 405
 406         if (in_nmi()) {
 407                 if (!spin_trylock_irqsave(&rb->spinlock, flags))
 408                         return NULL;
 409         } else {
 410                 spin_lock_irqsave(&rb->spinlock, flags);
 411         }
 412
 413         prod_pos = rb->producer_pos;
 414         new_prod_pos = prod_pos + len;
 415
 416         /* check for out of ringbuf space by ensuring producer position
 417          * doesn't advance more than (ringbuf_size - 1) ahead
 418          */
 419         if (new_prod_pos - cons_pos > rb->mask) {
 420                 spin_unlock_irqrestore(&rb->spinlock, flags);
 421                 return NULL;
 422         }
 423
 424         hdr = (void *)rb->data + (prod_pos & rb->mask);
 425         pg_off = bpf_ringbuf_rec_pg_off(rb, hdr);
 426         hdr->len = size | BPF_RINGBUF_BUSY_BIT;
 427         hdr->pg_off = pg_off;
 428
 429         /* pairs with consumer's smp_load_acquire() */
 430         smp_store_release(&rb->producer_pos, new_prod_pos);
 431
 432         spin_unlock_irqrestore(&rb->spinlock, flags);
 433
 434         return (void *)hdr + BPF_RINGBUF_HDR_SZ;
 435 }
 436
 437 BPF_CALL_3(bpf_ringbuf_reserve, struct bpf_map *, map, u64, size, u64, flags)
 438 {
 439         struct bpf_ringbuf_map *rb_map;
 440
 441         if (unlikely(flags))
 442                 return 0;
 443
 444         rb_map = container_of(map, struct bpf_ringbuf_map, map);
 445         return (unsigned long)__bpf_ringbuf_reserve(rb_map->rb, size);
 446 }
 447
 448 const struct bpf_func_proto bpf_ringbuf_reserve_proto = {
 449         .func           = bpf_ringbuf_reserve,
 450         .ret_type       = RET_PTR_TO_ALLOC_MEM_OR_NULL,
 451         .arg1_type      = ARG_CONST_MAP_PTR,
 452         .arg2_type      = ARG_CONST_ALLOC_SIZE_OR_ZERO,
 453         .arg3_type      = ARG_ANYTHING,
 454 };
 455
 456 static void bpf_ringbuf_commit(void *sample, u64 flags, bool discard)
 457 {
 458         unsigned long rec_pos, cons_pos;
 459         struct bpf_ringbuf_hdr *hdr;
 460         struct bpf_ringbuf *rb;
 461         u32 new_len;
 462
 463         hdr = sample - BPF_RINGBUF_HDR_SZ;
 464         rb = bpf_ringbuf_restore_from_rec(hdr);
 465         new_len = hdr->len ^ BPF_RINGBUF_BUSY_BIT;
 466         if (discard)
 467                 new_len |= BPF_RINGBUF_DISCARD_BIT;
 468
 469         /* update record header with correct final size prefix */
 470         xchg(&hdr->len, new_len);
 471
 472         /* if consumer caught up and is waiting for our record, notify about
 473          * new data availability
 474          */
 475         rec_pos = (void *)hdr - (void *)rb->data;
 476         cons_pos = smp_load_acquire(&rb->consumer_pos) & rb->mask;
 477
 478         if (flags & BPF_RB_FORCE_WAKEUP)
 479                 irq_work_queue(&rb->work);
 480         else if (cons_pos == rec_pos && !(flags & BPF_RB_NO_WAKEUP))
 481                 irq_work_queue(&rb->work);
 482 }
 483
 484 BPF_CALL_2(bpf_ringbuf_submit, void *, sample, u64, flags)
 485 {
 486         bpf_ringbuf_commit(sample, flags, false /* discard */);
 487         return 0;
 488 }
 489
 490 const struct bpf_func_proto bpf_ringbuf_submit_proto = {
 491         .func           = bpf_ringbuf_submit,
 492         .ret_type       = RET_VOID,
 493         .arg1_type      = ARG_PTR_TO_ALLOC_MEM | OBJ_RELEASE,
 494         .arg2_type      = ARG_ANYTHING,
 495 };
 496
 497 BPF_CALL_2(bpf_ringbuf_discard, void *, sample, u64, flags)
 498 {
 499         bpf_ringbuf_commit(sample, flags, true /* discard */);
 500         return 0;
 501 }
 502
 503 const struct bpf_func_proto bpf_ringbuf_discard_proto = {
 504         .func           = bpf_ringbuf_discard,
 505         .ret_type       = RET_VOID,
 506         .arg1_type      = ARG_PTR_TO_ALLOC_MEM | OBJ_RELEASE,
 507         .arg2_type      = ARG_ANYTHING,
 508 };
 509
 510 BPF_CALL_4(bpf_ringbuf_output, struct bpf_map *, map, void *, data, u64, size,
 511            u64, flags)
 512 {
 513         struct bpf_ringbuf_map *rb_map;
 514         void *rec;
 515
 516         if (unlikely(flags & ~(BPF_RB_NO_WAKEUP | BPF_RB_FORCE_WAKEUP)))
 517                 return -EINVAL;
 518
 519         rb_map = container_of(map, struct bpf_ringbuf_map, map);
 520         rec = __bpf_ringbuf_reserve(rb_map->rb, size);
 521         if (!rec)
 522                 return -EAGAIN;
 523
 524         memcpy(rec, data, size);
 525         bpf_ringbuf_commit(rec, flags, false /* discard */);
 526         return 0;
 527 }
 528
 529 const struct bpf_func_proto bpf_ringbuf_output_proto = {
 530         .func           = bpf_ringbuf_output,
 531         .ret_type       = RET_INTEGER,
 532         .arg1_type      = ARG_CONST_MAP_PTR,
 533         .arg2_type      = ARG_PTR_TO_MEM | MEM_RDONLY,
 534         .arg3_type      = ARG_CONST_SIZE_OR_ZERO,
 535         .arg4_type      = ARG_ANYTHING,
 536 };
 537
 538 BPF_CALL_2(bpf_ringbuf_query, struct bpf_map *, map, u64, flags)
 539 {
 540         struct bpf_ringbuf *rb;
 541
 542         rb = container_of(map, struct bpf_ringbuf_map, map)->rb;
 543
 544         switch (flags) {
 545         case BPF_RB_AVAIL_DATA:
 546                 return ringbuf_avail_data_sz(rb);
 547         case BPF_RB_RING_SIZE:
 548                 return ringbuf_total_data_sz(rb);
 549         case BPF_RB_CONS_POS:
 550                 return smp_load_acquire(&rb->consumer_pos);
 551         case BPF_RB_PROD_POS:
 552                 return smp_load_acquire(&rb->producer_pos);
 553         default:
 554                 return 0;
 555         }
 556 }
 557
 558 const struct bpf_func_proto bpf_ringbuf_query_proto = {
 559         .func           = bpf_ringbuf_query,
 560         .ret_type       = RET_INTEGER,
 561         .arg1_type      = ARG_CONST_MAP_PTR,
 562         .arg2_type      = ARG_ANYTHING,
 563 };
 564
 565 BPF_CALL_4(bpf_ringbuf_reserve_dynptr, struct bpf_map *, map, u32, size, u64, flags,
 566            struct bpf_dynptr_kern *, ptr)
 567 {
 568         struct bpf_ringbuf_map *rb_map;
 569         void *sample;
 570         int err;
 571
 572         if (unlikely(flags)) {
 573                 bpf_dynptr_set_null(ptr);
 574                 return -EINVAL;
 575         }
 576
 577         err = bpf_dynptr_check_size(size);
 578         if (err) {
 579                 bpf_dynptr_set_null(ptr);
 580                 return err;
 581         }
 582
 583         rb_map = container_of(map, struct bpf_ringbuf_map, map);
 584
 585         sample = __bpf_ringbuf_reserve(rb_map->rb, size);
 586         if (!sample) {
 587                 bpf_dynptr_set_null(ptr);
 588                 return -EINVAL;
 589         }
 590
 591         bpf_dynptr_init(ptr, sample, BPF_DYNPTR_TYPE_RINGBUF, 0, size);
 592
 593         return 0;
 594 }
 595
 596 const struct bpf_func_proto bpf_ringbuf_reserve_dynptr_proto = {
 597         .func           = bpf_ringbuf_reserve_dynptr,
 598         .ret_type       = RET_INTEGER,
 599         .arg1_type      = ARG_CONST_MAP_PTR,
 600         .arg2_type      = ARG_ANYTHING,
 601         .arg3_type      = ARG_ANYTHING,
 602         .arg4_type      = ARG_PTR_TO_DYNPTR | DYNPTR_TYPE_RINGBUF | MEM_UNINIT,
 603 };
 604
 605 BPF_CALL_2(bpf_ringbuf_submit_dynptr, struct bpf_dynptr_kern *, ptr, u64, flags)
 606 {
 607         if (!ptr->data)
 608                 return 0;
 609
 610         bpf_ringbuf_commit(ptr->data, flags, false /* discard */);
 611
 612         bpf_dynptr_set_null(ptr);
 613
 614         return 0;
 615 }
 616
 617 const struct bpf_func_proto bpf_ringbuf_submit_dynptr_proto = {
 618         .func           = bpf_ringbuf_submit_dynptr,
 619         .ret_type       = RET_VOID,
 620         .arg1_type      = ARG_PTR_TO_DYNPTR | DYNPTR_TYPE_RINGBUF | OBJ_RELEASE,
 621         .arg2_type      = ARG_ANYTHING,
 622 };
 623
 624 BPF_CALL_2(bpf_ringbuf_discard_dynptr, struct bpf_dynptr_kern *, ptr, u64, flags)
 625 {
 626         if (!ptr->data)
 627                 return 0;
 628
 629         bpf_ringbuf_commit(ptr->data, flags, true /* discard */);
 630
 631         bpf_dynptr_set_null(ptr);
 632
 633         return 0;
 634 }
 635
 636 const struct bpf_func_proto bpf_ringbuf_discard_dynptr_proto = {
 637         .func           = bpf_ringbuf_discard_dynptr,
 638         .ret_type       = RET_VOID,
 639         .arg1_type      = ARG_PTR_TO_DYNPTR | DYNPTR_TYPE_RINGBUF | OBJ_RELEASE,
 640         .arg2_type      = ARG_ANYTHING,
 641 };
 642
 643 static int __bpf_user_ringbuf_peek(struct bpf_ringbuf *rb, void **sample, u32 *size)
 644 {
 645         int err;
 646         u32 hdr_len, sample_len, total_len, flags, *hdr;
 647         u64 cons_pos, prod_pos;
 648
 649         /* Synchronizes with smp_store_release() in user-space producer. */
 650         prod_pos = smp_load_acquire(&rb->producer_pos);
 651         if (prod_pos % 8)
 652                 return -EINVAL;
 653
 654         /* Synchronizes with smp_store_release() in __bpf_user_ringbuf_sample_release() */
 655         cons_pos = smp_load_acquire(&rb->consumer_pos);
 656         if (cons_pos >= prod_pos)
 657                 return -ENODATA;
 658
 659         hdr = (u32 *)((uintptr_t)rb->data + (uintptr_t)(cons_pos & rb->mask));
 660         /* Synchronizes with smp_store_release() in user-space producer. */
 661         hdr_len = smp_load_acquire(hdr);
 662         flags = hdr_len & (BPF_RINGBUF_BUSY_BIT | BPF_RINGBUF_DISCARD_BIT);
 663         sample_len = hdr_len & ~flags;
 664         total_len = round_up(sample_len + BPF_RINGBUF_HDR_SZ, 8);
 665
 666         /* The sample must fit within the region advertised by the producer position. */
 667         if (total_len > prod_pos - cons_pos)
 668                 return -EINVAL;
 669
 670         /* The sample must fit within the data region of the ring buffer. */
 671         if (total_len > ringbuf_total_data_sz(rb))
 672                 return -E2BIG;
 673
 674         /* The sample must fit into a struct bpf_dynptr. */
 675         err = bpf_dynptr_check_size(sample_len);
 676         if (err)
 677                 return -E2BIG;
 678
 679         if (flags & BPF_RINGBUF_DISCARD_BIT) {
 680                 /* If the discard bit is set, the sample should be skipped.
 681                  *
 682                  * Update the consumer pos, and return -EAGAIN so the caller
 683                  * knows to skip this sample and try to read the next one.
 684                  */
 685                 smp_store_release(&rb->consumer_pos, cons_pos + total_len);
 686                 return -EAGAIN;
 687         }
 688
 689         if (flags & BPF_RINGBUF_BUSY_BIT)
 690                 return -ENODATA;
 691
 692         *sample = (void *)((uintptr_t)rb->data +
 693                            (uintptr_t)((cons_pos + BPF_RINGBUF_HDR_SZ) & rb->mask));
 694         *size = sample_len;
 695         return 0;
 696 }
 697
 698 static void __bpf_user_ringbuf_sample_release(struct bpf_ringbuf *rb, size_t size, u64 flags)
 699 {
 700         u64 consumer_pos;
 701         u32 rounded_size = round_up(size + BPF_RINGBUF_HDR_SZ, 8);
 702
 703         /* Using smp_load_acquire() is unnecessary here, as the busy-bit
 704          * prevents another task from writing to consumer_pos after it was read
 705          * by this task with smp_load_acquire() in __bpf_user_ringbuf_peek().
 706          */
 707         consumer_pos = rb->consumer_pos;
 708          /* Synchronizes with smp_load_acquire() in user-space producer. */
 709         smp_store_release(&rb->consumer_pos, consumer_pos + rounded_size);
 710 }
 711
 712 BPF_CALL_4(bpf_user_ringbuf_drain, struct bpf_map *, map,
 713            void *, callback_fn, void *, callback_ctx, u64, flags)
 714 {
 715         struct bpf_ringbuf *rb;
 716         long samples, discarded_samples = 0, ret = 0;
 717         bpf_callback_t callback = (bpf_callback_t)callback_fn;
 718         u64 wakeup_flags = BPF_RB_NO_WAKEUP | BPF_RB_FORCE_WAKEUP;
 719         int busy = 0;
 720
 721         if (unlikely(flags & ~wakeup_flags))
 722                 return -EINVAL;
 723
 724         rb = container_of(map, struct bpf_ringbuf_map, map)->rb;
 725
 726         /* If another consumer is already consuming a sample, wait for them to finish. */
 727         if (!atomic_try_cmpxchg(&rb->busy, &busy, 1))
 728                 return -EBUSY;
 729
 730         for (samples = 0; samples < BPF_MAX_USER_RINGBUF_SAMPLES && ret == 0; samples++) {
 731                 int err;
 732                 u32 size;
 733                 void *sample;
 734                 struct bpf_dynptr_kern dynptr;
 735
 736                 err = __bpf_user_ringbuf_peek(rb, &sample, &size);
 737                 if (err) {
 738                         if (err == -ENODATA) {
 739                                 break;
 740                         } else if (err == -EAGAIN) {
 741                                 discarded_samples++;
 742                                 continue;
 743                         } else {
 744                                 ret = err;
 745                                 goto schedule_work_return;
 746                         }
 747                 }
 748
 749                 bpf_dynptr_init(&dynptr, sample, BPF_DYNPTR_TYPE_LOCAL, 0, size);
 750                 ret = callback((uintptr_t)&dynptr, (uintptr_t)callback_ctx, 0, 0, 0);
 751                 __bpf_user_ringbuf_sample_release(rb, size, flags);
 752         }
 753         ret = samples - discarded_samples;
 754
 755 schedule_work_return:
 756         /* Prevent the clearing of the busy-bit from being reordered before the
 757          * storing of any rb consumer or producer positions.
 758          */
 759         smp_mb__before_atomic();
 760         atomic_set(&rb->busy, 0);
 761
 762         if (flags & BPF_RB_FORCE_WAKEUP)
 763                 irq_work_queue(&rb->work);
 764         else if (!(flags & BPF_RB_NO_WAKEUP) && samples > 0)
 765                 irq_work_queue(&rb->work);
 766         return ret;
 767 }
 768
 769 const struct bpf_func_proto bpf_user_ringbuf_drain_proto = {
 770         .func           = bpf_user_ringbuf_drain,
 771         .ret_type       = RET_INTEGER,
 772         .arg1_type      = ARG_CONST_MAP_PTR,
 773         .arg2_type      = ARG_PTR_TO_FUNC,
 774         .arg3_type      = ARG_PTR_TO_STACK_OR_NULL,
 775         .arg4_type      = ARG_ANYTHING,
 776 };