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_unmap_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 72 /* Add extra RINGBUF maps to this ring buffer manager */ 73 int ring_buffer__add(struct ring_buffer *rb, int map_fd, 74 ring_buffer_sample_fn sample_cb, void *ctx) 75 { 76 struct bpf_map_info info; 77 __u32 len = sizeof(info); 78 struct epoll_event *e; 79 struct ring *r; 80 __u64 mmap_sz; 81 void *tmp; 82 int err; 83 84 memset(&info, 0, sizeof(info)); 85 86 err = bpf_obj_get_info_by_fd(map_fd, &info, &len); 87 if (err) { 88 err = -errno; 89 pr_warn("ringbuf: failed to get map info for fd=%d: %d\n", 90 map_fd, err); 91 return libbpf_err(err); 92 } 93 94 if (info.type != BPF_MAP_TYPE_RINGBUF) { 95 pr_warn("ringbuf: map fd=%d is not BPF_MAP_TYPE_RINGBUF\n", 96 map_fd); 97 return libbpf_err(-EINVAL); 98 } 99 100 tmp = libbpf_reallocarray(rb->rings, rb->ring_cnt + 1, sizeof(*rb->rings)); 101 if (!tmp) 102 return libbpf_err(-ENOMEM); 103 rb->rings = tmp; 104 105 tmp = libbpf_reallocarray(rb->events, rb->ring_cnt + 1, sizeof(*rb->events)); 106 if (!tmp) 107 return libbpf_err(-ENOMEM); 108 rb->events = tmp; 109 110 r = &rb->rings[rb->ring_cnt]; 111 memset(r, 0, sizeof(*r)); 112 113 r->map_fd = map_fd; 114 r->sample_cb = sample_cb; 115 r->ctx = ctx; 116 r->mask = info.max_entries - 1; 117 118 /* Map writable consumer page */ 119 tmp = mmap(NULL, rb->page_size, PROT_READ | PROT_WRITE, MAP_SHARED, map_fd, 0); 120 if (tmp == MAP_FAILED) { 121 err = -errno; 122 pr_warn("ringbuf: failed to mmap consumer page for map fd=%d: %d\n", 123 map_fd, err); 124 return libbpf_err(err); 125 } 126 r->consumer_pos = tmp; 127 128 /* Map read-only producer page and data pages. We map twice as big 129 * data size to allow simple reading of samples that wrap around the 130 * end of a ring buffer. See kernel implementation for details. 131 * */ 132 mmap_sz = rb->page_size + 2 * (__u64)info.max_entries; 133 if (mmap_sz != (__u64)(size_t)mmap_sz) { 134 pr_warn("ringbuf: ring buffer size (%u) is too big\n", info.max_entries); 135 return libbpf_err(-E2BIG); 136 } 137 tmp = mmap(NULL, (size_t)mmap_sz, PROT_READ, MAP_SHARED, map_fd, rb->page_size); 138 if (tmp == MAP_FAILED) { 139 err = -errno; 140 ringbuf_unmap_ring(rb, r); 141 pr_warn("ringbuf: failed to mmap data pages for map fd=%d: %d\n", 142 map_fd, err); 143 return libbpf_err(err); 144 } 145 r->producer_pos = tmp; 146 r->data = tmp + rb->page_size; 147 148 e = &rb->events[rb->ring_cnt]; 149 memset(e, 0, sizeof(*e)); 150 151 e->events = EPOLLIN; 152 e->data.fd = rb->ring_cnt; 153 if (epoll_ctl(rb->epoll_fd, EPOLL_CTL_ADD, map_fd, e) < 0) { 154 err = -errno; 155 ringbuf_unmap_ring(rb, r); 156 pr_warn("ringbuf: failed to epoll add map fd=%d: %d\n", 157 map_fd, err); 158 return libbpf_err(err); 159 } 160 161 rb->ring_cnt++; 162 return 0; 163 } 164 165 void ring_buffer__free(struct ring_buffer *rb) 166 { 167 int i; 168 169 if (!rb) 170 return; 171 172 for (i = 0; i < rb->ring_cnt; ++i) 173 ringbuf_unmap_ring(rb, &rb->rings[i]); 174 if (rb->epoll_fd >= 0) 175 close(rb->epoll_fd); 176 177 free(rb->events); 178 free(rb->rings); 179 free(rb); 180 } 181 182 struct ring_buffer * 183 ring_buffer__new(int map_fd, ring_buffer_sample_fn sample_cb, void *ctx, 184 const struct ring_buffer_opts *opts) 185 { 186 struct ring_buffer *rb; 187 int err; 188 189 if (!OPTS_VALID(opts, ring_buffer_opts)) 190 return errno = EINVAL, NULL; 191 192 rb = calloc(1, sizeof(*rb)); 193 if (!rb) 194 return errno = ENOMEM, NULL; 195 196 rb->page_size = getpagesize(); 197 198 rb->epoll_fd = epoll_create1(EPOLL_CLOEXEC); 199 if (rb->epoll_fd < 0) { 200 err = -errno; 201 pr_warn("ringbuf: failed to create epoll instance: %d\n", err); 202 goto err_out; 203 } 204 205 err = ring_buffer__add(rb, map_fd, sample_cb, ctx); 206 if (err) 207 goto err_out; 208 209 return rb; 210 211 err_out: 212 ring_buffer__free(rb); 213 return errno = -err, NULL; 214 } 215 216 static inline int roundup_len(__u32 len) 217 { 218 /* clear out top 2 bits (discard and busy, if set) */ 219 len <<= 2; 220 len >>= 2; 221 /* add length prefix */ 222 len += BPF_RINGBUF_HDR_SZ; 223 /* round up to 8 byte alignment */ 224 return (len + 7) / 8 * 8; 225 } 226 227 static int64_t ringbuf_process_ring(struct ring* r) 228 { 229 int *len_ptr, len, err; 230 /* 64-bit to avoid overflow in case of extreme application behavior */ 231 int64_t cnt = 0; 232 unsigned long cons_pos, prod_pos; 233 bool got_new_data; 234 void *sample; 235 236 cons_pos = smp_load_acquire(r->consumer_pos); 237 do { 238 got_new_data = false; 239 prod_pos = smp_load_acquire(r->producer_pos); 240 while (cons_pos < prod_pos) { 241 len_ptr = r->data + (cons_pos & r->mask); 242 len = smp_load_acquire(len_ptr); 243 244 /* sample not committed yet, bail out for now */ 245 if (len & BPF_RINGBUF_BUSY_BIT) 246 goto done; 247 248 got_new_data = true; 249 cons_pos += roundup_len(len); 250 251 if ((len & BPF_RINGBUF_DISCARD_BIT) == 0) { 252 sample = (void *)len_ptr + BPF_RINGBUF_HDR_SZ; 253 err = r->sample_cb(r->ctx, sample, len); 254 if (err < 0) { 255 /* update consumer pos and bail out */ 256 smp_store_release(r->consumer_pos, 257 cons_pos); 258 return err; 259 } 260 cnt++; 261 } 262 263 smp_store_release(r->consumer_pos, cons_pos); 264 } 265 } while (got_new_data); 266 done: 267 return cnt; 268 } 269 270 /* Consume available ring buffer(s) data without event polling. 271 * Returns number of records consumed across all registered ring buffers (or 272 * INT_MAX, whichever is less), or negative number if any of the callbacks 273 * return error. 274 */ 275 int ring_buffer__consume(struct ring_buffer *rb) 276 { 277 int64_t err, res = 0; 278 int i; 279 280 for (i = 0; i < rb->ring_cnt; i++) { 281 struct ring *ring = &rb->rings[i]; 282 283 err = ringbuf_process_ring(ring); 284 if (err < 0) 285 return libbpf_err(err); 286 res += err; 287 } 288 if (res > INT_MAX) 289 return INT_MAX; 290 return res; 291 } 292 293 /* Poll for available data and consume records, if any are available. 294 * Returns number of records consumed (or INT_MAX, whichever is less), or 295 * negative number, if any of the registered callbacks returned error. 296 */ 297 int ring_buffer__poll(struct ring_buffer *rb, int timeout_ms) 298 { 299 int i, cnt; 300 int64_t err, res = 0; 301 302 cnt = epoll_wait(rb->epoll_fd, rb->events, rb->ring_cnt, timeout_ms); 303 if (cnt < 0) 304 return libbpf_err(-errno); 305 306 for (i = 0; i < cnt; i++) { 307 __u32 ring_id = rb->events[i].data.fd; 308 struct ring *ring = &rb->rings[ring_id]; 309 310 err = ringbuf_process_ring(ring); 311 if (err < 0) 312 return libbpf_err(err); 313 res += err; 314 } 315 if (res > INT_MAX) 316 return INT_MAX; 317 return res; 318 } 319 320 /* Get an fd that can be used to sleep until data is available in the ring(s) */ 321 int ring_buffer__epoll_fd(const struct ring_buffer *rb) 322 { 323 return rb->epoll_fd; 324 } 325 326 static void user_ringbuf_unmap_ring(struct user_ring_buffer *rb) 327 { 328 if (rb->consumer_pos) { 329 munmap(rb->consumer_pos, rb->page_size); 330 rb->consumer_pos = NULL; 331 } 332 if (rb->producer_pos) { 333 munmap(rb->producer_pos, rb->page_size + 2 * (rb->mask + 1)); 334 rb->producer_pos = NULL; 335 } 336 } 337 338 void user_ring_buffer__free(struct user_ring_buffer *rb) 339 { 340 if (!rb) 341 return; 342 343 user_ringbuf_unmap_ring(rb); 344 345 if (rb->epoll_fd >= 0) 346 close(rb->epoll_fd); 347 348 free(rb); 349 } 350 351 static int user_ringbuf_map(struct user_ring_buffer *rb, int map_fd) 352 { 353 struct bpf_map_info info; 354 __u32 len = sizeof(info); 355 __u64 mmap_sz; 356 void *tmp; 357 struct epoll_event *rb_epoll; 358 int err; 359 360 memset(&info, 0, sizeof(info)); 361 362 err = bpf_obj_get_info_by_fd(map_fd, &info, &len); 363 if (err) { 364 err = -errno; 365 pr_warn("user ringbuf: failed to get map info for fd=%d: %d\n", map_fd, err); 366 return err; 367 } 368 369 if (info.type != BPF_MAP_TYPE_USER_RINGBUF) { 370 pr_warn("user ringbuf: map fd=%d is not BPF_MAP_TYPE_USER_RINGBUF\n", map_fd); 371 return -EINVAL; 372 } 373 374 rb->map_fd = map_fd; 375 rb->mask = info.max_entries - 1; 376 377 /* Map read-only consumer page */ 378 tmp = mmap(NULL, rb->page_size, PROT_READ, MAP_SHARED, map_fd, 0); 379 if (tmp == MAP_FAILED) { 380 err = -errno; 381 pr_warn("user ringbuf: failed to mmap consumer page for map fd=%d: %d\n", 382 map_fd, err); 383 return err; 384 } 385 rb->consumer_pos = tmp; 386 387 /* Map read-write the producer page and data pages. We map the data 388 * region as twice the total size of the ring buffer to allow the 389 * simple reading and writing of samples that wrap around the end of 390 * the buffer. See the kernel implementation for details. 391 */ 392 mmap_sz = rb->page_size + 2 * (__u64)info.max_entries; 393 if (mmap_sz != (__u64)(size_t)mmap_sz) { 394 pr_warn("user ringbuf: ring buf size (%u) is too big\n", info.max_entries); 395 return -E2BIG; 396 } 397 tmp = mmap(NULL, (size_t)mmap_sz, PROT_READ | PROT_WRITE, MAP_SHARED, 398 map_fd, rb->page_size); 399 if (tmp == MAP_FAILED) { 400 err = -errno; 401 pr_warn("user ringbuf: failed to mmap data pages for map fd=%d: %d\n", 402 map_fd, err); 403 return err; 404 } 405 406 rb->producer_pos = tmp; 407 rb->data = tmp + rb->page_size; 408 409 rb_epoll = &rb->event; 410 rb_epoll->events = EPOLLOUT; 411 if (epoll_ctl(rb->epoll_fd, EPOLL_CTL_ADD, map_fd, rb_epoll) < 0) { 412 err = -errno; 413 pr_warn("user ringbuf: failed to epoll add map fd=%d: %d\n", map_fd, err); 414 return err; 415 } 416 417 return 0; 418 } 419 420 struct user_ring_buffer * 421 user_ring_buffer__new(int map_fd, const struct user_ring_buffer_opts *opts) 422 { 423 struct user_ring_buffer *rb; 424 int err; 425 426 if (!OPTS_VALID(opts, user_ring_buffer_opts)) 427 return errno = EINVAL, NULL; 428 429 rb = calloc(1, sizeof(*rb)); 430 if (!rb) 431 return errno = ENOMEM, NULL; 432 433 rb->page_size = getpagesize(); 434 435 rb->epoll_fd = epoll_create1(EPOLL_CLOEXEC); 436 if (rb->epoll_fd < 0) { 437 err = -errno; 438 pr_warn("user ringbuf: failed to create epoll instance: %d\n", err); 439 goto err_out; 440 } 441 442 err = user_ringbuf_map(rb, map_fd); 443 if (err) 444 goto err_out; 445 446 return rb; 447 448 err_out: 449 user_ring_buffer__free(rb); 450 return errno = -err, NULL; 451 } 452 453 static void user_ringbuf_commit(struct user_ring_buffer *rb, void *sample, bool discard) 454 { 455 __u32 new_len; 456 struct ringbuf_hdr *hdr; 457 uintptr_t hdr_offset; 458 459 hdr_offset = rb->mask + 1 + (sample - rb->data) - BPF_RINGBUF_HDR_SZ; 460 hdr = rb->data + (hdr_offset & rb->mask); 461 462 new_len = hdr->len & ~BPF_RINGBUF_BUSY_BIT; 463 if (discard) 464 new_len |= BPF_RINGBUF_DISCARD_BIT; 465 466 /* Synchronizes with smp_load_acquire() in __bpf_user_ringbuf_peek() in 467 * the kernel. 468 */ 469 __atomic_exchange_n(&hdr->len, new_len, __ATOMIC_ACQ_REL); 470 } 471 472 void user_ring_buffer__discard(struct user_ring_buffer *rb, void *sample) 473 { 474 user_ringbuf_commit(rb, sample, true); 475 } 476 477 void user_ring_buffer__submit(struct user_ring_buffer *rb, void *sample) 478 { 479 user_ringbuf_commit(rb, sample, false); 480 } 481 482 void *user_ring_buffer__reserve(struct user_ring_buffer *rb, __u32 size) 483 { 484 __u32 avail_size, total_size, max_size; 485 /* 64-bit to avoid overflow in case of extreme application behavior */ 486 __u64 cons_pos, prod_pos; 487 struct ringbuf_hdr *hdr; 488 489 /* The top two bits are used as special flags */ 490 if (size & (BPF_RINGBUF_BUSY_BIT | BPF_RINGBUF_DISCARD_BIT)) 491 return errno = E2BIG, NULL; 492 493 /* Synchronizes with smp_store_release() in __bpf_user_ringbuf_peek() in 494 * the kernel. 495 */ 496 cons_pos = smp_load_acquire(rb->consumer_pos); 497 /* Synchronizes with smp_store_release() in user_ringbuf_commit() */ 498 prod_pos = smp_load_acquire(rb->producer_pos); 499 500 max_size = rb->mask + 1; 501 avail_size = max_size - (prod_pos - cons_pos); 502 /* Round up total size to a multiple of 8. */ 503 total_size = (size + BPF_RINGBUF_HDR_SZ + 7) / 8 * 8; 504 505 if (total_size > max_size) 506 return errno = E2BIG, NULL; 507 508 if (avail_size < total_size) 509 return errno = ENOSPC, NULL; 510 511 hdr = rb->data + (prod_pos & rb->mask); 512 hdr->len = size | BPF_RINGBUF_BUSY_BIT; 513 hdr->pad = 0; 514 515 /* Synchronizes with smp_load_acquire() in __bpf_user_ringbuf_peek() in 516 * the kernel. 517 */ 518 smp_store_release(rb->producer_pos, prod_pos + total_size); 519 520 return (void *)rb->data + ((prod_pos + BPF_RINGBUF_HDR_SZ) & rb->mask); 521 } 522 523 static __u64 ns_elapsed_timespec(const struct timespec *start, const struct timespec *end) 524 { 525 __u64 start_ns, end_ns, ns_per_s = 1000000000; 526 527 start_ns = (__u64)start->tv_sec * ns_per_s + start->tv_nsec; 528 end_ns = (__u64)end->tv_sec * ns_per_s + end->tv_nsec; 529 530 return end_ns - start_ns; 531 } 532 533 void *user_ring_buffer__reserve_blocking(struct user_ring_buffer *rb, __u32 size, int timeout_ms) 534 { 535 void *sample; 536 int err, ms_remaining = timeout_ms; 537 struct timespec start; 538 539 if (timeout_ms < 0 && timeout_ms != -1) 540 return errno = EINVAL, NULL; 541 542 if (timeout_ms != -1) { 543 err = clock_gettime(CLOCK_MONOTONIC, &start); 544 if (err) 545 return NULL; 546 } 547 548 do { 549 int cnt, ms_elapsed; 550 struct timespec curr; 551 __u64 ns_per_ms = 1000000; 552 553 sample = user_ring_buffer__reserve(rb, size); 554 if (sample) 555 return sample; 556 else if (errno != ENOSPC) 557 return NULL; 558 559 /* The kernel guarantees at least one event notification 560 * delivery whenever at least one sample is drained from the 561 * ring buffer in an invocation to bpf_ringbuf_drain(). Other 562 * additional events may be delivered at any time, but only one 563 * event is guaranteed per bpf_ringbuf_drain() invocation, 564 * provided that a sample is drained, and the BPF program did 565 * not pass BPF_RB_NO_WAKEUP to bpf_ringbuf_drain(). If 566 * BPF_RB_FORCE_WAKEUP is passed to bpf_ringbuf_drain(), a 567 * wakeup event will be delivered even if no samples are 568 * drained. 569 */ 570 cnt = epoll_wait(rb->epoll_fd, &rb->event, 1, ms_remaining); 571 if (cnt < 0) 572 return NULL; 573 574 if (timeout_ms == -1) 575 continue; 576 577 err = clock_gettime(CLOCK_MONOTONIC, &curr); 578 if (err) 579 return NULL; 580 581 ms_elapsed = ns_elapsed_timespec(&start, &curr) / ns_per_ms; 582 ms_remaining = timeout_ms - ms_elapsed; 583 } while (ms_remaining > 0); 584 585 /* Try one more time to reserve a sample after the specified timeout has elapsed. */ 586 return user_ring_buffer__reserve(rb, size); 587 } 588