1 // SPDX-License-Identifier: (LGPL-2.1 OR BSD-2-Clause)
3 * Ring buffer operations.
5 * Copyright (C) 2020 Facebook, Inc.
14 #include <linux/err.h>
15 #include <linux/bpf.h>
16 #include <asm/barrier.h>
18 #include <sys/epoll.h>
21 #include "libbpf_internal.h"
25 ring_buffer_sample_fn sample_cb;
28 unsigned long *consumer_pos;
29 unsigned long *producer_pos;
35 struct epoll_event *events;
42 static void ringbuf_unmap_ring(struct ring_buffer *rb, struct ring *r)
44 if (r->consumer_pos) {
45 munmap(r->consumer_pos, rb->page_size);
46 r->consumer_pos = NULL;
48 if (r->producer_pos) {
49 munmap(r->producer_pos, rb->page_size + 2 * (r->mask + 1));
50 r->producer_pos = NULL;
54 /* Add extra RINGBUF maps to this ring buffer manager */
55 int ring_buffer__add(struct ring_buffer *rb, int map_fd,
56 ring_buffer_sample_fn sample_cb, void *ctx)
58 struct bpf_map_info info;
59 __u32 len = sizeof(info);
60 struct epoll_event *e;
65 memset(&info, 0, sizeof(info));
67 err = bpf_obj_get_info_by_fd(map_fd, &info, &len);
70 pr_warn("ringbuf: failed to get map info for fd=%d: %d\n",
75 if (info.type != BPF_MAP_TYPE_RINGBUF) {
76 pr_warn("ringbuf: map fd=%d is not BPF_MAP_TYPE_RINGBUF\n",
81 tmp = libbpf_reallocarray(rb->rings, rb->ring_cnt + 1, sizeof(*rb->rings));
86 tmp = libbpf_reallocarray(rb->events, rb->ring_cnt + 1, sizeof(*rb->events));
91 r = &rb->rings[rb->ring_cnt];
92 memset(r, 0, sizeof(*r));
95 r->sample_cb = sample_cb;
97 r->mask = info.max_entries - 1;
99 /* Map writable consumer page */
100 tmp = mmap(NULL, rb->page_size, PROT_READ | PROT_WRITE, MAP_SHARED,
102 if (tmp == MAP_FAILED) {
104 pr_warn("ringbuf: failed to mmap consumer page for map fd=%d: %d\n",
108 r->consumer_pos = tmp;
110 /* Map read-only producer page and data pages. We map twice as big
111 * data size to allow simple reading of samples that wrap around the
112 * end of a ring buffer. See kernel implementation for details.
114 tmp = mmap(NULL, rb->page_size + 2 * info.max_entries, PROT_READ,
115 MAP_SHARED, map_fd, rb->page_size);
116 if (tmp == MAP_FAILED) {
118 ringbuf_unmap_ring(rb, r);
119 pr_warn("ringbuf: failed to mmap data pages for map fd=%d: %d\n",
123 r->producer_pos = tmp;
124 r->data = tmp + rb->page_size;
126 e = &rb->events[rb->ring_cnt];
127 memset(e, 0, sizeof(*e));
130 e->data.fd = rb->ring_cnt;
131 if (epoll_ctl(rb->epoll_fd, EPOLL_CTL_ADD, map_fd, e) < 0) {
133 ringbuf_unmap_ring(rb, r);
134 pr_warn("ringbuf: failed to epoll add map fd=%d: %d\n",
143 void ring_buffer__free(struct ring_buffer *rb)
150 for (i = 0; i < rb->ring_cnt; ++i)
151 ringbuf_unmap_ring(rb, &rb->rings[i]);
152 if (rb->epoll_fd >= 0)
161 ring_buffer__new(int map_fd, ring_buffer_sample_fn sample_cb, void *ctx,
162 const struct ring_buffer_opts *opts)
164 struct ring_buffer *rb;
167 if (!OPTS_VALID(opts, ring_buffer_opts))
170 rb = calloc(1, sizeof(*rb));
174 rb->page_size = getpagesize();
176 rb->epoll_fd = epoll_create1(EPOLL_CLOEXEC);
177 if (rb->epoll_fd < 0) {
179 pr_warn("ringbuf: failed to create epoll instance: %d\n", err);
183 err = ring_buffer__add(rb, map_fd, sample_cb, ctx);
190 ring_buffer__free(rb);
194 static inline int roundup_len(__u32 len)
196 /* clear out top 2 bits (discard and busy, if set) */
199 /* add length prefix */
200 len += BPF_RINGBUF_HDR_SZ;
201 /* round up to 8 byte alignment */
202 return (len + 7) / 8 * 8;
205 static int ringbuf_process_ring(struct ring* r)
207 int *len_ptr, len, err, cnt = 0;
208 unsigned long cons_pos, prod_pos;
212 cons_pos = smp_load_acquire(r->consumer_pos);
214 got_new_data = false;
215 prod_pos = smp_load_acquire(r->producer_pos);
216 while (cons_pos < prod_pos) {
217 len_ptr = r->data + (cons_pos & r->mask);
218 len = smp_load_acquire(len_ptr);
220 /* sample not committed yet, bail out for now */
221 if (len & BPF_RINGBUF_BUSY_BIT)
225 cons_pos += roundup_len(len);
227 if ((len & BPF_RINGBUF_DISCARD_BIT) == 0) {
228 sample = (void *)len_ptr + BPF_RINGBUF_HDR_SZ;
229 err = r->sample_cb(r->ctx, sample, len);
231 /* update consumer pos and bail out */
232 smp_store_release(r->consumer_pos,
239 smp_store_release(r->consumer_pos, cons_pos);
241 } while (got_new_data);
246 /* Consume available ring buffer(s) data without event polling.
247 * Returns number of records consumed across all registered ring buffers, or
248 * negative number if any of the callbacks return error.
250 int ring_buffer__consume(struct ring_buffer *rb)
254 for (i = 0; i < rb->ring_cnt; i++) {
255 struct ring *ring = &rb->rings[i];
257 err = ringbuf_process_ring(ring);
265 /* Poll for available data and consume records, if any are available.
266 * Returns number of records consumed, or negative number, if any of the
267 * registered callbacks returned error.
269 int ring_buffer__poll(struct ring_buffer *rb, int timeout_ms)
271 int i, cnt, err, res = 0;
273 cnt = epoll_wait(rb->epoll_fd, rb->events, rb->ring_cnt, timeout_ms);
274 for (i = 0; i < cnt; i++) {
275 __u32 ring_id = rb->events[i].data.fd;
276 struct ring *ring = &rb->rings[ring_id];
278 err = ringbuf_process_ring(ring);
283 return cnt < 0 ? -errno : res;
286 /* Get an fd that can be used to sleep until data is available in the ring(s) */
287 int ring_buffer__epoll_fd(const struct ring_buffer *rb)