projects/linux·blob master · refs

880 lines · 25.3 KiB

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