X-Git-Url: http://git.efficios.com/?p=babeltrace.git;a=blobdiff_plain;f=formats%2Flttng-live%2Flttng-live-comm.c;h=d26d83eb4bbb01a9d383585d57fab2761543fb4a;hp=9feb7185c76d0d1cdea4c1e5dfd79dc300849dbc;hb=5ea98697b9f786ac73fd5152d96aebe02906a10d;hpb=e29395a8120118fa59d3e464442135b0b1940fc7 diff --git a/formats/lttng-live/lttng-live-comm.c b/formats/lttng-live/lttng-live-comm.c index 9feb7185..d26d83eb 100644 --- a/formats/lttng-live/lttng-live-comm.c +++ b/formats/lttng-live/lttng-live-comm.c @@ -26,7 +26,6 @@ #include #include #include -#include #include #include #include @@ -38,6 +37,7 @@ #include #include +#include #include #include #include @@ -51,6 +51,8 @@ #include #include +#include +#include #include "lttng-live.h" #include "lttng-viewer-abi.h" @@ -69,7 +71,7 @@ static void ctf_live_packet_seek(struct bt_stream_pos *stream_pos, size_t index, int whence); -static void add_traces(gpointer key, gpointer value, gpointer user_data); +static int add_traces(struct lttng_live_ctx *ctx); static int del_traces(gpointer key, gpointer value, gpointer user_data); static int get_new_metadata(struct lttng_live_ctx *ctx, struct lttng_live_viewer_stream *viewer_stream, @@ -102,7 +104,7 @@ ssize_t lttng_live_send(int fd, const void *buf, size_t len) ssize_t ret; do { - ret = send(fd, buf, len, MSG_NOSIGNAL); + ret = bt_send_nosigpipe(fd, buf, len); } while (ret < 0 && errno == EINTR); return ret; } @@ -133,7 +135,7 @@ int lttng_live_connect_viewer(struct lttng_live_ctx *ctx) server_addr.sin_family = AF_INET; server_addr.sin_port = htons(ctx->port); server_addr.sin_addr = *((struct in_addr *) host->h_addr); - bzero(&(server_addr.sin_zero), 8); + memset(&(server_addr.sin_zero), 0, 8); if (connect(ctx->control_sock, (struct sockaddr *) &server_addr, sizeof(struct sockaddr)) == -1) { @@ -201,6 +203,18 @@ int lttng_live_establish_connection(struct lttng_live_ctx *ctx) be64toh(connect.viewer_session_id)); printf_verbose("Relayd version : %u.%u\n", be32toh(connect.major), be32toh(connect.minor)); + + if (LTTNG_LIVE_MAJOR != be32toh(connect.major)) { + fprintf(stderr, "[error] Incompatible lttng-relayd protocol\n"); + goto error; + } + /* Use the smallest protocol version implemented. */ + if (LTTNG_LIVE_MINOR > be32toh(connect.minor)) { + ctx->minor = be32toh(connect.minor); + } else { + ctx->minor = LTTNG_LIVE_MINOR; + } + ctx->major = LTTNG_LIVE_MAJOR; ret = 0; end: return ret; @@ -250,9 +264,9 @@ void update_session_list(GPtrArray *session_list, char *hostname, for (i = 0; i < session_list->len; i++) { relay_session = g_ptr_array_index(session_list, i); - if ((strncmp(relay_session->hostname, hostname, NAME_MAX) == 0) && + if ((strncmp(relay_session->hostname, hostname, MAXNAMLEN) == 0) && strncmp(relay_session->name, - session_name, NAME_MAX) == 0) { + session_name, MAXNAMLEN) == 0) { relay_session->streams += streams; if (relay_session->clients < clients) relay_session->clients = clients; @@ -264,8 +278,8 @@ void update_session_list(GPtrArray *session_list, char *hostname, return; relay_session = g_new0(struct lttng_live_relay_session, 1); - relay_session->hostname = strndup(hostname, NAME_MAX); - relay_session->name = strndup(session_name, NAME_MAX); + relay_session->hostname = bt_strndup(hostname, MAXNAMLEN); + relay_session->name = bt_strndup(session_name, MAXNAMLEN); relay_session->clients = clients; relay_session->streams = streams; relay_session->timer = timer; @@ -339,8 +353,8 @@ int lttng_live_list_sessions(struct lttng_live_ctx *ctx, const char *path) be32toh(lsession.live_timer)); } else { if ((strncmp(lsession.session_name, ctx->session_name, - NAME_MAX) == 0) && (strncmp(lsession.hostname, - ctx->traced_hostname, NAME_MAX) == 0)) { + MAXNAMLEN) == 0) && (strncmp(lsession.hostname, + ctx->traced_hostname, MAXNAMLEN) == 0)) { printf_verbose("Reading from session %" PRIu64 "\n", session_id); g_array_append_val(ctx->session_ids, @@ -369,13 +383,13 @@ int lttng_live_ctf_trace_assign(struct lttng_live_viewer_stream *stream, int ret = 0; trace = g_hash_table_lookup(stream->session->ctf_traces, - (gpointer) ctf_trace_id); + &ctf_trace_id); if (!trace) { trace = g_new0(struct lttng_live_ctf_trace, 1); trace->ctf_trace_id = ctf_trace_id; trace->streams = g_ptr_array_new(); g_hash_table_insert(stream->session->ctf_traces, - (gpointer) ctf_trace_id, + &trace->ctf_trace_id, trace); } if (stream->metadata_flag) @@ -510,8 +524,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].first_read = 1; 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; @@ -531,18 +545,22 @@ error: return -1; } +/* + * Ask the relay for new streams. + * + * Returns the number of new streams received or a negative value on error. + */ static int ask_new_streams(struct lttng_live_ctx *ctx) { - int i, ret = 0; + int i, ret = 0, nb_streams = 0; uint64_t id; restart: for (i = 0; i < ctx->session_ids->len; i++) { 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); + printf_verbose("Asking for new streams returns %d\n", ret); if (ret < 0) { if (lttng_live_should_quit()) { goto end; @@ -565,8 +583,11 @@ restart: ret = -1; goto end; } + } else { + nb_streams += ret; } } + ret = nb_streams; end: return ret; @@ -661,8 +682,8 @@ retry: goto error; } if (ret_len != sizeof(rp)) { - fprintf(stderr, "[error] get_data_packet: expected %" PRId64 - ", received %" PRId64 "\n", sizeof(rp), + fprintf(stderr, "[error] get_data_packet: expected %zu" + ", received %zd\n", sizeof(rp), ret_len); goto error; } @@ -689,10 +710,14 @@ retry: if (rp.flags & LTTNG_VIEWER_FLAG_NEW_STREAM) { printf_verbose("get_data_packet: new streams needed\n"); ret = ask_new_streams(ctx); - if (ret < 0) + if (ret < 0) { goto error; - g_hash_table_foreach(ctx->session->ctf_traces, - add_traces, ctx->bt_ctx); + } else if (ret > 0) { + ret = add_traces(ctx); + if (ret < 0) { + goto error; + } + } } if (rp.flags & (LTTNG_VIEWER_FLAG_NEW_METADATA | LTTNG_VIEWER_FLAG_NEW_STREAM)) { @@ -874,7 +899,7 @@ int get_new_metadata(struct lttng_live_ctx *ctx, { int ret = 0; struct lttng_live_viewer_stream *metadata_stream; - size_t size, len_read = 0;; + size_t size, len_read = 0; metadata_stream = viewer_stream->ctf_trace->metadata_stream; if (!metadata_stream) { @@ -889,6 +914,10 @@ int get_new_metadata(struct lttng_live_ctx *ctx, } do { + if (lttng_live_should_quit()) { + ret = -1; + goto error; + } /* * get_one_metadata_packet returns the number of bytes * received, 0 when we have received everything, a @@ -911,6 +940,24 @@ error: 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. * @@ -919,13 +966,13 @@ error: 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_index rp; 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); @@ -953,7 +1000,7 @@ retry: } 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; @@ -962,37 +1009,40 @@ retry: 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)); - 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", - 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; } - 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) + if (ret < 0) { goto error; - g_hash_table_foreach(ctx->session->ctf_traces, - add_traces, ctx->bt_ctx); + } else if (ret > 0) { + ret = add_traces(ctx); + if (ret < 0) { + goto error; + } + } } break; case LTTNG_VIEWER_INDEX_RETRY: @@ -1020,6 +1070,84 @@ error: 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) @@ -1029,14 +1157,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; + uint64_t stream_id = -1ULL; 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; + 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); @@ -1064,13 +1199,46 @@ retry: 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; } @@ -1084,19 +1252,23 @@ retry: } 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; - 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 { - /* 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); @@ -1125,34 +1297,9 @@ retry: } 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; @@ -1220,15 +1367,14 @@ int del_traces(gpointer key, gpointer value, gpointer user_data) } static -void add_traces(gpointer key, gpointer value, gpointer user_data) +int add_one_trace(struct lttng_live_ctx *ctx, + struct lttng_live_ctf_trace *trace) { int i, ret; - struct bt_context *bt_ctx = user_data; - struct lttng_live_ctf_trace *trace = value; + struct bt_context *bt_ctx = ctx->bt_ctx; struct lttng_live_viewer_stream *stream; struct bt_mmap_stream *new_mmap_stream; struct bt_mmap_stream_list mmap_list; - struct lttng_live_ctx *ctx = NULL; struct bt_trace_descriptor *td; struct bt_trace_handle *handle; @@ -1241,14 +1387,15 @@ void add_traces(gpointer key, gpointer value, gpointer user_data) * times the same traces. * If a trace is already in the context, we just skip this function. */ - if (trace->in_use) - return; + if (trace->in_use) { + ret = 0; + goto end; + } BT_INIT_LIST_HEAD(&mmap_list.head); for (i = 0; i < trace->streams->len; i++) { stream = g_ptr_array_index(trace->streams, i); - ctx = stream->session->ctx; if (!stream->metadata_flag) { new_mmap_stream = zmalloc(sizeof(struct bt_mmap_stream)); @@ -1284,6 +1431,7 @@ void add_traces(gpointer key, gpointer value, gpointer user_data) if (!trace->metadata_fp) { fprintf(stderr, "[error] No metadata stream opened\n"); + ret = -1; goto end_free; } @@ -1291,6 +1439,7 @@ void add_traces(gpointer key, gpointer value, gpointer user_data) ctf_live_packet_seek, &mmap_list, trace->metadata_fp); if (ret < 0) { fprintf(stderr, "[error] Error adding trace\n"); + ret = -1; goto end_free; } trace->metadata_stream->metadata_len = 0; @@ -1312,16 +1461,44 @@ void add_traces(gpointer key, gpointer value, gpointer user_data) end_free: bt_context_put(bt_ctx); end: - return; + return ret; +} + +static +int add_traces(struct lttng_live_ctx *ctx) +{ + int ret; + struct lttng_live_ctf_trace *trace; + GHashTableIter it; + gpointer key; + gpointer value; + + g_hash_table_iter_init(&it, ctx->session->ctf_traces); + while (g_hash_table_iter_next(&it, &key, &value)) { + trace = (struct lttng_live_ctf_trace *) value; + ret = add_one_trace(ctx, trace); + if (ret < 0) { + goto end; + } + } + + ret = 0; + +end: + return ret; } +/* + * Request new streams for a session. + * Returns the number of streams received or a negative value on error. + */ int lttng_live_get_new_streams(struct lttng_live_ctx *ctx, uint64_t id) { struct lttng_viewer_cmd cmd; struct lttng_viewer_new_streams_request rq; struct lttng_viewer_new_streams_response rp; struct lttng_viewer_stream stream; - int ret, i; + int ret, i, nb_streams = 0; ssize_t ret_len; uint32_t stream_count; @@ -1415,8 +1592,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].first_read = 1; 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; @@ -1426,9 +1603,10 @@ int lttng_live_get_new_streams(struct lttng_live_ctx *ctx, uint64_t id) if (ret < 0) { goto error; } + nb_streams++; } - ret = 0; + ret = nb_streams; end: return ret; @@ -1479,7 +1657,7 @@ int lttng_live_read(struct lttng_live_ctx *ctx) for (i = 0; i < ctx->session_ids->len; i++) { id = g_array_index(ctx->session_ids, uint64_t, i); - printf_verbose("Attaching to session %lu\n", id); + printf_verbose("Attaching to session %" PRIu64 "\n", id); ret = lttng_live_attach_session(ctx, id); printf_verbose("Attaching session returns %d\n", ret); if (ret < 0) { @@ -1516,8 +1694,10 @@ int lttng_live_read(struct lttng_live_ctx *ctx) } } - g_hash_table_foreach(ctx->session->ctf_traces, add_traces, - ctx->bt_ctx); + ret = add_traces(ctx); + if (ret < 0) { + goto end_free; + } begin_pos.type = BT_SEEK_BEGIN; iter = bt_ctf_iter_create(ctx->bt_ctx, &begin_pos, NULL);