X-Git-Url: http://git.efficios.com/?p=babeltrace.git;a=blobdiff_plain;f=formats%2Flttng-live%2Flttng-live-comm.c;h=065af49fb06f5f1be0d02f8a39e9919ddafef50b;hp=fdecf31c92eaa70238611a0b138fdcf6baf7b764;hb=1ab659eb38221a7908ee4693616ba43cca651a91;hpb=857eaa682bde4acfb5b9437ac9ace2d429d7107a diff --git a/formats/lttng-live/lttng-live-comm.c b/formats/lttng-live/lttng-live-comm.c index fdecf31c..065af49f 100644 --- a/formats/lttng-live/lttng-live-comm.c +++ b/formats/lttng-live/lttng-live-comm.c @@ -31,7 +31,6 @@ #include #include #include -#include #include #include @@ -53,6 +52,7 @@ #include #include #include +#include #include "lttng-live.h" #include "lttng-viewer-abi.h" @@ -157,6 +157,8 @@ int lttng_live_establish_connection(struct lttng_live_ctx *ctx) { struct lttng_viewer_cmd cmd; struct lttng_viewer_connect connect; + const size_t cmd_buf_len = sizeof(cmd) + sizeof(connect); + char cmd_buf[cmd_buf_len]; int ret; ssize_t ret_len; @@ -166,7 +168,7 @@ int lttng_live_establish_connection(struct lttng_live_ctx *ctx) } cmd.cmd = htobe32(LTTNG_VIEWER_CONNECT); - cmd.data_size = htobe64(sizeof(connect)); + cmd.data_size = htobe64((uint64_t) sizeof(connect)); cmd.cmd_version = htobe32(0); connect.viewer_session_id = -1ULL; /* will be set on recv */ @@ -174,19 +176,20 @@ int lttng_live_establish_connection(struct lttng_live_ctx *ctx) connect.minor = htobe32(LTTNG_LIVE_MINOR); connect.type = htobe32(LTTNG_VIEWER_CLIENT_COMMAND); - ret_len = lttng_live_send(ctx->control_sock, &cmd, sizeof(cmd)); - if (ret_len < 0) { - perror("[error] Error sending cmd"); - goto error; - } - assert(ret_len == sizeof(cmd)); + /* + * Merge the cmd and connection request to prevent a write-write + * sequence on the TCP socket. Otherwise, a delayed ACK will prevent the + * second write to be performed quickly in presence of Nagle's algorithm. + */ + memcpy(cmd_buf, &cmd, sizeof(cmd)); + memcpy(cmd_buf + sizeof(cmd), &connect, sizeof(connect)); - ret_len = lttng_live_send(ctx->control_sock, &connect, sizeof(connect)); + ret_len = lttng_live_send(ctx->control_sock, cmd_buf, cmd_buf_len); if (ret_len < 0) { - perror("[error] Error sending version"); + perror("[error] Error sending cmd for establishing session"); goto error; } - assert(ret_len == sizeof(connect)); + assert(ret_len == cmd_buf_len); ret_len = lttng_live_recv(ctx->control_sock, &connect, sizeof(connect)); if (ret_len == 0) { @@ -307,7 +310,7 @@ int lttng_live_list_sessions(struct lttng_live_ctx *ctx, const char *path) } cmd.cmd = htobe32(LTTNG_VIEWER_LIST_SESSIONS); - cmd.data_size = htobe64(0); + cmd.data_size = htobe64((uint64_t) 0); cmd.cmd_version = htobe32(0); ret_len = lttng_live_send(ctx->control_sock, &cmd, sizeof(cmd)); @@ -387,7 +390,7 @@ int lttng_live_ctf_trace_assign(struct lttng_live_viewer_stream *stream, if (!trace) { trace = g_new0(struct lttng_live_ctf_trace, 1); trace->ctf_trace_id = ctf_trace_id; - trace->streams = g_ptr_array_new(); + BT_INIT_LIST_HEAD(&trace->stream_list); g_hash_table_insert(stream->session->ctf_traces, &trace->ctf_trace_id, trace); @@ -395,8 +398,10 @@ int lttng_live_ctf_trace_assign(struct lttng_live_viewer_stream *stream, if (stream->metadata_flag) trace->metadata_stream = stream; + assert(!stream->in_trace); + stream->in_trace = 1; stream->ctf_trace = trace; - g_ptr_array_add(trace->streams, stream); + bt_list_add(&stream->trace_stream_node, &trace->stream_list); return ret; } @@ -423,6 +428,8 @@ int lttng_live_attach_session(struct lttng_live_ctx *ctx, uint64_t id) struct lttng_viewer_attach_session_request rq; struct lttng_viewer_attach_session_response rp; struct lttng_viewer_stream stream; + const size_t cmd_buf_len = sizeof(cmd) + sizeof(rq); + char cmd_buf[cmd_buf_len]; int ret, i; ssize_t ret_len; @@ -432,7 +439,7 @@ int lttng_live_attach_session(struct lttng_live_ctx *ctx, uint64_t id) } cmd.cmd = htobe32(LTTNG_VIEWER_ATTACH_SESSION); - cmd.data_size = htobe64(sizeof(rq)); + cmd.data_size = htobe64((uint64_t) sizeof(rq)); cmd.cmd_version = htobe32(0); memset(&rq, 0, sizeof(rq)); @@ -441,19 +448,20 @@ int lttng_live_attach_session(struct lttng_live_ctx *ctx, uint64_t id) // rq.seek = htobe32(LTTNG_VIEWER_SEEK_BEGINNING); rq.seek = htobe32(LTTNG_VIEWER_SEEK_LAST); - ret_len = lttng_live_send(ctx->control_sock, &cmd, sizeof(cmd)); - if (ret_len < 0) { - perror("[error] Error sending cmd"); - goto error; - } - assert(ret_len == sizeof(cmd)); + /* + * Merge the cmd and connection request to prevent a write-write + * sequence on the TCP socket. Otherwise, a delayed ACK will prevent the + * second write to be performed quickly in presence of Nagle's algorithm. + */ + memcpy(cmd_buf, &cmd, sizeof(cmd)); + memcpy(cmd_buf + sizeof(cmd), &rq, sizeof(rq)); - ret_len = lttng_live_send(ctx->control_sock, &rq, sizeof(rq)); + ret_len = lttng_live_send(ctx->control_sock, cmd_buf, cmd_buf_len); if (ret_len < 0) { - perror("[error] Error sending attach request"); + perror("[error] Error sending attach command and request"); goto error; } - assert(ret_len == sizeof(rq)); + assert(ret_len == cmd_buf_len); ret_len = lttng_live_recv(ctx->control_sock, &rp, sizeof(rp)); if (ret_len == 0) { @@ -500,18 +508,22 @@ int lttng_live_attach_session(struct lttng_live_ctx *ctx, uint64_t id) ret = 0; goto end; } - printf_verbose("Waiting for %" PRIu64 " streams:\n", - ctx->session->stream_count); - ctx->session->streams = g_new0(struct lttng_live_viewer_stream, - ctx->session->stream_count); + printf_verbose("Waiting for %d streams:\n", + be32toh(rp.streams_count)); for (i = 0; i < be32toh(rp.streams_count); i++) { - ret_len = lttng_live_recv(ctx->control_sock, &stream, sizeof(stream)); + struct lttng_live_viewer_stream *lvstream; + + lvstream = g_new0(struct lttng_live_viewer_stream, 1); + ret_len = lttng_live_recv(ctx->control_sock, &stream, + sizeof(stream)); if (ret_len == 0) { fprintf(stderr, "[error] Remote side has closed connection\n"); + g_free(lvstream); goto error; } if (ret_len < 0) { perror("[error] Error receiving stream"); + g_free(lvstream); goto error; } assert(ret_len == sizeof(stream)); @@ -521,21 +533,23 @@ int lttng_live_attach_session(struct lttng_live_ctx *ctx, uint64_t id) printf_verbose(" stream %" PRIu64 " : %s/%s\n", be64toh(stream.id), stream.path_name, stream.channel_name); - ctx->session->streams[i].id = be64toh(stream.id); - ctx->session->streams[i].session = ctx->session; + lvstream->id = be64toh(stream.id); + lvstream->session = ctx->session; - ctx->session->streams[i].mmap_size = 0; - ctx->session->streams[i].ctf_stream_id = -1ULL; + lvstream->mmap_size = 0; + lvstream->ctf_stream_id = -1ULL; if (be32toh(stream.metadata_flag)) { - ctx->session->streams[i].metadata_flag = 1; + lvstream->metadata_flag = 1; } - ret = lttng_live_ctf_trace_assign(&ctx->session->streams[i], + ret = lttng_live_ctf_trace_assign(lvstream, be64toh(stream.ctf_trace_id)); if (ret < 0) { + g_free(lvstream); goto error; } - + bt_list_add(&lvstream->session_stream_node, + &ctx->session->stream_list); } ret = 0; end: @@ -561,10 +575,11 @@ restart: id = g_array_index(ctx->session_ids, uint64_t, i); ret = lttng_live_get_new_streams(ctx, id); printf_verbose("Asking for new streams returns %d\n", ret); + if (lttng_live_should_quit()) { + ret = -1; + goto end; + } if (ret < 0) { - if (lttng_live_should_quit()) { - goto end; - } if (ret == -LTTNG_VIEWER_NEW_STREAMS_HUP) { printf_verbose("Session %" PRIu64 " closed\n", id); @@ -587,7 +602,12 @@ restart: nb_streams += ret; } } - ret = nb_streams; + if (ctx->session_ids->len == 0) { + /* All sessions are closed. */ + ret = -1; + } else { + ret = nb_streams; + } end: return ret; @@ -601,6 +621,11 @@ int append_metadata(struct lttng_live_ctx *ctx, struct lttng_live_viewer_stream *metadata; char *metadata_buf = NULL; + if (!viewer_stream->ctf_trace->handle) { + printf_verbose("append_metadata: trace handle not ready yet.\n"); + return 0; + } + printf_verbose("get_next_index: new metadata needed\n"); ret = get_new_metadata(ctx, viewer_stream, &metadata_buf); if (ret < 0) { @@ -641,6 +666,8 @@ int get_data_packet(struct lttng_live_ctx *ctx, struct lttng_viewer_cmd cmd; struct lttng_viewer_get_packet rq; struct lttng_viewer_trace_packet rp; + const size_t cmd_buf_len = sizeof(cmd) + sizeof(rq); + char cmd_buf[cmd_buf_len]; ssize_t ret_len; int ret; @@ -650,19 +677,8 @@ retry: goto end; } - /* - * Flush the output between attempts to grab a packet, thus - * ensuring we flush at least at the periodical timer period. - * This ensures the output remains reactive for interactive users and - * that the output is flushed when redirected to a file by the shell. - */ - if (fflush(LTTNG_LIVE_OUTPUT_FP) < 0) { - perror("fflush"); - goto error; - } - cmd.cmd = htobe32(LTTNG_VIEWER_GET_PACKET); - cmd.data_size = htobe64(sizeof(rq)); + cmd.data_size = htobe64((uint64_t) sizeof(rq)); cmd.cmd_version = htobe32(0); memset(&rq, 0, sizeof(rq)); @@ -670,19 +686,20 @@ retry: rq.offset = htobe64(offset); rq.len = htobe32(len); - ret_len = lttng_live_send(ctx->control_sock, &cmd, sizeof(cmd)); - if (ret_len < 0) { - perror("[error] Error sending cmd"); - goto error; - } - assert(ret_len == sizeof(cmd)); + /* + * Merge the cmd and connection request to prevent a write-write + * sequence on the TCP socket. Otherwise, a delayed ACK will prevent the + * second write to be performed quickly in presence of Nagle's algorithm. + */ + memcpy(cmd_buf, &cmd, sizeof(cmd)); + memcpy(cmd_buf + sizeof(cmd), &rq, sizeof(rq)); - ret_len = lttng_live_send(ctx->control_sock, &rq, sizeof(rq)); + ret_len = lttng_live_send(ctx->control_sock, cmd_buf, cmd_buf_len); if (ret_len < 0) { - perror("[error] Error sending get_data_packet request"); + perror("[error] Error sending get_data_packet cmd and request"); goto error; } - assert(ret_len == sizeof(rq)); + assert(ret_len == cmd_buf_len); ret_len = lttng_live_recv(ctx->control_sock, &rp, sizeof(rp)); if (ret_len == 0) { @@ -806,6 +823,8 @@ int get_one_metadata_packet(struct lttng_live_ctx *ctx, struct lttng_viewer_metadata_packet rp; char *data = NULL; ssize_t ret_len; + const size_t cmd_buf_len = sizeof(cmd) + sizeof(rq); + char cmd_buf[cmd_buf_len]; if (lttng_live_should_quit()) { ret = -1; @@ -814,22 +833,23 @@ int get_one_metadata_packet(struct lttng_live_ctx *ctx, rq.stream_id = htobe64(metadata_stream->id); cmd.cmd = htobe32(LTTNG_VIEWER_GET_METADATA); - cmd.data_size = htobe64(sizeof(rq)); + cmd.data_size = htobe64((uint64_t) sizeof(rq)); cmd.cmd_version = htobe32(0); - ret_len = lttng_live_send(ctx->control_sock, &cmd, sizeof(cmd)); - if (ret_len < 0) { - perror("[error] Error sending cmd"); - goto error; - } - assert(ret_len == sizeof(cmd)); + /* + * Merge the cmd and connection request to prevent a write-write + * sequence on the TCP socket. Otherwise, a delayed ACK will prevent the + * second write to be performed quickly in presence of Nagle's algorithm. + */ + memcpy(cmd_buf, &cmd, sizeof(cmd)); + memcpy(cmd_buf + sizeof(cmd), &rq, sizeof(rq)); - ret_len = lttng_live_send(ctx->control_sock, &rq, sizeof(rq)); + ret_len = lttng_live_send(ctx->control_sock, cmd_buf, cmd_buf_len); if (ret_len < 0) { - perror("[error] Error sending get_metadata request"); + perror("[error] Error sending get_metadata cmd and request"); goto error; } - assert(ret_len == sizeof(rq)); + assert(ret_len == cmd_buf_len); ret_len = lttng_live_recv(ctx->control_sock, &rp, sizeof(rp)); if (ret_len == 0) { @@ -942,6 +962,9 @@ int get_new_metadata(struct lttng_live_ctx *ctx, if (!len_read) { (void) poll(NULL, 0, ACTIVE_POLL_DELAY); } + if (ret < 0) { + break; /* Stop on error. */ + } } while (ret > 0 || !len_read); if (babeltrace_close_memstream(metadata_buf, &size, @@ -987,32 +1010,34 @@ int get_next_index(struct lttng_live_ctx *ctx, int ret; ssize_t ret_len; struct lttng_viewer_index *rp = &viewer_stream->current_index; + const size_t cmd_buf_len = sizeof(cmd) + sizeof(rq); + char cmd_buf[cmd_buf_len]; cmd.cmd = htobe32(LTTNG_VIEWER_GET_NEXT_INDEX); - cmd.data_size = htobe64(sizeof(rq)); + cmd.data_size = htobe64((uint64_t) sizeof(rq)); cmd.cmd_version = htobe32(0); memset(&rq, 0, sizeof(rq)); rq.stream_id = htobe64(viewer_stream->id); + /* + * Merge the cmd and connection request to prevent a write-write + * sequence on the TCP socket. Otherwise, a delayed ACK will prevent the + * second write to be performed quickly in presence of Nagle's algorithm. + */ + memcpy(cmd_buf, &cmd, sizeof(cmd)); + memcpy(cmd_buf + sizeof(cmd), &rq, sizeof(rq)); retry: if (lttng_live_should_quit()) { ret = -1; goto end; } - ret_len = lttng_live_send(ctx->control_sock, &cmd, sizeof(cmd)); + ret_len = lttng_live_send(ctx->control_sock, &cmd_buf, cmd_buf_len); if (ret_len < 0) { - perror("[error] Error sending cmd"); + perror("[error] Error sending get_next_index cmd and request"); goto error; } - assert(ret_len == sizeof(cmd)); - - ret_len = lttng_live_send(ctx->control_sock, &rq, sizeof(rq)); - if (ret_len < 0) { - perror("[error] Error sending get_next_index request"); - goto error; - } - assert(ret_len == sizeof(rq)); + assert(ret_len == cmd_buf_len); ret_len = lttng_live_recv(ctx->control_sock, rp, sizeof(*rp)); if (ret_len == 0) { @@ -1030,6 +1055,13 @@ retry: switch (be32toh(rp->status)) { case LTTNG_VIEWER_INDEX_INACTIVE: printf_verbose("get_next_index: inactive\n"); + + if (index->ts_cycles.timestamp_end == + be64toh(rp->timestamp_end)) { + /* Already seen this timestamp. */ + (void) poll(NULL, 0, ACTIVE_POLL_DELAY); + } + memset(index, 0, sizeof(struct packet_index)); index->ts_cycles.timestamp_end = be64toh(rp->timestamp_end); *stream_id = be64toh(rp->stream_id); @@ -1068,6 +1100,11 @@ retry: viewer_stream->id = -1ULL; index->offset = EOF; ctx->session->stream_count--; + viewer_stream->in_trace = 0; + bt_list_del(&viewer_stream->trace_stream_node); + bt_list_del(&viewer_stream->session_stream_node); + g_free(viewer_stream); + *stream_id = be64toh(rp->stream_id); break; case LTTNG_VIEWER_INDEX_ERR: fprintf(stderr, "[error] get_next_index: error\n"); @@ -1266,14 +1303,30 @@ retry: } if (cur_index->content_size == 0) { + /* Beacon packet index */ if (file_stream->parent.stream_class) { file_stream->parent.cycles_timestamp = cur_index->ts_cycles.timestamp_end; file_stream->parent.real_timestamp = ctf_get_real_timestamp( &file_stream->parent, cur_index->ts_cycles.timestamp_end); + + /* + * Duplicate the data from the previous index, because + * the one we just received is only a beacon with no + * relevant information except the timestamp_end. We + * don't need to keep this timestamp_end because we already + * updated the file_stream timestamps, so we only need + * to keep the last real index data as prev_index. That + * way, we keep the original prev timestamps and + * discarded events counter. This is the same behaviour + * as if we were reading a local trace, we would not + * have fake indexes between real indexes. + */ + memcpy(cur_index, prev_index, sizeof(struct packet_index)); } } else { + /* Real packet index */ 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( @@ -1287,12 +1340,33 @@ retry: ctf_update_current_packet_index(&file_stream->parent, prev_index, cur_index); + /* + * We need to check if we are in trace read or called + * from packet indexing. In this last case, the + * collection is not there, so we cannot print the + * timestamps. + */ + if ((&file_stream->parent)->stream_class->trace->parent.collection) { + ctf_print_discarded_lost(stderr, &file_stream->parent); + } + file_stream->parent.cycles_timestamp = cur_index->ts_cycles.timestamp_begin; file_stream->parent.real_timestamp = cur_index->ts_real.timestamp_begin; } + /* + * Flush the output between attempts to grab a packet, thus + * ensuring we flush at least at the periodical timer period. + * This ensures the output remains reactive for interactive users and + * that the output is flushed when redirected to a file by the shell. + */ + if (fflush(LTTNG_LIVE_OUTPUT_FP) < 0) { + perror("fflush"); + goto end; + } + if (pos->packet_size == 0 || pos->offset == EOF) { goto end; } @@ -1332,7 +1406,7 @@ int lttng_live_create_viewer_session(struct lttng_live_ctx *ctx) } cmd.cmd = htobe32(LTTNG_VIEWER_CREATE_SESSION); - cmd.data_size = htobe64(0); + cmd.data_size = htobe64((uint64_t) 0); cmd.cmd_version = htobe32(0); ret_len = lttng_live_send(ctx->control_sock, &cmd, sizeof(cmd)); @@ -1371,10 +1445,22 @@ int del_traces(gpointer key, gpointer value, gpointer user_data) struct bt_context *bt_ctx = user_data; struct lttng_live_ctf_trace *trace = value; int ret; + struct lttng_live_viewer_stream *lvstream, *tmp; - ret = bt_context_remove_trace(bt_ctx, trace->trace_id); - if (ret < 0) - fprintf(stderr, "[error] removing trace from context\n"); + /* + * We don't have ownership of the live viewer stream, just + * remove them from our list. + */ + bt_list_for_each_entry_safe(lvstream, tmp, &trace->stream_list, + trace_stream_node) { + lvstream->in_trace = 0; + bt_list_del(&lvstream->trace_stream_node); + } + if (trace->in_use) { + ret = bt_context_remove_trace(bt_ctx, trace->trace_id); + if (ret < 0) + fprintf(stderr, "[error] removing trace from context\n"); + } /* remove the key/value pair from the HT. */ return 1; @@ -1384,7 +1470,7 @@ static int add_one_trace(struct lttng_live_ctx *ctx, struct lttng_live_ctf_trace *trace) { - int i, ret; + int ret; struct bt_context *bt_ctx = ctx->bt_ctx; struct lttng_live_viewer_stream *stream; struct bt_mmap_stream *new_mmap_stream; @@ -1408,9 +1494,7 @@ int add_one_trace(struct lttng_live_ctx *ctx, BT_INIT_LIST_HEAD(&mmap_list.head); - for (i = 0; i < trace->streams->len; i++) { - stream = g_ptr_array_index(trace->streams, i); - + bt_list_for_each_entry(stream, &trace->stream_list, trace_stream_node) { if (!stream->metadata_flag) { new_mmap_stream = zmalloc(sizeof(struct bt_mmap_stream)); new_mmap_stream->priv = (void *) stream; @@ -1515,6 +1599,8 @@ int lttng_live_get_new_streams(struct lttng_live_ctx *ctx, uint64_t id) int ret, i, nb_streams = 0; ssize_t ret_len; uint32_t stream_count; + const size_t cmd_buf_len = sizeof(cmd) + sizeof(rq); + char cmd_buf[cmd_buf_len]; if (lttng_live_should_quit()) { ret = -1; @@ -1522,25 +1608,26 @@ int lttng_live_get_new_streams(struct lttng_live_ctx *ctx, uint64_t id) } cmd.cmd = htobe32(LTTNG_VIEWER_GET_NEW_STREAMS); - cmd.data_size = htobe64(sizeof(rq)); + cmd.data_size = htobe64((uint64_t) sizeof(rq)); cmd.cmd_version = htobe32(0); memset(&rq, 0, sizeof(rq)); rq.session_id = htobe64(id); - ret_len = lttng_live_send(ctx->control_sock, &cmd, sizeof(cmd)); - if (ret_len < 0) { - perror("[error] Error sending cmd"); - goto error; - } - assert(ret_len == sizeof(cmd)); + /* + * Merge the cmd and connection request to prevent a write-write + * sequence on the TCP socket. Otherwise, a delayed ACK will prevent the + * second write to be performed quickly in presence of Nagle's algorithm. + */ + memcpy(cmd_buf, &cmd, sizeof(cmd)); + memcpy(cmd_buf + sizeof(cmd), &rq, sizeof(rq)); - ret_len = lttng_live_send(ctx->control_sock, &rq, sizeof(rq)); + ret_len = lttng_live_send(ctx->control_sock, cmd_buf, cmd_buf_len); if (ret_len < 0) { - perror("[error] Error sending get_new_streams request"); + perror("[error] Error sending get_new_streams cmd and request"); goto error; } - assert(ret_len == sizeof(rq)); + assert(ret_len == cmd_buf_len); ret_len = lttng_live_recv(ctx->control_sock, &rp, sizeof(rp)); if (ret_len == 0) { @@ -1582,18 +1669,22 @@ int lttng_live_get_new_streams(struct lttng_live_ctx *ctx, uint64_t id) ret = 0; goto end; } - printf_verbose("Waiting for %" PRIu64 " streams:\n", - ctx->session->stream_count); - ctx->session->streams = g_new0(struct lttng_live_viewer_stream, - ctx->session->stream_count); + printf_verbose("Waiting for %d streams:\n", stream_count); + for (i = 0; i < stream_count; i++) { - ret_len = lttng_live_recv(ctx->control_sock, &stream, sizeof(stream)); + struct lttng_live_viewer_stream *lvstream; + + lvstream = g_new0(struct lttng_live_viewer_stream, 1); + ret_len = lttng_live_recv(ctx->control_sock, &stream, + sizeof(stream)); if (ret_len == 0) { fprintf(stderr, "[error] Remote side has closed connection\n"); + g_free(lvstream); goto error; } if (ret_len < 0) { perror("[error] Error receiving stream"); + g_free(lvstream); goto error; } assert(ret_len == sizeof(stream)); @@ -1603,22 +1694,24 @@ int lttng_live_get_new_streams(struct lttng_live_ctx *ctx, uint64_t id) printf_verbose(" stream %" PRIu64 " : %s/%s\n", be64toh(stream.id), stream.path_name, stream.channel_name); - ctx->session->streams[i].id = be64toh(stream.id); - ctx->session->streams[i].session = ctx->session; + lvstream->id = be64toh(stream.id); + lvstream->session = ctx->session; - ctx->session->streams[i].mmap_size = 0; - ctx->session->streams[i].ctf_stream_id = -1ULL; + lvstream->mmap_size = 0; + lvstream->ctf_stream_id = -1ULL; if (be32toh(stream.metadata_flag)) { - ctx->session->streams[i].metadata_flag = 1; + lvstream->metadata_flag = 1; } - ret = lttng_live_ctf_trace_assign(&ctx->session->streams[i], + ret = lttng_live_ctf_trace_assign(lvstream, be64toh(stream.ctf_trace_id)); if (ret < 0) { + g_free(lvstream); goto error; } nb_streams++; - + bt_list_add(&lvstream->session_stream_node, + &ctx->session->stream_list); } ret = nb_streams; end: @@ -1649,7 +1742,7 @@ int lttng_live_read(struct lttng_live_ctx *ctx) fmt_write = bt_lookup_format(g_quark_from_static_string("text")); if (!fmt_write) { fprintf(stderr, "[error] ctf-text error\n"); - goto end; + goto end_free; } td_write = fmt_write->open_trace(NULL, O_RDWR, NULL, NULL); @@ -1701,6 +1794,7 @@ int lttng_live_read(struct lttng_live_ctx *ctx) } ret = ask_new_streams(ctx); if (ret < 0) { + ret = 0; goto end_free; } if (!ctx->session->stream_count) { @@ -1718,10 +1812,10 @@ int lttng_live_read(struct lttng_live_ctx *ctx) if (!iter) { if (lttng_live_should_quit()) { ret = 0; - goto end; + goto end_free; } fprintf(stderr, "[error] Iterator creation error\n"); - goto end; + goto end_free; } for (;;) { if (lttng_live_should_quit()) { @@ -1754,6 +1848,8 @@ int lttng_live_read(struct lttng_live_ctx *ctx) } end_free: + g_hash_table_foreach_remove(ctx->session->ctf_traces, + del_traces, ctx->bt_ctx); bt_context_put(ctx->bt_ctx); end: if (lttng_live_should_quit()) {