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 <uapi/linux/btf.h> 12 13 #define RINGBUF_CREATE_FLAG_MASK (BPF_F_NUMA_NODE) 14 15 /* non-mmap()'able part of bpf_ringbuf (everything up to consumer page) */ 16 #define RINGBUF_PGOFF \ 17 (offsetof(struct bpf_ringbuf, consumer_pos) >> PAGE_SHIFT) 18 /* consumer page and producer page */ 19 #define RINGBUF_POS_PAGES 2 20 21 #define RINGBUF_MAX_RECORD_SZ (UINT_MAX/4) 22 23 /* Maximum size of ring buffer area is limited by 32-bit page offset within 24 * record header, counted in pages. Reserve 8 bits for extensibility, and take 25 * into account few extra pages for consumer/producer pages and 26 * non-mmap()'able parts. This gives 64GB limit, which seems plenty for single 27 * ring buffer. 28 */ 29 #define RINGBUF_MAX_DATA_SZ \ 30 (((1ULL << 24) - RINGBUF_POS_PAGES - RINGBUF_PGOFF) * PAGE_SIZE) 31 32 struct bpf_ringbuf { 33 wait_queue_head_t waitq; 34 struct irq_work work; 35 u64 mask; 36 struct page **pages; 37 int nr_pages; 38 spinlock_t spinlock ____cacheline_aligned_in_smp; 39 /* Consumer and producer counters are put into separate pages to allow 40 * mapping consumer page as r/w, but restrict producer page to r/o. 41 * This protects producer position from being modified by user-space 42 * application and ruining in-kernel position tracking. 43 */ 44 unsigned long consumer_pos __aligned(PAGE_SIZE); 45 unsigned long producer_pos __aligned(PAGE_SIZE); 46 char data[] __aligned(PAGE_SIZE); 47 }; 48 49 struct bpf_ringbuf_map { 50 struct bpf_map map; 51 struct bpf_map_memory memory; 52 struct bpf_ringbuf *rb; 53 }; 54 55 /* 8-byte ring buffer record header structure */ 56 struct bpf_ringbuf_hdr { 57 u32 len; 58 u32 pg_off; 59 }; 60 61 static struct bpf_ringbuf *bpf_ringbuf_area_alloc(size_t data_sz, int numa_node) 62 { 63 const gfp_t flags = GFP_KERNEL | __GFP_RETRY_MAYFAIL | __GFP_NOWARN | 64 __GFP_ZERO; 65 int nr_meta_pages = RINGBUF_PGOFF + RINGBUF_POS_PAGES; 66 int nr_data_pages = data_sz >> PAGE_SHIFT; 67 int nr_pages = nr_meta_pages + nr_data_pages; 68 struct page **pages, *page; 69 struct bpf_ringbuf *rb; 70 size_t array_size; 71 int i; 72 73 /* Each data page is mapped twice to allow "virtual" 74 * continuous read of samples wrapping around the end of ring 75 * buffer area: 76 * ------------------------------------------------------ 77 * | meta pages | real data pages | same data pages | 78 * ------------------------------------------------------ 79 * | | 1 2 3 4 5 6 7 8 9 | 1 2 3 4 5 6 7 8 9 | 80 * ------------------------------------------------------ 81 * | | TA DA | TA DA | 82 * ------------------------------------------------------ 83 * ^^^^^^^ 84 * | 85 * Here, no need to worry about special handling of wrapped-around 86 * data due to double-mapped data pages. This works both in kernel and 87 * when mmap()'ed in user-space, simplifying both kernel and 88 * user-space implementations significantly. 89 */ 90 array_size = (nr_meta_pages + 2 * nr_data_pages) * sizeof(*pages); 91 if (array_size > PAGE_SIZE) 92 pages = vmalloc_node(array_size, numa_node); 93 else 94 pages = kmalloc_node(array_size, flags, numa_node); 95 if (!pages) 96 return NULL; 97 98 for (i = 0; i < nr_pages; i++) { 99 page = alloc_pages_node(numa_node, flags, 0); 100 if (!page) { 101 nr_pages = i; 102 goto err_free_pages; 103 } 104 pages[i] = page; 105 if (i >= nr_meta_pages) 106 pages[nr_data_pages + i] = page; 107 } 108 109 rb = vmap(pages, nr_meta_pages + 2 * nr_data_pages, 110 VM_ALLOC | VM_USERMAP, PAGE_KERNEL); 111 if (rb) { 112 rb->pages = pages; 113 rb->nr_pages = nr_pages; 114 return rb; 115 } 116 117 err_free_pages: 118 for (i = 0; i < nr_pages; i++) 119 __free_page(pages[i]); 120 kvfree(pages); 121 return NULL; 122 } 123 124 static void bpf_ringbuf_notify(struct irq_work *work) 125 { 126 struct bpf_ringbuf *rb = container_of(work, struct bpf_ringbuf, work); 127 128 wake_up_all(&rb->waitq); 129 } 130 131 static struct bpf_ringbuf *bpf_ringbuf_alloc(size_t data_sz, int numa_node) 132 { 133 struct bpf_ringbuf *rb; 134 135 if (!data_sz || !PAGE_ALIGNED(data_sz)) 136 return ERR_PTR(-EINVAL); 137 138 #ifdef CONFIG_64BIT 139 /* on 32-bit arch, it's impossible to overflow record's hdr->pgoff */ 140 if (data_sz > RINGBUF_MAX_DATA_SZ) 141 return ERR_PTR(-E2BIG); 142 #endif 143 144 rb = bpf_ringbuf_area_alloc(data_sz, numa_node); 145 if (!rb) 146 return ERR_PTR(-ENOMEM); 147 148 spin_lock_init(&rb->spinlock); 149 init_waitqueue_head(&rb->waitq); 150 init_irq_work(&rb->work, bpf_ringbuf_notify); 151 152 rb->mask = data_sz - 1; 153 rb->consumer_pos = 0; 154 rb->producer_pos = 0; 155 156 return rb; 157 } 158 159 static struct bpf_map *ringbuf_map_alloc(union bpf_attr *attr) 160 { 161 struct bpf_ringbuf_map *rb_map; 162 u64 cost; 163 int err; 164 165 if (attr->map_flags & ~RINGBUF_CREATE_FLAG_MASK) 166 return ERR_PTR(-EINVAL); 167 168 if (attr->key_size || attr->value_size || 169 attr->max_entries == 0 || !PAGE_ALIGNED(attr->max_entries)) 170 return ERR_PTR(-EINVAL); 171 172 rb_map = kzalloc(sizeof(*rb_map), GFP_USER); 173 if (!rb_map) 174 return ERR_PTR(-ENOMEM); 175 176 bpf_map_init_from_attr(&rb_map->map, attr); 177 178 cost = sizeof(struct bpf_ringbuf_map) + 179 sizeof(struct bpf_ringbuf) + 180 attr->max_entries; 181 err = bpf_map_charge_init(&rb_map->map.memory, cost); 182 if (err) 183 goto err_free_map; 184 185 rb_map->rb = bpf_ringbuf_alloc(attr->max_entries, rb_map->map.numa_node); 186 if (IS_ERR(rb_map->rb)) { 187 err = PTR_ERR(rb_map->rb); 188 goto err_uncharge; 189 } 190 191 return &rb_map->map; 192 193 err_uncharge: 194 bpf_map_charge_finish(&rb_map->map.memory); 195 err_free_map: 196 kfree(rb_map); 197 return ERR_PTR(err); 198 } 199 200 static void bpf_ringbuf_free(struct bpf_ringbuf *rb) 201 { 202 /* copy pages pointer and nr_pages to local variable, as we are going 203 * to unmap rb itself with vunmap() below 204 */ 205 struct page **pages = rb->pages; 206 int i, nr_pages = rb->nr_pages; 207 208 vunmap(rb); 209 for (i = 0; i < nr_pages; i++) 210 __free_page(pages[i]); 211 kvfree(pages); 212 } 213 214 static void ringbuf_map_free(struct bpf_map *map) 215 { 216 struct bpf_ringbuf_map *rb_map; 217 218 /* at this point bpf_prog->aux->refcnt == 0 and this map->refcnt == 0, 219 * so the programs (can be more than one that used this map) were 220 * disconnected from events. Wait for outstanding critical sections in 221 * these programs to complete 222 */ 223 synchronize_rcu(); 224 225 rb_map = container_of(map, struct bpf_ringbuf_map, map); 226 bpf_ringbuf_free(rb_map->rb); 227 kfree(rb_map); 228 } 229 230 static void *ringbuf_map_lookup_elem(struct bpf_map *map, void *key) 231 { 232 return ERR_PTR(-ENOTSUPP); 233 } 234 235 static int ringbuf_map_update_elem(struct bpf_map *map, void *key, void *value, 236 u64 flags) 237 { 238 return -ENOTSUPP; 239 } 240 241 static int ringbuf_map_delete_elem(struct bpf_map *map, void *key) 242 { 243 return -ENOTSUPP; 244 } 245 246 static int ringbuf_map_get_next_key(struct bpf_map *map, void *key, 247 void *next_key) 248 { 249 return -ENOTSUPP; 250 } 251 252 static size_t bpf_ringbuf_mmap_page_cnt(const struct bpf_ringbuf *rb) 253 { 254 size_t data_pages = (rb->mask + 1) >> PAGE_SHIFT; 255 256 /* consumer page + producer page + 2 x data pages */ 257 return RINGBUF_POS_PAGES + 2 * data_pages; 258 } 259 260 static int ringbuf_map_mmap(struct bpf_map *map, struct vm_area_struct *vma) 261 { 262 struct bpf_ringbuf_map *rb_map; 263 size_t mmap_sz; 264 265 rb_map = container_of(map, struct bpf_ringbuf_map, map); 266 mmap_sz = bpf_ringbuf_mmap_page_cnt(rb_map->rb) << PAGE_SHIFT; 267 268 if (vma->vm_pgoff * PAGE_SIZE + (vma->vm_end - vma->vm_start) > mmap_sz) 269 return -EINVAL; 270 271 return remap_vmalloc_range(vma, rb_map->rb, 272 vma->vm_pgoff + RINGBUF_PGOFF); 273 } 274 275 static unsigned long ringbuf_avail_data_sz(struct bpf_ringbuf *rb) 276 { 277 unsigned long cons_pos, prod_pos; 278 279 cons_pos = smp_load_acquire(&rb->consumer_pos); 280 prod_pos = smp_load_acquire(&rb->producer_pos); 281 return prod_pos - cons_pos; 282 } 283 284 static __poll_t ringbuf_map_poll(struct bpf_map *map, struct file *filp, 285 struct poll_table_struct *pts) 286 { 287 struct bpf_ringbuf_map *rb_map; 288 289 rb_map = container_of(map, struct bpf_ringbuf_map, map); 290 poll_wait(filp, &rb_map->rb->waitq, pts); 291 292 if (ringbuf_avail_data_sz(rb_map->rb)) 293 return EPOLLIN | EPOLLRDNORM; 294 return 0; 295 } 296 297 static int ringbuf_map_btf_id; 298 const struct bpf_map_ops ringbuf_map_ops = { 299 .map_alloc = ringbuf_map_alloc, 300 .map_free = ringbuf_map_free, 301 .map_mmap = ringbuf_map_mmap, 302 .map_poll = ringbuf_map_poll, 303 .map_lookup_elem = ringbuf_map_lookup_elem, 304 .map_update_elem = ringbuf_map_update_elem, 305 .map_delete_elem = ringbuf_map_delete_elem, 306 .map_get_next_key = ringbuf_map_get_next_key, 307 .map_btf_name = "bpf_ringbuf_map", 308 .map_btf_id = &ringbuf_map_btf_id, 309 }; 310 311 /* Given pointer to ring buffer record metadata and struct bpf_ringbuf itself, 312 * calculate offset from record metadata to ring buffer in pages, rounded 313 * down. This page offset is stored as part of record metadata and allows to 314 * restore struct bpf_ringbuf * from record pointer. This page offset is 315 * stored at offset 4 of record metadata header. 316 */ 317 static size_t bpf_ringbuf_rec_pg_off(struct bpf_ringbuf *rb, 318 struct bpf_ringbuf_hdr *hdr) 319 { 320 return ((void *)hdr - (void *)rb) >> PAGE_SHIFT; 321 } 322 323 /* Given pointer to ring buffer record header, restore pointer to struct 324 * bpf_ringbuf itself by using page offset stored at offset 4 325 */ 326 static struct bpf_ringbuf * 327 bpf_ringbuf_restore_from_rec(struct bpf_ringbuf_hdr *hdr) 328 { 329 unsigned long addr = (unsigned long)(void *)hdr; 330 unsigned long off = (unsigned long)hdr->pg_off << PAGE_SHIFT; 331 332 return (void*)((addr & PAGE_MASK) - off); 333 } 334 335 static void *__bpf_ringbuf_reserve(struct bpf_ringbuf *rb, u64 size) 336 { 337 unsigned long cons_pos, prod_pos, new_prod_pos, flags; 338 u32 len, pg_off; 339 struct bpf_ringbuf_hdr *hdr; 340 341 if (unlikely(size > RINGBUF_MAX_RECORD_SZ)) 342 return NULL; 343 344 len = round_up(size + BPF_RINGBUF_HDR_SZ, 8); 345 cons_pos = smp_load_acquire(&rb->consumer_pos); 346 347 if (in_nmi()) { 348 if (!spin_trylock_irqsave(&rb->spinlock, flags)) 349 return NULL; 350 } else { 351 spin_lock_irqsave(&rb->spinlock, flags); 352 } 353 354 prod_pos = rb->producer_pos; 355 new_prod_pos = prod_pos + len; 356 357 /* check for out of ringbuf space by ensuring producer position 358 * doesn't advance more than (ringbuf_size - 1) ahead 359 */ 360 if (new_prod_pos - cons_pos > rb->mask) { 361 spin_unlock_irqrestore(&rb->spinlock, flags); 362 return NULL; 363 } 364 365 hdr = (void *)rb->data + (prod_pos & rb->mask); 366 pg_off = bpf_ringbuf_rec_pg_off(rb, hdr); 367 hdr->len = size | BPF_RINGBUF_BUSY_BIT; 368 hdr->pg_off = pg_off; 369 370 /* pairs with consumer's smp_load_acquire() */ 371 smp_store_release(&rb->producer_pos, new_prod_pos); 372 373 spin_unlock_irqrestore(&rb->spinlock, flags); 374 375 return (void *)hdr + BPF_RINGBUF_HDR_SZ; 376 } 377 378 BPF_CALL_3(bpf_ringbuf_reserve, struct bpf_map *, map, u64, size, u64, flags) 379 { 380 struct bpf_ringbuf_map *rb_map; 381 382 if (unlikely(flags)) 383 return 0; 384 385 rb_map = container_of(map, struct bpf_ringbuf_map, map); 386 return (unsigned long)__bpf_ringbuf_reserve(rb_map->rb, size); 387 } 388 389 const struct bpf_func_proto bpf_ringbuf_reserve_proto = { 390 .func = bpf_ringbuf_reserve, 391 .ret_type = RET_PTR_TO_ALLOC_MEM_OR_NULL, 392 .arg1_type = ARG_CONST_MAP_PTR, 393 .arg2_type = ARG_CONST_ALLOC_SIZE_OR_ZERO, 394 .arg3_type = ARG_ANYTHING, 395 }; 396 397 static void bpf_ringbuf_commit(void *sample, u64 flags, bool discard) 398 { 399 unsigned long rec_pos, cons_pos; 400 struct bpf_ringbuf_hdr *hdr; 401 struct bpf_ringbuf *rb; 402 u32 new_len; 403 404 hdr = sample - BPF_RINGBUF_HDR_SZ; 405 rb = bpf_ringbuf_restore_from_rec(hdr); 406 new_len = hdr->len ^ BPF_RINGBUF_BUSY_BIT; 407 if (discard) 408 new_len |= BPF_RINGBUF_DISCARD_BIT; 409 410 /* update record header with correct final size prefix */ 411 xchg(&hdr->len, new_len); 412 413 /* if consumer caught up and is waiting for our record, notify about 414 * new data availability 415 */ 416 rec_pos = (void *)hdr - (void *)rb->data; 417 cons_pos = smp_load_acquire(&rb->consumer_pos) & rb->mask; 418 419 if (flags & BPF_RB_FORCE_WAKEUP) 420 irq_work_queue(&rb->work); 421 else if (cons_pos == rec_pos && !(flags & BPF_RB_NO_WAKEUP)) 422 irq_work_queue(&rb->work); 423 } 424 425 BPF_CALL_2(bpf_ringbuf_submit, void *, sample, u64, flags) 426 { 427 bpf_ringbuf_commit(sample, flags, false /* discard */); 428 return 0; 429 } 430 431 const struct bpf_func_proto bpf_ringbuf_submit_proto = { 432 .func = bpf_ringbuf_submit, 433 .ret_type = RET_VOID, 434 .arg1_type = ARG_PTR_TO_ALLOC_MEM, 435 .arg2_type = ARG_ANYTHING, 436 }; 437 438 BPF_CALL_2(bpf_ringbuf_discard, void *, sample, u64, flags) 439 { 440 bpf_ringbuf_commit(sample, flags, true /* discard */); 441 return 0; 442 } 443 444 const struct bpf_func_proto bpf_ringbuf_discard_proto = { 445 .func = bpf_ringbuf_discard, 446 .ret_type = RET_VOID, 447 .arg1_type = ARG_PTR_TO_ALLOC_MEM, 448 .arg2_type = ARG_ANYTHING, 449 }; 450 451 BPF_CALL_4(bpf_ringbuf_output, struct bpf_map *, map, void *, data, u64, size, 452 u64, flags) 453 { 454 struct bpf_ringbuf_map *rb_map; 455 void *rec; 456 457 if (unlikely(flags & ~(BPF_RB_NO_WAKEUP | BPF_RB_FORCE_WAKEUP))) 458 return -EINVAL; 459 460 rb_map = container_of(map, struct bpf_ringbuf_map, map); 461 rec = __bpf_ringbuf_reserve(rb_map->rb, size); 462 if (!rec) 463 return -EAGAIN; 464 465 memcpy(rec, data, size); 466 bpf_ringbuf_commit(rec, flags, false /* discard */); 467 return 0; 468 } 469 470 const struct bpf_func_proto bpf_ringbuf_output_proto = { 471 .func = bpf_ringbuf_output, 472 .ret_type = RET_INTEGER, 473 .arg1_type = ARG_CONST_MAP_PTR, 474 .arg2_type = ARG_PTR_TO_MEM, 475 .arg3_type = ARG_CONST_SIZE_OR_ZERO, 476 .arg4_type = ARG_ANYTHING, 477 }; 478 479 BPF_CALL_2(bpf_ringbuf_query, struct bpf_map *, map, u64, flags) 480 { 481 struct bpf_ringbuf *rb; 482 483 rb = container_of(map, struct bpf_ringbuf_map, map)->rb; 484 485 switch (flags) { 486 case BPF_RB_AVAIL_DATA: 487 return ringbuf_avail_data_sz(rb); 488 case BPF_RB_RING_SIZE: 489 return rb->mask + 1; 490 case BPF_RB_CONS_POS: 491 return smp_load_acquire(&rb->consumer_pos); 492 case BPF_RB_PROD_POS: 493 return smp_load_acquire(&rb->producer_pos); 494 default: 495 return 0; 496 } 497 } 498 499 const struct bpf_func_proto bpf_ringbuf_query_proto = { 500 .func = bpf_ringbuf_query, 501 .ret_type = RET_INTEGER, 502 .arg1_type = ARG_CONST_MAP_PTR, 503 .arg2_type = ARG_ANYTHING, 504 }; 505