| Message ID | 20231219184556.1552951-1-vdonnefort@google.com (mailing list archive) |
|---|---|
| Headers | show |
| Series | Introducing trace buffer mapping by user-space | expand |
On Tue, 19 Dec 2023 18:45:54 +0000 Vincent Donnefort <vdonnefort@google.com> wrote: > The tracing ring-buffers can be stored on disk or sent to network > without any copy via splice. However the later doesn't allow real time > processing of the traces. A solution is to give userspace direct access > to the ring-buffer pages via a mapping. An application can now become a > consumer of the ring-buffer, in a similar fashion to what trace_pipe > offers. > > Attached to this cover letter an example of consuming read for a > ring-buffer, using libtracefs. > I'm still testing this, but I needed to add this patch to fix two bugs. One is that you are calling rb_wakeup_waiters() for both the buffer and the cpu_buffer, and it needs to know which one to use the container_of() macro. The other is a "goto unlock" that unlocks two locks where only one was taken. -- Steve diff --git a/kernel/trace/ring_buffer.c b/kernel/trace/ring_buffer.c index 35f3736f660b..987ad7bd1e8b 100644 --- a/kernel/trace/ring_buffer.c +++ b/kernel/trace/ring_buffer.c @@ -389,6 +389,7 @@ struct rb_irq_work { bool waiters_pending; bool full_waiters_pending; bool wakeup_full; + bool is_cpu_buffer; }; /* @@ -771,10 +772,20 @@ static void rb_update_meta_page(struct ring_buffer_per_cpu *cpu_buffer) static void rb_wake_up_waiters(struct irq_work *work) { struct rb_irq_work *rbwork = container_of(work, struct rb_irq_work, work); - struct ring_buffer_per_cpu *cpu_buffer = - container_of(rbwork, struct ring_buffer_per_cpu, irq_work); + struct ring_buffer_per_cpu *cpu_buffer; + struct trace_buffer *buffer; + int cpu; - rb_update_meta_page(cpu_buffer); + if (rbwork->is_cpu_buffer) { + cpu_buffer = container_of(rbwork, struct ring_buffer_per_cpu, irq_work); + rb_update_meta_page(cpu_buffer); + } else { + buffer = container_of(rbwork, struct trace_buffer, irq_work); + for_each_buffer_cpu(buffer, cpu) { + cpu_buffer = buffer->buffers[cpu]; + rb_update_meta_page(cpu_buffer); + } + } wake_up_all(&rbwork->waiters); if (rbwork->full_waiters_pending || rbwork->wakeup_full) { @@ -1569,6 +1580,7 @@ rb_allocate_cpu_buffer(struct trace_buffer *buffer, long nr_pages, int cpu) init_waitqueue_head(&cpu_buffer->irq_work.waiters); init_waitqueue_head(&cpu_buffer->irq_work.full_waiters); mutex_init(&cpu_buffer->mapping_lock); + cpu_buffer->irq_work.is_cpu_buffer = true; bpage = kzalloc_node(ALIGN(sizeof(*bpage), cache_line_size()), GFP_KERNEL, cpu_to_node(cpu)); @@ -6209,7 +6221,8 @@ int ring_buffer_map(struct trace_buffer *buffer, int cpu) if (cpu_buffer->mapped) { WRITE_ONCE(cpu_buffer->mapped, cpu_buffer->mapped + 1); - goto unlock; + mutex_unlock(&cpu_buffer->mapping_lock); + return 0; } /* prevent another thread from changing buffer sizes */
On Tue, Dec 19, 2023 at 03:39:24PM -0500, Steven Rostedt wrote: > On Tue, 19 Dec 2023 18:45:54 +0000 > Vincent Donnefort <vdonnefort@google.com> wrote: > > > The tracing ring-buffers can be stored on disk or sent to network > > without any copy via splice. However the later doesn't allow real time > > processing of the traces. A solution is to give userspace direct access > > to the ring-buffer pages via a mapping. An application can now become a > > consumer of the ring-buffer, in a similar fashion to what trace_pipe > > offers. > > > > Attached to this cover letter an example of consuming read for a > > ring-buffer, using libtracefs. > > > > I'm still testing this, but I needed to add this patch to fix two bugs. One > is that you are calling rb_wakeup_waiters() for both the buffer and the > cpu_buffer, and it needs to know which one to use the container_of() macro. > > The other is a "goto unlock" that unlocks two locks where only one was taken. > > -- Steve > > diff --git a/kernel/trace/ring_buffer.c b/kernel/trace/ring_buffer.c > index 35f3736f660b..987ad7bd1e8b 100644 > --- a/kernel/trace/ring_buffer.c > +++ b/kernel/trace/ring_buffer.c > @@ -389,6 +389,7 @@ struct rb_irq_work { > bool waiters_pending; > bool full_waiters_pending; > bool wakeup_full; > + bool is_cpu_buffer; > }; > > /* > @@ -771,10 +772,20 @@ static void rb_update_meta_page(struct ring_buffer_per_cpu *cpu_buffer) > static void rb_wake_up_waiters(struct irq_work *work) > { > struct rb_irq_work *rbwork = container_of(work, struct rb_irq_work, work); > - struct ring_buffer_per_cpu *cpu_buffer = > - container_of(rbwork, struct ring_buffer_per_cpu, irq_work); > + struct ring_buffer_per_cpu *cpu_buffer; > + struct trace_buffer *buffer; > + int cpu; > > - rb_update_meta_page(cpu_buffer); > + if (rbwork->is_cpu_buffer) { > + cpu_buffer = container_of(rbwork, struct ring_buffer_per_cpu, irq_work); > + rb_update_meta_page(cpu_buffer); > + } else { > + buffer = container_of(rbwork, struct trace_buffer, irq_work); > + for_each_buffer_cpu(buffer, cpu) { > + cpu_buffer = buffer->buffers[cpu]; > + rb_update_meta_page(cpu_buffer); > + } > + } Arg, somehow never reproduced the problem :-\. I suppose you need to cat trace/trace_pipe and mmap(trace/cpuX/trace_pipe) at the same time? Updating the meta-page is only useful if the reader we are waking up is a user-space one, which would only happen with the cpu_buffer version of this function. We could limit the update of the meta_page only to this case? > > wake_up_all(&rbwork->waiters); > if (rbwork->full_waiters_pending || rbwork->wakeup_full) { > @@ -1569,6 +1580,7 @@ rb_allocate_cpu_buffer(struct trace_buffer *buffer, long nr_pages, int cpu) > init_waitqueue_head(&cpu_buffer->irq_work.waiters); > init_waitqueue_head(&cpu_buffer->irq_work.full_waiters); > mutex_init(&cpu_buffer->mapping_lock); > + cpu_buffer->irq_work.is_cpu_buffer = true; > > bpage = kzalloc_node(ALIGN(sizeof(*bpage), cache_line_size()), > GFP_KERNEL, cpu_to_node(cpu)); > @@ -6209,7 +6221,8 @@ int ring_buffer_map(struct trace_buffer *buffer, int cpu) > > if (cpu_buffer->mapped) { > WRITE_ONCE(cpu_buffer->mapped, cpu_buffer->mapped + 1); > - goto unlock; > + mutex_unlock(&cpu_buffer->mapping_lock); > + return 0; > } > > /* prevent another thread from changing buffer sizes */
On Wed, 20 Dec 2023 13:06:06 +0000 Vincent Donnefort <vdonnefort@google.com> wrote: > > @@ -771,10 +772,20 @@ static void rb_update_meta_page(struct ring_buffer_per_cpu *cpu_buffer) > > static void rb_wake_up_waiters(struct irq_work *work) > > { > > struct rb_irq_work *rbwork = container_of(work, struct rb_irq_work, work); > > - struct ring_buffer_per_cpu *cpu_buffer = > > - container_of(rbwork, struct ring_buffer_per_cpu, irq_work); > > + struct ring_buffer_per_cpu *cpu_buffer; > > + struct trace_buffer *buffer; > > + int cpu; > > > > - rb_update_meta_page(cpu_buffer); > > + if (rbwork->is_cpu_buffer) { > > + cpu_buffer = container_of(rbwork, struct ring_buffer_per_cpu, irq_work); > > + rb_update_meta_page(cpu_buffer); > > + } else { > > + buffer = container_of(rbwork, struct trace_buffer, irq_work); > > + for_each_buffer_cpu(buffer, cpu) { > > + cpu_buffer = buffer->buffers[cpu]; > > + rb_update_meta_page(cpu_buffer); > > + } > > + } > > Arg, somehow never reproduced the problem :-\. I suppose you need to cat > trace/trace_pipe and mmap(trace/cpuX/trace_pipe) at the same time? It triggered as soon as I ran "trace-cmd start -e sched_switch" In other words, it broke the non mmap case. This function gets called for both the buffer and cpu_buffer irq_work entries. You added the container_of() to get access to cpu_buffer, when the rbwork could also be for the main buffer too. The main buffer has no meta page, and it triggered a NULL pointer dereference, as "cpu_buffer->mapped" returned true (because it was on something of the buffer structure that wasn't zero), and then here: if (cpu_buffer->mapped) { WRITE_ONCE(cpu_buffer->meta_page->reader.read, 0); It dereferenced cpu_buffer->meta_page->reader which is only God knows what! > > Updating the meta-page is only useful if the reader we are waking up is a > user-space one, which would only happen with the cpu_buffer version of this > function. We could limit the update of the meta_page only to this case? I rather not add another irq_work entry. This workaround should be good enough. Thanks, -- Steve
On Wed, Dec 20, 2023 at 08:29:32AM -0500, Steven Rostedt wrote: > On Wed, 20 Dec 2023 13:06:06 +0000 > Vincent Donnefort <vdonnefort@google.com> wrote: > > > > @@ -771,10 +772,20 @@ static void rb_update_meta_page(struct ring_buffer_per_cpu *cpu_buffer) > > > static void rb_wake_up_waiters(struct irq_work *work) > > > { > > > struct rb_irq_work *rbwork = container_of(work, struct rb_irq_work, work); > > > - struct ring_buffer_per_cpu *cpu_buffer = > > > - container_of(rbwork, struct ring_buffer_per_cpu, irq_work); > > > + struct ring_buffer_per_cpu *cpu_buffer; > > > + struct trace_buffer *buffer; > > > + int cpu; > > > > > > - rb_update_meta_page(cpu_buffer); > > > + if (rbwork->is_cpu_buffer) { > > > + cpu_buffer = container_of(rbwork, struct ring_buffer_per_cpu, irq_work); > > > + rb_update_meta_page(cpu_buffer); > > > + } else { > > > + buffer = container_of(rbwork, struct trace_buffer, irq_work); > > > + for_each_buffer_cpu(buffer, cpu) { > > > + cpu_buffer = buffer->buffers[cpu]; > > > + rb_update_meta_page(cpu_buffer); > > > + } > > > + } > > > > Arg, somehow never reproduced the problem :-\. I suppose you need to cat > > trace/trace_pipe and mmap(trace/cpuX/trace_pipe) at the same time? > > It triggered as soon as I ran "trace-cmd start -e sched_switch" > > In other words, it broke the non mmap case. This function gets called for > both the buffer and cpu_buffer irq_work entries. You added the > container_of() to get access to cpu_buffer, when the rbwork could also be > for the main buffer too. The main buffer has no meta page, and it triggered > a NULL pointer dereference, as "cpu_buffer->mapped" returned true (because > it was on something of the buffer structure that wasn't zero), and then here: > > if (cpu_buffer->mapped) { > WRITE_ONCE(cpu_buffer->meta_page->reader.read, 0); > > It dereferenced cpu_buffer->meta_page->reader > > which is only God knows what! > > > > > Updating the meta-page is only useful if the reader we are waking up is a > > user-space one, which would only happen with the cpu_buffer version of this > > function. We could limit the update of the meta_page only to this case? > > I rather not add another irq_work entry. This workaround should be good > enough. I meant, to only do in rb_wake_up_waiters() if (rbwork->is_cpu_buffer) rb_update_meta_page(cpu_buffer) And skip the meta-page update for the !is_cpu_buffer case? > > Thanks, > > -- Steve
On Wed, 20 Dec 2023 13:49:30 +0000 Vincent Donnefort <vdonnefort@google.com> wrote: > I meant, to only do in rb_wake_up_waiters() > > if (rbwork->is_cpu_buffer) > rb_update_meta_page(cpu_buffer) > > And skip the meta-page update for the !is_cpu_buffer case? Ah yeah, that works. -- Steve
The tracing ring-buffers can be stored on disk or sent to network without any copy via splice. However the later doesn't allow real time processing of the traces. A solution is to give userspace direct access to the ring-buffer pages via a mapping. An application can now become a consumer of the ring-buffer, in a similar fashion to what trace_pipe offers. Attached to this cover letter an example of consuming read for a ring-buffer, using libtracefs. Vincent v7 -> v8: * Drop the subbufs renaming into bpages * Use subbuf as a name when relevant v6 -> v7: * Rebase onto lore.kernel.org/lkml/20231215175502.106587604@goodmis.org/ * Support for subbufs * Rename subbufs into bpages v5 -> v6: * Rebase on next-20230802. * (unsigned long) -> (void *) cast for virt_to_page(). * Add a wait for the GET_READER_PAGE ioctl. * Move writer fields update (overrun/pages_lost/entries/pages_touched) in the irq_work. * Rearrange id in struct buffer_page. * Rearrange the meta-page. * ring_buffer_meta_page -> trace_buffer_meta_page. * Add meta_struct_len into the meta-page. v4 -> v5: * Trivial rebase onto 6.5-rc3 (previously 6.4-rc3) v3 -> v4: * Add to the meta-page: - pages_lost / pages_read (allow to compute how full is the ring-buffer) - read (allow to compute how many entries can be read) - A reader_page struct. * Rename ring_buffer_meta_header -> ring_buffer_meta * Rename ring_buffer_get_reader_page -> ring_buffer_map_get_reader_page * Properly consume events on ring_buffer_map_get_reader_page() with rb_advance_reader(). v2 -> v3: * Remove data page list (for non-consuming read) ** Implies removing order > 0 meta-page * Add a new meta page field ->read * Rename ring_buffer_meta_page_header into ring_buffer_meta_header v1 -> v2: * Hide data_pages from the userspace struct * Fix META_PAGE_MAX_PAGES * Support for order > 0 meta-page * Add missing page->mapping. --- /* Need to access private struct to save counters */ struct kbuffer { unsigned long long timestamp; long long lost_events; unsigned long flags; void *subbuffer; void *data; unsigned int index; unsigned int curr; unsigned int next; unsigned int size; unsigned int start; unsigned int first; unsigned int (*read_4)(void *ptr); unsigned long long (*read_8)(void *ptr); unsigned long long (*read_long)(struct kbuffer *kbuf, void *ptr); int (*next_event)(struct kbuffer *kbuf); }; struct trace_buffer_meta { unsigned long entries; unsigned long overrun; unsigned long read; unsigned long subbufs_touched; unsigned long subbufs_lost; unsigned long subbufs_read; struct { unsigned long lost_events; /* Events lost at the time of the reader swap */ __u32 id; /* Reader subbuf ID from 0 to nr_subbufs - 1 */ __u32 read; /* Number of bytes read on the reader subbuf */ } reader; __u32 subbuf_size; __u32 nr_subbufs; /* Number of subbufs in the ring-buffer */ __u32 meta_page_size; /* Size of the meta-page */ __u32 meta_struct_len; /* Len of this struct */ }; static char *argv0; static bool exit_requested; static char *get_this_name(void) { static char *this_name; char *arg; char *p; if (this_name) return this_name; arg = argv0; p = arg+strlen(arg); while (p >= arg && *p != '/') p--; p++; this_name = p; return p; } static void __vdie(const char *fmt, va_list ap, int err) { int ret = errno; char *p = get_this_name(); if (err && errno) perror(p); else ret = -1; fprintf(stderr, " "); vfprintf(stderr, fmt, ap); fprintf(stderr, "\n"); exit(ret); } void pdie(const char *fmt, ...) { va_list ap; va_start(ap, fmt); __vdie(fmt, ap, 1); va_end(ap); } static void read_subbuf(struct tep_handle *tep, struct kbuffer *kbuf) { static struct trace_seq seq; struct tep_record record; if (seq.buffer) trace_seq_reset(&seq); else trace_seq_init(&seq); while ((record.data = kbuffer_read_event(kbuf, &record.ts))) { record.size = kbuffer_event_size(kbuf); kbuffer_next_event(kbuf, NULL); tep_print_event(tep, &seq, &record, "%s-%d %9d\t%s: %s\n", TEP_PRINT_COMM, TEP_PRINT_PID, TEP_PRINT_TIME, TEP_PRINT_NAME, TEP_PRINT_INFO); trace_seq_do_printf(&seq); trace_seq_reset(&seq); } } static int next_reader_subbuf(int fd, struct trace_buffer_meta *meta, unsigned long *read) { __u32 prev_read, prev_reader, new_reader; prev_read = READ_ONCE(meta->reader.read); prev_reader = READ_ONCE(meta->reader.id); if (ioctl(fd, TRACE_MMAP_IOCTL_GET_READER) < 0) pdie("ioctl"); new_reader = READ_ONCE(meta->reader.id); if (prev_reader != new_reader) *read = 0; else *read = prev_read; return new_reader; } static void signal_handler(int unused) { printf("Exit!\n"); exit_requested = true; } int main(int argc, char **argv) { int page_size, meta_len, data_len, subbuf, fd; struct trace_buffer_meta *map; struct tep_handle *tep; struct kbuffer *kbuf; unsigned long read; void *meta, *data; char path[32]; int cpu; if (argc != 2) return -EINVAL; argv0 = argv[0]; cpu = atoi(argv[1]); snprintf(path, 32, "per_cpu/cpu%d/trace_pipe_raw", cpu); tep = tracefs_local_events(NULL); kbuf = tep_kbuffer(tep); page_size = getpagesize(); fd = tracefs_instance_file_open(NULL, path, O_RDONLY); if (fd < 0) pdie("raw"); meta = mmap(NULL, page_size, PROT_READ, MAP_SHARED, fd, 0); if (meta == MAP_FAILED) pdie("mmap"); map = (struct trace_buffer_meta *)meta; meta_len = map->meta_page_size; printf("entries: %lu\n", map->entries); printf("overrun: %lu\n", map->overrun); printf("read: %lu\n", map->read); printf("subbufs_touched:%lu\n", map->subbufs_touched); printf("subbufs_lost: %lu\n", map->subbufs_lost); printf("subbufs_read: %lu\n", map->subbufs_read); printf("nr_subbufs: %u\n", map->nr_subbufs); data_len = map->subbuf_size * map->nr_subbufs; data = mmap(NULL, data_len, PROT_READ, MAP_SHARED, fd, meta_len); if (data == MAP_FAILED) pdie("mmap data"); signal(SIGINT, signal_handler); while (!exit_requested) { subbuf = next_reader_subbuf(fd, map, &read); kbuffer_load_subbuffer(kbuf, data + map->subbuf_size * subbuf); while (kbuf->curr < read) kbuffer_next_event(kbuf, NULL); read_subbuf(tep, kbuf); } munmap(data, data_len); munmap(meta, page_size); close(fd); return 0; } Vincent Donnefort (2): ring-buffer: Introducing ring-buffer mapping functions tracing: Allow user-space mapping of the ring-buffer include/linux/ring_buffer.h | 7 + include/uapi/linux/trace_mmap.h | 31 +++ kernel/trace/ring_buffer.c | 371 +++++++++++++++++++++++++++++++- kernel/trace/trace.c | 79 ++++++- 4 files changed, 484 insertions(+), 4 deletions(-) create mode 100644 include/uapi/linux/trace_mmap.h