Add CPU hotplug notifier for ltt debugfs ABI, add pollwait exclusive wrapper

[lttng-modules.git] / lib / ringbuffer / ring_buffer_frontend.c

diff --git a/lib/ringbuffer/ring_buffer_frontend.c b/lib/ringbuffer/ring_buffer_frontend.c
index d48d66b32779e90c7baa8dacf7475d57da5cf252..ecc72ab84f7e3143f930e77cac0bc6ad3de16a8e 100644 (file)
--- a/lib/ringbuffer/ring_buffer_frontend.c
+++ b/lib/ringbuffer/ring_buffer_frontend.c
@@ -410,6 +410,7 @@ int __cpuinit lib_ring_buffer_cpu_hp_callback(struct notifier_block *nb,
        case CPU_DOWN_FAILED_FROZEN:
        case CPU_ONLINE:
        case CPU_ONLINE_FROZEN:
+               wake_up_interruptible(&chan->hp_wait);
                lib_ring_buffer_start_switch_timer(buf);
                lib_ring_buffer_start_read_timer(buf);
                return NOTIFY_OK;
@@ -624,7 +625,9 @@ struct channel *channel_create(const struct lib_ring_buffer_config *config,
        chan->commit_count_mask = (~0UL >> chan->backend.num_subbuf_order);
        chan->switch_timer_interval = usecs_to_jiffies(switch_timer_interval);
        chan->read_timer_interval = usecs_to_jiffies(read_timer_interval);
+       kref_init(&chan->ref);
        init_waitqueue_head(&chan->read_wait);
+       init_waitqueue_head(&chan->hp_wait);
 
        if (config->alloc == RING_BUFFER_ALLOC_PER_CPU) {
 #if defined(CONFIG_NO_HZ) && defined(CONFIG_LIB_RING_BUFFER)
@@ -685,6 +688,13 @@ error:
 }
 EXPORT_SYMBOL_GPL(channel_create);
 
+static
+void channel_release(struct kref *kref)
+{
+       struct channel *chan = container_of(kref, struct channel, ref);
+       channel_free(chan);
+}
+
 /**
  * channel_destroy - Finalize, wait for q.s. and destroy channel.
  * @chan: channel to destroy
@@ -740,14 +750,11 @@ void *channel_destroy(struct channel *chan)
                ACCESS_ONCE(buf->finalized) = 1;
                wake_up_interruptible(&buf->read_wait);
        }
+       ACCESS_ONCE(chan->finalized) = 1;
+       wake_up_interruptible(&chan->hp_wait);
        wake_up_interruptible(&chan->read_wait);
-
-       while (atomic_long_read(&chan->read_ref) > 0)
-               msleep(100);
-       /* Finish waiting for refcount before free */
-       smp_mb();
+       kref_put(&chan->ref, channel_release);
        priv = chan->backend.priv;
-       channel_free(chan);
        return priv;
 }
 EXPORT_SYMBOL_GPL(channel_destroy);
@@ -769,7 +776,7 @@ int lib_ring_buffer_open_read(struct lib_ring_buffer *buf)
 
        if (!atomic_long_add_unless(&buf->active_readers, 1, 1))
                return -EBUSY;
-       atomic_long_inc(&chan->read_ref);
+       kref_get(&chan->ref);
        smp_mb__after_atomic_inc();
        return 0;
 }
@@ -781,8 +788,8 @@ void lib_ring_buffer_release_read(struct lib_ring_buffer *buf)
 
        CHAN_WARN_ON(chan, atomic_long_read(&buf->active_readers) != 1);
        smp_mb__before_atomic_dec();
-       atomic_long_dec(&chan->read_ref);
        atomic_long_dec(&buf->active_readers);
+       kref_put(&chan->ref, channel_release);
 }
 EXPORT_SYMBOL_GPL(lib_ring_buffer_release_read);
 
@@ -1482,7 +1489,9 @@ EXPORT_SYMBOL_GPL(lib_ring_buffer_switch_slow);
 /*
  * Returns :
  * 0 if ok
- * !0 if execution must be aborted.
+ * -ENOSPC if event size is too large for packet.
+ * -ENOBUFS if there is currently not enough space in buffer for the event.
+ * -EIO if data cannot be written into the buffer for any other reason.
  */
 static
 int lib_ring_buffer_try_reserve_slow(struct lib_ring_buffer *buf,
@@ -1501,18 +1510,19 @@ int lib_ring_buffer_try_reserve_slow(struct lib_ring_buffer *buf,
        offsets->pre_header_padding = 0;
 
        ctx->tsc = config->cb.ring_buffer_clock_read(chan);
+       if ((int64_t) ctx->tsc == -EIO)
+               return -EIO;
 
        if (last_tsc_overflow(config, buf, ctx->tsc))
-               ctx->rflags = RING_BUFFER_RFLAG_FULL_TSC;
+               ctx->rflags |= RING_BUFFER_RFLAG_FULL_TSC;
 
        if (unlikely(subbuf_offset(offsets->begin, ctx->chan) == 0)) {
                offsets->switch_new_start = 1;          /* For offsets->begin */
        } else {
                offsets->size = config->cb.record_header_size(config, chan,
                                                offsets->begin,
-                                               ctx->data_size,
                                                &offsets->pre_header_padding,
-                                               ctx->rflags, ctx);
+                                               ctx);
                offsets->size +=
                        lib_ring_buffer_align(offsets->begin + offsets->size,
                                              ctx->largest_align)
@@ -1553,7 +1563,7 @@ int lib_ring_buffer_try_reserve_slow(struct lib_ring_buffer *buf,
                                 * and we are full : record is lost.
                                 */
                                v_inc(config, &buf->records_lost_full);
-                               return -1;
+                               return -ENOBUFS;
                        } else {
                                /*
                                 * Next subbuffer not being written to, and we
@@ -1570,14 +1580,13 @@ int lib_ring_buffer_try_reserve_slow(struct lib_ring_buffer *buf,
                         * many nested writes over a reserve/commit pair.
                         */
                        v_inc(config, &buf->records_lost_wrap);
-                       return -1;
+                       return -EIO;
                }
                offsets->size =
                        config->cb.record_header_size(config, chan,
                                                offsets->begin,
-                                               ctx->data_size,
                                                &offsets->pre_header_padding,
-                                               ctx->rflags, ctx);
+                                               ctx);
                offsets->size +=
                        lib_ring_buffer_align(offsets->begin + offsets->size,
                                              ctx->largest_align)
@@ -1589,7 +1598,7 @@ int lib_ring_buffer_try_reserve_slow(struct lib_ring_buffer *buf,
                         * complete the sub-buffer switch.
                         */
                        v_inc(config, &buf->records_lost_big);
-                       return -1;
+                       return -ENOSPC;
                } else {
                        /*
                         * We just made a successful buffer switch and the
@@ -1618,7 +1627,8 @@ int lib_ring_buffer_try_reserve_slow(struct lib_ring_buffer *buf,
  * lib_ring_buffer_reserve_slow - Atomic slot reservation in a buffer.
  * @ctx: ring buffer context.
  *
- * Return : -ENOSPC if not enough space, else returns 0.
+ * Return : -NOBUFS if not enough space, -ENOSPC if event size too large,
+ * -EIO for other errors, else returns 0.
  * It will take care of sub-buffer switching.
  */
 int lib_ring_buffer_reserve_slow(struct lib_ring_buffer_ctx *ctx)
@@ -1627,6 +1637,7 @@ int lib_ring_buffer_reserve_slow(struct lib_ring_buffer_ctx *ctx)
        const struct lib_ring_buffer_config *config = chan->backend.config;
        struct lib_ring_buffer *buf;
        struct switch_offsets offsets;
+       int ret;
 
        if (config->alloc == RING_BUFFER_ALLOC_PER_CPU)
                buf = per_cpu_ptr(chan->backend.buf, ctx->cpu);
@@ -1637,9 +1648,10 @@ int lib_ring_buffer_reserve_slow(struct lib_ring_buffer_ctx *ctx)
        offsets.size = 0;
 
        do {
-               if (unlikely(lib_ring_buffer_try_reserve_slow(buf, chan, &offsets,
-                                                         ctx)))
-                       return -ENOSPC;
+               ret = lib_ring_buffer_try_reserve_slow(buf, chan, &offsets,
+                                                      ctx);
+               if (unlikely(ret))
+                       return ret;
        } while (unlikely(v_cmpxchg(config, &buf->offset, offsets.old,
                                    offsets.end)
                          != offsets.old));