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 b30b0f5814553f84915ab20a1982e062d5a20cc7..44a8bc24a0b867f1365b74bdd83521b83c1fe934 100644 (file)
--- a/lib/ringbuffer/vfs.h
+++ b/lib/ringbuffer/vfs.h
@@ -112,6 +112,12 @@ ssize_t vfs_lib_ring_buffer_splice_read(struct file *in, loff_t *ppos,
  * so it can be read again.
  */
 #define RING_BUFFER_METADATA_CACHE_DUMP                _IO(0xF6, 0x10)
+/*
+ * 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 */
@@ -156,6 +162,13 @@ ssize_t vfs_lib_ring_buffer_splice_read(struct file *in, loff_t *ppos,
 /* Flush the current sub-buffer, even if empty. */
 #define RING_BUFFER_COMPAT_FLUSH_EMPTY                 \
        RING_BUFFER_FLUSH_EMPTY
+/*
+ * 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 816e6fddf57848d4e68e323b3f902fe3e2bff515..b0b638d4f5f61168004a5d2b95e75428a1273345 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,
@@ -744,6 +745,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:
@@ -752,7 +760,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;
@@ -778,7 +786,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;
@@ -795,13 +803,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;
 
@@ -812,6 +835,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;
        }
@@ -827,6 +854,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:
@@ -835,7 +869,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;
@@ -861,7 +895,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;
@@ -878,13 +912,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;
 
@@ -895,6 +944,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;
        }
@@ -1061,6 +1114,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
@@ -1597,6 +1652,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 d78e8efabe50fdf87f55dfbb6ede91176bbc4b9f..481abeb320dff552f966c0b85701ca7305c819e6 100644 (file)
--- a/lttng-events.c
+++ b/lttng-events.c
@@ -1588,7 +1588,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;
@@ -1627,6 +1627,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,
@@ -1634,18 +1635,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, ...)
@@ -1665,6 +1691,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;
@@ -1690,6 +1717,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);
 
@@ -1699,6 +1727,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;
@@ -2236,6 +2265,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,
@@ -2249,6 +2280,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"
@@ -2298,12 +2331,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,
@@ -2317,6 +2353,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"
@@ -2350,6 +2388,7 @@ int _lttng_channel_metadata_statedump(struct lttng_session *session,
 
        chan->metadata_dumped = 1;
 end:
+       lttng_metadata_end(session);
        return ret;
 }
 
@@ -2536,6 +2575,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;
 
@@ -2697,6 +2739,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 769ec695feaedba7d338e10d50adbf287f2e590d..2c6df40a7b7b328140a451c00cdcc3115ca9e598 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
@@ -526,6 +527,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) */
@@ -606,7 +608,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_pid_tracker_get_node_pid(const struct lttng_pid_hash_node *node);
 struct lttng_pid_tracker *lttng_pid_tracker_create(void);