lttng lib: ring buffer: remove stale null-pointer

[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 7d4dd8380638adf09262fbe30ddd55b34a486c65..957d7f3d7ac6184f5401d44099275b6893403e45 100644 (file)
--- a/lib/ringbuffer/ring_buffer_frontend.c
+++ b/lib/ringbuffer/ring_buffer_frontend.c
@@ -164,7 +164,6 @@ int lib_ring_buffer_create(struct lib_ring_buffer *buf,
        const struct lib_ring_buffer_config *config = chanb->config;
        struct channel *chan = container_of(chanb, struct channel, backend);
        void *priv = chanb->priv;
-       unsigned int num_subbuf;
        size_t subbuf_header_size;
        u64 tsc;
        int ret;
@@ -203,8 +202,8 @@ int lib_ring_buffer_create(struct lib_ring_buffer *buf,
                goto free_commit;
        }
 
-       num_subbuf = chan->backend.num_subbuf;
        init_waitqueue_head(&buf->read_wait);
+       init_waitqueue_head(&buf->write_wait);
        raw_spin_lock_init(&buf->raw_tick_nohz_spinlock);
 
        /*
@@ -410,6 +409,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;
@@ -626,6 +626,7 @@ struct channel *channel_create(const struct lib_ring_buffer_config *config,
        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)
@@ -698,11 +699,11 @@ void channel_release(struct kref *kref)
  * @chan: channel to destroy
  *
  * Holds cpu hotplug.
- * Call "destroy" callback, finalize channels, wait for readers to release their
- * reference, then destroy ring buffer data. Note that when readers have
- * completed data consumption of finalized channels, get_subbuf() will return
- * -ENODATA. They should release their handle at that point.
- * Returns the private data pointer.
+ * Call "destroy" callback, finalize channels, and then decrement the
+ * channel reference count.  Note that when readers have completed data
+ * consumption of finalized channels, get_subbuf() will return -ENODATA.
+ * They should release their handle at that point.  Returns the private
+ * data pointer.
  */
 void *channel_destroy(struct channel *chan)
 {
@@ -748,9 +749,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);
-       kref_put(&chan->ref, channel_release);
        priv = chan->backend.priv;
+       kref_put(&chan->ref, channel_release);
        return priv;
 }
 EXPORT_SYMBOL_GPL(channel_destroy);
@@ -865,6 +868,8 @@ EXPORT_SYMBOL_GPL(lib_ring_buffer_snapshot);
 
 /**
  * lib_ring_buffer_put_snapshot - move consumed counter forward
+ *
+ * Should only be called from consumer context.
  * @buf: ring buffer
  * @consumed_new: new consumed count value
  */
@@ -886,6 +891,8 @@ void lib_ring_buffer_move_consumer(struct lib_ring_buffer *buf,
        while ((long) consumed - (long) consumed_new < 0)
                consumed = atomic_long_cmpxchg(&buf->consumed, consumed,
                                               consumed_new);
+       /* Wake-up the metadata producer */
+       wake_up_interruptible(&buf->write_wait);
 }
 EXPORT_SYMBOL_GPL(lib_ring_buffer_move_consumer);
 
@@ -1136,12 +1143,6 @@ void lib_ring_buffer_print_buffer_errors(struct lib_ring_buffer *buf,
        const struct lib_ring_buffer_config *config = chan->backend.config;
        unsigned long write_offset, cons_offset;
 
-       /*
-        * Can be called in the error path of allocation when
-        * trans_channel_data is not yet set.
-        */
-       if (!chan)
-               return;
        /*
         * No need to order commit_count, write_offset and cons_offset reads
         * because we execute at teardown when no more writer nor reader
@@ -1150,7 +1151,7 @@ void lib_ring_buffer_print_buffer_errors(struct lib_ring_buffer *buf,
        write_offset = v_read(config, &buf->offset);
        cons_offset = atomic_long_read(&buf->consumed);
        if (write_offset != cons_offset)
-               printk(KERN_WARNING
+               printk(KERN_DEBUG
                       "ring buffer %s, cpu %d: "
                       "non-consumed data\n"
                       "  [ %lu bytes written, %lu bytes read ]\n",
@@ -1510,16 +1511,15 @@ int lib_ring_buffer_try_reserve_slow(struct lib_ring_buffer *buf,
                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)
@@ -1582,9 +1582,8 @@ int lib_ring_buffer_try_reserve_slow(struct lib_ring_buffer *buf,
                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)
@@ -1697,3 +1696,20 @@ int lib_ring_buffer_reserve_slow(struct lib_ring_buffer_ctx *ctx)
        return 0;
 }
 EXPORT_SYMBOL_GPL(lib_ring_buffer_reserve_slow);
+
+int __init init_lib_ring_buffer_frontend(void)
+{
+       int cpu;
+
+       for_each_possible_cpu(cpu)
+               spin_lock_init(&per_cpu(ring_buffer_nohz_lock, cpu));
+       return 0;
+}
+
+module_init(init_lib_ring_buffer_frontend);
+
+void __exit exit_lib_ring_buffer_frontend(void)
+{
+}
+
+module_exit(exit_lib_ring_buffer_frontend);