X-Git-Url: http://git.efficios.com/?p=babeltrace.git;a=blobdiff_plain;f=formats%2Flttng-live%2Flttng-live-comm.c;h=065af49fb06f5f1be0d02f8a39e9919ddafef50b;hp=8cdac9188577bc95df8f0dbbc2855ffd80c6628a;hb=1ab659eb38221a7908ee4693616ba43cca651a91;hpb=6baf4b8ee0e699b92dc697e01287d56577bd4397 diff --git a/formats/lttng-live/lttng-live-comm.c b/formats/lttng-live/lttng-live-comm.c index 8cdac918..065af49f 100644 --- a/formats/lttng-live/lttng-live-comm.c +++ b/formats/lttng-live/lttng-live-comm.c @@ -390,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); @@ -398,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; } @@ -506,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)); @@ -527,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: @@ -594,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; @@ -949,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, @@ -1084,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"); @@ -1424,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; @@ -1437,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; @@ -1461,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; @@ -1638,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)); @@ -1659,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: @@ -1705,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); @@ -1775,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()) { @@ -1811,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()) {