lib ring buffer: use irq_work for wakeup by writer

Using irq_work (like perf does) allows using an interrupt handler
firing soon after the instrumentation execution to issue the wakeups.

This allows the RING_BUFFER_WAKEUP_BY_WRITER ring buffer configuration
to be entirely lock-free, which allows using it in NMI context for
general tracing purposes.

Signed-off-by: Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
Change-Id: I842ff15736f53d1283cf953804d803f70779652b

diff --git a/include/ringbuffer/config.h b/include/ringbuffer/config.h
index a17d22024ddc13f0374e6bb7c18ba6ba0b2eaa27..ccb683371cd9ec52176909f59e8a348dbfb01776 100644 (file)
--- a/include/ringbuffer/config.h
+++ b/include/ringbuffer/config.h
@@ -101,7 +101,9 @@ struct lib_ring_buffer_client_cb {
  *
  * RING_BUFFER_WAKEUP_BY_WRITER directly wakes up readers when a subbuffer is
  * ready to read. Lower latencies before the reader is woken up. Mainly suitable
- * for drivers.
+ * for drivers. Going through an "irq_work" allows triggering this type of wakeup
+ * even from NMI context: the wakeup will be slightly delayed until the next
+ * interrupts are handled.
  *
  * RING_BUFFER_WAKEUP_NONE does not perform any wakeup whatsoever. The client
  * has the responsibility to perform wakeups.
@@ -142,9 +144,8 @@ struct lib_ring_buffer_config {
        enum {
                RING_BUFFER_WAKEUP_BY_TIMER,    /* wake up performed by timer */
                RING_BUFFER_WAKEUP_BY_WRITER,   /*
-                                                * writer wakes up reader,
-                                                * not lock-free
-                                                * (takes spinlock).
+                                                * writer wakes up reader through
+                                                * irq_work.
                                                 */
        } wakeup;
        /*

diff --git a/include/ringbuffer/frontend_types.h b/include/ringbuffer/frontend_types.h
index 07be81aacf95c98de11c3763f1bc9e7e4bcdd1bd..de205337ce9c8ccdc53f7c7fcb93428f44a38588 100644 (file)
--- a/include/ringbuffer/frontend_types.h
+++ b/include/ringbuffer/frontend_types.h
@@ -13,6 +13,7 @@
 #define _LIB_RING_BUFFER_FRONTEND_TYPES_H
 
 #include <linux/kref.h>
+#include <linux/irq_work.h>
 #include <ringbuffer/config.h>
 #include <ringbuffer/backend_types.h>
 #include <lttng/prio_heap.h>   /* For per-CPU read-side iterator */
@@ -66,6 +67,7 @@ struct channel {
        struct notifier_block tick_nohz_notifier; /* CPU nohz notifier */
        wait_queue_head_t read_wait;            /* reader wait queue */
        wait_queue_head_t hp_wait;              /* CPU hotplug wait queue */
+       struct irq_work wakeup_pending;         /* Pending wakeup irq work */
        int finalized;                          /* Has channel been finalized */
        struct channel_iter iter;               /* Channel read-side iterator */
        struct kref ref;                        /* Reference count */
@@ -146,6 +148,7 @@ struct lib_ring_buffer {
        union v_atomic records_overrun; /* Number of overwritten records */
        wait_queue_head_t read_wait;    /* reader buffer-level wait queue */
        wait_queue_head_t write_wait;   /* writer buffer-level wait queue (for metadata only) */
+       struct irq_work wakeup_pending;         /* Pending wakeup irq work */
        int finalized;                  /* buffer has been finalized */
        struct timer_list switch_timer; /* timer for periodical switch */
        struct timer_list read_timer;   /* timer for read poll */

diff --git a/src/lib/ringbuffer/ring_buffer_frontend.c b/src/lib/ringbuffer/ring_buffer_frontend.c
index cc5ac836420fb0f9a54a86de07dd36b417b2a2a4..c298296aa6b7e70c5406a397211ad37be4eef973 100644 (file)
--- a/src/lib/ringbuffer/ring_buffer_frontend.c
+++ b/src/lib/ringbuffer/ring_buffer_frontend.c
@@ -133,6 +133,8 @@ void lib_ring_buffer_free(struct lib_ring_buffer *buf)
 {
        struct channel *chan = buf->backend.chan;
 
+       irq_work_sync(&buf->wakeup_pending);
+
        lib_ring_buffer_print_errors(chan, buf, buf->backend.cpu);
        lttng_kvfree(buf->commit_hot);
        lttng_kvfree(buf->commit_cold);
@@ -206,6 +208,19 @@ void channel_reset(struct channel *chan)
 }
 EXPORT_SYMBOL_GPL(channel_reset);
 
+static void lib_ring_buffer_pending_wakeup_buf(struct irq_work *entry)
+{
+       struct lib_ring_buffer *buf = container_of(entry, struct lib_ring_buffer,
+                                                  wakeup_pending);
+       wake_up_interruptible(&buf->read_wait);
+}
+
+static void lib_ring_buffer_pending_wakeup_chan(struct irq_work *entry)
+{
+       struct channel *chan = container_of(entry, struct channel, wakeup_pending);
+       wake_up_interruptible(&chan->read_wait);
+}
+
 /*
  * Must be called under cpu hotplug protection.
  */
@@ -268,6 +283,7 @@ int lib_ring_buffer_create(struct lib_ring_buffer *buf,
 
        init_waitqueue_head(&buf->read_wait);
        init_waitqueue_head(&buf->write_wait);
+       init_irq_work(&buf->wakeup_pending, lib_ring_buffer_pending_wakeup_buf);
        raw_spin_lock_init(&buf->raw_tick_nohz_spinlock);
 
        /*
@@ -854,6 +870,7 @@ struct channel *channel_create(const struct lib_ring_buffer_config *config,
        kref_init(&chan->ref);
        init_waitqueue_head(&chan->read_wait);
        init_waitqueue_head(&chan->hp_wait);
+       init_irq_work(&chan->wakeup_pending, lib_ring_buffer_pending_wakeup_chan);
 
        if (config->alloc == RING_BUFFER_ALLOC_PER_CPU) {
 #if (LINUX_VERSION_CODE >= KERNEL_VERSION(4,10,0))
@@ -963,6 +980,8 @@ void *channel_destroy(struct channel *chan)
        const struct lib_ring_buffer_config *config = &chan->backend.config;
        void *priv;
 
+       irq_work_sync(&chan->wakeup_pending);
+
        channel_unregister_notifiers(chan);
 
        if (config->alloc == RING_BUFFER_ALLOC_PER_CPU) {
@@ -2356,13 +2375,14 @@ void lib_ring_buffer_check_deliver_slow(const struct lib_ring_buffer_config *con
                                                 commit_count, idx);
 
                /*
-                * RING_BUFFER_WAKEUP_BY_WRITER wakeup is not lock-free.
+                * RING_BUFFER_WAKEUP_BY_WRITER uses an irq_work to issue
+                * the wakeups.
                 */
                if (config->wakeup == RING_BUFFER_WAKEUP_BY_WRITER
                    && atomic_long_read(&buf->active_readers)
                    && lib_ring_buffer_poll_deliver(config, buf, chan)) {
-                       wake_up_interruptible(&buf->read_wait);
-                       wake_up_interruptible(&chan->read_wait);
+                       irq_work_queue(&buf->wakeup_pending);
+                       irq_work_queue(&chan->wakeup_pending);
                }
 
        }