[PATCH bpf-next v1 0/6] File descriptor interface for BPF streams

[PATCH bpf-next v1 0/6] File descriptor interface for BPF streams

[Kumar Kartikeya Dwivedi]

BPF program stdout and stderr streams can currently be consumed only
through BPF_PROG_STREAM_READ_BY_FD. This requires repeated bpf() calls and
provides no way to block for output or integrate with poll-based event
loops.

Add BPF_PROG_STREAM_OPEN to return a read-only, close-on-exec descriptor
for a program stream. Reads block by default, with an option for
non-blocking mode. poll and epoll report readable data and report hangup
when the program is released, while allowing buffered data to be drained
before EOF. Stream descriptors retain the stream storage without retaining
the program itself.

Expose the command through libbpf and switch bpftool tracelog to blocking
stream reads. bpftool falls back to BPF_PROG_STREAM_READ_BY_FD on older
kernels, where an unknown bpf() command returns EINVAL. The legacy command
remains supported.

See commit logs for details.

Kumar Kartikeya Dwivedi (6):
  bpf: Add file descriptor interface for program streams
  bpf: Defer stream file notifications from NMI context
  bpf: Separate stream readiness from capacity accounting
  libbpf: Add bpf_prog_stream_open()
  bpftool: Read program streams through file descriptors
  selftests/bpf: Test program stream file descriptors

 include/linux/bpf.h                           |  13 +-
 include/uapi/linux/bpf.h                      |  36 ++
 kernel/bpf/core.c                             |  13 +-
 kernel/bpf/stream.c                           | 180 ++++++++-
 kernel/bpf/syscall.c                          |  27 ++
 tools/bpf/bpftool/prog.c                      |  29 +-
 tools/include/uapi/linux/bpf.h                |  36 ++
 tools/lib/bpf/bpf.c                           |  19 +
 tools/lib/bpf/bpf.h                           |  24 ++
 tools/lib/bpf/libbpf.map                      |   1 +
 .../testing/selftests/bpf/prog_tests/stream.c | 376 ++++++++++++++++++
 tools/testing/selftests/bpf/progs/stream.c    |  10 +
 12 files changed, 733 insertions(+), 31 deletions(-)


base-commit: fbc547151dbcd9b5eee54d6b29e924c760039638
-- 
2.53.0

[PATCH bpf-next v1 1/6] bpf: Add file descriptor interface for program streams

[Kumar Kartikeya Dwivedi]

The existing BPF_PROG_STREAM_READ_BY_FD command only supports polling a
program stream through repeated bpf() calls. It cannot block for new data
or integrate with poll-based event loops.

Add BPF_PROG_STREAM_OPEN to return a read-only, close-on-exec file
descriptor for a selected program stream. Reads block by default and
BPF_F_STREAM_NONBLOCK provides non-blocking behavior. poll reports readable
data and reports hangup once the program is released.

A stream descriptor deliberately does not retain the program. Move each
stream into a separately refcounted allocation so program teardown can mark
it dead and wake descriptor users while outstanding descriptors drain
buffered data safely.

Signed-off-by: Kumar Kartikeya Dwivedi <memxor@gmail.com>
---
 include/linux/bpf.h            |   8 +-
 include/uapi/linux/bpf.h       |  36 +++++++++
 kernel/bpf/core.c              |  13 +--
 kernel/bpf/stream.c            | 139 ++++++++++++++++++++++++++++++---
 kernel/bpf/syscall.c           |  27 +++++++
 tools/include/uapi/linux/bpf.h |  36 +++++++++
 6 files changed, 241 insertions(+), 18 deletions(-)

diff --git a/include/linux/bpf.h b/include/linux/bpf.h
index 3a7eb2185c35..1745686331be 100644
--- a/include/linux/bpf.h
+++ b/include/linux/bpf.h
@@ -1711,12 +1711,15 @@ enum {
 };
 
 struct bpf_stream {
+	refcount_t refcnt;
 	atomic_t capacity;
 	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;
+	bool dead;
 };
 
 struct bpf_stream_stage {
@@ -1855,7 +1858,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;
 };
@@ -4112,9 +4115,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 732b35cc08d1..da2370036bf0 100644
--- a/include/uapi/linux/bpf.h
+++ b/include/uapi/linux/bpf.h
@@ -936,6 +936,30 @@ 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 is released. 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. *flags* may contain **BPF_F_RDONLY** and
+ *		**BPF_F_STREAM_NONBLOCK**. **BPF_F_WRONLY** is not supported.
+ *
+ *	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 +1017,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 +1549,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 +1980,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 5db77d7915df..afa030b01ed3 100644
--- a/kernel/bpf/core.c
+++ b/kernel/bpf/core.c
@@ -142,7 +142,10 @@ struct bpf_prog *bpf_prog_alloc_no_stats(unsigned int size, gfp_t gfp_extra_flag
 	mutex_init(&fp->aux->st_ops_assoc_mutex);
 
 #ifdef CONFIG_BPF_SYSCALL
-	bpf_prog_stream_init(fp);
+	if (bpf_prog_stream_init(fp, gfp_extra_flags)) {
+		__bpf_prog_free(fp);
+		return NULL;
+	}
 #endif
 
 	return fp;
@@ -160,9 +163,7 @@ struct bpf_prog *bpf_prog_alloc(unsigned int size, gfp_t gfp_extra_flags)
 
 	prog->stats = alloc_percpu_gfp(struct bpf_prog_stats, gfp_flags);
 	if (!prog->stats) {
-		free_percpu(prog->active);
-		kfree(prog->aux);
-		vfree(prog);
+		__bpf_prog_free(prog);
 		return NULL;
 	}
 
@@ -288,6 +289,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);
@@ -3067,7 +3071,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..d3dbb1aca792 100644
--- a/kernel/bpf/stream.c
+++ b/kernel/bpf/stream.c
@@ -2,11 +2,14 @@
 /* Copyright (c) 2025 Meta Platforms, Inc. and affiliates. */
 
 #include <linux/bpf.h>
+#include <linux/anon_inodes.h>
 #include <linux/filter.h>
 #include <linux/bpf_mem_alloc.h>
 #include <linux/gfp.h>
 #include <linux/memory.h>
 #include <linux/mutex.h>
+#include <linux/poll.h>
+#include <linux/refcount.h>
 
 static void bpf_stream_elem_init(struct bpf_stream_elem *elem, int len)
 {
@@ -83,6 +86,8 @@ static int bpf_stream_push_str(struct bpf_stream *stream, const char *str, int l
 	ret = __bpf_stream_push_str(&stream->log, str, len);
 	if (ret)
 		bpf_stream_release_capacity(stream, len);
+	else if (len)
+		wake_up_interruptible_poll(&stream->waitq, EPOLLIN | EPOLLRDNORM);
 
 	return ret;
 }
@@ -91,7 +96,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)
@@ -215,6 +220,102 @@ 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 atomic_read(&stream->capacity) > 0;
+}
+
+static void bpf_stream_put(struct bpf_stream *stream)
+{
+	if (refcount_dec_and_test(&stream->refcnt)) {
+		struct llist_node *list;
+
+		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;
+	int ret;
+
+	if (len > INT_MAX)
+		return -EINVAL;
+	if (!len)
+		return 0;
+
+	for (;;) {
+		ret = bpf_stream_read(stream, buf, len);
+		if (ret)
+			return ret;
+		if (READ_ONCE(stream->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,
+	.llseek = noop_llseek,
+};
+
+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 +383,43 @@ __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;
+
+		stream = kzalloc_obj(*stream,
+				     bpf_memcg_flags(GFP_KERNEL | gfp_extra_flags));
+		if (!stream) {
+			bpf_prog_stream_free(prog);
+			return -ENOMEM;
+		}
+
+		refcount_set(&stream->refcnt, 1);
+		atomic_set(&stream->capacity, 0);
+		init_llist_head(&stream->log);
+		mutex_init(&stream->lock);
+		init_waitqueue_head(&stream->waitq);
+		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;
+		WRITE_ONCE(stream->dead, true);
+		wake_up_interruptible_poll(&stream->waitq, EPOLLHUP);
+		bpf_stream_put(stream);
+		prog->aux->stream[i] = NULL;
 	}
 }
 
@@ -366,6 +482,7 @@ int bpf_stream_stage_commit(struct bpf_stream_stage *ss, struct bpf_prog *prog,
 		list = tail;
 	}
 	llist_add_batch(head, tail, &stream->log);
+	wake_up_interruptible_poll(&stream->waitq, EPOLLIN | EPOLLRDNORM);
 	return 0;
 }
 
diff --git a/kernel/bpf/syscall.c b/kernel/bpf/syscall.c
index 6874ba1424af..b5405f7270d1 100644
--- a/kernel/bpf/syscall.c
+++ b/kernel/bpf/syscall.c
@@ -6300,6 +6300,30 @@ 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_RDONLY | BPF_F_WRONLY | BPF_F_STREAM_NONBLOCK))
+		return -EINVAL;
+	if (flags & BPF_F_WRONLY)
+		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)
 {
@@ -6472,6 +6496,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/include/uapi/linux/bpf.h b/tools/include/uapi/linux/bpf.h
index 732b35cc08d1..da2370036bf0 100644
--- a/tools/include/uapi/linux/bpf.h
+++ b/tools/include/uapi/linux/bpf.h
@@ -936,6 +936,30 @@ 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 is released. 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. *flags* may contain **BPF_F_RDONLY** and
+ *		**BPF_F_STREAM_NONBLOCK**. **BPF_F_WRONLY** is not supported.
+ *
+ *	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 +1017,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 +1549,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 +1980,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
-- 
2.53.0

Re: [PATCH bpf-next v1 1/6] bpf: Add file descriptor interface for program streams

[sashiko-bot]

Thank you for your contribution! Sashiko AI review found 1 potential issue(s) to consider:
- [Medium] Race condition in bpf_stream_file_read() leading to premature EOF and data loss.
--

commit 15d530fc84e28348bbaf5fc7ae04276f6622bc6f
Author: Kumar Kartikeya Dwivedi <memxor@gmail.com>
Date:   Sun Aug 30 09:35:06 2026 +0000

bpf: Add file descriptor interface for program streams

This commit adds a BPF_PROG_STREAM_OPEN command to return a read-only file
descriptor for a selected program stream. It supports blocking and
non-blocking reads, as well as polling, and handles program teardown by
marking the stream dead and waking up descriptor users.

> diff --git a/kernel/bpf/stream.c b/kernel/bpf/stream.c
> index 2b80a0599865e..d3dbb1aca792f 100644
> --- a/kernel/bpf/stream.c
> +++ b/kernel/bpf/stream.c
[ ... ]
> @@ -215,6 +220,102 @@ int bpf_prog_stream_read(struct bpf_prog *prog, enum bpf_stream_id stream_id, vo
>  	return bpf_stream_read(stream, buf, len);
>  }
>  
[ ... ]
> +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;
> +	int ret;
> +
> +	if (len > INT_MAX)
> +		return -EINVAL;
> +	if (!len)
> +		return 0;
> +
> +	for (;;) {
> +		ret = bpf_stream_read(stream, buf, len);
> +		if (ret)
> +			return ret;
> +		if (READ_ONCE(stream->dead))
> +			return 0;

[Severity: Medium]
Does this code have a race condition that could lead to premature EOF and
missing the final stream events?

If a reader thread gets preempted right after bpf_stream_read() returns 0,
but before checking the stream->dead flag, it leaves a small race window.
Concurrently, a BPF program could emit new data and then be immediately
destroyed, which marks the stream as dead.

When the reader thread resumes, it will see READ_ONCE(stream->dead) as true
and return 0 (EOF) immediately. Would this cause the reader to completely
miss the newly published data that was emitted just before the teardown?

> +		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;
> +	}
> +}

-- 
Sashiko AI review · https://sashiko.dev/#/patchset/20260830093514.4105972-1-memxor@gmail.com?part=1

Re: [PATCH bpf-next v1 1/6] bpf: Add file descriptor interface for program streams

[bot+bpf-ci]

[-- Attachment #1: Type: text/plain, Size: 7193 bytes --]

> diff --git a/include/linux/bpf.h b/include/linux/bpf.h
> index 3a7eb2185c354..1745686331be0 100644
> --- a/include/linux/bpf.h
> +++ b/include/linux/bpf.h

[ ... ]

> diff --git a/kernel/bpf/stream.c b/kernel/bpf/stream.c
> index 2b80a0599865e..d3dbb1aca792f 100644
> --- a/kernel/bpf/stream.c
> +++ b/kernel/bpf/stream.c
> @@ -2,11 +2,14 @@
>  /* Copyright (c) 2025 Meta Platforms, Inc. and affiliates. */
>
>  #include <linux/bpf.h>
> +#include <linux/anon_inodes.h>
>  #include <linux/filter.h>
>  #include <linux/bpf_mem_alloc.h>
>  #include <linux/gfp.h>
>  #include <linux/memory.h>
>  #include <linux/mutex.h>
> +#include <linux/poll.h>
> +#include <linux/refcount.h>
>
>  static void bpf_stream_elem_init(struct bpf_stream_elem *elem, int len)
>  {
> @@ -83,6 +86,8 @@ static int bpf_stream_push_str(struct bpf_stream *stream, const char *str, int l
>  	ret = __bpf_stream_push_str(&stream->log, str, len);
>  	if (ret)
>  		bpf_stream_release_capacity(stream, len);
> +	else if (len)
> +		wake_up_interruptible_poll(&stream->waitq, EPOLLIN | EPOLLRDNORM);
>
>  	return ret;
>  }

Can this wake_up_interruptible_poll call run from NMI context?
bpf_stream_push_str is reached from bpf_stream_vprintk, which is registered
in common_btf_ids for BPF_PROG_TYPE_UNSPEC (kernel/bpf/helpers.c:4968,
5007), making it callable from NMI-capable perf_event and kprobe programs.

wake_up_interruptible_poll unconditionally does
spin_lock_irqsave(&wq_head->lock) (kernel/sched/wait.c:118-130), which can
deadlock if a reader blocked in wait_event_interruptible holds stream->waitq
lock when an NMI fires on the same CPU and tries to re-acquire it.

On PREEMPT_RT the breakage is unconditional: spinlock_t is a sleeping
rt_mutex, and rt_spin_lock calls might_sleep, which BUGs when called with
IRQs disabled and preempt_count != 0.

The bpf_stream_vprintk kfunc also carries KF_SPINLOCK_SAFE, meaning it is
allowed inside bpf_spin_lock-ed regions after local_irq_save +
preempt_disable + arch_spin_lock, where acquiring a spinlock_t is similarly
illegal.

bpf_ringbuf handles this by calling irq_work_queue instead of waking the
wait queue directly:

kernel/bpf/ringbuf.c:bpf_ringbuf_commit() {
    ...
    irq_work_queue(&rb->work);
}

kernel/bpf/ringbuf.c:bpf_ringbuf_notify() {
    wake_up_all(&rb->waitq);
}

Should the wakeup be deferred to an irq_work callback?

> @@ -215,6 +220,102 @@ 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 atomic_read(&stream->capacity) > 0;
> +}
> +
> +static void bpf_stream_put(struct bpf_stream *stream)
> +{
> +	if (refcount_dec_and_test(&stream->refcnt)) {
> +		struct llist_node *list;
> +
> +		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;
> +	int ret;
> +
> +	if (len > INT_MAX)
> +		return -EINVAL;
> +	if (!len)
> +		return 0;
> +
> +	for (;;) {
> +		ret = bpf_stream_read(stream, buf, len);
> +		if (ret)
> +			return ret;
> +		if (READ_ONCE(stream->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;
> +}

Does bpf_stream_has_data reliably indicate readable data?

The function tests stream->capacity, but capacity is a reservation counter
charged before data is published. Both producers call
bpf_stream_consume_capacity first, then allocate the element and llist_add
it:

kernel/bpf/stream.c:bpf_stream_push_str() {
    ret = bpf_stream_consume_capacity(stream, len);
    if (ret)
        return ret;

    ret = __bpf_stream_push_str(&stream->log, str, len);
    ...
}

kernel/bpf/stream.c:bpf_stream_stage_commit() {
    ret = bpf_stream_consume_capacity(stream, ss->len);
    if (ret)
        return ret;
    ...
    llist_add_batch(head, tail, &stream->log);
}

So capacity > 0 is observable while stream->log is still empty and nothing
is readable. If a producer is preempted or descheduled between the
reservation and the publication, the window widens.

For poll, this means reporting EPOLLIN with no data present. A level-
triggered epoll consumer wakes, calls read, gets -EAGAIN, and re-polls
immediately, spinning at 100% CPU until the data is published or the
reservation is released.

For the blocking read loop above, wait_event_interruptible returns
immediately because its condition is already true. The for loop re-runs
bpf_stream_read (returns 0), checks dead (false), checks O_NONBLOCK (clear),
and calls wait_event_interruptible again with the same result. The loop
spins without reaching schedule, so it is not interruptible by a signal and
only preemptible.

Per locking.md section 6 (Multi-Variable Races), the readiness flag and the
data must be published together. Could the code use a dedicated atomic
counter incremented after llist_add / llist_add_batch and decremented by the
consumer, so the readiness predicate reflects actual readable bytes?

> @@ -366,6 +482,7 @@ int bpf_stream_stage_commit(struct bpf_stream_stage *ss, struct bpf_prog *prog,
>  		list = tail;
>  	}
>  	llist_add_batch(head, tail, &stream->log);
> +	wake_up_interruptible_poll(&stream->waitq, EPOLLIN | EPOLLRDNORM);
>  	return 0;
>  }

This second wake_up_interruptible_poll call has the same NMI-context
concern. bpf_stream_stage_commit is reached from
bpf_prog_report_rqspinlock_violation (kernel/bpf/rqspinlock.c:673), which
runs from bpf_res_spin_lock_irqsave after preempt_disable + local_irq_save
with an rqspinlock held, and from __bpf_prog_report_arena_violation
(kernel/bpf/arena.c:1137) invoked from the JIT exception handlers.

Both paths are hard-atomic contexts where acquiring a spinlock_t is illegal.

Should this wakeup also be deferred to irq_work?

[ ... ]


---
AI reviewed your patch. Please fix the bug or email reply why it's not a bug.
See: https://github.com/kernel-patches/vmtest/blob/master/ci/claude/README.md

CI run summary: https://github.com/kernel-patches/bpf/actions/runs/33304989971

[PATCH bpf-next v1 2/6] bpf: Defer stream file notifications from NMI context

[Kumar Kartikeya Dwivedi]

bpf_stream_vprintk() and staged stream writers can run in NMI context.
The stream file interface currently wakes its wait queue directly after
publishing data. Wait queue wakeups take a spin lock and can invoke epoll
callbacks that take further locks, so calling them from NMI context can
deadlock.

Give each stream an irq_work item and queue it after publishing data. The
irq_work callback reports readable data to poll waiters outside NMI
context, while naturally coalescing concurrent notifications. This matches
poll and epoll readiness semantics: notifications do not count records,
but prompt waiters to re-evaluate persistent readable state. Publications
made before a coalesced queue attempt are ordered before the pending
callback, and the work can be queued again once that callback begins.
EPOLLET consumers drain until EAGAIN, so one wakeup may safely represent a
batch.

Synchronize pending work before releasing the final stream reference so
the callback cannot outlive the stream.

Signed-off-by: Kumar Kartikeya Dwivedi <memxor@gmail.com>
---
 include/linux/bpf.h |  2 ++
 kernel/bpf/stream.c | 18 ++++++++++++++++--
 2 files changed, 18 insertions(+), 2 deletions(-)

diff --git a/include/linux/bpf.h b/include/linux/bpf.h
index 1745686331be..0af3c79f5d03 100644
--- a/include/linux/bpf.h
+++ b/include/linux/bpf.h
@@ -17,6 +17,7 @@
 #include <linux/numa.h>
 #include <linux/mm_types.h>
 #include <linux/wait.h>
+#include <linux/irq_work_types.h>
 #include <linux/refcount.h>
 #include <linux/mutex.h>
 #include <linux/module.h>
@@ -1719,6 +1720,7 @@ struct bpf_stream {
 	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 dead;
 };
 
diff --git a/kernel/bpf/stream.c b/kernel/bpf/stream.c
index d3dbb1aca792..99a89533eaef 100644
--- a/kernel/bpf/stream.c
+++ b/kernel/bpf/stream.c
@@ -6,6 +6,7 @@
 #include <linux/filter.h>
 #include <linux/bpf_mem_alloc.h>
 #include <linux/gfp.h>
+#include <linux/irq_work.h>
 #include <linux/memory.h>
 #include <linux/mutex.h>
 #include <linux/poll.h>
@@ -76,6 +77,17 @@ 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);
+
+	/*
+	 * Stream writers can run in NMI context, while wait queue callbacks may
+	 * acquire locks. Defer those callbacks to irq_work context.
+	 */
+	wake_up_interruptible_poll(&stream->waitq, EPOLLIN | EPOLLRDNORM);
+}
+
 static int bpf_stream_push_str(struct bpf_stream *stream, const char *str, int len)
 {
 	int ret = bpf_stream_consume_capacity(stream, len);
@@ -87,7 +99,7 @@ static int bpf_stream_push_str(struct bpf_stream *stream, const char *str, int l
 	if (ret)
 		bpf_stream_release_capacity(stream, len);
 	else if (len)
-		wake_up_interruptible_poll(&stream->waitq, EPOLLIN | EPOLLRDNORM);
+		irq_work_queue(&stream->notify_work);
 
 	return ret;
 }
@@ -230,6 +242,7 @@ static void bpf_stream_put(struct bpf_stream *stream)
 	if (refcount_dec_and_test(&stream->refcnt)) {
 		struct llist_node *list;
 
+		irq_work_sync(&stream->notify_work);
 		list = llist_del_all(&stream->log);
 		bpf_stream_free_list(list);
 		bpf_stream_free_list(stream->backlog_head);
@@ -402,6 +415,7 @@ int bpf_prog_stream_init(struct bpf_prog *prog, gfp_t gfp_extra_flags)
 		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;
@@ -482,7 +496,7 @@ int bpf_stream_stage_commit(struct bpf_stream_stage *ss, struct bpf_prog *prog,
 		list = tail;
 	}
 	llist_add_batch(head, tail, &stream->log);
-	wake_up_interruptible_poll(&stream->waitq, EPOLLIN | EPOLLRDNORM);
+	irq_work_queue(&stream->notify_work);
 	return 0;
 }
 
-- 
2.53.0

[PATCH bpf-next v1 3/6] bpf: Separate stream readiness from capacity accounting

[Kumar Kartikeya Dwivedi]

Stream capacity is charged before allocation and before an element is
published to the stream log. Using that reservation counter as the read and
poll readiness condition can therefore report readable data while no
element exists. A blocking reader can then repeatedly retry instead of
sleeping, while a non-blocking reader can observe POLLIN followed by
EAGAIN.

Add a separate readable byte counter. Publish bytes with release ordering
after adding elements to the lockless log, and use acquire loads before
consuming them or reporting readiness. Limit each read to its readable
snapshot and subtract only bytes actually copied; this keeps the aggregate
count correct even when concurrent publishers update it out of publication
order.

Keep the existing capacity counter solely for enforcing the stream size
limit.

Signed-off-by: Kumar Kartikeya Dwivedi <memxor@gmail.com>
---
 include/linux/bpf.h |  3 ++-
 kernel/bpf/stream.c | 35 +++++++++++++++++++++++++++--------
 2 files changed, 29 insertions(+), 9 deletions(-)

diff --git a/include/linux/bpf.h b/include/linux/bpf.h
index 0af3c79f5d03..2ed1e82391b9 100644
--- a/include/linux/bpf.h
+++ b/include/linux/bpf.h
@@ -1713,7 +1713,8 @@ enum {
 
 struct bpf_stream {
 	refcount_t refcnt;
-	atomic_t capacity;
+	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} */
diff --git a/kernel/bpf/stream.c b/kernel/bpf/stream.c
index 99a89533eaef..b1e6767d8754 100644
--- a/kernel/bpf/stream.c
+++ b/kernel/bpf/stream.c
@@ -88,6 +88,21 @@ static void bpf_stream_notify(struct irq_work *work)
 	wake_up_interruptible_poll(&stream->waitq, EPOLLIN | EPOLLRDNORM);
 }
 
+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)
+{
+	if (!len)
+		return;
+
+	/* Pairs with atomic_read_acquire() in bpf_stream_readable_bytes(). */
+	(void)atomic_add_return_release(len, &stream->readable);
+	irq_work_queue(&stream->notify_work);
+}
+
 static int bpf_stream_push_str(struct bpf_stream *stream, const char *str, int len)
 {
 	int ret = bpf_stream_consume_capacity(stream, len);
@@ -98,8 +113,8 @@ static int bpf_stream_push_str(struct bpf_stream *stream, const char *str, int l
 	ret = __bpf_stream_push_str(&stream->log, str, len);
 	if (ret)
 		bpf_stream_release_capacity(stream, len);
-	else if (len)
-		irq_work_queue(&stream->notify_work);
+	else
+		bpf_stream_publish(stream, len);
 
 	return ret;
 }
@@ -176,14 +191,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;
 
@@ -205,7 +222,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;
 		}
 
@@ -216,8 +233,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)
@@ -234,7 +252,7 @@ int bpf_prog_stream_read(struct bpf_prog *prog, enum bpf_stream_id stream_id, vo
 
 static bool bpf_stream_has_data(struct bpf_stream *stream)
 {
-	return atomic_read(&stream->capacity) > 0;
+	return bpf_stream_readable_bytes(stream) > 0;
 }
 
 static void bpf_stream_put(struct bpf_stream *stream)
@@ -412,6 +430,7 @@ int bpf_prog_stream_init(struct bpf_prog *prog, gfp_t gfp_extra_flags)
 
 		refcount_set(&stream->refcnt, 1);
 		atomic_set(&stream->capacity, 0);
+		atomic_set(&stream->readable, 0);
 		init_llist_head(&stream->log);
 		mutex_init(&stream->lock);
 		init_waitqueue_head(&stream->waitq);
@@ -496,7 +515,7 @@ int bpf_stream_stage_commit(struct bpf_stream_stage *ss, struct bpf_prog *prog,
 		list = tail;
 	}
 	llist_add_batch(head, tail, &stream->log);
-	irq_work_queue(&stream->notify_work);
+	bpf_stream_publish(stream, ss->len);
 	return 0;
 }
 
-- 
2.53.0

Re: [PATCH bpf-next v1 3/6] bpf: Separate stream readiness from capacity accounting

[sashiko-bot]

Thank you for your contribution! Sashiko AI review found 1 potential issue(s) to consider:
- [High] Unbounded accumulation of zero-length stream elements leading to memory leak and potential kernel OOM.
--

commit 0e18588b75a5e2877e4350149476217ae3dc124c
Author: Kumar Kartikeya Dwivedi <memxor@gmail.com>

bpf: Separate stream readiness from capacity accounting

This commit separates stream capacity accounting from readiness by
introducing a new readable byte counter. The new counter tracks how many
bytes are actually available for userspace to read, while the existing
capacity counter remains solely for enforcing the stream limit.

> diff --git a/kernel/bpf/stream.c b/kernel/bpf/stream.c
> index 99a89533eaef6..b1e6767d8754d 100644
> --- a/kernel/bpf/stream.c
> +++ b/kernel/bpf/stream.c
[ ... ]
> @@ -88,6 +88,21 @@ static void bpf_stream_notify(struct irq_work *work)
>  	wake_up_interruptible_poll(&stream->waitq, EPOLLIN | EPOLLRDNORM);
>  }
>  
> +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)
> +{
> +	if (!len)
> +		return;

[Severity: High]
If a 0-length string is pushed, does this early return bypass incrementing
the readable counter? It appears the 0-length element has already been
allocated and added to the log in __bpf_stream_push_str() prior to this.

> +
> +	/* Pairs with atomic_read_acquire() in bpf_stream_readable_bytes(). */
> +	(void)atomic_add_return_release(len, &stream->readable);
> +	irq_work_queue(&stream->notify_work);
> +}
[ ... ]
> @@ -176,14 +191,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) {

[Severity: High]
Can this lead to an unbounded accumulation of 0-length elements in the log?

Because 0-length pushes skip incrementing the readable counter in
bpf_stream_publish(), bpf_stream_readable_bytes() can return 0 here. This
makes read_len and rem_len both 0, which entirely skips the while loop.

Since the loop is bypassed, readers cannot drain and free these 0-length
elements. Furthermore, because bpf_stream_consume_capacity() adds 0, the
allocations never hit the stream limit. Could a BPF program repeatedly
pushing empty strings cause a memory leak and potential kernel OOM?

> -		int pos = len - rem_len;
> +		int pos = read_len - rem_len;
>  		int chunk, n;
>  		bool cont;
[ ... ]

-- 
Sashiko AI review · https://sashiko.dev/#/patchset/20260830093514.4105972-1-memxor@gmail.com?part=3

Re: [PATCH bpf-next v1 3/6] bpf: Separate stream readiness from capacity accounting

[bot+bpf-ci]

[-- Attachment #1: Type: text/plain, Size: 4313 bytes --]

> commit 7ff803d1d406e5f7340f74f25183895b96ce7628
> Author: Kumar Kartikeya Dwivedi <memxor@gmail.com>
> Date:   Sun Aug 30 11:35:08 2026 +0200
>
>     bpf: Separate stream readiness from capacity accounting
>
>     Stream capacity is charged before allocation and before an element is
>     published to the stream log. Using that reservation counter as the read and
>     poll readiness condition can therefore report readable data while no
>     element exists. A blocking reader can then repeatedly retry instead of
>     sleeping, while a non-blocking reader can observe POLLIN followed by
>     EAGAIN.
>
>     Add a separate readable byte counter. Publish bytes with release ordering
>     after adding elements to the lockless log, and use acquire loads before
>     consuming them or reporting readiness. Limit each read to its readable
>     snapshot and subtract only bytes actually copied; this keeps the aggregate
>     count correct even when concurrent publishers update it out of publication
>     order.
>
>     Keep the existing capacity counter solely for enforcing the stream size
>     limit.
>
>     Signed-off-by: Kumar Kartikeya Dwivedi <memxor@gmail.com>

This looks like a bug fix for stream readiness tracking. Should this
include:

  Fixes: 05755d4f1f98 ("bpf: Add file descriptor interface for program streams")

> diff --git a/include/linux/bpf.h b/include/linux/bpf.h
> index 0af3c79f5d03a..2ed1e82391b93 100644
> --- a/include/linux/bpf.h
> +++ b/include/linux/bpf.h
> @@ -1713,7 +1713,8 @@ enum {
>
>  struct bpf_stream {
>  	refcount_t refcnt;
> -	atomic_t capacity;
> +	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} */

[ ... ]

> diff --git a/kernel/bpf/stream.c b/kernel/bpf/stream.c
> index 99a89533eaef6..b1e6767d8754d 100644
> --- a/kernel/bpf/stream.c
> +++ b/kernel/bpf/stream.c

[ ... ]

> @@ -234,7 +252,7 @@ int bpf_prog_stream_read(struct bpf_prog *prog, enum bpf_stream_id stream_id, vo
>
>  static bool bpf_stream_has_data(struct bpf_stream *stream)
>  {
> -	return atomic_read(&stream->capacity) > 0;
> +	return bpf_stream_readable_bytes(stream) > 0;
>  }

A subsystem pattern flags this as potentially concerning: the readiness
check here happens outside the exclusion point that guards the consuming
action. bpf_stream_has_data() does an unlocked atomic_read_acquire() and
is called from bpf_stream_poll() and from the wait_event_interruptible()
condition in bpf_stream_file_read(), while bpf_stream_read() only takes
stream->lock afterwards and re-samples readable inside the lock.

Can this interleaving occur with two readers on the same stream?

  CPU0 (poller)                          CPU1 (reader)
  -------------                          -------------
  poll(): bpf_stream_has_data()
    readable == 50 -> EPOLLIN
                                         mutex_lock(&stream->lock)
                                         read_len = min(len, 50)
                                         ... copies 50 bytes ...
                                         atomic_sub(50) -> readable == 0
                                         mutex_unlock()
  read(): bpf_stream_read()
    read_len = min(len, 0) == 0 -> 0
  bpf_stream_file_read(): O_NONBLOCK -> -EAGAIN

The commit message states that "a non-blocking reader can observe POLLIN
followed by EAGAIN" as the defect being eliminated. Does the new counter
fully prevent this, or does it narrow the window without moving the check
inside the exclusion point?

Note this is about the structural placement of the check relative to the
lock, which was introduced two commits earlier in this series in commit
05755d4f1f98. poll(2) readiness is advisory, and the single-reader case is
fully addressed by this patch. The question is whether the changelog claim
about POLLIN/EAGAIN being fixed needs qualification for the concurrent
reader case.


---
AI reviewed your patch. Please fix the bug or email reply why it's not a bug.
See: https://github.com/kernel-patches/vmtest/blob/master/ci/claude/README.md

CI run summary: https://github.com/kernel-patches/bpf/actions/runs/33304989971

[PATCH bpf-next v1 4/6] libbpf: Add bpf_prog_stream_open()

[Kumar Kartikeya Dwivedi]

Expose the BPF_PROG_STREAM_OPEN command through a low-level libbpf wrapper.
The options structure carries stream open flags, including non-blocking
mode, while leaving room for future extensions.

Signed-off-by: Kumar Kartikeya Dwivedi <memxor@gmail.com>
---
 tools/lib/bpf/bpf.c      | 19 +++++++++++++++++++
 tools/lib/bpf/bpf.h      | 24 ++++++++++++++++++++++++
 tools/lib/bpf/libbpf.map |  1 +
 3 files changed, 44 insertions(+)

diff --git a/tools/lib/bpf/bpf.c b/tools/lib/bpf/bpf.c
index 96819c082c77..5186f6245879 100644
--- a/tools/lib/bpf/bpf.c
+++ b/tools/lib/bpf/bpf.c
@@ -1464,6 +1464,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 7534a593edae..860c5644ec1e 100644
--- a/tools/lib/bpf/bpf.h
+++ b/tools/lib/bpf/bpf.h
@@ -758,10 +758,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 08ab2ea881fb..a89e10811a73 100644
--- a/tools/lib/bpf/libbpf.map
+++ b/tools/lib/bpf/libbpf.map
@@ -458,6 +458,7 @@ LIBBPF_1.7.0 {
 
 LIBBPF_1.8.0 {
 	global:
+		bpf_prog_stream_open;
 		bpf_program__attach_tracing_multi;
 		bpf_program__clone;
 		btf__find_by_name_kind_own;
-- 
2.53.0

[PATCH bpf-next v1 5/6] bpftool: Read program streams through file descriptors

[Kumar Kartikeya Dwivedi]

Use bpf_prog_stream_open() in its default blocking mode when following a
program stream. This lets bpftool wait for later output instead of exiting
as soon as it drains the currently buffered bytes. Flush each chunk as it
arrives so redirected output is not held in stdio buffers while the next
read blocks.

Fall back to BPF_PROG_STREAM_READ_BY_FD when the open command returns
EINVAL. The bpf() syscall uses EINVAL for unknown commands, so this keeps a
new bpftool compatible with kernels that provide program streams but
predate their file descriptor interface.

Signed-off-by: Kumar Kartikeya Dwivedi <memxor@gmail.com>
---
 tools/bpf/bpftool/prog.c | 29 +++++++++++++++++++++--------
 1 file changed, 21 insertions(+), 8 deletions(-)

diff --git a/tools/bpf/bpftool/prog.c b/tools/bpf/bpftool/prog.c
index a9f730d407a9..6b43cb43b27e 100644
--- a/tools/bpf/bpftool/prog.c
+++ b/tools/bpf/bpftool/prog.c
@@ -1125,14 +1125,27 @@ prog_tracelog_stream(int prog_fd, enum prog_tracelog_mode mode)
 	FILE *file = mode == TRACE_STDOUT ? stdout : stderr;
 	int stream_id = mode == TRACE_STDOUT ? 1 : 2;
 	char buf[512];
-	int 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);
+	int fd, ret;
+
+	fd = bpf_prog_stream_open(prog_fd, stream_id, NULL);
+	if (fd == -EINVAL) {
+		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);
+	} else {
+		if (fd < 0)
+			return -1;
+		do {
+			ret = read(fd, buf, sizeof(buf));
+			if (ret > 0) {
+				fwrite(buf, sizeof(buf[0]), ret, file);
+				fflush(file);
+			}
+		} while (ret > 0);
+		close(fd);
+	}
 
 	fflush(file);
 	return ret ? -1 : 0;
-- 
2.53.0

Re: [PATCH bpf-next v1 5/6] bpftool: Read program streams through file descriptors

[bot+bpf-ci]

[-- Attachment #1: Type: text/plain, Size: 3164 bytes --]

> diff --git a/tools/bpf/bpftool/prog.c b/tools/bpf/bpftool/prog.c
> index a9f730d407a92..6b43cb43b27ed 100644
> --- a/tools/bpf/bpftool/prog.c
> +++ b/tools/bpf/bpftool/prog.c
> @@ -1125,14 +1125,27 @@ prog_tracelog_stream(int prog_fd, enum prog_tracelog_mode mode)
>  	FILE *file = mode == TRACE_STDOUT ? stdout : stderr;
>  	int stream_id = mode == TRACE_STDOUT ? 1 : 2;
>  	char buf[512];
> -	int 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);
> +	int fd, ret;
> +
> +	fd = bpf_prog_stream_open(prog_fd, stream_id, NULL);
> +	if (fd == -EINVAL) {

[ ... ]

> +	} else {
> +		if (fd < 0)
> +			return -1;
> +		do {
> +			ret = read(fd, buf, sizeof(buf));
> +			if (ret > 0) {
> +				fwrite(buf, sizeof(buf[0]), ret, file);
> +				fflush(file);
> +			}
> +		} while (ret > 0);
> +		close(fd);
> +	}
>
>  	fflush(file);
>  	return ret ? -1 : 0;

Can bpftool prog tracelog stdout|stderr PROG terminate normally under
the new blocking descriptor path?

The loop exits when read(fd, ...) returns <= 0. Looking at
kernel/bpf/stream.c in bpf_stream_file_read(), read() returns 0 only
when READ_ONCE(stream->dead) is true:

    ret = bpf_stream_read(stream, buf, len);
    if (ret)
            return ret;
    if (READ_ONCE(stream->dead))
            return 0;
    if (file->f_flags & O_NONBLOCK)
            return -EAGAIN;
    ret = wait_event_interruptible(stream->waitq, ...);

stream->dead is set only by bpf_prog_stream_free(), which is reached
from __bpf_prog_free() when the last program reference is dropped.

prog_tracelog_stream() is called from do_tracelog_any(), which obtains
prog_fd = prog_parse_fd(&argc, &argv) and passes it without ever
closing it. This means bpftool holds the program reference for the
entire call, so the program cannot be freed while
prog_tracelog_stream() is running.

Consequently read() never returns 0 and the do/while loop has no
reachable normal exit. The close(fd) and return are only reachable via
a read() error (like EINTR), in which case the command exits with
failure (-1) instead of 0.

Before this commit, the same command drained the buffered bytes with
BPF_PROG_STREAM_READ_BY_FD and exited 0. After it, on any kernel that
implements BPF_PROG_STREAM_OPEN, the command blocks indefinitely and
must be killed by a signal. This breaks scripts and CI jobs that expect
the command to exit after draining current output. The behavior change
is also undocumented in tools/bpf/bpftool/Documentation/bpftool-prog.rst.

Should the follow behavior be opt-in (e.g. a --follow flag that passes
BPF_F_STREAM_NONBLOCK by default), so the existing invocation keeps
working? If the blocking behavior is intended, should
prog_tracelog_stream() install signal handlers that exit cleanly, like
do_tracelog() does in tools/bpf/bpftool/tracelog.c?


---
AI reviewed your patch. Please fix the bug or email reply why it's not a bug.
See: https://github.com/kernel-patches/vmtest/blob/master/ci/claude/README.md

CI run summary: https://github.com/kernel-patches/bpf/actions/runs/33304989971

[PATCH bpf-next v1 6/6] selftests/bpf: Test program stream file descriptors

[Kumar Kartikeya Dwivedi]

Exercise validation and normal file semantics of BPF_PROG_STREAM_OPEN.
Cover read-only and close-on-exec state, write rejection, non-blocking
reads, partial reads, poll readiness, and a blocking reader woken by new
stream data.

Register the descriptor with EPOLLET, consume the initial data event
without draining buffered output, then release the program. Require a
second edge carrying POLLIN and POLLHUP, drain the buffered bytes, and
verify that poll continues to report HUP before read returns EOF. This
ensures program teardown notifies edge-triggered loops even when
readability never transitions.

Also block a reader on an empty live stream, release the program, and
verify that it wakes and returns EOF. This directly covers the teardown
wait condition and its no-lost-wakeup ordering.

Finally, attach a stream-writing program to a hardware perf event and
busy-poll its epoll registration until output arrives. This both drives the
per-thread event into NMI context and directly verifies that the deferred
irq_work notification reaches epoll, without a separate BSS completion
handshake.

Signed-off-by: Kumar Kartikeya Dwivedi <memxor@gmail.com>
---
 .../testing/selftests/bpf/prog_tests/stream.c | 376 ++++++++++++++++++
 tools/testing/selftests/bpf/progs/stream.c    |  10 +
 2 files changed, 386 insertions(+)

diff --git a/tools/testing/selftests/bpf/prog_tests/stream.c b/tools/testing/selftests/bpf/prog_tests/stream.c
index 74bd15c4bfba..655aab35c03e 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 <test_progs.h>
+#include <linux/perf_event.h>
+#include <poll.h>
+#include <sys/epoll.h>
 #include <sys/mman.h>
+#include <sys/syscall.h>
 
 #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,376 @@ 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_WRONLY;
+	fd = bpf_prog_stream_open(prog_fd, BPF_STREAM_STDOUT, &opts);
+	ASSERT_EQ(fd, -EINVAL, "writable");
+
+	opts.flags = 1U << 31;
+	fd = bpf_prog_stream_open(prog_fd, BPF_STREAM_STDOUT, &opts);
+	ASSERT_EQ(fd, -EINVAL, "unknown_flag");
+
+	opts.flags = BPF_F_RDONLY;
+	fd = bpf_prog_stream_open(prog_fd, BPF_STREAM_STDERR, &opts);
+	if (ASSERT_OK_FD(fd, "readonly"))
+		close(fd);
+
+	stream__destroy(skel);
+}
+
+static void test_stream_fd_nonblock(void)
+{
+	LIBBPF_OPTS(bpf_prog_stream_open_opts, opts,
+		.flags = BPF_F_RDONLY | 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 = 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);
+}
+
+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_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("blocking"))
+		test_stream_fd_blocking();
+	if (test__start_subtest("hup"))
+		test_stream_fd_hup();
+	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..00843127bca8 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,15 @@ int stream_syscall(void *ctx)
 	return 0;
 }
 
+SEC("perf_event")
+int stream_nmi(void *ctx)
+{
+	if (nmi_stream_prints)
+		return 0;
+	nmi_stream_prints = 1;
+	return bpf_stream_printk(BPF_STDOUT, "nmi");
+}
+
 SEC("syscall")
 __success __retval(0)
 int stream_oversize(void *ctx)
-- 
2.53.0