xref: /linux/tools/lib/bpf/ringbuf.c (revision 87c9c16317882dd6dbbc07e349bc3223e14f3244)
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 
20 #include "libbpf.h"
21 #include "libbpf_internal.h"
22 #include "bpf.h"
23 
24 struct ring {
25 	ring_buffer_sample_fn sample_cb;
26 	void *ctx;
27 	void *data;
28 	unsigned long *consumer_pos;
29 	unsigned long *producer_pos;
30 	unsigned long mask;
31 	int map_fd;
32 };
33 
34 struct ring_buffer {
35 	struct epoll_event *events;
36 	struct ring *rings;
37 	size_t page_size;
38 	int epoll_fd;
39 	int ring_cnt;
40 };
41 
42 static void ringbuf_unmap_ring(struct ring_buffer *rb, struct ring *r)
43 {
44 	if (r->consumer_pos) {
45 		munmap(r->consumer_pos, rb->page_size);
46 		r->consumer_pos = NULL;
47 	}
48 	if (r->producer_pos) {
49 		munmap(r->producer_pos, rb->page_size + 2 * (r->mask + 1));
50 		r->producer_pos = NULL;
51 	}
52 }
53 
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)
57 {
58 	struct bpf_map_info info;
59 	__u32 len = sizeof(info);
60 	struct epoll_event *e;
61 	struct ring *r;
62 	void *tmp;
63 	int err;
64 
65 	memset(&info, 0, sizeof(info));
66 
67 	err = bpf_obj_get_info_by_fd(map_fd, &info, &len);
68 	if (err) {
69 		err = -errno;
70 		pr_warn("ringbuf: failed to get map info for fd=%d: %d\n",
71 			map_fd, err);
72 		return err;
73 	}
74 
75 	if (info.type != BPF_MAP_TYPE_RINGBUF) {
76 		pr_warn("ringbuf: map fd=%d is not BPF_MAP_TYPE_RINGBUF\n",
77 			map_fd);
78 		return -EINVAL;
79 	}
80 
81 	tmp = libbpf_reallocarray(rb->rings, rb->ring_cnt + 1, sizeof(*rb->rings));
82 	if (!tmp)
83 		return -ENOMEM;
84 	rb->rings = tmp;
85 
86 	tmp = libbpf_reallocarray(rb->events, rb->ring_cnt + 1, sizeof(*rb->events));
87 	if (!tmp)
88 		return -ENOMEM;
89 	rb->events = tmp;
90 
91 	r = &rb->rings[rb->ring_cnt];
92 	memset(r, 0, sizeof(*r));
93 
94 	r->map_fd = map_fd;
95 	r->sample_cb = sample_cb;
96 	r->ctx = ctx;
97 	r->mask = info.max_entries - 1;
98 
99 	/* Map writable consumer page */
100 	tmp = mmap(NULL, rb->page_size, PROT_READ | PROT_WRITE, MAP_SHARED,
101 		   map_fd, 0);
102 	if (tmp == MAP_FAILED) {
103 		err = -errno;
104 		pr_warn("ringbuf: failed to mmap consumer page for map fd=%d: %d\n",
105 			map_fd, err);
106 		return err;
107 	}
108 	r->consumer_pos = tmp;
109 
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.
113 	 * */
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) {
117 		err = -errno;
118 		ringbuf_unmap_ring(rb, r);
119 		pr_warn("ringbuf: failed to mmap data pages for map fd=%d: %d\n",
120 			map_fd, err);
121 		return err;
122 	}
123 	r->producer_pos = tmp;
124 	r->data = tmp + rb->page_size;
125 
126 	e = &rb->events[rb->ring_cnt];
127 	memset(e, 0, sizeof(*e));
128 
129 	e->events = EPOLLIN;
130 	e->data.fd = rb->ring_cnt;
131 	if (epoll_ctl(rb->epoll_fd, EPOLL_CTL_ADD, map_fd, e) < 0) {
132 		err = -errno;
133 		ringbuf_unmap_ring(rb, r);
134 		pr_warn("ringbuf: failed to epoll add map fd=%d: %d\n",
135 			map_fd, err);
136 		return err;
137 	}
138 
139 	rb->ring_cnt++;
140 	return 0;
141 }
142 
143 void ring_buffer__free(struct ring_buffer *rb)
144 {
145 	int i;
146 
147 	if (!rb)
148 		return;
149 
150 	for (i = 0; i < rb->ring_cnt; ++i)
151 		ringbuf_unmap_ring(rb, &rb->rings[i]);
152 	if (rb->epoll_fd >= 0)
153 		close(rb->epoll_fd);
154 
155 	free(rb->events);
156 	free(rb->rings);
157 	free(rb);
158 }
159 
160 struct ring_buffer *
161 ring_buffer__new(int map_fd, ring_buffer_sample_fn sample_cb, void *ctx,
162 		 const struct ring_buffer_opts *opts)
163 {
164 	struct ring_buffer *rb;
165 	int err;
166 
167 	if (!OPTS_VALID(opts, ring_buffer_opts))
168 		return NULL;
169 
170 	rb = calloc(1, sizeof(*rb));
171 	if (!rb)
172 		return NULL;
173 
174 	rb->page_size = getpagesize();
175 
176 	rb->epoll_fd = epoll_create1(EPOLL_CLOEXEC);
177 	if (rb->epoll_fd < 0) {
178 		err = -errno;
179 		pr_warn("ringbuf: failed to create epoll instance: %d\n", err);
180 		goto err_out;
181 	}
182 
183 	err = ring_buffer__add(rb, map_fd, sample_cb, ctx);
184 	if (err)
185 		goto err_out;
186 
187 	return rb;
188 
189 err_out:
190 	ring_buffer__free(rb);
191 	return NULL;
192 }
193 
194 static inline int roundup_len(__u32 len)
195 {
196 	/* clear out top 2 bits (discard and busy, if set) */
197 	len <<= 2;
198 	len >>= 2;
199 	/* add length prefix */
200 	len += BPF_RINGBUF_HDR_SZ;
201 	/* round up to 8 byte alignment */
202 	return (len + 7) / 8 * 8;
203 }
204 
205 static int64_t ringbuf_process_ring(struct ring* r)
206 {
207 	int *len_ptr, len, err;
208 	/* 64-bit to avoid overflow in case of extreme application behavior */
209 	int64_t cnt = 0;
210 	unsigned long cons_pos, prod_pos;
211 	bool got_new_data;
212 	void *sample;
213 
214 	cons_pos = smp_load_acquire(r->consumer_pos);
215 	do {
216 		got_new_data = false;
217 		prod_pos = smp_load_acquire(r->producer_pos);
218 		while (cons_pos < prod_pos) {
219 			len_ptr = r->data + (cons_pos & r->mask);
220 			len = smp_load_acquire(len_ptr);
221 
222 			/* sample not committed yet, bail out for now */
223 			if (len & BPF_RINGBUF_BUSY_BIT)
224 				goto done;
225 
226 			got_new_data = true;
227 			cons_pos += roundup_len(len);
228 
229 			if ((len & BPF_RINGBUF_DISCARD_BIT) == 0) {
230 				sample = (void *)len_ptr + BPF_RINGBUF_HDR_SZ;
231 				err = r->sample_cb(r->ctx, sample, len);
232 				if (err < 0) {
233 					/* update consumer pos and bail out */
234 					smp_store_release(r->consumer_pos,
235 							  cons_pos);
236 					return err;
237 				}
238 				cnt++;
239 			}
240 
241 			smp_store_release(r->consumer_pos, cons_pos);
242 		}
243 	} while (got_new_data);
244 done:
245 	return cnt;
246 }
247 
248 /* Consume available ring buffer(s) data without event polling.
249  * Returns number of records consumed across all registered ring buffers (or
250  * INT_MAX, whichever is less), or negative number if any of the callbacks
251  * return error.
252  */
253 int ring_buffer__consume(struct ring_buffer *rb)
254 {
255 	int64_t err, res = 0;
256 	int i;
257 
258 	for (i = 0; i < rb->ring_cnt; i++) {
259 		struct ring *ring = &rb->rings[i];
260 
261 		err = ringbuf_process_ring(ring);
262 		if (err < 0)
263 			return err;
264 		res += err;
265 	}
266 	if (res > INT_MAX)
267 		return INT_MAX;
268 	return res;
269 }
270 
271 /* Poll for available data and consume records, if any are available.
272  * Returns number of records consumed (or INT_MAX, whichever is less), or
273  * negative number, if any of the registered callbacks returned error.
274  */
275 int ring_buffer__poll(struct ring_buffer *rb, int timeout_ms)
276 {
277 	int i, cnt;
278 	int64_t err, res = 0;
279 
280 	cnt = epoll_wait(rb->epoll_fd, rb->events, rb->ring_cnt, timeout_ms);
281 	if (cnt < 0)
282 		return -errno;
283 
284 	for (i = 0; i < cnt; i++) {
285 		__u32 ring_id = rb->events[i].data.fd;
286 		struct ring *ring = &rb->rings[ring_id];
287 
288 		err = ringbuf_process_ring(ring);
289 		if (err < 0)
290 			return err;
291 		res += err;
292 	}
293 	if (res > INT_MAX)
294 		return INT_MAX;
295 	return res;
296 }
297 
298 /* Get an fd that can be used to sleep until data is available in the ring(s) */
299 int ring_buffer__epoll_fd(const struct ring_buffer *rb)
300 {
301 	return rb->epoll_fd;
302 }
303