[RFC PATCH v1 06/25] printk-rb: add blocking reader support

[John Ogness]

Add a blocking read function for readers. An irq_work function is
used to signal the wait queue so that write notification can
be triggered from any context.

Signed-off-by: John Ogness <john.ogness@xxxxxxxxxxxxx>
---
 include/linux/printk_ringbuffer.h | 20 ++++++++++++++++
 lib/printk_ringbuffer.c           | 49 +++++++++++++++++++++++++++++++++++++++
 2 files changed, 69 insertions(+)

diff --git a/include/linux/printk_ringbuffer.h b/include/linux/printk_ringbuffer.h
index 5fdaf632c111..106f20ef8b4d 100644
--- a/include/linux/printk_ringbuffer.h
+++ b/include/linux/printk_ringbuffer.h
@@ -2,8 +2,10 @@
 #ifndef _LINUX_PRINTK_RINGBUFFER_H
 #define _LINUX_PRINTK_RINGBUFFER_H
 
+#include <linux/irq_work.h>
 #include <linux/atomic.h>
 #include <linux/percpu.h>
+#include <linux/wait.h>
 
 struct prb_cpulock {
 	atomic_t owner;
@@ -22,6 +24,10 @@ struct printk_ringbuffer {
 
 	struct prb_cpulock *cpulock;
 	atomic_t ctx;
+
+	struct wait_queue_head *wq;
+	atomic_long_t wq_counter;
+	struct irq_work *wq_work;
 };
 
 struct prb_entry {
@@ -59,6 +65,15 @@ struct prb_iterator {
 #define DECLARE_STATIC_PRINTKRB(name, szbits, cpulockptr)		\
 static char _##name##_buffer[1 << (szbits)]				\
 	__aligned(__alignof__(long));					\
+static DECLARE_WAIT_QUEUE_HEAD(_##name##_wait);				\
+static void _##name##_wake_work_func(struct irq_work *irq_work)		\
+{									\
+	wake_up_interruptible_all(&_##name##_wait);			\
+}									\
+static struct irq_work _##name##_wake_work = {				\
+	.func = _##name##_wake_work_func,				\
+	.flags = IRQ_WORK_LAZY,						\
+};									\
 static struct printk_ringbuffer name = {				\
 	.buffer = &_##name##_buffer[0],					\
 	.size_bits = szbits,						\
@@ -68,6 +83,9 @@ static struct printk_ringbuffer name = {				\
 	.reserve = ATOMIC_LONG_INIT(-111 * sizeof(long)),		\
 	.cpulock = cpulockptr,						\
 	.ctx = ATOMIC_INIT(0),						\
+	.wq = &_##name##_wait,						\
+	.wq_counter = ATOMIC_LONG_INIT(0),				\
+	.wq_work = &_##name##_wake_work,				\
 }
 
 /* writer interface */
@@ -80,6 +98,8 @@ void prb_iter_init(struct prb_iterator *iter, struct printk_ringbuffer *rb,
 		   u64 *seq);
 void prb_iter_copy(struct prb_iterator *dest, struct prb_iterator *src);
 int prb_iter_next(struct prb_iterator *iter, char *buf, int size, u64 *seq);
+int prb_iter_wait_next(struct prb_iterator *iter, char *buf, int size,
+		       u64 *seq);
 int prb_iter_data(struct prb_iterator *iter, char *buf, int size, u64 *seq);
 
 /* utility functions */
diff --git a/lib/printk_ringbuffer.c b/lib/printk_ringbuffer.c
index 1d1e886a0966..c2ddf4cb9f92 100644
--- a/lib/printk_ringbuffer.c
+++ b/lib/printk_ringbuffer.c
@@ -1,4 +1,5 @@
 // SPDX-License-Identifier: GPL-2.0
+#include <linux/sched.h>
 #include <linux/smp.h>
 #include <linux/string.h>
 #include <linux/errno.h>
@@ -154,6 +155,7 @@ static bool push_tail(struct printk_ringbuffer *rb, unsigned long tail)
 void prb_commit(struct prb_handle *h)
 {
 	struct printk_ringbuffer *rb = h->rb;
+	bool changed = false;
 	struct prb_entry *e;
 	unsigned long head;
 	unsigned long res;
@@ -175,6 +177,7 @@ void prb_commit(struct prb_handle *h)
 			}
 			e->seq = ++rb->seq;
 			head += e->size;
+			changed = true;
 		}
 		atomic_long_set_release(&rb->head, res);
 		atomic_dec(&rb->ctx);
@@ -185,6 +188,12 @@ void prb_commit(struct prb_handle *h)
 	}
 
 	prb_unlock(rb->cpulock, h->cpu);
+
+	if (changed) {
+		atomic_long_inc(&rb->wq_counter);
+		if (wq_has_sleeper(rb->wq))
+			irq_work_queue(rb->wq_work);
+	}
 }
 
 /*
@@ -437,3 +446,43 @@ int prb_iter_next(struct prb_iterator *iter, char *buf, int size, u64 *seq)
 
 	return 1;
 }
+
+/*
+ * prb_iter_wait_next: Advance to the next record, blocking if none available.
+ * @iter: Iterator tracking the current position.
+ * @buf: A buffer to store the data of the next record. May be NULL.
+ * @size: The size of @buf. (Ignored if @buf is NULL.)
+ * @seq: The sequence number of the next record. May be NULL.
+ *
+ * If a next record is already available, this function works like
+ * prb_iter_next(). Otherwise block interruptible until a next record is
+ * available.
+ *
+ * When a next record is available, @iter is advanced and (if specified)
+ * the data and/or sequence number of that record are provided.
+ *
+ * This function might sleep.
+ *
+ * Returns 1 if @iter was advanced, -EINVAL if @iter is now invalid, or
+ * -ERESTARTSYS if interrupted by a signal.
+ */
+int prb_iter_wait_next(struct prb_iterator *iter, char *buf, int size, u64 *seq)
+{
+	unsigned long last_seen;
+	int ret;
+
+	for (;;) {
+		last_seen = atomic_long_read(&iter->rb->wq_counter);
+
+		ret = prb_iter_next(iter, buf, size, seq);
+		if (ret != 0)
+			break;
+
+		ret = wait_event_interruptible(*iter->rb->wq,
+			last_seen != atomic_long_read(&iter->rb->wq_counter));
+		if (ret < 0)
+			break;
+	}
+
+	return ret;
+}
-- 
2.11.0