Fix: mmap trace read the stream_id from the first packet

We were hardcoding the stream_id as 0 for mmap traces, causing a
problem for live traces with multiple channels.
A fix enters in lttng-tools v2.4.2 to send the CTF stream id in the beacon
packets, but until then, we cannot support traces with multiple CTF
streams (LTTng channels).

Fixes: #811
Signed-off-by: Julien Desfossez <jdesfossez@efficios.com>
Signed-off-by: Mathieu Desnoyers <mathieu.desnoyers@efficios.com>

diff --git a/formats/ctf/ctf.c b/formats/ctf/ctf.c
index e31808c3843d27380a63b0d173a949aa027e336e..8318a96a0781742837f9f4c754754d83c25d9f68 100644 (file)
--- a/formats/ctf/ctf.c
+++ b/formats/ctf/ctf.c
@@ -2245,13 +2245,17 @@ void ctf_init_mmap_pos(struct ctf_stream_pos *pos,
 
 static
 int prepare_mmap_stream_definition(struct ctf_trace *td,
 
 static
 int prepare_mmap_stream_definition(struct ctf_trace *td,

-               struct ctf_file_stream *file_stream)
+               struct ctf_file_stream *file_stream,
+               void (*packet_seek)(struct bt_stream_pos *pos, size_t index,
+                       int whence))

 {
        struct ctf_stream_declaration *stream;
 {
        struct ctf_stream_declaration *stream;

-       uint64_t stream_id = 0;
+       uint64_t stream_id;

        int ret;
 
        int ret;
 

-       file_stream->parent.stream_id = stream_id;
+       /* Ask for the first packet to get the stream_id. */
+       packet_seek(&file_stream->pos.parent, 0, SEEK_SET);
+       stream_id = file_stream->parent.stream_id;

        if (stream_id >= td->streams->len) {
                fprintf(stderr, "[error] Stream %" PRIu64 " is not declared "
                                "in metadata.\n", stream_id);
        if (stream_id >= td->streams->len) {
                fprintf(stderr, "[error] Stream %" PRIu64 " is not declared "
                                "in metadata.\n", stream_id);


@@ -2281,6 +2285,7 @@ int ctf_open_mmap_stream_read(struct ctf_trace *td,
        struct ctf_file_stream *file_stream;
 
        file_stream = g_new0(struct ctf_file_stream, 1);
        struct ctf_file_stream *file_stream;
 
        file_stream = g_new0(struct ctf_file_stream, 1);

+       file_stream->parent.stream_id = -1ULL;

        file_stream->pos.last_offset = LAST_OFFSET_POISON;
        ctf_init_mmap_pos(&file_stream->pos, mmap_info);
 
        file_stream->pos.last_offset = LAST_OFFSET_POISON;
        ctf_init_mmap_pos(&file_stream->pos, mmap_info);
 


@@ -2291,7 +2296,7 @@ int ctf_open_mmap_stream_read(struct ctf_trace *td,
                goto error_def;
        }
 
                goto error_def;
        }
 

-       ret = prepare_mmap_stream_definition(td, file_stream);
+       ret = prepare_mmap_stream_definition(td, file_stream, packet_seek);

        if (ret)
                goto error_index;
 
        if (ret)
                goto error_index;
 

diff --git a/formats/lttng-live/lttng-live-comm.c b/formats/lttng-live/lttng-live-comm.c
index 070674aa67c13298032d6b0fc5c4411ef03085f5..a3a02a2cc19ad1bc9741b8340d83fb92978136f6 100644 (file)
--- a/formats/lttng-live/lttng-live-comm.c
+++ b/formats/lttng-live/lttng-live-comm.c
@@ -523,8 +523,8 @@ int lttng_live_attach_session(struct lttng_live_ctx *ctx, uint64_t id)
                ctx->session->streams[i].id = be64toh(stream.id);
                ctx->session->streams[i].session = ctx->session;
 
                ctx->session->streams[i].id = be64toh(stream.id);
                ctx->session->streams[i].session = ctx->session;
 

-               ctx->session->streams[i].first_read = 1;

                ctx->session->streams[i].mmap_size = 0;
                ctx->session->streams[i].mmap_size = 0;

+               ctx->session->streams[i].ctf_stream_id = -1ULL;

 
                if (be32toh(stream.metadata_flag)) {
                        ctx->session->streams[i].metadata_flag = 1;
 
                if (be32toh(stream.metadata_flag)) {
                        ctx->session->streams[i].metadata_flag = 1;


@@ -932,6 +932,24 @@ error:
        return ret;
 }
 
        return ret;
 }
 

+/*
+ * Assign the fields from a lttng_viewer_index to a packet_index.
+ */
+static
+void lttng_index_to_packet_index(struct lttng_viewer_index *lindex,
+               struct packet_index *pindex)
+{
+       assert(lindex);
+       assert(pindex);
+
+       pindex->offset = be64toh(lindex->offset);
+       pindex->packet_size = be64toh(lindex->packet_size);
+       pindex->content_size = be64toh(lindex->content_size);
+       pindex->ts_cycles.timestamp_begin = be64toh(lindex->timestamp_begin);
+       pindex->ts_cycles.timestamp_end = be64toh(lindex->timestamp_end);
+       pindex->events_discarded = be64toh(lindex->events_discarded);
+}
+

 /*
  * Get one index for a stream.
  *
 /*
  * Get one index for a stream.
  *


@@ -940,13 +958,13 @@ error:
 static
 int get_next_index(struct lttng_live_ctx *ctx,
                struct lttng_live_viewer_stream *viewer_stream,
 static
 int get_next_index(struct lttng_live_ctx *ctx,
                struct lttng_live_viewer_stream *viewer_stream,

-               struct packet_index *index)
+               struct packet_index *index, uint64_t *stream_id)

 {
        struct lttng_viewer_cmd cmd;
        struct lttng_viewer_get_next_index rq;
 {
        struct lttng_viewer_cmd cmd;
        struct lttng_viewer_get_next_index rq;

-       struct lttng_viewer_index rp;

        int ret;
        ssize_t ret_len;
        int ret;
        ssize_t ret_len;

+       struct lttng_viewer_index *rp = &viewer_stream->current_index;

 
        cmd.cmd = htobe32(LTTNG_VIEWER_GET_NEXT_INDEX);
        cmd.data_size = sizeof(rq);
 
        cmd.cmd = htobe32(LTTNG_VIEWER_GET_NEXT_INDEX);
        cmd.data_size = sizeof(rq);


@@ -974,7 +992,7 @@ retry:
        }
        assert(ret_len == sizeof(rq));
 
        }
        assert(ret_len == sizeof(rq));
 

-       ret_len = lttng_live_recv(ctx->control_sock, &rp, sizeof(rp));
+       ret_len = lttng_live_recv(ctx->control_sock, rp, sizeof(*rp));

        if (ret_len == 0) {
                fprintf(stderr, "[error] Remote side has closed connection\n");
                goto error;
        if (ret_len == 0) {
                fprintf(stderr, "[error] Remote side has closed connection\n");
                goto error;


@@ -983,32 +1001,30 @@ retry:
                perror("[error] Error receiving index response");
                goto error;
        }
                perror("[error] Error receiving index response");
                goto error;
        }

-       assert(ret_len == sizeof(rp));
+       assert(ret_len == sizeof(*rp));

 
 

-       rp.flags = be32toh(rp.flags);
+       rp->flags = be32toh(rp->flags);

 
 

-       switch (be32toh(rp.status)) {
+       switch (be32toh(rp->status)) {

        case LTTNG_VIEWER_INDEX_INACTIVE:
                printf_verbose("get_next_index: inactive\n");
                memset(index, 0, sizeof(struct packet_index));
        case LTTNG_VIEWER_INDEX_INACTIVE:
                printf_verbose("get_next_index: inactive\n");
                memset(index, 0, sizeof(struct packet_index));

-               index->ts_cycles.timestamp_end = be64toh(rp.timestamp_end);
+               index->ts_cycles.timestamp_end = be64toh(rp->timestamp_end);
+               *stream_id = be64toh(rp->stream_id);

                break;
        case LTTNG_VIEWER_INDEX_OK:
                printf_verbose("get_next_index: Ok, need metadata update : %u\n",
                break;
        case LTTNG_VIEWER_INDEX_OK:
                printf_verbose("get_next_index: Ok, need metadata update : %u\n",

-                               rp.flags & LTTNG_VIEWER_FLAG_NEW_METADATA);
-               index->offset = be64toh(rp.offset);
-               index->packet_size = be64toh(rp.packet_size);
-               index->content_size = be64toh(rp.content_size);
-               index->ts_cycles.timestamp_begin = be64toh(rp.timestamp_begin);
-               index->ts_cycles.timestamp_end = be64toh(rp.timestamp_end);
-               index->events_discarded = be64toh(rp.events_discarded);
+                               rp->flags & LTTNG_VIEWER_FLAG_NEW_METADATA);
+               lttng_index_to_packet_index(rp, index);
+               *stream_id = be64toh(rp->stream_id);
+               viewer_stream->data_pending = 1;

 
 

-               if (rp.flags & LTTNG_VIEWER_FLAG_NEW_METADATA) {
+               if (rp->flags & LTTNG_VIEWER_FLAG_NEW_METADATA) {

                        ret = append_metadata(ctx, viewer_stream);
                        if (ret)
                                goto error;
                }
                        ret = append_metadata(ctx, viewer_stream);
                        if (ret)
                                goto error;
                }

-               if (rp.flags & LTTNG_VIEWER_FLAG_NEW_STREAM) {
+               if (rp->flags & LTTNG_VIEWER_FLAG_NEW_STREAM) {

                        printf_verbose("get_next_index: need new streams\n");
                        ret = ask_new_streams(ctx);
                        if (ret < 0)
                        printf_verbose("get_next_index: need new streams\n");
                        ret = ask_new_streams(ctx);
                        if (ret < 0)


@@ -1043,6 +1059,82 @@ error:
        return -1;
 }
 
        return -1;
 }
 

+static
+void read_packet_header(struct ctf_stream_pos *pos,
+               struct ctf_file_stream *file_stream)
+{
+       int ret;
+
+       /* update trace_packet_header and stream_packet_context */
+       if (pos->prot != PROT_WRITE && file_stream->parent.trace_packet_header) {
+               /* Read packet header */
+               ret = generic_rw(&pos->parent,
+                               &file_stream->parent.trace_packet_header->p);
+               if (ret) {
+                       pos->offset = EOF;
+                       fprintf(stderr, "[error] trace packet "
+                                       "header read failed\n");
+                       goto end;
+               }
+       }
+       if (pos->prot != PROT_WRITE && file_stream->parent.stream_packet_context) {
+               /* Read packet context */
+               ret = generic_rw(&pos->parent,
+                               &file_stream->parent.stream_packet_context->p);
+               if (ret) {
+                       pos->offset = EOF;
+                       fprintf(stderr, "[error] stream packet "
+                                       "context read failed\n");
+                       goto end;
+               }
+       }
+       pos->data_offset = pos->offset;
+
+end:
+       return;
+}
+
+/*
+ * Handle the seek parameters.
+ * Returns 0 if the packet_seek can continue, a positive value to
+ * cleanly exit the packet_seek, a negative value on error.
+ */
+static
+int handle_seek_position(size_t index, int whence,
+               struct lttng_live_viewer_stream *viewer_stream,
+               struct ctf_stream_pos *pos,
+               struct ctf_file_stream *file_stream)
+{
+       int ret;
+
+       switch (whence) {
+       case SEEK_CUR:
+               ret = 0;
+               goto end;
+       case SEEK_SET:
+               /*
+                * We only allow to seek to 0.
+                */
+               if (index != 0) {
+                       fprintf(stderr, "[error] Arbitrary seek in lttng-live "
+                                       "trace not supported\n");
+                       pos->offset = EOF;
+                       ret = -1;
+                       goto end;
+               }
+
+               ret = 0;
+               goto end;
+
+       default:
+               fprintf(stderr, "[error] Invalid seek parameter\n");
+               assert(0);
+       }
+
+end:
+       return ret;
+}
+

 static
 void ctf_live_packet_seek(struct bt_stream_pos *stream_pos, size_t index,
                int whence)
 static
 void ctf_live_packet_seek(struct bt_stream_pos *stream_pos, size_t index,
                int whence)


@@ -1052,14 +1144,21 @@ void ctf_live_packet_seek(struct bt_stream_pos *stream_pos, size_t index,
        struct packet_index *prev_index = NULL, *cur_index;
        struct lttng_live_viewer_stream *viewer_stream;
        struct lttng_live_session *session;
        struct packet_index *prev_index = NULL, *cur_index;
        struct lttng_live_viewer_stream *viewer_stream;
        struct lttng_live_session *session;

+       uint64_t stream_id = -1ULL;

        int ret;
 
        int ret;
 

-retry:

        pos = ctf_pos(stream_pos);
        file_stream = container_of(pos, struct ctf_file_stream, pos);
        viewer_stream = (struct lttng_live_viewer_stream *) pos->priv;
        session = viewer_stream->session;
 
        pos = ctf_pos(stream_pos);
        file_stream = container_of(pos, struct ctf_file_stream, pos);
        viewer_stream = (struct lttng_live_viewer_stream *) pos->priv;
        session = viewer_stream->session;
 

+       ret = handle_seek_position(index, whence, viewer_stream, pos,
+                       file_stream);
+       if (ret != 0) {
+               return;
+       }
+
+retry:

        switch (pos->packet_index->len) {
        case 0:
                g_array_set_size(pos->packet_index, 1);
        switch (pos->packet_index->len) {
        case 0:
                g_array_set_size(pos->packet_index, 1);


@@ -1087,13 +1186,46 @@ retry:
                abort();
                break;
        }
                abort();
                break;
        }

-       printf_verbose("get_next_index for stream %" PRIu64 "\n", viewer_stream->id);
-       ret = get_next_index(session->ctx, viewer_stream, cur_index);
-       if (ret < 0) {
-               pos->offset = EOF;
-               if (!lttng_live_should_quit()) {
-                       fprintf(stderr, "[error] get_next_index failed\n");
+
+       if (viewer_stream->data_pending) {
+               lttng_index_to_packet_index(&viewer_stream->current_index, cur_index);
+       } else {
+               printf_verbose("get_next_index for stream %" PRIu64 "\n", viewer_stream->id);
+               ret = get_next_index(session->ctx, viewer_stream, cur_index, &stream_id);
+               if (ret < 0) {
+                       pos->offset = EOF;
+                       if (!lttng_live_should_quit()) {
+                               fprintf(stderr, "[error] get_next_index failed\n");
+                       }
+                       return;

                }
                }

+               printf_verbose("Index received : packet_size : %" PRIu64
+                               ", offset %" PRIu64 ", content_size %" PRIu64
+                               ", timestamp_end : %" PRIu64 "\n",
+                               cur_index->packet_size, cur_index->offset,
+                               cur_index->content_size,
+                               cur_index->ts_cycles.timestamp_end);
+
+       }
+
+       /*
+        * On the first time we receive an index, the stream_id needs to
+        * be set for the stream in order to use it, we don't want any
+        * data at this stage.
+        */
+       if (file_stream->parent.stream_id == -1ULL) {
+               /*
+                * Warning: with lttng-tools < 2.4.2, the beacon does not
+                * contain the real stream ID, it is memset to 0, so this
+                * might create a problem when a session has multiple
+                * channels. We can't detect it at this stage, lttng-tools
+                * has to be upgraded to fix this problem.
+                */
+               printf_verbose("Assigning stream_id %" PRIu64 "\n",
+                               stream_id);
+               file_stream->parent.stream_id = stream_id;
+               viewer_stream->ctf_stream_id = stream_id;
+

                return;
        }
 
                return;
        }
 


@@ -1107,19 +1239,23 @@ retry:
        }
 
        if (cur_index->content_size == 0) {
        }
 
        if (cur_index->content_size == 0) {

-               file_stream->parent.cycles_timestamp =
+               if (file_stream->parent.stream_class) {
+                       file_stream->parent.cycles_timestamp =

                                cur_index->ts_cycles.timestamp_end;
                                cur_index->ts_cycles.timestamp_end;

-               file_stream->parent.real_timestamp = ctf_get_real_timestamp(
-                               &file_stream->parent,
-                               cur_index->ts_cycles.timestamp_end);
+                       file_stream->parent.real_timestamp = ctf_get_real_timestamp(
+                                       &file_stream->parent,
+                                       cur_index->ts_cycles.timestamp_end);
+               }

        } else {
        } else {

-               /* Convert the timestamps and append to the real_index. */
-               cur_index->ts_real.timestamp_begin = ctf_get_real_timestamp(
-                               &file_stream->parent,
-                               cur_index->ts_cycles.timestamp_begin);
-               cur_index->ts_real.timestamp_end = ctf_get_real_timestamp(
-                               &file_stream->parent,
-                               cur_index->ts_cycles.timestamp_end);
+               if (file_stream->parent.stream_class) {
+                       /* Convert the timestamps and append to the real_index. */
+                       cur_index->ts_real.timestamp_begin = ctf_get_real_timestamp(
+                                       &file_stream->parent,
+                                       cur_index->ts_cycles.timestamp_begin);
+                       cur_index->ts_real.timestamp_end = ctf_get_real_timestamp(
+                                       &file_stream->parent,
+                                       cur_index->ts_cycles.timestamp_end);
+               }

 
                ctf_update_current_packet_index(&file_stream->parent,
                                prev_index, cur_index);
 
                ctf_update_current_packet_index(&file_stream->parent,
                                prev_index, cur_index);


@@ -1148,34 +1284,9 @@ retry:
                }
                return;
        }
                }
                return;
        }

+       viewer_stream->data_pending = 0;

 
 

-       printf_verbose("Index received : packet_size : %" PRIu64
-                       ", offset %" PRIu64 ", content_size %" PRIu64
-                       ", timestamp_end : %" PRIu64 "\n",
-                       cur_index->packet_size, cur_index->offset,
-                       cur_index->content_size,
-                       cur_index->ts_cycles.timestamp_end);
-
-       /* update trace_packet_header and stream_packet_context */
-       if (pos->prot != PROT_WRITE && file_stream->parent.trace_packet_header) {
-               /* Read packet header */
-               ret = generic_rw(&pos->parent, &file_stream->parent.trace_packet_header->p);
-               if (ret) {
-                       pos->offset = EOF;
-                       fprintf(stderr, "[error] trace packet header read failed\n");
-                       goto end;
-               }
-       }
-       if (pos->prot != PROT_WRITE && file_stream->parent.stream_packet_context) {
-               /* Read packet context */
-               ret = generic_rw(&pos->parent, &file_stream->parent.stream_packet_context->p);
-               if (ret) {
-                       pos->offset = EOF;
-                       fprintf(stderr, "[error] stream packet context read failed\n");
-                       goto end;
-               }
-       }
-       pos->data_offset = pos->offset;
+       read_packet_header(pos, file_stream);

 
 end:
        return;
 
 end:
        return;


@@ -1442,8 +1553,8 @@ int lttng_live_get_new_streams(struct lttng_live_ctx *ctx, uint64_t id)
                ctx->session->streams[i].id = be64toh(stream.id);
                ctx->session->streams[i].session = ctx->session;
 
                ctx->session->streams[i].id = be64toh(stream.id);
                ctx->session->streams[i].session = ctx->session;
 

-               ctx->session->streams[i].first_read = 1;

                ctx->session->streams[i].mmap_size = 0;
                ctx->session->streams[i].mmap_size = 0;

+               ctx->session->streams[i].ctf_stream_id = -1ULL;

 
                if (be32toh(stream.metadata_flag)) {
                        ctx->session->streams[i].metadata_flag = 1;
 
                if (be32toh(stream.metadata_flag)) {
                        ctx->session->streams[i].metadata_flag = 1;

diff --git a/formats/lttng-live/lttng-live.h b/formats/lttng-live/lttng-live.h
index d7b0e860b14786950ca5dee227447d6b8dd35a83..c4641df7c62bc934896ea3ac0d81d49ca8bc5d92 100644 (file)
--- a/formats/lttng-live/lttng-live.h
+++ b/formats/lttng-live/lttng-live.h
@@ -25,6 +25,7 @@
  */
 
 #include <stdint.h>
  */
 
 #include <stdint.h>

+#include "lttng-viewer-abi.h"

 
 #define LTTNG_DEFAULT_NETWORK_VIEWER_PORT      5344
 
 
 #define LTTNG_DEFAULT_NETWORK_VIEWER_PORT      5344
 


@@ -48,12 +49,14 @@ struct lttng_live_ctx {
 struct lttng_live_viewer_stream {
        uint64_t id;
        uint64_t mmap_size;
 struct lttng_live_viewer_stream {
        uint64_t id;
        uint64_t mmap_size;

+       uint64_t ctf_stream_id;

        FILE *metadata_fp_write;
        ssize_t metadata_len;
        int metadata_flag;
        FILE *metadata_fp_write;
        ssize_t metadata_len;
        int metadata_flag;

-       int first_read;
+       int data_pending;

        struct lttng_live_session *session;
        struct lttng_live_ctf_trace *ctf_trace;
        struct lttng_live_session *session;
        struct lttng_live_ctf_trace *ctf_trace;

+       struct lttng_viewer_index current_index;

        char path[PATH_MAX];
 };
 
        char path[PATH_MAX];
 };