1 // SPDX-License-Identifier: (LGPL-2.1 OR BSD-2-Clause) 2 /* 3 * Ring buffer operations. 4 * 5 * Copyright (C) 2020 Facebook, Inc. 6 */ 7 #ifndef _GNU_SOURCE 8 #define _GNU_SOURCE 9 #endif 10 #include <stdlib.h> 11 #include <stdio.h> 12 #include <errno.h> 13 #include <unistd.h> 14 #include <linux/err.h> 15 #include <linux/bpf.h> 16 #include <asm/barrier.h> 17 #include <sys/mman.h> 18 #include <sys/epoll.h> 19 #include <time.h> 20 21 #include "libbpf.h" 22 #include "libbpf_internal.h" 23 #include "bpf.h" 24 25 struct ring { 26 ring_buffer_sample_fn sample_cb; 27 void *ctx; 28 void *data; 29 unsigned long *consumer_pos; 30 unsigned long *producer_pos; 31 unsigned long mask; 32 int map_fd; 33 }; 34 35 struct ring_buffer { 36 struct epoll_event *events; 37 struct ring **rings; 38 size_t page_size; 39 int epoll_fd; 40 int ring_cnt; 41 }; 42 43 struct user_ring_buffer { 44 struct epoll_event event; 45 unsigned long *consumer_pos; 46 unsigned long *producer_pos; 47 void *data; 48 unsigned long mask; 49 size_t page_size; 50 int map_fd; 51 int epoll_fd; 52 }; 53 54 /* 8-byte ring buffer header structure */ 55 struct ringbuf_hdr { 56 __u32 len; 57 __u32 pad; 58 }; 59 60 static void ringbuf_free_ring(struct ring_buffer *rb, struct ring *r) 61 { 62 if (r->consumer_pos) { 63 munmap(r->consumer_pos, rb->page_size); 64 r->consumer_pos = NULL; 65 } 66 if (r->producer_pos) { 67 munmap(r->producer_pos, rb->page_size + 2 * (r->mask + 1)); 68 r->producer_pos = NULL; 69 } 70 71 free(r); 72 } 73 74 /* Add extra RINGBUF maps to this ring buffer manager */ 75 int ring_buffer__add(struct ring_buffer *rb, int map_fd, 76 ring_buffer_sample_fn sample_cb, void *ctx) 77 { 78 struct bpf_map_info info; 79 __u32 len = sizeof(info); 80 struct epoll_event *e; 81 struct ring *r; 82 __u64 mmap_sz; 83 void *tmp; 84 int err; 85 86 memset(&info, 0, sizeof(info)); 87 88 err = bpf_map_get_info_by_fd(map_fd, &info, &len); 89 if (err) { 90 err = -errno; 91 pr_warn("ringbuf: failed to get map info for fd=%d: %d\n", 92 map_fd, err); 93 return libbpf_err(err); 94 } 95 96 if (info.type != BPF_MAP_TYPE_RINGBUF) { 97 pr_warn("ringbuf: map fd=%d is not BPF_MAP_TYPE_RINGBUF\n", 98 map_fd); 99 return libbpf_err(-EINVAL); 100 } 101 102 tmp = libbpf_reallocarray(rb->rings, rb->ring_cnt + 1, sizeof(*rb->rings)); 103 if (!tmp) 104 return libbpf_err(-ENOMEM); 105 rb->rings = tmp; 106 107 tmp = libbpf_reallocarray(rb->events, rb->ring_cnt + 1, sizeof(*rb->events)); 108 if (!tmp) 109 return libbpf_err(-ENOMEM); 110 rb->events = tmp; 111 112 r = calloc(1, sizeof(*r)); 113 if (!r) 114 return libbpf_err(-ENOMEM); 115 rb->rings[rb->ring_cnt] = r; 116 117 r->map_fd = map_fd; 118 r->sample_cb = sample_cb; 119 r->ctx = ctx; 120 r->mask = info.max_entries - 1; 121 122 /* Map writable consumer page */ 123 tmp = mmap(NULL, rb->page_size, PROT_READ | PROT_WRITE, MAP_SHARED, map_fd, 0); 124 if (tmp == MAP_FAILED) { 125 err = -errno; 126 pr_warn("ringbuf: failed to mmap consumer page for map fd=%d: %d\n", 127 map_fd, err); 128 goto err_out; 129 } 130 r->consumer_pos = tmp; 131 132 /* Map read-only producer page and data pages. We map twice as big 133 * data size to allow simple reading of samples that wrap around the 134 * end of a ring buffer. See kernel implementation for details. 135 */ 136 mmap_sz = rb->page_size + 2 * (__u64)info.max_entries; 137 if (mmap_sz != (__u64)(size_t)mmap_sz) { 138 err = -E2BIG; 139 pr_warn("ringbuf: ring buffer size (%u) is too big\n", info.max_entries); 140 goto err_out; 141 } 142 tmp = mmap(NULL, (size_t)mmap_sz, PROT_READ, MAP_SHARED, map_fd, rb->page_size); 143 if (tmp == MAP_FAILED) { 144 err = -errno; 145 pr_warn("ringbuf: failed to mmap data pages for map fd=%d: %d\n", 146 map_fd, err); 147 goto err_out; 148 } 149 r->producer_pos = tmp; 150 r->data = tmp + rb->page_size; 151 152 e = &rb->events[rb->ring_cnt]; 153 memset(e, 0, sizeof(*e)); 154 155 e->events = EPOLLIN; 156 e->data.fd = rb->ring_cnt; 157 if (epoll_ctl(rb->epoll_fd, EPOLL_CTL_ADD, map_fd, e) < 0) { 158 err = -errno; 159 pr_warn("ringbuf: failed to epoll add map fd=%d: %d\n", 160 map_fd, err); 161 goto err_out; 162 } 163 164 rb->ring_cnt++; 165 return 0; 166 167 err_out: 168 ringbuf_free_ring(rb, r); 169 return libbpf_err(err); 170 } 171 172 void ring_buffer__free(struct ring_buffer *rb) 173 { 174 int i; 175 176 if (!rb) 177 return; 178 179 for (i = 0; i < rb->ring_cnt; ++i) 180 ringbuf_free_ring(rb, rb->rings[i]); 181 if (rb->epoll_fd >= 0) 182 close(rb->epoll_fd); 183 184 free(rb->events); 185 free(rb->rings); 186 free(rb); 187 } 188 189 struct ring_buffer * 190 ring_buffer__new(int map_fd, ring_buffer_sample_fn sample_cb, void *ctx, 191 const struct ring_buffer_opts *opts) 192 { 193 struct ring_buffer *rb; 194 int err; 195 196 if (!OPTS_VALID(opts, ring_buffer_opts)) 197 return errno = EINVAL, NULL; 198 199 rb = calloc(1, sizeof(*rb)); 200 if (!rb) 201 return errno = ENOMEM, NULL; 202 203 rb->page_size = getpagesize(); 204 205 rb->epoll_fd = epoll_create1(EPOLL_CLOEXEC); 206 if (rb->epoll_fd < 0) { 207 err = -errno; 208 pr_warn("ringbuf: failed to create epoll instance: %d\n", err); 209 goto err_out; 210 } 211 212 err = ring_buffer__add(rb, map_fd, sample_cb, ctx); 213 if (err) 214 goto err_out; 215 216 return rb; 217 218 err_out: 219 ring_buffer__free(rb); 220 return errno = -err, NULL; 221 } 222 223 static inline int roundup_len(__u32 len) 224 { 225 /* clear out top 2 bits (discard and busy, if set) */ 226 len <<= 2; 227 len >>= 2; 228 /* add length prefix */ 229 len += BPF_RINGBUF_HDR_SZ; 230 /* round up to 8 byte alignment */ 231 return (len + 7) / 8 * 8; 232 } 233 234 static int64_t ringbuf_process_ring(struct ring *r, size_t n) 235 { 236 int *len_ptr, len, err; 237 /* 64-bit to avoid overflow in case of extreme application behavior */ 238 int64_t cnt = 0; 239 unsigned long cons_pos, prod_pos; 240 bool got_new_data; 241 void *sample; 242 243 cons_pos = smp_load_acquire(r->consumer_pos); 244 do { 245 got_new_data = false; 246 prod_pos = smp_load_acquire(r->producer_pos); 247 while (cons_pos < prod_pos) { 248 len_ptr = r->data + (cons_pos & r->mask); 249 len = smp_load_acquire(len_ptr); 250 251 /* sample not committed yet, bail out for now */ 252 if (len & BPF_RINGBUF_BUSY_BIT) 253 goto done; 254 255 got_new_data = true; 256 cons_pos += roundup_len(len); 257 258 if ((len & BPF_RINGBUF_DISCARD_BIT) == 0) { 259 sample = (void *)len_ptr + BPF_RINGBUF_HDR_SZ; 260 err = r->sample_cb(r->ctx, sample, len); 261 if (err < 0) { 262 /* update consumer pos and bail out */ 263 smp_store_release(r->consumer_pos, 264 cons_pos); 265 return err; 266 } 267 cnt++; 268 } 269 270 smp_store_release(r->consumer_pos, cons_pos); 271 272 if (cnt >= n) 273 goto done; 274 } 275 } while (got_new_data); 276 done: 277 return cnt; 278 } 279 280 /* Consume available ring buffer(s) data without event polling, up to n 281 * records. 282 * 283 * Returns number of records consumed across all registered ring buffers (or 284 * n, whichever is less), or negative number if any of the callbacks return 285 * error. 286 */ 287 int ring_buffer__consume_n(struct ring_buffer *rb, size_t n) 288 { 289 int64_t err, res = 0; 290 int i; 291 292 for (i = 0; i < rb->ring_cnt; i++) { 293 struct ring *ring = rb->rings[i]; 294 295 err = ringbuf_process_ring(ring, n); 296 if (err < 0) 297 return libbpf_err(err); 298 res += err; 299 n -= err; 300 301 if (n == 0) 302 break; 303 } 304 return res > INT_MAX ? INT_MAX : res; 305 } 306 307 /* Consume available ring buffer(s) data without event polling. 308 * Returns number of records consumed across all registered ring buffers (or 309 * INT_MAX, whichever is less), or negative number if any of the callbacks 310 * return error. 311 */ 312 int ring_buffer__consume(struct ring_buffer *rb) 313 { 314 int64_t err, res = 0; 315 int i; 316 317 for (i = 0; i < rb->ring_cnt; i++) { 318 struct ring *ring = rb->rings[i]; 319 320 err = ringbuf_process_ring(ring, INT_MAX); 321 if (err < 0) 322 return libbpf_err(err); 323 res += err; 324 if (res > INT_MAX) { 325 res = INT_MAX; 326 break; 327 } 328 } 329 return res; 330 } 331 332 /* Poll for available data and consume records, if any are available. 333 * Returns number of records consumed (or INT_MAX, whichever is less), or 334 * negative number, if any of the registered callbacks returned error. 335 */ 336 int ring_buffer__poll(struct ring_buffer *rb, int timeout_ms) 337 { 338 int i, cnt; 339 int64_t err, res = 0; 340 341 cnt = epoll_wait(rb->epoll_fd, rb->events, rb->ring_cnt, timeout_ms); 342 if (cnt < 0) 343 return libbpf_err(-errno); 344 345 for (i = 0; i < cnt; i++) { 346 __u32 ring_id = rb->events[i].data.fd; 347 struct ring *ring = rb->rings[ring_id]; 348 349 err = ringbuf_process_ring(ring, INT_MAX); 350 if (err < 0) 351 return libbpf_err(err); 352 res += err; 353 } 354 if (res > INT_MAX) 355 res = INT_MAX; 356 return res; 357 } 358 359 /* Get an fd that can be used to sleep until data is available in the ring(s) */ 360 int ring_buffer__epoll_fd(const struct ring_buffer *rb) 361 { 362 return rb->epoll_fd; 363 } 364 365 struct ring *ring_buffer__ring(struct ring_buffer *rb, unsigned int idx) 366 { 367 if (idx >= rb->ring_cnt) 368 return errno = ERANGE, NULL; 369 370 return rb->rings[idx]; 371 } 372 373 unsigned long ring__consumer_pos(const struct ring *r) 374 { 375 /* Synchronizes with smp_store_release() in ringbuf_process_ring(). */ 376 return smp_load_acquire(r->consumer_pos); 377 } 378 379 unsigned long ring__producer_pos(const struct ring *r) 380 { 381 /* Synchronizes with smp_store_release() in __bpf_ringbuf_reserve() in 382 * the kernel. 383 */ 384 return smp_load_acquire(r->producer_pos); 385 } 386 387 size_t ring__avail_data_size(const struct ring *r) 388 { 389 unsigned long cons_pos, prod_pos; 390 391 cons_pos = ring__consumer_pos(r); 392 prod_pos = ring__producer_pos(r); 393 return prod_pos - cons_pos; 394 } 395 396 size_t ring__size(const struct ring *r) 397 { 398 return r->mask + 1; 399 } 400 401 int ring__map_fd(const struct ring *r) 402 { 403 return r->map_fd; 404 } 405 406 int ring__consume_n(struct ring *r, size_t n) 407 { 408 int64_t res; 409 410 res = ringbuf_process_ring(r, n); 411 if (res < 0) 412 return libbpf_err(res); 413 414 return res > INT_MAX ? INT_MAX : res; 415 } 416 417 int ring__consume(struct ring *r) 418 { 419 return ring__consume_n(r, INT_MAX); 420 } 421 422 static void user_ringbuf_unmap_ring(struct user_ring_buffer *rb) 423 { 424 if (rb->consumer_pos) { 425 munmap(rb->consumer_pos, rb->page_size); 426 rb->consumer_pos = NULL; 427 } 428 if (rb->producer_pos) { 429 munmap(rb->producer_pos, rb->page_size + 2 * (rb->mask + 1)); 430 rb->producer_pos = NULL; 431 } 432 } 433 434 void user_ring_buffer__free(struct user_ring_buffer *rb) 435 { 436 if (!rb) 437 return; 438 439 user_ringbuf_unmap_ring(rb); 440 441 if (rb->epoll_fd >= 0) 442 close(rb->epoll_fd); 443 444 free(rb); 445 } 446 447 static int user_ringbuf_map(struct user_ring_buffer *rb, int map_fd) 448 { 449 struct bpf_map_info info; 450 __u32 len = sizeof(info); 451 __u64 mmap_sz; 452 void *tmp; 453 struct epoll_event *rb_epoll; 454 int err; 455 456 memset(&info, 0, sizeof(info)); 457 458 err = bpf_map_get_info_by_fd(map_fd, &info, &len); 459 if (err) { 460 err = -errno; 461 pr_warn("user ringbuf: failed to get map info for fd=%d: %d\n", map_fd, err); 462 return err; 463 } 464 465 if (info.type != BPF_MAP_TYPE_USER_RINGBUF) { 466 pr_warn("user ringbuf: map fd=%d is not BPF_MAP_TYPE_USER_RINGBUF\n", map_fd); 467 return -EINVAL; 468 } 469 470 rb->map_fd = map_fd; 471 rb->mask = info.max_entries - 1; 472 473 /* Map read-only consumer page */ 474 tmp = mmap(NULL, rb->page_size, PROT_READ, MAP_SHARED, map_fd, 0); 475 if (tmp == MAP_FAILED) { 476 err = -errno; 477 pr_warn("user ringbuf: failed to mmap consumer page for map fd=%d: %d\n", 478 map_fd, err); 479 return err; 480 } 481 rb->consumer_pos = tmp; 482 483 /* Map read-write the producer page and data pages. We map the data 484 * region as twice the total size of the ring buffer to allow the 485 * simple reading and writing of samples that wrap around the end of 486 * the buffer. See the kernel implementation for details. 487 */ 488 mmap_sz = rb->page_size + 2 * (__u64)info.max_entries; 489 if (mmap_sz != (__u64)(size_t)mmap_sz) { 490 pr_warn("user ringbuf: ring buf size (%u) is too big\n", info.max_entries); 491 return -E2BIG; 492 } 493 tmp = mmap(NULL, (size_t)mmap_sz, PROT_READ | PROT_WRITE, MAP_SHARED, 494 map_fd, rb->page_size); 495 if (tmp == MAP_FAILED) { 496 err = -errno; 497 pr_warn("user ringbuf: failed to mmap data pages for map fd=%d: %d\n", 498 map_fd, err); 499 return err; 500 } 501 502 rb->producer_pos = tmp; 503 rb->data = tmp + rb->page_size; 504 505 rb_epoll = &rb->event; 506 rb_epoll->events = EPOLLOUT; 507 if (epoll_ctl(rb->epoll_fd, EPOLL_CTL_ADD, map_fd, rb_epoll) < 0) { 508 err = -errno; 509 pr_warn("user ringbuf: failed to epoll add map fd=%d: %d\n", map_fd, err); 510 return err; 511 } 512 513 return 0; 514 } 515 516 struct user_ring_buffer * 517 user_ring_buffer__new(int map_fd, const struct user_ring_buffer_opts *opts) 518 { 519 struct user_ring_buffer *rb; 520 int err; 521 522 if (!OPTS_VALID(opts, user_ring_buffer_opts)) 523 return errno = EINVAL, NULL; 524 525 rb = calloc(1, sizeof(*rb)); 526 if (!rb) 527 return errno = ENOMEM, NULL; 528 529 rb->page_size = getpagesize(); 530 531 rb->epoll_fd = epoll_create1(EPOLL_CLOEXEC); 532 if (rb->epoll_fd < 0) { 533 err = -errno; 534 pr_warn("user ringbuf: failed to create epoll instance: %d\n", err); 535 goto err_out; 536 } 537 538 err = user_ringbuf_map(rb, map_fd); 539 if (err) 540 goto err_out; 541 542 return rb; 543 544 err_out: 545 user_ring_buffer__free(rb); 546 return errno = -err, NULL; 547 } 548 549 static void user_ringbuf_commit(struct user_ring_buffer *rb, void *sample, bool discard) 550 { 551 __u32 new_len; 552 struct ringbuf_hdr *hdr; 553 uintptr_t hdr_offset; 554 555 hdr_offset = rb->mask + 1 + (sample - rb->data) - BPF_RINGBUF_HDR_SZ; 556 hdr = rb->data + (hdr_offset & rb->mask); 557 558 new_len = hdr->len & ~BPF_RINGBUF_BUSY_BIT; 559 if (discard) 560 new_len |= BPF_RINGBUF_DISCARD_BIT; 561 562 /* Synchronizes with smp_load_acquire() in __bpf_user_ringbuf_peek() in 563 * the kernel. 564 */ 565 __atomic_exchange_n(&hdr->len, new_len, __ATOMIC_ACQ_REL); 566 } 567 568 void user_ring_buffer__discard(struct user_ring_buffer *rb, void *sample) 569 { 570 user_ringbuf_commit(rb, sample, true); 571 } 572 573 void user_ring_buffer__submit(struct user_ring_buffer *rb, void *sample) 574 { 575 user_ringbuf_commit(rb, sample, false); 576 } 577 578 void *user_ring_buffer__reserve(struct user_ring_buffer *rb, __u32 size) 579 { 580 __u32 avail_size, total_size, max_size; 581 /* 64-bit to avoid overflow in case of extreme application behavior */ 582 __u64 cons_pos, prod_pos; 583 struct ringbuf_hdr *hdr; 584 585 /* The top two bits are used as special flags */ 586 if (size & (BPF_RINGBUF_BUSY_BIT | BPF_RINGBUF_DISCARD_BIT)) 587 return errno = E2BIG, NULL; 588 589 /* Synchronizes with smp_store_release() in __bpf_user_ringbuf_peek() in 590 * the kernel. 591 */ 592 cons_pos = smp_load_acquire(rb->consumer_pos); 593 /* Synchronizes with smp_store_release() in user_ringbuf_commit() */ 594 prod_pos = smp_load_acquire(rb->producer_pos); 595 596 max_size = rb->mask + 1; 597 avail_size = max_size - (prod_pos - cons_pos); 598 /* Round up total size to a multiple of 8. */ 599 total_size = (size + BPF_RINGBUF_HDR_SZ + 7) / 8 * 8; 600 601 if (total_size > max_size) 602 return errno = E2BIG, NULL; 603 604 if (avail_size < total_size) 605 return errno = ENOSPC, NULL; 606 607 hdr = rb->data + (prod_pos & rb->mask); 608 hdr->len = size | BPF_RINGBUF_BUSY_BIT; 609 hdr->pad = 0; 610 611 /* Synchronizes with smp_load_acquire() in __bpf_user_ringbuf_peek() in 612 * the kernel. 613 */ 614 smp_store_release(rb->producer_pos, prod_pos + total_size); 615 616 return (void *)rb->data + ((prod_pos + BPF_RINGBUF_HDR_SZ) & rb->mask); 617 } 618 619 static __u64 ns_elapsed_timespec(const struct timespec *start, const struct timespec *end) 620 { 621 __u64 start_ns, end_ns, ns_per_s = 1000000000; 622 623 start_ns = (__u64)start->tv_sec * ns_per_s + start->tv_nsec; 624 end_ns = (__u64)end->tv_sec * ns_per_s + end->tv_nsec; 625 626 return end_ns - start_ns; 627 } 628 629 void *user_ring_buffer__reserve_blocking(struct user_ring_buffer *rb, __u32 size, int timeout_ms) 630 { 631 void *sample; 632 int err, ms_remaining = timeout_ms; 633 struct timespec start; 634 635 if (timeout_ms < 0 && timeout_ms != -1) 636 return errno = EINVAL, NULL; 637 638 if (timeout_ms != -1) { 639 err = clock_gettime(CLOCK_MONOTONIC, &start); 640 if (err) 641 return NULL; 642 } 643 644 do { 645 int cnt, ms_elapsed; 646 struct timespec curr; 647 __u64 ns_per_ms = 1000000; 648 649 sample = user_ring_buffer__reserve(rb, size); 650 if (sample) 651 return sample; 652 else if (errno != ENOSPC) 653 return NULL; 654 655 /* The kernel guarantees at least one event notification 656 * delivery whenever at least one sample is drained from the 657 * ring buffer in an invocation to bpf_ringbuf_drain(). Other 658 * additional events may be delivered at any time, but only one 659 * event is guaranteed per bpf_ringbuf_drain() invocation, 660 * provided that a sample is drained, and the BPF program did 661 * not pass BPF_RB_NO_WAKEUP to bpf_ringbuf_drain(). If 662 * BPF_RB_FORCE_WAKEUP is passed to bpf_ringbuf_drain(), a 663 * wakeup event will be delivered even if no samples are 664 * drained. 665 */ 666 cnt = epoll_wait(rb->epoll_fd, &rb->event, 1, ms_remaining); 667 if (cnt < 0) 668 return NULL; 669 670 if (timeout_ms == -1) 671 continue; 672 673 err = clock_gettime(CLOCK_MONOTONIC, &curr); 674 if (err) 675 return NULL; 676 677 ms_elapsed = ns_elapsed_timespec(&start, &curr) / ns_per_ms; 678 ms_remaining = timeout_ms - ms_elapsed; 679 } while (ms_remaining > 0); 680 681 /* Try one more time to reserve a sample after the specified timeout has elapsed. */ 682 return user_ring_buffer__reserve(rb, size); 683 } 684