1 files changed, 236 insertions, 37 deletions

diff --git a/kernel/bpf/stream.c b/kernel/bpf/stream.c
index be9ce98e9469..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 <linux/bpf.h>
+#include <linux/anon_inodes.h>
 #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>
+#include <linux/refcount.h>
 
 static void bpf_stream_elem_init(struct bpf_stream_elem *elem, int len)
 {
@@ -22,11 +26,11 @@ static struct bpf_stream_elem *bpf_stream_elem_alloc(int len)
 	size_t alloc_size;
 
 	/*
-	 * Length denotes the amount of data to be written as part of stream element,
-	 * thus includes '\0' byte. We're capped by how much bpf_bprintf_buffers can
-	 * accomodate, therefore deny allocations that won't fit into them.
+	 * Length is the payload pushed into the stream, excluding the
+	 * trailing NUL of the bprintf buffer. Reject anything that cannot
+	 * fit without copying that NUL into the stream element.
 	 */
-	if (len < 0 || len > max_len)
+	if (len < 0 || len >= max_len)
 		return NULL;
 
 	alloc_size = offsetof(struct bpf_stream_elem, str[len]);
@@ -68,25 +72,82 @@ static int bpf_stream_consume_capacity(struct bpf_stream *stream, int len)
 	return 0;
 }
 
-static void bpf_stream_release_capacity(struct bpf_stream *stream, struct bpf_stream_elem *elem)
+static void bpf_stream_release_capacity(struct bpf_stream *stream, int len)
 {
-	int len = elem->total_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 ?: __bpf_stream_push_str(&stream->log, str, len);
+	return ret;
 }
 
 static struct bpf_stream *bpf_stream_get(enum bpf_stream_id stream_id, struct bpf_prog_aux *aux)
 {
 	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)
@@ -154,14 +215,17 @@ 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;
 
 		node = bpf_stream_backlog_peek(stream);
@@ -175,37 +239,146 @@ static int bpf_stream_read(struct bpf_stream *stream, void __user *buf, int len)
 
 		cons_len = elem->consumed_len;
 		cont = bpf_stream_consume_elem(elem, &rem_len) == false;
-
-		ret = copy_to_user(buf + pos, elem->str + cons_len,
-				   elem->consumed_len - cons_len);
-		/* Restore in case of error. */
-		if (ret) {
-			ret = -EFAULT;
-			elem->consumed_len = cons_len;
+		chunk = elem->consumed_len - cons_len;
+
+		n = copy_to_user(buf + pos, elem->str + cons_len, chunk);
+		if (n) {
+			/* Keep any successfully copied bytes; -EFAULT only if none. */
+			elem->consumed_len -= n;
+			rem_len += n;
+			ret = (read_len == rem_len) ? -EFAULT : 0;
 			break;
 		}
 
 		if (cont)
 			continue;
 		bpf_stream_backlog_pop(stream);
-		bpf_stream_release_capacity(stream, elem);
+		bpf_stream_release_capacity(stream, elem->total_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, int len)
+int bpf_prog_stream_read(struct bpf_prog *prog, enum bpf_stream_id stream_id, void __user *buf, u32 len)
 {
 	struct bpf_stream *stream;
 
 	stream = bpf_stream_get(stream_id, prog->aux);
 	if (!stream)
 		return -ENOENT;
+	if (len > INT_MAX)
+		return -EINVAL;
 	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();
 
 /*
@@ -238,6 +411,11 @@ __bpf_kfunc int bpf_stream_vprintk(int stream_id, const char *fmt__str, const vo
 		return ret;
 
 	ret = bstr_printf(data.buf, MAX_BPRINTF_BUF, fmt__str, data.bin_args);
+	/* Truncation: reject before capacity charge (not -ENOMEM). */
+	if (ret >= MAX_BPRINTF_BUF) {
+		bpf_bprintf_cleanup(&data);
+		return -E2BIG;
+	}
 	/* Exclude NULL byte during push. */
 	ret = bpf_stream_push_str(stream, data.buf, ret);
 	bpf_bprintf_cleanup(&data);
@@ -268,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;
 	}
 }
 
@@ -311,17 +508,18 @@ int bpf_stream_stage_printk(struct bpf_stream_stage *ss, const char *fmt, ...)
 {
 	struct bpf_bprintf_buffers *buf;
 	va_list args;
-	int ret;
+	int len, ret;
 
 	if (bpf_try_get_buffers(&buf))
 		return -EBUSY;
 
 	va_start(args, fmt);
-	ret = vsnprintf(buf->buf, ARRAY_SIZE(buf->buf), fmt, args);
+	len = vscnprintf(buf->buf, ARRAY_SIZE(buf->buf), fmt, args);
 	va_end(args);
-	ss->len += ret;
-	/* Exclude NULL byte during push. */
-	ret = __bpf_stream_push_str(&ss->log, buf->buf, ret);
+	/* 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();
 	return ret;
 }
@@ -351,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;
 }