tools/lib/bpf/ringbuf.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_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 }