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/lib/ringbuffer/vfs.h b/lib/ringbuffer/vfs.h
index 2061156dd1b19aba0ebab7cf01815ab4e7645f78..26cda485e8f8f82c3aadf3bd969932b916cc0156 100644 (file)
--- a/lib/ringbuffer/vfs.h
+++ b/lib/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/lttng-abi.c b/lttng-abi.c
index 826ecab465b83ab10b2a712da1704c8f772e6374..0996f2f198532924704266d6014ab98cd1f55318 100644 (file)
--- a/lttng-abi.c
+++ b/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/lttng-events.c b/lttng-events.c
index 5a64f5b2e915950233c695b35540344718c3a78f..cfe2a3529de33efb0184409d4dbdbd105eb93b8a 100644 (file)
--- a/lttng-events.c
+++ b/lttng-events.c
@@ -1622,7 +1622,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;
@@ -1661,6 +1661,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,
@@ -1668,18 +1669,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, ...)
@@ -1699,6 +1725,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;
@@ -1724,6 +1751,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);
 
@@ -1733,6 +1761,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;
@@ -2270,6 +2299,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,
@@ -2283,6 +2314,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"
@@ -2332,12 +2365,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,
@@ -2351,6 +2387,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"
@@ -2384,6 +2422,7 @@ int _lttng_channel_metadata_statedump(struct lttng_session *session,
 
        chan->metadata_dumped = 1;
 end:
+       lttng_metadata_end(session);
        return ret;
 }
 
@@ -2571,6 +2610,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;
 
@@ -2743,6 +2785,7 @@ skip_session:
        }
        session->metadata_dumped = 1;
 end:
+       lttng_metadata_end(session);
        return ret;
 }
 

diff --git a/lttng-events.h b/lttng-events.h
index 6148fe12f0e1c90ce35a3c9d56cabe8f22c30761..2ee5fdcff7772d2e6d9912f179b54a14aaa7a3dd 100644 (file)
--- a/lttng-events.h
+++ b/lttng-events.h
@@ -474,6 +474,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
@@ -548,6 +549,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) */
@@ -628,7 +630,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);