684 lines · c
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_SOURCE8#define _GNU_SOURCE9#endif10#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 25struct 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 35struct 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 43struct 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 */55struct ringbuf_hdr {56 __u32 len;57 __u32 pad;58};59 60static 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 */75int 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 big133 * data size to allow simple reading of samples that wrap around the134 * 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 167err_out:168 ringbuf_free_ring(rb, r);169 return libbpf_err(err);170}171 172void 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 189struct ring_buffer *190ring_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 218err_out:219 ring_buffer__free(rb);220 return errno = -err, NULL;221}222 223static 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 234static 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);276done:277 return cnt;278}279 280/* Consume available ring buffer(s) data without event polling, up to n281 * records.282 *283 * Returns number of records consumed across all registered ring buffers (or284 * n, whichever is less), or negative number if any of the callbacks return285 * error.286 */287int 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 (or309 * INT_MAX, whichever is less), or negative number if any of the callbacks310 * return error.311 */312int 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), or334 * negative number, if any of the registered callbacks returned error.335 */336int 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) */360int ring_buffer__epoll_fd(const struct ring_buffer *rb)361{362 return rb->epoll_fd;363}364 365struct 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 373unsigned 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 379unsigned long ring__producer_pos(const struct ring *r)380{381 /* Synchronizes with smp_store_release() in __bpf_ringbuf_reserve() in382 * the kernel.383 */384 return smp_load_acquire(r->producer_pos);385}386 387size_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 396size_t ring__size(const struct ring *r)397{398 return r->mask + 1;399}400 401int ring__map_fd(const struct ring *r)402{403 return r->map_fd;404}405 406int 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 417int ring__consume(struct ring *r)418{419 return ring__consume_n(r, INT_MAX);420}421 422static 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 434void 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 447static 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 data484 * region as twice the total size of the ring buffer to allow the485 * simple reading and writing of samples that wrap around the end of486 * 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 516struct user_ring_buffer *517user_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 544err_out:545 user_ring_buffer__free(rb);546 return errno = -err, NULL;547}548 549static 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() in563 * the kernel.564 */565 __atomic_exchange_n(&hdr->len, new_len, __ATOMIC_ACQ_REL);566}567 568void user_ring_buffer__discard(struct user_ring_buffer *rb, void *sample)569{570 user_ringbuf_commit(rb, sample, true);571}572 573void user_ring_buffer__submit(struct user_ring_buffer *rb, void *sample)574{575 user_ringbuf_commit(rb, sample, false);576}577 578void *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() in590 * 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() in612 * 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 619static __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 629void *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 notification656 * delivery whenever at least one sample is drained from the657 * ring buffer in an invocation to bpf_ringbuf_drain(). Other658 * additional events may be delivered at any time, but only one659 * event is guaranteed per bpf_ringbuf_drain() invocation,660 * provided that a sample is drained, and the BPF program did661 * not pass BPF_RB_NO_WAKEUP to bpf_ringbuf_drain(). If662 * BPF_RB_FORCE_WAKEUP is passed to bpf_ringbuf_drain(), a663 * wakeup event will be delivered even if no samples are664 * 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