diff --git a/include/linux/bpf.h b/include/linux/bpf.h index 2cf04bba16bb..4bae3796c42f 100644 --- a/include/linux/bpf.h +++ b/include/linux/bpf.h @@ -17,6 +17,7 @@ #include #include #include +#include #include #include #include @@ -1771,12 +1772,18 @@ enum { }; struct bpf_stream { - atomic_t capacity; + refcount_t refcnt; + atomic_t capacity; /* bytes reserved against the stream limit */ + atomic_t readable; /* published bytes available to readers */ struct llist_head log; /* list of in-flight stream elements in LIFO order */ struct mutex lock; /* lock protecting backlog_{head,tail} */ struct llist_node *backlog_head; /* list of in-flight stream elements in FIFO order */ struct llist_node *backlog_tail; /* tail of the list above */ + wait_queue_head_t waitq; + struct irq_work notify_work; + bool notify_used; /* notify_work was queued at least once */ + bool dead; }; struct bpf_stream_stage { @@ -1915,7 +1922,7 @@ struct bpf_prog_aux { struct work_struct work; struct rcu_head rcu; }; - struct bpf_stream stream[2]; + struct bpf_stream *stream[2]; struct mutex st_ops_assoc_mutex; struct bpf_map __rcu *st_ops_assoc; }; @@ -4224,9 +4231,10 @@ void bpf_bprintf_cleanup(struct bpf_bprintf_data *data); int bpf_try_get_buffers(struct bpf_bprintf_buffers **bufs); void bpf_put_buffers(void); -void bpf_prog_stream_init(struct bpf_prog *prog); +int bpf_prog_stream_init(struct bpf_prog *prog, gfp_t gfp_extra_flags); void bpf_prog_stream_free(struct bpf_prog *prog); int bpf_prog_stream_read(struct bpf_prog *prog, enum bpf_stream_id stream_id, void __user *buf, u32 len); +int bpf_prog_stream_new_fd(struct bpf_prog *prog, enum bpf_stream_id stream_id, u32 flags); void bpf_stream_stage_init(struct bpf_stream_stage *ss); void bpf_stream_stage_free(struct bpf_stream_stage *ss); __printf(2, 3) diff --git a/include/uapi/linux/bpf.h b/include/uapi/linux/bpf.h index 0aaa54359aeb..4687c3310996 100644 --- a/include/uapi/linux/bpf.h +++ b/include/uapi/linux/bpf.h @@ -936,6 +936,33 @@ union bpf_iter_link_info { * 0 on success or -1 if an error occurred (in which case, * *errno* is set appropriately). * + * BPF_PROG_STREAM_OPEN + * Description + * Open a file descriptor for one of the BPF streams associated + * with the program identified by *prog_fd*. The stream is selected + * by *stream_id*. + * + * The returned file descriptor supports **read**\ (2) and + * **poll**\ (2). Reads block while the stream is empty unless + * **BPF_F_STREAM_NONBLOCK** is specified in *flags*. A non-blocking + * read of an empty stream fails with **EAGAIN**. + * + * **poll**\ (2) reports **POLLIN** when data is available and + * **POLLHUP** once the program has been freed, that is, after every + * reference to it, including links and other file descriptors, has + * been dropped. Hangup may lag the final release because program + * teardown is deferred. Buffered data remains readable after + * **POLLHUP** and a read returns zero after all such data has been + * consumed. + * + * The file descriptor is read-only and has the close-on-exec flag + * set. It is not seekable and **lseek**\ (2) fails with **ESPIPE**. + * *flags* may only contain **BPF_F_STREAM_NONBLOCK**. + * + * Return + * A new file descriptor (a nonnegative integer), or -1 if an + * error occurred (in which case, *errno* is set appropriately). + * * NOTES * eBPF objects (maps and programs) can be shared between processes. * @@ -993,6 +1020,7 @@ enum bpf_cmd { BPF_TOKEN_CREATE, BPF_PROG_STREAM_READ_BY_FD, BPF_PROG_ASSOC_STRUCT_OPS, + BPF_PROG_STREAM_OPEN, __MAX_BPF_CMD, BPF_COMMON_ATTRS = 1 << 16, /* Indicate carrying syscall common attrs. */ }; @@ -1524,6 +1552,11 @@ enum { BPF_STREAM_STDERR = 2, }; +/* flags for BPF_PROG_STREAM_OPEN command */ +enum { + BPF_F_STREAM_NONBLOCK = (1U << 0), +}; + union bpf_attr { struct { /* anonymous struct used by BPF_MAP_CREATE command */ __u32 map_type; /* one of enum bpf_map_type */ @@ -1950,6 +1983,12 @@ union bpf_attr { __u32 flags; } prog_assoc_struct_ops; + struct { + __u32 prog_fd; + __u32 stream_id; + __u32 flags; + } prog_stream_open; + } __attribute__((aligned(8))); /* The description below is an attempt at providing documentation to eBPF diff --git a/kernel/bpf/core.c b/kernel/bpf/core.c index aa72ad3ec689..d3b8b626ec0f 100644 --- a/kernel/bpf/core.c +++ b/kernel/bpf/core.c @@ -142,10 +142,6 @@ struct bpf_prog *bpf_prog_alloc_no_stats(unsigned int size, gfp_t gfp_extra_flag mutex_init(&fp->aux->dst_mutex); mutex_init(&fp->aux->st_ops_assoc_mutex); -#ifdef CONFIG_BPF_SYSCALL - bpf_prog_stream_init(fp); -#endif - return fp; } @@ -289,6 +285,9 @@ struct bpf_prog *bpf_prog_realloc(struct bpf_prog *fp_old, unsigned int size, void __bpf_prog_free(struct bpf_prog *fp) { if (fp->aux) { +#ifdef CONFIG_BPF_SYSCALL + bpf_prog_stream_free(fp); +#endif mutex_destroy(&fp->aux->used_maps_mutex); mutex_destroy(&fp->aux->dst_mutex); mutex_destroy(&fp->aux->st_ops_assoc_mutex); @@ -3086,7 +3085,6 @@ static void bpf_prog_free_deferred(struct work_struct *work) aux = container_of(work, struct bpf_prog_aux, work); #ifdef CONFIG_BPF_SYSCALL bpf_free_kfunc_btf_tab(aux->kfunc_btf_tab); - bpf_prog_stream_free(aux->prog); #endif #ifdef CONFIG_CGROUP_BPF if (aux->cgroup_atype != CGROUP_BPF_ATTACH_TYPE_INVALID) diff --git a/kernel/bpf/stream.c b/kernel/bpf/stream.c index 2b80a0599865..83e803ea54ef 100644 --- a/kernel/bpf/stream.c +++ b/kernel/bpf/stream.c @@ -2,11 +2,15 @@ /* Copyright (c) 2025 Meta Platforms, Inc. and affiliates. */ #include +#include #include #include #include +#include #include #include +#include +#include static void bpf_stream_elem_init(struct bpf_stream_elem *elem, int len) { @@ -73,16 +77,68 @@ static void bpf_stream_release_capacity(struct bpf_stream *stream, int len) atomic_sub(len, &stream->capacity); } +static void bpf_stream_notify(struct irq_work *work) +{ + struct bpf_stream *stream = container_of(work, struct bpf_stream, notify_work); + + /* + * Writers run in arbitrary program contexts, including NMI and regions + * that already hold wait queue or epoll locks. Wake waiters from + * irq_work instead, where taking those locks is safe. + */ + wake_up_interruptible_poll(&stream->waitq, EPOLLIN | EPOLLRDNORM); +} + +static void bpf_stream_queue_notify(struct bpf_stream *stream) +{ + /* + * Record that the work has been used so that teardown only pays for + * irq_work_sync(), which may wait for an RCU grace period, when a + * callback could actually be in flight. + */ + if (!READ_ONCE(stream->notify_used)) + WRITE_ONCE(stream->notify_used, true); + irq_work_queue(&stream->notify_work); +} + +static int bpf_stream_readable_bytes(struct bpf_stream *stream) +{ + return atomic_read_acquire(&stream->readable); +} + +static void bpf_stream_publish(struct bpf_stream *stream, int len) +{ + /* + * Pairs with atomic_read_acquire() in bpf_stream_readable_bytes(). + * + * Notify only when the stream turns from empty to readable. Readers + * block and pollers wait only after finding it empty, and a read never + * takes the count below zero, so the publication that makes it positive + * is the one they wait for. Publishing into a stream that already holds + * data would only raise an interrupt for waiters that are being woken + * already, or for nobody at all. + */ + if (atomic_add_return_release(len, &stream->readable) == len) + bpf_stream_queue_notify(stream); +} + static int bpf_stream_push_str(struct bpf_stream *stream, const char *str, int len) { - int ret = bpf_stream_consume_capacity(stream, len); + int ret; + /* Nothing to publish; do not allocate an element for it. */ + if (!len) + return 0; + + ret = bpf_stream_consume_capacity(stream, len); if (ret) return ret; ret = __bpf_stream_push_str(&stream->log, str, len); if (ret) bpf_stream_release_capacity(stream, len); + else + bpf_stream_publish(stream, len); return ret; } @@ -91,7 +147,7 @@ static struct bpf_stream *bpf_stream_get(enum bpf_stream_id stream_id, struct bp { if (stream_id != BPF_STDOUT && stream_id != BPF_STDERR) return NULL; - return &aux->stream[stream_id - 1]; + return aux->stream[stream_id - 1]; } static void bpf_stream_free_elem(struct bpf_stream_elem *elem) @@ -159,14 +215,16 @@ static bool bpf_stream_consume_elem(struct bpf_stream_elem *elem, int *len) static int bpf_stream_read(struct bpf_stream *stream, void __user *buf, int len) { - int rem_len = len, cons_len, ret = 0; + int read_len, rem_len, cons_len, ret = 0; struct bpf_stream_elem *elem = NULL; struct llist_node *node; mutex_lock(&stream->lock); + read_len = min(len, bpf_stream_readable_bytes(stream)); + rem_len = read_len; while (rem_len) { - int pos = len - rem_len; + int pos = read_len - rem_len; int chunk, n; bool cont; @@ -188,7 +246,7 @@ static int bpf_stream_read(struct bpf_stream *stream, void __user *buf, int len) /* Keep any successfully copied bytes; -EFAULT only if none. */ elem->consumed_len -= n; rem_len += n; - ret = (len == rem_len) ? -EFAULT : 0; + ret = (read_len == rem_len) ? -EFAULT : 0; break; } @@ -199,8 +257,9 @@ static int bpf_stream_read(struct bpf_stream *stream, void __user *buf, int len) bpf_stream_free_elem(elem); } + atomic_sub(read_len - rem_len, &stream->readable); mutex_unlock(&stream->lock); - return ret ? ret : len - rem_len; + return ret ? ret : read_len - rem_len; } int bpf_prog_stream_read(struct bpf_prog *prog, enum bpf_stream_id stream_id, void __user *buf, u32 len) @@ -215,6 +274,111 @@ int bpf_prog_stream_read(struct bpf_prog *prog, enum bpf_stream_id stream_id, vo return bpf_stream_read(stream, buf, len); } +static bool bpf_stream_has_data(struct bpf_stream *stream) +{ + return bpf_stream_readable_bytes(stream) > 0; +} + +static void bpf_stream_put(struct bpf_stream *stream) +{ + if (refcount_dec_and_test(&stream->refcnt)) { + struct llist_node *list; + + /* Only a stream that ever queued its work can have a callback in flight. */ + if (READ_ONCE(stream->notify_used)) + irq_work_sync(&stream->notify_work); + list = llist_del_all(&stream->log); + bpf_stream_free_list(list); + bpf_stream_free_list(stream->backlog_head); + mutex_destroy(&stream->lock); + kfree(stream); + } +} + +static int bpf_stream_release(struct inode *inode, struct file *file) +{ + bpf_stream_put(file->private_data); + return 0; +} + +static ssize_t bpf_stream_file_read(struct file *file, char __user *buf, size_t len, + loff_t *ppos) +{ + struct bpf_stream *stream = file->private_data; + bool dead; + int ret; + + if (!len) + return 0; + + for (;;) { + /* + * Sample teardown state before looking for data. Nothing is + * published once the program is gone, so finding the stream + * empty after observing dead means EOF. The opposite order could + * report EOF while data published just before teardown is still + * buffered. + */ + dead = smp_load_acquire(&stream->dead); + ret = bpf_stream_read(stream, buf, len); + if (ret) + return ret; + if (dead) + return 0; + if (file->f_flags & O_NONBLOCK) + return -EAGAIN; + + ret = wait_event_interruptible(stream->waitq, + bpf_stream_has_data(stream) || + READ_ONCE(stream->dead)); + if (ret) + return ret; + } +} + +static __poll_t bpf_stream_poll(struct file *file, struct poll_table_struct *pts) +{ + struct bpf_stream *stream = file->private_data; + __poll_t events = 0; + + /* + * poll_wait() only registers the wait queue callback. Register before + * checking persistent state so a concurrent publication or teardown is + * observed either by the callback or by the checks below. + */ + poll_wait(file, &stream->waitq, pts); + if (bpf_stream_has_data(stream)) + events |= EPOLLIN | EPOLLRDNORM; + if (READ_ONCE(stream->dead)) + events |= EPOLLHUP; + return events; +} + +static const struct file_operations bpf_stream_fops = { + .release = bpf_stream_release, + .read = bpf_stream_file_read, + .poll = bpf_stream_poll, +}; + +int bpf_prog_stream_new_fd(struct bpf_prog *prog, enum bpf_stream_id stream_id, u32 flags) +{ + struct bpf_stream *stream; + int fd_flags = O_RDONLY | O_CLOEXEC; + int fd; + + stream = bpf_stream_get(stream_id, prog->aux); + if (!stream) + return -ENOENT; + if (flags & BPF_F_STREAM_NONBLOCK) + fd_flags |= O_NONBLOCK; + + refcount_inc(&stream->refcnt); + fd = anon_inode_getfd("bpf-stream", &bpf_stream_fops, stream, fd_flags); + if (fd < 0) + bpf_stream_put(stream); + return fd; +} + __bpf_kfunc_start_defs(); /* @@ -282,28 +446,47 @@ __bpf_kfunc_end_defs(); /* Added kfunc to common_btf_ids */ -void bpf_prog_stream_init(struct bpf_prog *prog) +int bpf_prog_stream_init(struct bpf_prog *prog, gfp_t gfp_extra_flags) { int i; for (i = 0; i < ARRAY_SIZE(prog->aux->stream); i++) { - atomic_set(&prog->aux->stream[i].capacity, 0); - init_llist_head(&prog->aux->stream[i].log); - mutex_init(&prog->aux->stream[i].lock); - prog->aux->stream[i].backlog_head = NULL; - prog->aux->stream[i].backlog_tail = NULL; + struct bpf_stream *stream; + + /* On failure, bpf_prog_stream_free() releases the streams allocated so far. */ + stream = kzalloc_obj(*stream, + bpf_memcg_flags(GFP_KERNEL | gfp_extra_flags)); + if (!stream) + return -ENOMEM; + + refcount_set(&stream->refcnt, 1); + init_llist_head(&stream->log); + mutex_init(&stream->lock); + init_waitqueue_head(&stream->waitq); + init_irq_work(&stream->notify_work, bpf_stream_notify); + prog->aux->stream[i] = stream; } + return 0; } void bpf_prog_stream_free(struct bpf_prog *prog) { - struct llist_node *list; int i; for (i = 0; i < ARRAY_SIZE(prog->aux->stream); i++) { - list = llist_del_all(&prog->aux->stream[i].log); - bpf_stream_free_list(list); - bpf_stream_free_list(prog->aux->stream[i].backlog_head); + struct bpf_stream *stream = prog->aux->stream[i]; + + if (!stream) + continue; + /* + * Pairs with smp_load_acquire() in bpf_stream_file_read(): every + * publication precedes the dead flag, so a reader that observes + * it also observes all buffered data. + */ + smp_store_release(&stream->dead, true); + wake_up_interruptible_poll(&stream->waitq, EPOLLHUP); + bpf_stream_put(stream); + prog->aux->stream[i] = NULL; } } @@ -333,8 +516,8 @@ int bpf_stream_stage_printk(struct bpf_stream_stage *ss, const char *fmt, ...) va_start(args, fmt); len = vscnprintf(buf->buf, ARRAY_SIZE(buf->buf), fmt, args); va_end(args); - /* Exclude NULL byte during push. */ - ret = __bpf_stream_push_str(&ss->log, buf->buf, len); + /* Exclude NULL byte during push; skip empty output entirely. */ + ret = len ? __bpf_stream_push_str(&ss->log, buf->buf, len) : 0; if (!ret) ss->len += len; bpf_put_buffers(); @@ -366,6 +549,7 @@ int bpf_stream_stage_commit(struct bpf_stream_stage *ss, struct bpf_prog *prog, list = tail; } llist_add_batch(head, tail, &stream->log); + bpf_stream_publish(stream, ss->len); return 0; } diff --git a/kernel/bpf/syscall.c b/kernel/bpf/syscall.c index e1bcfd43f54f..ac52f4ae414c 100644 --- a/kernel/bpf/syscall.c +++ b/kernel/bpf/syscall.c @@ -3080,6 +3080,10 @@ static int bpf_prog_load(union bpf_attr *attr, bpfptr_t uattr, struct bpf_log_at prog->aux->user = get_current_user(); prog->len = attr->insn_cnt; + err = bpf_prog_stream_init(prog, GFP_USER); + if (err) + goto free_prog; + err = -EFAULT; if (copy_from_bpfptr(prog->insns, make_bpfptr(attr->insns, uattr.is_kernel), @@ -6316,6 +6320,28 @@ static int prog_assoc_struct_ops(union bpf_attr *attr) return ret; } +#define BPF_PROG_STREAM_OPEN_LAST_FIELD prog_stream_open.flags + +static int prog_stream_open(union bpf_attr *attr) +{ + struct bpf_prog *prog; + u32 flags = attr->prog_stream_open.flags; + int ret; + + if (CHECK_ATTR(BPF_PROG_STREAM_OPEN)) + return -EINVAL; + if (flags & ~BPF_F_STREAM_NONBLOCK) + return -EINVAL; + + prog = bpf_prog_get(attr->prog_stream_open.prog_fd); + if (IS_ERR(prog)) + return PTR_ERR(prog); + + ret = bpf_prog_stream_new_fd(prog, attr->prog_stream_open.stream_id, flags); + bpf_prog_put(prog); + return ret; +} + static int __sys_bpf(enum bpf_cmd cmd, bpfptr_t uattr, unsigned int size, bpfptr_t uattr_common, unsigned int size_common) { @@ -6488,6 +6514,9 @@ static int __sys_bpf(enum bpf_cmd cmd, bpfptr_t uattr, unsigned int size, case BPF_PROG_ASSOC_STRUCT_OPS: err = prog_assoc_struct_ops(&attr); break; + case BPF_PROG_STREAM_OPEN: + err = prog_stream_open(&attr); + break; default: err = -EINVAL; break; diff --git a/tools/bpf/bpftool/Documentation/bpftool-prog.rst b/tools/bpf/bpftool/Documentation/bpftool-prog.rst index 90fe8c61bf42..17aa0d560e32 100644 --- a/tools/bpf/bpftool/Documentation/bpftool-prog.rst +++ b/tools/bpf/bpftool/Documentation/bpftool-prog.rst @@ -18,7 +18,8 @@ SYNOPSIS *OPTIONS* := { |COMMON_OPTIONS| | { **-f** | **--bpffs** } | { **-m** | **--mapcompat** } | { **-n** | **--nomount** } | -{ **-L** | **--use-loader** } | [ { **-S** | **--sign** } **-k** **-i** ] } +{ **-L** | **--use-loader** } | [ { **-S** | **--sign** } **-k** **-i** ] | +{ **-w** | **--wait** } } *COMMANDS* := { **show** | **list** | **dump xlated** | **dump jited** | **pin** | **load** | @@ -186,6 +187,11 @@ bpftool prog tracelog { stdout | stderr } *PROG* error messages to the standard error stream. This facility should be used only for debugging purposes. + By default, bpftool prints the output buffered so far and exits. With + **-w** or **--wait**, it keeps printing new output as the program produces + it, until the program is unloaded or is hit. Waiting requires a + kernel that supports opening a stream as a file descriptor. + bpftool prog run *PROG* data_in *FILE* [data_out *FILE* [data_size_out *L*]] [ctx_in *FILE* [ctx_out *FILE* [ctx_size_out *M*]]] [repeat *N*] Run BPF program *PROG* in the kernel testing infrastructure for BPF, meaning that the program works on the data and context provided by the @@ -267,6 +273,10 @@ OPTIONS Path to the X.509 certificate file in PEM or DER format, required when signing. +-w, --wait + When dumping a program stream with **bpftool prog tracelog**, wait for new + output instead of exiting once the buffered output has been printed. + EXAMPLES ======== **# bpftool prog show** diff --git a/tools/bpf/bpftool/bash-completion/bpftool b/tools/bpf/bpftool/bash-completion/bpftool index 9d9ced270685..a28af31937ad 100644 --- a/tools/bpf/bpftool/bash-completion/bpftool +++ b/tools/bpf/bpftool/bash-completion/bpftool @@ -269,7 +269,7 @@ _bpftool() # Deal with options if [[ ${words[cword]} == -* ]]; then local c='--version --json --pretty --bpffs --mapcompat --debug \ - --use-loader --base-btf --sign -i -k' + --use-loader --base-btf --sign -i -k --wait' COMPREPLY=( $( compgen -W "$c" -- "$cur" ) ) return 0 fi diff --git a/tools/bpf/bpftool/main.c b/tools/bpf/bpftool/main.c index 83884b21e708..d15fec71b64e 100644 --- a/tools/bpf/bpftool/main.c +++ b/tools/bpf/bpftool/main.c @@ -31,6 +31,7 @@ bool block_mount; bool verifier_logs; bool relaxed_maps; bool use_loader; +bool wait_output; struct btf *base_btf; struct hashmap *refs_table; bool sign_progs; @@ -462,6 +463,7 @@ int main(int argc, char **argv) { "nomount", no_argument, NULL, 'n' }, { "debug", no_argument, NULL, 'd' }, { "use-loader", no_argument, NULL, 'L' }, + { "wait", no_argument, NULL, 'w' }, { "sign", no_argument, NULL, 'S' }, { "base-btf", required_argument, NULL, 'B' }, { 0 } @@ -480,7 +482,7 @@ int main(int argc, char **argv) bin_name = "bpftool"; opterr = 0; - while ((opt = getopt_long(argc, argv, "VhpjfLmndSi:k:B:l", + while ((opt = getopt_long(argc, argv, "VhpjfLmndSi:k:B:lw", options, NULL)) >= 0) { switch (opt) { case 'V': @@ -530,6 +532,9 @@ int main(int argc, char **argv) case 'L': use_loader = true; break; + case 'w': + wait_output = true; + break; case 'S': sign_progs = true; use_loader = true; diff --git a/tools/bpf/bpftool/main.h b/tools/bpf/bpftool/main.h index 48eedcb9f6c0..fa2c877a3cbf 100644 --- a/tools/bpf/bpftool/main.h +++ b/tools/bpf/bpftool/main.h @@ -89,6 +89,7 @@ extern bool block_mount; extern bool verifier_logs; extern bool relaxed_maps; extern bool use_loader; +extern bool wait_output; extern struct btf *base_btf; extern struct hashmap *refs_table; extern bool sign_progs; diff --git a/tools/bpf/bpftool/prog.c b/tools/bpf/bpftool/prog.c index 24e40dfab469..5e27b22444f2 100644 --- a/tools/bpf/bpftool/prog.c +++ b/tools/bpf/bpftool/prog.c @@ -1119,21 +1119,66 @@ enum prog_tracelog_mode { TRACE_STDERR, }; +static void exit_stream(int signo) +{ + exit(0); +} + +/* Consumes prog_fd. */ static int prog_tracelog_stream(int prog_fd, enum prog_tracelog_mode mode) { + const struct sigaction act = { .sa_handler = exit_stream }; + const int signals[] = { SIGHUP, SIGINT, SIGTERM }; FILE *file = mode == TRACE_STDOUT ? stdout : stderr; int stream_id = mode == TRACE_STDOUT ? 1 : 2; char buf[512]; - int ret; + unsigned int i; + int fd, ret; - ret = 0; - do { - ret = bpf_prog_stream_read(prog_fd, stream_id, buf, sizeof(buf), NULL); - if (ret > 0) - fwrite(buf, sizeof(buf[0]), ret, file); - } while (ret > 0); + if (!wait_output) { + do { + ret = bpf_prog_stream_read(prog_fd, stream_id, buf, sizeof(buf), NULL); + if (ret > 0) + fwrite(buf, sizeof(buf[0]), ret, file); + } while (ret > 0); + if (ret < 0) + p_err("failed to read stream: %s", strerror(-ret)); + close(prog_fd); + goto out; + } + fd = bpf_prog_stream_open(prog_fd, stream_id, NULL); + /* + * The stream descriptor does not keep the program alive. Drop the + * program reference so that reads return EOF once the program is gone. + */ + close(prog_fd); + if (fd == -EINVAL) { + /* The bpf() syscall rejects unknown commands with EINVAL. */ + p_err("waiting for stream output is not supported by this kernel"); + return -1; + } + if (fd < 0) { + p_err("failed to open stream: %s", strerror(-fd)); + return -1; + } + + /* + * Exit from the handler like the trace pipe variant does. A flag checked + * between reads would miss a signal that lands before read() blocks and + * only end the loop at the next print. + */ + for (i = 0; i < ARRAY_SIZE(signals); i++) + sigaction(signals[i], &act, NULL); + while ((ret = read(fd, buf, sizeof(buf))) > 0) { + fwrite(buf, sizeof(buf[0]), ret, file); + fflush(file); + } + if (ret < 0) + p_err("failed to read stream: %s", strerror(errno)); + close(fd); +out: fflush(file); return ret ? -1 : 0; } @@ -2659,7 +2704,8 @@ static int do_help(int argc, char **argv) " METRIC := { cycles | instructions | l1d_loads | llc_misses | itlb_misses | dtlb_misses }\n" " " HELP_SPEC_OPTIONS " |\n" " {-f|--bpffs} | {-m|--mapcompat} | {-n|--nomount} |\n" - " {-L|--use-loader} | [ {-S|--sign } {-k} {-i} ] \n" + " {-L|--use-loader} | [ {-S|--sign } {-k} {-i} ] |\n" + " {-w|--wait} }\n" "", bin_name, argv[-2]); diff --git a/tools/include/uapi/linux/bpf.h b/tools/include/uapi/linux/bpf.h index 0aaa54359aeb..4687c3310996 100644 --- a/tools/include/uapi/linux/bpf.h +++ b/tools/include/uapi/linux/bpf.h @@ -936,6 +936,33 @@ union bpf_iter_link_info { * 0 on success or -1 if an error occurred (in which case, * *errno* is set appropriately). * + * BPF_PROG_STREAM_OPEN + * Description + * Open a file descriptor for one of the BPF streams associated + * with the program identified by *prog_fd*. The stream is selected + * by *stream_id*. + * + * The returned file descriptor supports **read**\ (2) and + * **poll**\ (2). Reads block while the stream is empty unless + * **BPF_F_STREAM_NONBLOCK** is specified in *flags*. A non-blocking + * read of an empty stream fails with **EAGAIN**. + * + * **poll**\ (2) reports **POLLIN** when data is available and + * **POLLHUP** once the program has been freed, that is, after every + * reference to it, including links and other file descriptors, has + * been dropped. Hangup may lag the final release because program + * teardown is deferred. Buffered data remains readable after + * **POLLHUP** and a read returns zero after all such data has been + * consumed. + * + * The file descriptor is read-only and has the close-on-exec flag + * set. It is not seekable and **lseek**\ (2) fails with **ESPIPE**. + * *flags* may only contain **BPF_F_STREAM_NONBLOCK**. + * + * Return + * A new file descriptor (a nonnegative integer), or -1 if an + * error occurred (in which case, *errno* is set appropriately). + * * NOTES * eBPF objects (maps and programs) can be shared between processes. * @@ -993,6 +1020,7 @@ enum bpf_cmd { BPF_TOKEN_CREATE, BPF_PROG_STREAM_READ_BY_FD, BPF_PROG_ASSOC_STRUCT_OPS, + BPF_PROG_STREAM_OPEN, __MAX_BPF_CMD, BPF_COMMON_ATTRS = 1 << 16, /* Indicate carrying syscall common attrs. */ }; @@ -1524,6 +1552,11 @@ enum { BPF_STREAM_STDERR = 2, }; +/* flags for BPF_PROG_STREAM_OPEN command */ +enum { + BPF_F_STREAM_NONBLOCK = (1U << 0), +}; + union bpf_attr { struct { /* anonymous struct used by BPF_MAP_CREATE command */ __u32 map_type; /* one of enum bpf_map_type */ @@ -1950,6 +1983,12 @@ union bpf_attr { __u32 flags; } prog_assoc_struct_ops; + struct { + __u32 prog_fd; + __u32 stream_id; + __u32 flags; + } prog_stream_open; + } __attribute__((aligned(8))); /* The description below is an attempt at providing documentation to eBPF diff --git a/tools/lib/bpf/bpf.c b/tools/lib/bpf/bpf.c index a9de7f107cf7..b49822d212ae 100644 --- a/tools/lib/bpf/bpf.c +++ b/tools/lib/bpf/bpf.c @@ -1466,6 +1466,25 @@ int bpf_prog_stream_read(int prog_fd, __u32 stream_id, void *buf, __u32 buf_len, return libbpf_err_errno(err); } +int bpf_prog_stream_open(int prog_fd, __u32 stream_id, + const struct bpf_prog_stream_open_opts *opts) +{ + const size_t attr_sz = offsetofend(union bpf_attr, prog_stream_open); + union bpf_attr attr; + int fd; + + if (!OPTS_VALID(opts, bpf_prog_stream_open_opts)) + return libbpf_err(-EINVAL); + + memset(&attr, 0, attr_sz); + attr.prog_stream_open.prog_fd = prog_fd; + attr.prog_stream_open.stream_id = stream_id; + attr.prog_stream_open.flags = OPTS_GET(opts, flags, 0); + + fd = sys_bpf_fd(BPF_PROG_STREAM_OPEN, &attr, attr_sz); + return libbpf_err_errno(fd); +} + int bpf_prog_assoc_struct_ops(int prog_fd, int map_fd, struct bpf_prog_assoc_struct_ops_opts *opts) { diff --git a/tools/lib/bpf/bpf.h b/tools/lib/bpf/bpf.h index 490e8cb4ba53..826d9cc9ab65 100644 --- a/tools/lib/bpf/bpf.h +++ b/tools/lib/bpf/bpf.h @@ -759,10 +759,34 @@ struct bpf_prog_stream_read_opts { * * @return The number of bytes read, on success; negative error code, otherwise * (errno is also set to the error code) + * + * For blocking reads and polling, prefer **bpf_prog_stream_open**. */ LIBBPF_API int bpf_prog_stream_read(int prog_fd, __u32 stream_id, void *buf, __u32 buf_len, struct bpf_prog_stream_read_opts *opts); +struct bpf_prog_stream_open_opts { + size_t sz; + __u32 flags; + size_t :0; +}; +#define bpf_prog_stream_open_opts__last_field flags + +/** + * @brief **bpf_prog_stream_open** opens a file descriptor for a BPF stream of + * a given BPF program. + * + * @param prog_fd FD for the BPF program whose BPF stream is to be opened. + * @param stream_id ID of the BPF stream to be opened. + * @param opts optional options, can be NULL. BPF_F_STREAM_NONBLOCK requests a + * non-blocking descriptor. + * + * @return A new stream FD, on success; negative error code, otherwise (errno + * is also set to the error code) + */ +LIBBPF_API int bpf_prog_stream_open(int prog_fd, __u32 stream_id, + const struct bpf_prog_stream_open_opts *opts); + struct bpf_prog_assoc_struct_ops_opts { size_t sz; __u32 flags; diff --git a/tools/lib/bpf/libbpf.map b/tools/lib/bpf/libbpf.map index 18d27d20102e..53b7b591325f 100644 --- a/tools/lib/bpf/libbpf.map +++ b/tools/lib/bpf/libbpf.map @@ -459,6 +459,7 @@ LIBBPF_1.7.0 { LIBBPF_1.8.0 { global: bpf_map__attach_cgroup_opts; + bpf_prog_stream_open; bpf_program__add_flags; bpf_program__attach_tracing_multi; bpf_program__clear_flags; diff --git a/tools/testing/selftests/bpf/prog_tests/stream.c b/tools/testing/selftests/bpf/prog_tests/stream.c index 74bd15c4bfba..ae400dcfbf24 100644 --- a/tools/testing/selftests/bpf/prog_tests/stream.c +++ b/tools/testing/selftests/bpf/prog_tests/stream.c @@ -1,11 +1,17 @@ // SPDX-License-Identifier: GPL-2.0 /* Copyright (c) 2025 Meta Platforms, Inc. and affiliates. */ #include +#include +#include +#include #include +#include #include "stream.skel.h" #include "stream_fail.skel.h" +#define NMI_TIMEOUT_NS (5ULL * 1000 * 1000 * 1000) + void test_stream_failure(void) { RUN_TESTS(stream_fail); @@ -62,6 +68,486 @@ void test_stream_syscall(void) stream__destroy(skel); } +static bool stream_fd_trigger(struct bpf_program *prog) +{ + LIBBPF_OPTS(bpf_test_run_opts, opts); + int ret; + + ret = bpf_prog_test_run_opts(bpf_program__fd(prog), &opts); + return ASSERT_OK(ret, "test_run") && ASSERT_OK(opts.retval, "retval"); +} + +static void test_stream_fd_open(void) +{ + LIBBPF_OPTS(bpf_prog_stream_open_opts, opts); + struct stream *skel; + int fd, prog_fd; + + skel = stream__open_and_load(); + if (!ASSERT_OK_PTR(skel, "stream__open_and_load")) + return; + + prog_fd = bpf_program__fd(skel->progs.stream_syscall); + fd = bpf_prog_stream_open(0, BPF_STREAM_STDOUT, NULL); + ASSERT_EQ(fd, -EINVAL, "bad_prog_fd"); + + fd = bpf_prog_stream_open(prog_fd, 0, NULL); + ASSERT_EQ(fd, -ENOENT, "bad_stream_id"); + + opts.flags = BPF_F_RDONLY; + fd = bpf_prog_stream_open(prog_fd, BPF_STREAM_STDOUT, &opts); + ASSERT_EQ(fd, -EINVAL, "access_flag"); + + opts.flags = 1U << 31; + fd = bpf_prog_stream_open(prog_fd, BPF_STREAM_STDOUT, &opts); + ASSERT_EQ(fd, -EINVAL, "unknown_flag"); + + fd = bpf_prog_stream_open(prog_fd, BPF_STREAM_STDERR, NULL); + if (ASSERT_OK_FD(fd, "stderr")) + close(fd); + + stream__destroy(skel); +} + +static void test_stream_fd_nonblock(void) +{ + LIBBPF_OPTS(bpf_prog_stream_open_opts, opts, + .flags = BPF_F_STREAM_NONBLOCK, + ); + struct pollfd pfd = { .events = POLLIN | POLLHUP }; + struct stream *skel; + char buf[4] = {}; + int fd, flags, ret; + + skel = stream__open_and_load(); + if (!ASSERT_OK_PTR(skel, "stream__open_and_load")) + return; + + fd = bpf_prog_stream_open(bpf_program__fd(skel->progs.stream_syscall), + BPF_STREAM_STDOUT, &opts); + if (!ASSERT_OK_FD(fd, "stream_open")) + goto out_destroy; + pfd.fd = fd; + + flags = fcntl(fd, F_GETFD); + ASSERT_GE(flags, 0, "getfd"); + ASSERT_NEQ(flags & FD_CLOEXEC, 0, "cloexec"); + flags = fcntl(fd, F_GETFL); + ASSERT_GE(flags, 0, "getfl"); + ASSERT_EQ(flags & O_ACCMODE, O_RDONLY, "readonly"); + ASSERT_NEQ(flags & O_NONBLOCK, 0, "nonblock"); + + ret = write(fd, "x", 1); + ASSERT_EQ(ret, -1, "write"); + ASSERT_EQ(errno, EBADF, "write_errno"); + ret = lseek(fd, 0, SEEK_SET); + ASSERT_EQ(ret, -1, "lseek"); + ASSERT_EQ(errno, ESPIPE, "lseek_errno"); + + ret = poll(&pfd, 1, 0); + ASSERT_EQ(ret, 0, "poll_empty"); + ret = read(fd, buf, sizeof(buf)); + ASSERT_EQ(ret, -1, "read_empty"); + ASSERT_EQ(errno, EAGAIN, "read_empty_errno"); + + if (!stream_fd_trigger(skel->progs.stream_syscall)) + goto out_close; + ret = poll(&pfd, 1, 0); + ASSERT_EQ(ret, 1, "poll_data"); + ASSERT_NEQ(pfd.revents & POLLIN, 0, "pollin"); + ASSERT_EQ(pfd.revents & POLLHUP, 0, "no_pollhup"); + + ret = read(fd, buf, 2); + ASSERT_EQ(ret, 2, "read_first"); + ASSERT_OK(memcmp(buf, "fo", 2), "read_first_data"); + pfd.revents = 0; + ret = poll(&pfd, 1, 0); + ASSERT_EQ(ret, 1, "poll_partial"); + ASSERT_NEQ(pfd.revents & POLLIN, 0, "pollin_partial"); + + ret = read(fd, buf, sizeof(buf)); + ASSERT_EQ(ret, 1, "read_rest"); + ASSERT_EQ(buf[0], 'o', "read_rest_data"); + ret = read(fd, buf, sizeof(buf)); + ASSERT_EQ(ret, -1, "read_drained"); + ASSERT_EQ(errno, EAGAIN, "read_drained_errno"); + +out_close: + close(fd); +out_destroy: + stream__destroy(skel); +} + +static void test_stream_fd_empty(void) +{ + LIBBPF_OPTS(bpf_prog_stream_open_opts, opts, + .flags = BPF_F_STREAM_NONBLOCK, + ); + struct pollfd pfd = { .events = POLLIN | POLLHUP }; + struct stream *skel; + char buf[4]; + int fd, ret; + + skel = stream__open_and_load(); + if (!ASSERT_OK_PTR(skel, "stream__open_and_load")) + return; + + fd = bpf_prog_stream_open(bpf_program__fd(skel->progs.stream_empty), + BPF_STREAM_STDOUT, &opts); + if (!ASSERT_OK_FD(fd, "stream_open")) + goto out_destroy; + pfd.fd = fd; + + /* Empty output produces neither data nor readiness. */ + if (!stream_fd_trigger(skel->progs.stream_empty)) + goto out_close; + ret = poll(&pfd, 1, 0); + ASSERT_EQ(ret, 0, "poll_empty_write"); + ret = read(fd, buf, sizeof(buf)); + ASSERT_EQ(ret, -1, "read_empty_write"); + ASSERT_EQ(errno, EAGAIN, "read_empty_write_errno"); + +out_close: + close(fd); +out_destroy: + stream__destroy(skel); +} + +struct stream_fd_read_ctx { + int fd; + ssize_t ret; + char buf[4]; +}; + +static void *stream_fd_read_thread(void *arg) +{ + struct stream_fd_read_ctx *ctx = arg; + + ctx->ret = read(ctx->fd, ctx->buf, sizeof(ctx->buf)); + return NULL; +} + +static int stream_fd_timed_join(pthread_t thread) +{ + struct timespec timeout; + + clock_gettime(CLOCK_REALTIME, &timeout); + timeout.tv_sec += 5; + return pthread_timedjoin_np(thread, NULL, &timeout); +} + +static void test_stream_fd_blocking(void) +{ + struct stream_fd_read_ctx ctx = {}; + struct stream *skel; + pthread_t thread; + bool thread_live = false; + int fd, flags, ret; + + skel = stream__open_and_load(); + if (!ASSERT_OK_PTR(skel, "stream__open_and_load")) + return; + + fd = bpf_prog_stream_open(bpf_program__fd(skel->progs.stream_syscall), + BPF_STREAM_STDOUT, NULL); + if (!ASSERT_OK_FD(fd, "stream_open")) + goto out_destroy; + ctx.fd = fd; + + flags = fcntl(fd, F_GETFL); + ASSERT_GE(flags, 0, "getfl"); + ASSERT_EQ(flags & O_NONBLOCK, 0, "blocking"); + + ret = pthread_create(&thread, NULL, stream_fd_read_thread, &ctx); + if (!ASSERT_OK(ret, "pthread_create")) + goto out_close; + thread_live = true; + + usleep(50000); + ret = pthread_tryjoin_np(thread, NULL); + if (!ASSERT_EQ(ret, EBUSY, "read_blocks")) { + thread_live = ret != 0; + goto out_thread; + } + if (!stream_fd_trigger(skel->progs.stream_syscall)) + goto out_thread; + + ret = stream_fd_timed_join(thread); + if (!ASSERT_OK(ret, "pthread_join")) + goto out_thread; + thread_live = false; + ASSERT_EQ(ctx.ret, 3, "read_len"); + ASSERT_OK(memcmp(ctx.buf, "foo", 3), "read_data"); + + memset(&ctx, 0, sizeof(ctx)); + ctx.fd = fd; + ret = pthread_create(&thread, NULL, stream_fd_read_thread, &ctx); + if (!ASSERT_OK(ret, "pthread_create_eof")) + goto out_close; + thread_live = true; + + usleep(50000); + ret = pthread_tryjoin_np(thread, NULL); + if (!ASSERT_EQ(ret, EBUSY, "read_eof_blocks")) { + thread_live = ret != 0; + goto out_thread; + } + + stream__destroy(skel); + skel = NULL; + ret = stream_fd_timed_join(thread); + if (!ASSERT_OK(ret, "pthread_join_eof")) + goto out_thread; + thread_live = false; + ASSERT_EQ(ctx.ret, 0, "read_eof"); + +out_thread: + if (thread_live) { + stream__destroy(skel); + skel = NULL; + ret = stream_fd_timed_join(thread); + if (ret) { + pthread_cancel(thread); + pthread_join(thread, NULL); + } + } +out_close: + close(fd); +out_destroy: + stream__destroy(skel); +} + +static void test_stream_fd_hup(void) +{ + LIBBPF_OPTS(bpf_prog_stream_open_opts, opts, + .flags = BPF_F_STREAM_NONBLOCK, + ); + struct epoll_event event = { + .events = EPOLLIN | EPOLLET, + }; + struct pollfd pfd = { .events = POLLIN | POLLHUP }; + struct stream *skel; + char buf[4] = {}; + int epfd = -1, fd, ret; + + skel = stream__open_and_load(); + if (!ASSERT_OK_PTR(skel, "stream__open_and_load")) + return; + + fd = bpf_prog_stream_open(bpf_program__fd(skel->progs.stream_syscall), + BPF_STREAM_STDOUT, &opts); + if (!ASSERT_OK_FD(fd, "stream_open")) + goto out_destroy; + pfd.fd = fd; + epfd = epoll_create1(EPOLL_CLOEXEC); + if (!ASSERT_OK_FD(epfd, "epoll_create")) + goto out_close; + event.data.fd = fd; + ret = epoll_ctl(epfd, EPOLL_CTL_ADD, fd, &event); + if (!ASSERT_OK(ret, "epoll_ctl")) + goto out_close; + + if (!stream_fd_trigger(skel->progs.stream_syscall)) + goto out_close; + ret = epoll_wait(epfd, &event, 1, 5000); + if (!ASSERT_EQ(ret, 1, "epoll_wait_data")) + goto out_close; + ASSERT_NEQ(event.events & EPOLLIN, 0, "epollin"); + ASSERT_EQ(event.events & EPOLLHUP, 0, "no_epollhup"); + + stream__destroy(skel); + skel = NULL; + event.events = 0; + ret = epoll_wait(epfd, &event, 1, 5000); + if (!ASSERT_EQ(ret, 1, "epoll_wait_hup")) + goto out_close; + ASSERT_NEQ(event.events & EPOLLIN, 0, "epollin_with_hup"); + ASSERT_NEQ(event.events & EPOLLHUP, 0, "epollhup"); + + ret = read(fd, buf, sizeof(buf)); + ASSERT_EQ(ret, 3, "read_buffered"); + ASSERT_OK(memcmp(buf, "foo", 3), "read_buffered_data"); + pfd.revents = 0; + ret = poll(&pfd, 1, 0); + ASSERT_EQ(ret, 1, "poll_drained_hup"); + ASSERT_EQ(pfd.revents & POLLIN, 0, "no_pollin_after_drain"); + ASSERT_NEQ(pfd.revents & POLLHUP, 0, "pollhup_after_drain"); + ret = read(fd, buf, sizeof(buf)); + ASSERT_EQ(ret, 0, "read_eof"); + +out_close: + if (epfd >= 0) + close(epfd); + close(fd); +out_destroy: + stream__destroy(skel); +} + +static void test_stream_fd_edge(void) +{ + LIBBPF_OPTS(bpf_prog_stream_open_opts, opts, + .flags = BPF_F_STREAM_NONBLOCK, + ); + struct epoll_event event = { + .events = EPOLLIN | EPOLLET, + }; + struct stream *skel; + char buf[8] = {}; + int epfd = -1, fd, ret; + + skel = stream__open_and_load(); + if (!ASSERT_OK_PTR(skel, "stream__open_and_load")) + return; + + fd = bpf_prog_stream_open(bpf_program__fd(skel->progs.stream_syscall), + BPF_STREAM_STDOUT, &opts); + if (!ASSERT_OK_FD(fd, "stream_open")) + goto out_destroy; + epfd = epoll_create1(EPOLL_CLOEXEC); + if (!ASSERT_OK_FD(epfd, "epoll_create")) + goto out_close; + event.data.fd = fd; + ret = epoll_ctl(epfd, EPOLL_CTL_ADD, fd, &event); + if (!ASSERT_OK(ret, "epoll_ctl")) + goto out_close; + + /* Output into an empty stream raises an edge. */ + if (!stream_fd_trigger(skel->progs.stream_syscall)) + goto out_close; + ret = epoll_wait(epfd, &event, 1, 5000); + if (!ASSERT_EQ(ret, 1, "epoll_wait_first")) + goto out_close; + ASSERT_NEQ(event.events & EPOLLIN, 0, "epollin_first"); + + /* + * Leave data behind, add more, then drain everything. Whether output + * into a non-empty stream raises an edge of its own is not part of the + * contract, so discard any pending event before re-arming the stream. + */ + ret = read(fd, buf, 2); + ASSERT_EQ(ret, 2, "read_partial"); + if (!stream_fd_trigger(skel->progs.stream_syscall)) + goto out_close; + ret = read(fd, buf, sizeof(buf)); + ASSERT_EQ(ret, 4, "read_rest"); + ASSERT_OK(memcmp(buf, "ofoo", 4), "read_rest_data"); + ret = read(fd, buf, sizeof(buf)); + ASSERT_EQ(ret, -1, "read_drained"); + ASSERT_EQ(errno, EAGAIN, "read_drained_errno"); + epoll_wait(epfd, &event, 1, 0); + + /* Output into the drained stream raises an edge again. */ + if (!stream_fd_trigger(skel->progs.stream_syscall)) + goto out_close; + ret = epoll_wait(epfd, &event, 1, 5000); + if (!ASSERT_EQ(ret, 1, "epoll_wait_after_drain")) + goto out_close; + ASSERT_NEQ(event.events & EPOLLIN, 0, "epollin_after_drain"); + +out_close: + if (epfd >= 0) + close(epfd); + close(fd); +out_destroy: + stream__destroy(skel); +} + +static void test_stream_fd_nmi_epoll(void) +{ + LIBBPF_OPTS(bpf_prog_stream_open_opts, opts, + .flags = BPF_F_STREAM_NONBLOCK, + ); + struct perf_event_attr attr = { + .size = sizeof(attr), + .type = PERF_TYPE_HARDWARE, + .config = PERF_COUNT_HW_CPU_CYCLES, + .freq = 1, + .sample_freq = 10, + }; + struct epoll_event event = { + .events = EPOLLIN, + }; + struct bpf_link *link = NULL; + struct stream *skel; + __u64 deadline; + char buf[4] = {}; + int epfd = -1, fd = -1, pmu_fd = -1; + int ret; + + skel = stream__open_and_load(); + if (!ASSERT_OK_PTR(skel, "stream__open_and_load")) + return; + fd = bpf_prog_stream_open(bpf_program__fd(skel->progs.stream_nmi), + BPF_STREAM_STDOUT, &opts); + if (!ASSERT_OK_FD(fd, "stream_open")) + goto out; + + epfd = epoll_create1(EPOLL_CLOEXEC); + if (!ASSERT_OK_FD(epfd, "epoll_create")) + goto out; + event.data.fd = fd; + ret = epoll_ctl(epfd, EPOLL_CTL_ADD, fd, &event); + if (!ASSERT_OK(ret, "epoll_ctl")) + goto out; + + pmu_fd = syscall(__NR_perf_event_open, &attr, 0, -1, -1, + PERF_FLAG_FD_CLOEXEC); + if (pmu_fd < 0 && (errno == ENOENT || errno == EOPNOTSUPP)) { + printf("%s:SKIP:no PERF_COUNT_HW_CPU_CYCLES\n", __func__); + test__skip(); + goto out; + } + if (!ASSERT_GE(pmu_fd, 0, "perf_event_open")) + goto out; + + link = bpf_program__attach_perf_event(skel->progs.stream_nmi, pmu_fd); + if (!ASSERT_OK_PTR(link, "attach_perf_event")) { + link = NULL; + goto out; + } + pmu_fd = -1; + + deadline = get_time_ns() + NMI_TIMEOUT_NS; + do { + ret = epoll_wait(epfd, &event, 1, 0); + } while (ret == 0 && get_time_ns() < deadline); + if (!ASSERT_EQ(ret, 1, "epoll_wait")) + goto out; + ASSERT_NEQ(event.events & EPOLLIN, 0, "epollin"); + + ret = read(fd, buf, sizeof(buf)); + ASSERT_EQ(ret, 3, "read_len"); + ASSERT_OK(memcmp(buf, "nmi", 3), "read_data"); + +out: + bpf_link__destroy(link); + if (pmu_fd >= 0) + close(pmu_fd); + if (epfd >= 0) + close(epfd); + if (fd >= 0) + close(fd); + stream__destroy(skel); +} + +void test_stream_fd(void) +{ + if (test__start_subtest("open")) + test_stream_fd_open(); + if (test__start_subtest("nonblock")) + test_stream_fd_nonblock(); + if (test__start_subtest("empty")) + test_stream_fd_empty(); + if (test__start_subtest("blocking")) + test_stream_fd_blocking(); + if (test__start_subtest("hup")) + test_stream_fd_hup(); + if (test__start_subtest("edge")) + test_stream_fd_edge(); + if (test__start_subtest("nmi_epoll")) + test_stream_fd_nmi_epoll(); +} + void test_stream_oversize(void) { LIBBPF_OPTS(bpf_test_run_opts, opts); diff --git a/tools/testing/selftests/bpf/progs/stream.c b/tools/testing/selftests/bpf/progs/stream.c index 12fc29e45487..1ae015c4e6b4 100644 --- a/tools/testing/selftests/bpf/progs/stream.c +++ b/tools/testing/selftests/bpf/progs/stream.c @@ -44,6 +44,7 @@ struct { _X64 _X64 _X64 _X64 _X64 _X64 _X64 _X64) int size; +int nmi_stream_prints; u64 fault_addr; void *arena_ptr; @@ -124,6 +125,25 @@ int stream_syscall(void *ctx) return 0; } +SEC("syscall") +__success __retval(0) +int stream_empty(void *ctx) +{ + return bpf_stream_printk(BPF_STDOUT, ""); +} + +SEC("perf_event") +int stream_nmi(void *ctx) +{ + if (nmi_stream_prints) + return 0; + /* Retry on a later sample if the write failed, e.g. with -EBUSY. */ + if (bpf_stream_printk(BPF_STDOUT, "nmi")) + return 0; + nmi_stream_prints = 1; + return 0; +} + SEC("syscall") __success __retval(0) int stream_oversize(void *ctx)