Fix: Implement RING_BUFFER_GET_NEXT_SUBBUF_METADATA_CHECK

Get next metadata subbuffer, returning a flag indicating whether the
metadata is guaranteed to be in a consistent state at the end of this
sub-buffer (can be parsed).

This can be used by the consumer to know whether the metadata can be
parsed at the end of this sub-buffer, which is useful to distinguish
between errors and incomplete metadata in live tracing.

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

diff --git a/include/lttng/events.h b/include/lttng/events.h
index c9d664f1db86de4e36cb401ad560c8fbd6e3bbd6..f010ff7698ba707114497991146b56fe7c6cc9f6 100644 (file)
--- a/include/lttng/events.h
+++ b/include/lttng/events.h
@@ -455,6 +455,7 @@ struct lttng_metadata_stream {
        struct list_head list;          /* Stream list */
        struct lttng_transport *transport;
        uint64_t version;               /* Current version of the metadata cache */
+       bool coherent;                  /* Stream in a coherent state */
 };
 
 #define LTTNG_DYNAMIC_LEN_STACK_SIZE   128
@@ -529,6 +530,7 @@ struct lttng_metadata_cache {
        char *data;                     /* Metadata cache */
        unsigned int cache_alloc;       /* Metadata allocated size (bytes) */
        unsigned int metadata_written;  /* Number of bytes written in metadata cache */
+       int producing;                  /* Metadata being produced (incomplete) */
        struct kref refcount;           /* Metadata cache usage */
        struct list_head metadata_stream;       /* Metadata stream list */
        uuid_le uuid;                   /* Trace session unique ID (copy) */
@@ -609,7 +611,7 @@ int lttng_probes_init(void);
 void lttng_probes_exit(void);
 
 int lttng_metadata_output_channel(struct lttng_metadata_stream *stream,
-               struct channel *chan);
+               struct channel *chan, bool *coherent);
 
 int lttng_id_tracker_get_node_id(const struct lttng_id_hash_node *node);
 int lttng_id_tracker_empty_set(struct lttng_id_tracker *lf);

diff --git a/include/ringbuffer/vfs.h b/include/ringbuffer/vfs.h
index 5e13068cb40b4a2192fb4b2f0a01dc9c82076530..b0e8bff6f96bae9b522d142650e9aed2608aaef2 100644 (file)
--- a/include/ringbuffer/vfs.h
+++ b/include/ringbuffer/vfs.h
@@ -114,6 +114,12 @@ ssize_t vfs_lib_ring_buffer_splice_read(struct file *in, loff_t *ppos,
 #define RING_BUFFER_METADATA_CACHE_DUMP                _IO(0xF6, 0x10)
 /* Clear ring buffer content. */
 #define RING_BUFFER_CLEAR                      _IO(0xF6, 0x11)
+/*
+ * Get next metadata subbuffer, returning a flag indicating whether the
+ * metadata is guaranteed to be in a consistent state at the end of this
+ * sub-buffer (can be parsed).
+ */
+#define RING_BUFFER_GET_NEXT_SUBBUF_METADATA_CHECK     _IOR(0xF6, 0x12, uint32_t)
 
 #ifdef CONFIG_COMPAT
 /* Get a snapshot of the current ring buffer producer and consumer positions */
@@ -161,6 +167,13 @@ ssize_t vfs_lib_ring_buffer_splice_read(struct file *in, loff_t *ppos,
 /* Clear ring buffer content. */
 #define RING_BUFFER_COMPAT_CLEAR                       \
        RING_BUFFER_CLEAR
+/*
+ * Get next metadata subbuffer, returning a flag indicating whether the
+ * metadata is guaranteed to be in a consistent state at the end of this
+ * sub-buffer (can be parsed).
+ */
+#define RING_BUFFER_COMPAT_GET_NEXT_SUBBUF_METADATA_CHECK \
+       RING_BUFFER_GET_NEXT_SUBBUF_METADATA_CHECK
 #endif /* CONFIG_COMPAT */
 
 #endif /* _LIB_RING_BUFFER_VFS_H */

diff --git a/src/lttng-abi.c b/src/lttng-abi.c
index 01eb4d55addea3ea41b5def0ed79ca9efd91fcf1..c3721b93e428ce7d48bf94f40ff9ebd0a6585ef4 100644 (file)
--- a/src/lttng-abi.c
+++ b/src/lttng-abi.c
@@ -65,6 +65,7 @@ static const struct file_operations lttng_event_fops;
 static struct file_operations lttng_stream_ring_buffer_file_operations;
 
 static int put_u64(uint64_t val, unsigned long arg);
+static int put_u32(uint32_t val, unsigned long arg);
 
 /*
  * Teardown management: opened file descriptors keep a refcount on the module,
@@ -850,6 +851,13 @@ long lttng_metadata_ring_buffer_ioctl(struct file *filp,
        int ret;
        struct lttng_metadata_stream *stream = filp->private_data;
        struct lib_ring_buffer *buf = stream->priv;
+       unsigned int rb_cmd;
+       bool coherent;
+
+       if (cmd == RING_BUFFER_GET_NEXT_SUBBUF_METADATA_CHECK)
+               rb_cmd = RING_BUFFER_GET_NEXT_SUBBUF;
+       else
+               rb_cmd = cmd;
 
        switch (cmd) {
        case RING_BUFFER_GET_NEXT_SUBBUF:
@@ -858,7 +866,7 @@ long lttng_metadata_ring_buffer_ioctl(struct file *filp,
                struct lib_ring_buffer *buf = stream->priv;
                struct channel *chan = buf->backend.chan;
 
-               ret = lttng_metadata_output_channel(stream, chan);
+               ret = lttng_metadata_output_channel(stream, chan, NULL);
                if (ret > 0) {
                        lib_ring_buffer_switch_slow(buf, SWITCH_ACTIVE);
                        ret = 0;
@@ -884,7 +892,7 @@ long lttng_metadata_ring_buffer_ioctl(struct file *filp,
                 * Before doing the actual ring buffer flush, write up to one
                 * packet of metadata in the ring buffer.
                 */
-               ret = lttng_metadata_output_channel(stream, chan);
+               ret = lttng_metadata_output_channel(stream, chan, NULL);
                if (ret < 0)
                        goto err;
                break;
@@ -901,13 +909,28 @@ long lttng_metadata_ring_buffer_ioctl(struct file *filp,
 
                return lttng_metadata_cache_dump(stream);
        }
+       case RING_BUFFER_GET_NEXT_SUBBUF_METADATA_CHECK:
+       {
+               struct lttng_metadata_stream *stream = filp->private_data;
+               struct lib_ring_buffer *buf = stream->priv;
+               struct channel *chan = buf->backend.chan;
+
+               ret = lttng_metadata_output_channel(stream, chan, &coherent);
+               if (ret > 0) {
+                       lib_ring_buffer_switch_slow(buf, SWITCH_ACTIVE);
+                       ret = 0;
+               } else if (ret < 0) {
+                       goto err;
+               }
+               break;
+       }
        default:
                break;
        }
        /* PUT_SUBBUF is the one from lib ring buffer, unmodified. */
 
        /* Performing lib ring buffer ioctl after our own. */
-       ret = lib_ring_buffer_ioctl(filp, cmd, arg, buf);
+       ret = lib_ring_buffer_ioctl(filp, rb_cmd, arg, buf);
        if (ret < 0)
                goto err;
 
@@ -918,6 +941,10 @@ long lttng_metadata_ring_buffer_ioctl(struct file *filp,
                                cmd, arg);
                break;
        }
+       case RING_BUFFER_GET_NEXT_SUBBUF_METADATA_CHECK:
+       {
+               return put_u32(coherent, arg);
+       }
        default:
                break;
        }
@@ -933,6 +960,13 @@ long lttng_metadata_ring_buffer_compat_ioctl(struct file *filp,
        int ret;
        struct lttng_metadata_stream *stream = filp->private_data;
        struct lib_ring_buffer *buf = stream->priv;
+       unsigned int rb_cmd;
+       bool coherent;
+
+       if (cmd == RING_BUFFER_GET_NEXT_SUBBUF_METADATA_CHECK)
+               rb_cmd = RING_BUFFER_GET_NEXT_SUBBUF;
+       else
+               rb_cmd = cmd;
 
        switch (cmd) {
        case RING_BUFFER_GET_NEXT_SUBBUF:
@@ -941,7 +975,7 @@ long lttng_metadata_ring_buffer_compat_ioctl(struct file *filp,
                struct lib_ring_buffer *buf = stream->priv;
                struct channel *chan = buf->backend.chan;
 
-               ret = lttng_metadata_output_channel(stream, chan);
+               ret = lttng_metadata_output_channel(stream, chan, NULL);
                if (ret > 0) {
                        lib_ring_buffer_switch_slow(buf, SWITCH_ACTIVE);
                        ret = 0;
@@ -967,7 +1001,7 @@ long lttng_metadata_ring_buffer_compat_ioctl(struct file *filp,
                 * Before doing the actual ring buffer flush, write up to one
                 * packet of metadata in the ring buffer.
                 */
-               ret = lttng_metadata_output_channel(stream, chan);
+               ret = lttng_metadata_output_channel(stream, chan, NULL);
                if (ret < 0)
                        goto err;
                break;
@@ -984,13 +1018,28 @@ long lttng_metadata_ring_buffer_compat_ioctl(struct file *filp,
 
                return lttng_metadata_cache_dump(stream);
        }
+       case RING_BUFFER_GET_NEXT_SUBBUF_METADATA_CHECK:
+       {
+               struct lttng_metadata_stream *stream = filp->private_data;
+               struct lib_ring_buffer *buf = stream->priv;
+               struct channel *chan = buf->backend.chan;
+
+               ret = lttng_metadata_output_channel(stream, chan, &coherent);
+               if (ret > 0) {
+                       lib_ring_buffer_switch_slow(buf, SWITCH_ACTIVE);
+                       ret = 0;
+               } else if (ret < 0) {
+                       goto err;
+               }
+               break;
+       }
        default:
                break;
        }
        /* PUT_SUBBUF is the one from lib ring buffer, unmodified. */
 
        /* Performing lib ring buffer ioctl after our own. */
-       ret = lib_ring_buffer_compat_ioctl(filp, cmd, arg, buf);
+       ret = lib_ring_buffer_compat_ioctl(filp, rb_cmd, arg, buf);
        if (ret < 0)
                goto err;
 
@@ -1001,6 +1050,10 @@ long lttng_metadata_ring_buffer_compat_ioctl(struct file *filp,
                                cmd, arg);
                break;
        }
+       case RING_BUFFER_GET_NEXT_SUBBUF_METADATA_CHECK:
+       {
+               return put_u32(coherent, arg);
+       }
        default:
                break;
        }
@@ -1167,6 +1220,8 @@ int lttng_abi_open_metadata_stream(struct file *channel_file)
        metadata_stream->priv = buf;
        stream_priv = metadata_stream;
        metadata_stream->transport = channel->transport;
+       /* Initial state is an empty metadata, considered as incoherent. */
+       metadata_stream->coherent = false;
 
        /*
         * Since life-time of metadata cache differs from that of
@@ -1703,6 +1758,11 @@ static int put_u64(uint64_t val, unsigned long arg)
        return put_user(val, (uint64_t __user *) arg);
 }
 
+static int put_u32(uint32_t val, unsigned long arg)
+{
+       return put_user(val, (uint32_t __user *) arg);
+}
+
 static long lttng_stream_ring_buffer_ioctl(struct file *filp,
                unsigned int cmd, unsigned long arg)
 {

diff --git a/src/lttng-events.c b/src/lttng-events.c
index 6bef77c934ccc765418a5ed9d9d2a6be4bfe4082..3b9b907beb85196b36301cbb6df6456ab195b23b 100644 (file)
--- a/src/lttng-events.c
+++ b/src/lttng-events.c
@@ -1626,7 +1626,7 @@ void lttng_session_lazy_sync_enablers(struct lttng_session *session)
  * was written and a negative value on error.
  */
 int lttng_metadata_output_channel(struct lttng_metadata_stream *stream,
-               struct channel *chan)
+               struct channel *chan, bool *coherent)
 {
        struct lib_ring_buffer_ctx ctx;
        int ret = 0;
@@ -1665,6 +1665,7 @@ int lttng_metadata_output_channel(struct lttng_metadata_stream *stream,
        ret = stream->transport->ops.event_reserve(&ctx, 0);
        if (ret != 0) {
                printk(KERN_WARNING "LTTng: Metadata event reservation failed\n");
+               stream->coherent = false;
                goto end;
        }
        stream->transport->ops.event_write(&ctx,
@@ -1672,18 +1673,43 @@ int lttng_metadata_output_channel(struct lttng_metadata_stream *stream,
                        reserve_len);
        stream->transport->ops.event_commit(&ctx);
        stream->metadata_in += reserve_len;
+       if (reserve_len < len || stream->metadata_cache->producing != 0)
+               stream->coherent = false;
+       else
+               stream->coherent = true;
        ret = reserve_len;
 
 end:
+       if (coherent)
+               *coherent = stream->coherent;
        mutex_unlock(&stream->metadata_cache->lock);
        return ret;
 }
 
+static
+void lttng_metadata_begin(struct lttng_session *session)
+{
+       mutex_lock(&session->metadata_cache->lock);
+       session->metadata_cache->producing++;
+       mutex_unlock(&session->metadata_cache->lock);
+}
+
+static
+void lttng_metadata_end(struct lttng_session *session)
+{
+       mutex_lock(&session->metadata_cache->lock);
+       WARN_ON_ONCE(!session->metadata_cache->producing);
+       session->metadata_cache->producing--;
+       mutex_unlock(&session->metadata_cache->lock);
+}
+
 /*
  * Write the metadata to the metadata cache.
  * Must be called with sessions_mutex held.
  * The metadata cache lock protects us from concurrent read access from
  * thread outputting metadata content to ring buffer.
+ * The content of the printf is printed as a single atomic metadata
+ * transaction.
  */
 int lttng_metadata_printf(struct lttng_session *session,
                          const char *fmt, ...)
@@ -1703,6 +1729,7 @@ int lttng_metadata_printf(struct lttng_session *session,
 
        len = strlen(str);
        mutex_lock(&session->metadata_cache->lock);
+       session->metadata_cache->producing++;
        if (session->metadata_cache->metadata_written + len >
                        session->metadata_cache->cache_alloc) {
                char *tmp_cache_realloc;
@@ -1728,6 +1755,7 @@ int lttng_metadata_printf(struct lttng_session *session,
                        session->metadata_cache->metadata_written,
                        str, len);
        session->metadata_cache->metadata_written += len;
+       session->metadata_cache->producing--;
        mutex_unlock(&session->metadata_cache->lock);
        kfree(str);
 
@@ -1737,6 +1765,7 @@ int lttng_metadata_printf(struct lttng_session *session,
        return 0;
 
 err:
+       session->metadata_cache->producing--;
        mutex_unlock(&session->metadata_cache->lock);
        kfree(str);
        return -ENOMEM;
@@ -2312,6 +2341,8 @@ int _lttng_fields_metadata_statedump(struct lttng_session *session,
 
 /*
  * Must be called with sessions_mutex held.
+ * The entire event metadata is printed as a single atomic metadata
+ * transaction.
  */
 static
 int _lttng_event_metadata_statedump(struct lttng_session *session,
@@ -2325,6 +2356,8 @@ int _lttng_event_metadata_statedump(struct lttng_session *session,
        if (chan->channel_type == METADATA_CHANNEL)
                return 0;
 
+       lttng_metadata_begin(session);
+
        ret = lttng_metadata_printf(session,
                "event {\n"
                "       name = \"%s\";\n"
@@ -2374,12 +2407,15 @@ int _lttng_event_metadata_statedump(struct lttng_session *session,
 
        event->metadata_dumped = 1;
 end:
+       lttng_metadata_end(session);
        return ret;
 
 }
 
 /*
  * Must be called with sessions_mutex held.
+ * The entire channel metadata is printed as a single atomic metadata
+ * transaction.
  */
 static
 int _lttng_channel_metadata_statedump(struct lttng_session *session,
@@ -2393,6 +2429,8 @@ int _lttng_channel_metadata_statedump(struct lttng_session *session,
        if (chan->channel_type == METADATA_CHANNEL)
                return 0;
 
+       lttng_metadata_begin(session);
+
        WARN_ON_ONCE(!chan->header_type);
        ret = lttng_metadata_printf(session,
                "stream {\n"
@@ -2426,6 +2464,7 @@ int _lttng_channel_metadata_statedump(struct lttng_session *session,
 
        chan->metadata_dumped = 1;
 end:
+       lttng_metadata_end(session);
        return ret;
 }
 
@@ -2613,6 +2652,9 @@ int _lttng_session_metadata_statedump(struct lttng_session *session)
 
        if (!READ_ONCE(session->active))
                return 0;
+
+       lttng_metadata_begin(session);
+
        if (session->metadata_dumped)
                goto skip_session;
 
@@ -2785,6 +2827,7 @@ skip_session:
        }
        session->metadata_dumped = 1;
 end:
+       lttng_metadata_end(session);
        return ret;
 }