X-Git-Url: http://git.efficios.com/?p=babeltrace.git;a=blobdiff_plain;f=formats%2Flttng-live%2Flttng-live-comm.c;h=d26d83eb4bbb01a9d383585d57fab2761543fb4a;hp=3970ac22ee6ed43fa33dae4664061cc478811836;hb=5ea98697b9f786ac73fd5152d96aebe02906a10d;hpb=85bb466398bbf32d8334066011bfe7be7ba71d8a diff --git a/formats/lttng-live/lttng-live-comm.c b/formats/lttng-live/lttng-live-comm.c index 3970ac22..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; } @@ -113,34 +115,42 @@ int lttng_live_connect_viewer(struct lttng_live_ctx *ctx) struct sockaddr_in server_addr; int ret; - host = gethostbyname(ctx->relay_hostname); - if (!host) { + if (lttng_live_should_quit()) { ret = -1; goto end; } + host = gethostbyname(ctx->relay_hostname); + if (!host) { + fprintf(stderr, "[error] Cannot lookup hostname %s\n", + ctx->relay_hostname); + goto error; + } + if ((ctx->control_sock = socket(AF_INET, SOCK_STREAM, 0)) == -1) { perror("Socket"); - ret = -1; - goto end; + goto error; } 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) { perror("Connect"); - ret = -1; - goto end; + goto error; } ret = 0; end: return ret; + +error: + fprintf(stderr, "[error] Connection failed\n"); + return -1; } int lttng_live_establish_connection(struct lttng_live_ctx *ctx) @@ -150,6 +160,11 @@ int lttng_live_establish_connection(struct lttng_live_ctx *ctx) int ret; ssize_t ret_len; + if (lttng_live_should_quit()) { + ret = -1; + goto end; + } + cmd.cmd = htobe32(LTTNG_VIEWER_CONNECT); cmd.data_size = sizeof(connect); cmd.cmd_version = 0; @@ -162,7 +177,6 @@ int lttng_live_establish_connection(struct lttng_live_ctx *ctx) ret_len = lttng_live_send(ctx->control_sock, &cmd, sizeof(cmd)); if (ret_len < 0) { perror("[error] Error sending cmd"); - ret = ret_len; goto error; } assert(ret_len == sizeof(cmd)); @@ -170,7 +184,6 @@ int lttng_live_establish_connection(struct lttng_live_ctx *ctx) ret_len = lttng_live_send(ctx->control_sock, &connect, sizeof(connect)); if (ret_len < 0) { perror("[error] Error sending version"); - ret = ret_len; goto error; } assert(ret_len == sizeof(connect)); @@ -178,12 +191,10 @@ int lttng_live_establish_connection(struct lttng_live_ctx *ctx) ret_len = lttng_live_recv(ctx->control_sock, &connect, sizeof(connect)); if (ret_len == 0) { fprintf(stderr, "[error] Remote side has closed connection\n"); - ret = -1; goto error; } if (ret_len < 0) { perror("[error] Error receiving version"); - ret = ret_len; goto error; } assert(ret_len == sizeof(connect)); @@ -193,10 +204,24 @@ int lttng_live_establish_connection(struct lttng_live_ctx *ctx) 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; error: - return ret; + fprintf(stderr, "[error] Unable to establish connection\n"); + return -1; } static @@ -239,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; @@ -253,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; @@ -271,6 +296,11 @@ int lttng_live_list_sessions(struct lttng_live_ctx *ctx, const char *path) uint64_t session_id; GPtrArray *session_list = NULL; + if (lttng_live_should_quit()) { + ret = -1; + goto end; + } + if (strlen(ctx->session_name) == 0) { print_list = 1; session_list = g_ptr_array_new(); @@ -283,7 +313,6 @@ int lttng_live_list_sessions(struct lttng_live_ctx *ctx, const char *path) ret_len = lttng_live_send(ctx->control_sock, &cmd, sizeof(cmd)); if (ret_len < 0) { perror("[error] Error sending cmd"); - ret = ret_len; goto error; } assert(ret_len == sizeof(cmd)); @@ -291,12 +320,10 @@ int lttng_live_list_sessions(struct lttng_live_ctx *ctx, const char *path) ret_len = lttng_live_recv(ctx->control_sock, &list, sizeof(list)); if (ret_len == 0) { fprintf(stderr, "[error] Remote side has closed connection\n"); - ret = -1; goto error; } if (ret_len < 0) { perror("[error] Error receiving session list"); - ret = ret_len; goto error; } assert(ret_len == sizeof(list)); @@ -306,12 +333,10 @@ int lttng_live_list_sessions(struct lttng_live_ctx *ctx, const char *path) ret_len = lttng_live_recv(ctx->control_sock, &lsession, sizeof(lsession)); if (ret_len == 0) { fprintf(stderr, "[error] Remote side has closed connection\n"); - ret = -1; goto error; } if (ret_len < 0) { perror("[error] Error receiving session"); - ret = ret_len; goto error; } assert(ret_len == sizeof(lsession)); @@ -328,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, @@ -342,11 +367,13 @@ int lttng_live_list_sessions(struct lttng_live_ctx *ctx, const char *path) print_session_list(session_list, path); free_session_list(session_list); } - ret = 0; +end: + return ret; error: - return ret; + fprintf(stderr, "[error] Unable to list sessions\n"); + return -1; } int lttng_live_ctf_trace_assign(struct lttng_live_viewer_stream *stream, @@ -356,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) @@ -399,6 +426,11 @@ int lttng_live_attach_session(struct lttng_live_ctx *ctx, uint64_t id) int ret, i; ssize_t ret_len; + if (lttng_live_should_quit()) { + ret = -1; + goto end; + } + cmd.cmd = htobe32(LTTNG_VIEWER_ATTACH_SESSION); cmd.data_size = sizeof(rq); cmd.cmd_version = 0; @@ -412,7 +444,6 @@ int lttng_live_attach_session(struct lttng_live_ctx *ctx, uint64_t id) ret_len = lttng_live_send(ctx->control_sock, &cmd, sizeof(cmd)); if (ret_len < 0) { perror("[error] Error sending cmd"); - ret = ret_len; goto error; } assert(ret_len == sizeof(cmd)); @@ -420,7 +451,6 @@ int lttng_live_attach_session(struct lttng_live_ctx *ctx, uint64_t id) ret_len = lttng_live_send(ctx->control_sock, &rq, sizeof(rq)); if (ret_len < 0) { perror("[error] Error sending attach request"); - ret = ret_len; goto error; } assert(ret_len == sizeof(rq)); @@ -428,12 +458,10 @@ int lttng_live_attach_session(struct lttng_live_ctx *ctx, uint64_t id) ret_len = lttng_live_recv(ctx->control_sock, &rp, sizeof(rp)); if (ret_len == 0) { fprintf(stderr, "[error] Remote side has closed connection\n"); - ret = -1; goto error; } if (ret_len < 0) { perror("[error] Error receiving attach response"); - ret = ret_len; goto error; } assert(ret_len == sizeof(rp)); @@ -446,25 +474,20 @@ int lttng_live_attach_session(struct lttng_live_ctx *ctx, uint64_t id) goto end; case LTTNG_VIEWER_ATTACH_ALREADY: fprintf(stderr, "[error] There is already a viewer attached to this session\n"); - ret = -1; - goto end; + goto error; case LTTNG_VIEWER_ATTACH_NOT_LIVE: fprintf(stderr, "[error] Not a live session\n"); - ret = -1; - goto end; + goto error; case LTTNG_VIEWER_ATTACH_SEEK_ERR: fprintf(stderr, "[error] Wrong seek parameter\n"); - ret = -1; - goto end; + goto error; default: fprintf(stderr, "[error] Unknown attach return code %u\n", be32toh(rp.status)); - ret = -1; - goto end; + goto error; } if (be32toh(rp.status) != LTTNG_VIEWER_ATTACH_OK) { - ret = -1; - goto end; + goto error; } ctx->session->stream_count += be32toh(rp.streams_count); @@ -485,12 +508,10 @@ int lttng_live_attach_session(struct lttng_live_ctx *ctx, uint64_t id) ret_len = lttng_live_recv(ctx->control_sock, &stream, sizeof(stream)); if (ret_len == 0) { fprintf(stderr, "[error] Remote side has closed connection\n"); - ret = -1; goto error; } if (ret_len < 0) { perror("[error] Error receiving stream"); - ret = ret_len; goto error; } assert(ret_len == sizeof(stream)); @@ -503,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; @@ -517,25 +538,33 @@ int lttng_live_attach_session(struct lttng_live_ctx *ctx, uint64_t id) } ret = 0; - end: -error: return ret; + +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; + } if (ret == -LTTNG_VIEWER_NEW_STREAMS_HUP) { printf_verbose("Session %" PRIu64 " closed\n", id); @@ -554,8 +583,11 @@ restart: ret = -1; goto end; } + } else { + nb_streams += ret; } } + ret = nb_streams; end: return ret; @@ -613,20 +645,22 @@ int get_data_packet(struct lttng_live_ctx *ctx, int ret; retry: + if (lttng_live_should_quit()) { + ret = -1; + goto end; + } cmd.cmd = htobe32(LTTNG_VIEWER_GET_PACKET); cmd.data_size = sizeof(rq); cmd.cmd_version = 0; memset(&rq, 0, sizeof(rq)); rq.stream_id = htobe64(stream->id); - /* Already in big endian. */ - rq.offset = offset; + 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"); - ret = ret_len; goto error; } assert(ret_len == sizeof(cmd)); @@ -634,7 +668,6 @@ retry: ret_len = lttng_live_send(ctx->control_sock, &rq, sizeof(rq)); if (ret_len < 0) { perror("[error] Error sending get_data_packet request"); - ret = ret_len; goto error; } assert(ret_len == sizeof(rq)); @@ -642,19 +675,16 @@ retry: ret_len = lttng_live_recv(ctx->control_sock, &rp, sizeof(rp)); if (ret_len == 0) { fprintf(stderr, "[error] Remote side has closed connection\n"); - ret = -1; goto error; } if (ret_len < 0) { perror("[error] Error receiving data response"); - ret = ret_len; goto error; } if (ret_len != sizeof(rp)) { - fprintf(stderr, "[error] get_data_packet: expected %" PRId64 - ", received %" PRId64 "\n", ret_len, - sizeof(rp)); - ret = -1; + fprintf(stderr, "[error] get_data_packet: expected %zu" + ", received %zd\n", sizeof(rp), + ret_len); goto error; } @@ -667,9 +697,9 @@ retry: "\n", len); break; case LTTNG_VIEWER_GET_PACKET_RETRY: + /* Unimplemented by relay daemon */ printf_verbose("get_data_packet: retry\n"); - ret = -1; - goto end; + goto error; case LTTNG_VIEWER_GET_PACKET_ERR: if (rp.flags & LTTNG_VIEWER_FLAG_NEW_METADATA) { printf_verbose("get_data_packet: new metadata needed\n"); @@ -680,31 +710,31 @@ 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)) { goto retry; } fprintf(stderr, "[error] get_data_packet: error\n"); - ret = -1; - goto end; + goto error; case LTTNG_VIEWER_GET_PACKET_EOF: ret = -2; - goto error; + goto end; default: printf_verbose("get_data_packet: unknown\n"); - assert(0); - ret = -1; - goto end; + goto error; } - if (len <= 0) { - ret = -1; - goto end; + if (len == 0) { + goto error; } if (len > stream->mmap_size) { @@ -716,7 +746,6 @@ retry: ret = munmap_align(pos->base_mma); if (ret) { perror("[error] Unable to unmap old base"); - ret = -1; goto error; } pos->base_mma = NULL; @@ -727,7 +756,6 @@ retry: if (pos->base_mma == MAP_FAILED) { perror("[error] mmap error"); pos->base_mma = NULL; - ret = -1; goto error; } @@ -740,19 +768,19 @@ retry: mmap_align_addr(pos->base_mma), len); if (ret_len == 0) { fprintf(stderr, "[error] Remote side has closed connection\n"); - ret = -1; goto error; } if (ret_len < 0) { perror("[error] Error receiving trace packet"); - ret = ret_len; goto error; } assert(ret_len == len); ret = 0; end: -error: return ret; + +error: + return -1; } static @@ -767,6 +795,11 @@ int get_one_metadata_packet(struct lttng_live_ctx *ctx, char *data = NULL; ssize_t ret_len; + if (lttng_live_should_quit()) { + ret = -1; + goto end; + } + rq.stream_id = htobe64(metadata_stream->id); cmd.cmd = htobe32(LTTNG_VIEWER_GET_METADATA); cmd.data_size = sizeof(rq); @@ -775,7 +808,6 @@ int get_one_metadata_packet(struct lttng_live_ctx *ctx, ret_len = lttng_live_send(ctx->control_sock, &cmd, sizeof(cmd)); if (ret_len < 0) { perror("[error] Error sending cmd"); - ret = ret_len; goto error; } assert(ret_len == sizeof(cmd)); @@ -783,7 +815,6 @@ int get_one_metadata_packet(struct lttng_live_ctx *ctx, ret_len = lttng_live_send(ctx->control_sock, &rq, sizeof(rq)); if (ret_len < 0) { perror("[error] Error sending get_metadata request"); - ret = ret_len; goto error; } assert(ret_len == sizeof(rq)); @@ -791,12 +822,10 @@ int get_one_metadata_packet(struct lttng_live_ctx *ctx, ret_len = lttng_live_recv(ctx->control_sock, &rp, sizeof(rp)); if (ret_len == 0) { fprintf(stderr, "[error] Remote side has closed connection\n"); - ret = -1; goto error; } if (ret_len < 0) { perror("[error] Error receiving metadata response"); - ret = ret_len; goto error; } assert(ret_len == sizeof(rp)); @@ -811,39 +840,31 @@ int get_one_metadata_packet(struct lttng_live_ctx *ctx, goto end; case LTTNG_VIEWER_METADATA_ERR: printf_verbose("get_metadata : ERR\n"); - ret = -1; - goto end; + goto error; default: printf_verbose("get_metadata : UNKNOWN\n"); - ret = -1; - goto end; + goto error; } len = be64toh(rp.len); printf_verbose("Writing %" PRIu64" bytes to metadata\n", len); if (len <= 0) { - ret = -1; - goto end; + goto error; } data = zmalloc(len); if (!data) { perror("relay data zmalloc"); - ret = -1; goto error; } ret_len = lttng_live_recv(ctx->control_sock, data, len); if (ret_len == 0) { fprintf(stderr, "[error] Remote side has closed connection\n"); - ret = -1; - free(data); - goto error; + goto error_free_data; } if (ret_len < 0) { perror("[error] Error receiving trace packet"); - ret = ret_len; - free(data); - goto error; + goto error_free_data; } assert(ret_len == len); @@ -853,19 +874,19 @@ int get_one_metadata_packet(struct lttng_live_ctx *ctx, } while (ret_len < 0 && errno == EINTR); if (ret_len < 0) { fprintf(stderr, "[error] Writing in the metadata fp\n"); - free(data); - ret = ret_len; - goto error; + goto error_free_data; } assert(ret_len == len); metadata_stream->metadata_len += len; + free(data); ret = len; +end: + return ret; +error_free_data: free(data); - -end: error: - return ret; + return -1; } /* @@ -878,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) { @@ -893,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 @@ -915,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. * @@ -923,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); @@ -939,10 +982,13 @@ int get_next_index(struct lttng_live_ctx *ctx, rq.stream_id = htobe64(viewer_stream->id); retry: + if (lttng_live_should_quit()) { + ret = -1; + goto end; + } ret_len = lttng_live_send(ctx->control_sock, &cmd, sizeof(cmd)); if (ret_len < 0) { perror("[error] Error sending cmd"); - ret = ret_len; goto error; } assert(ret_len == sizeof(cmd)); @@ -950,53 +996,53 @@ retry: ret_len = lttng_live_send(ctx->control_sock, &rq, sizeof(rq)); if (ret_len < 0) { perror("[error] Error sending get_next_index request"); - ret = ret_len; goto error; } 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"); - ret = -1; goto error; } if (ret_len < 0) { perror("[error] Error receiving index response"); - ret = ret_len; 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: @@ -1011,17 +1057,94 @@ retry: break; case LTTNG_VIEWER_INDEX_ERR: fprintf(stderr, "[error] get_next_index: error\n"); - ret = -1; goto error; default: fprintf(stderr, "[error] get_next_index: unkwown value\n"); - ret = -1; goto error; } - ret = 0; +end: + return ret; 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; } @@ -1034,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); @@ -1069,11 +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; - 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; } @@ -1087,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); @@ -1117,43 +1286,20 @@ retry: printf_verbose("get_data_packet for stream %" PRIu64 "\n", viewer_stream->id); ret = get_data_packet(session->ctx, pos, viewer_stream, - be64toh(cur_index->offset), + cur_index->offset, cur_index->packet_size / CHAR_BIT); if (ret == -2) { goto retry; } else if (ret < 0) { pos->offset = EOF; - fprintf(stderr, "[error] get_data_packet failed\n"); + if (!lttng_live_should_quit()) { + fprintf(stderr, "[error] get_data_packet failed\n"); + } 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; @@ -1166,6 +1312,11 @@ int lttng_live_create_viewer_session(struct lttng_live_ctx *ctx) int ret; ssize_t ret_len; + if (lttng_live_should_quit()) { + ret = -1; + goto end; + } + cmd.cmd = htobe32(LTTNG_VIEWER_CREATE_SESSION); cmd.data_size = 0; cmd.cmd_version = 0; @@ -1173,7 +1324,6 @@ int lttng_live_create_viewer_session(struct lttng_live_ctx *ctx) ret_len = lttng_live_send(ctx->control_sock, &cmd, sizeof(cmd)); if (ret_len < 0) { perror("[error] Error sending cmd"); - ret = ret_len; goto error; } assert(ret_len == sizeof(cmd)); @@ -1181,25 +1331,24 @@ int lttng_live_create_viewer_session(struct lttng_live_ctx *ctx) ret_len = lttng_live_recv(ctx->control_sock, &resp, sizeof(resp)); if (ret_len == 0) { fprintf(stderr, "[error] Remote side has closed connection\n"); - ret = -1; goto error; } if (ret_len < 0) { perror("[error] Error receiving create session reply"); - ret = ret_len; goto error; } assert(ret_len == sizeof(resp)); if (be32toh(resp.status) != LTTNG_VIEWER_CREATE_SESSION_OK) { fprintf(stderr, "[error] Error creating viewer session\n"); - ret = -1; goto error; } ret = 0; +end: + return ret; error: - return ret; + return -1; } static @@ -1218,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; @@ -1239,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)); @@ -1282,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; } @@ -1289,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; @@ -1310,19 +1461,52 @@ 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; + if (lttng_live_should_quit()) { + ret = -1; + goto end; + } + cmd.cmd = htobe32(LTTNG_VIEWER_GET_NEW_STREAMS); cmd.data_size = sizeof(rq); cmd.cmd_version = 0; @@ -1333,7 +1517,6 @@ int lttng_live_get_new_streams(struct lttng_live_ctx *ctx, uint64_t id) ret_len = lttng_live_send(ctx->control_sock, &cmd, sizeof(cmd)); if (ret_len < 0) { perror("[error] Error sending cmd"); - ret = ret_len; goto error; } assert(ret_len == sizeof(cmd)); @@ -1341,7 +1524,6 @@ int lttng_live_get_new_streams(struct lttng_live_ctx *ctx, uint64_t id) ret_len = lttng_live_send(ctx->control_sock, &rq, sizeof(rq)); if (ret_len < 0) { perror("[error] Error sending get_new_streams request"); - ret = ret_len; goto error; } assert(ret_len == sizeof(rq)); @@ -1349,12 +1531,10 @@ int lttng_live_get_new_streams(struct lttng_live_ctx *ctx, uint64_t id) ret_len = lttng_live_recv(ctx->control_sock, &rp, sizeof(rp)); if (ret_len == 0) { fprintf(stderr, "[error] Remote side has closed connection\n"); - ret = -1; goto error; } if (ret_len < 0) { perror("[error] Error receiving get_new_streams response"); - ret = ret_len; goto error; } assert(ret_len == sizeof(rp)); @@ -1370,13 +1550,11 @@ int lttng_live_get_new_streams(struct lttng_live_ctx *ctx, uint64_t id) goto end; case LTTNG_VIEWER_NEW_STREAMS_ERR: fprintf(stderr, "[error] get_new_streams error\n"); - ret = -1; - goto end; + goto error; default: fprintf(stderr, "[error] Unknown return code %u\n", be32toh(rp.status)); - ret = -1; - goto end; + goto error; } stream_count = be32toh(rp.streams_count); @@ -1398,12 +1576,10 @@ int lttng_live_get_new_streams(struct lttng_live_ctx *ctx, uint64_t id) ret_len = lttng_live_recv(ctx->control_sock, &stream, sizeof(stream)); if (ret_len == 0) { fprintf(stderr, "[error] Remote side has closed connection\n"); - ret = -1; goto error; } if (ret_len < 0) { perror("[error] Error receiving stream"); - ret = ret_len; goto error; } assert(ret_len == sizeof(stream)); @@ -1416,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; @@ -1427,18 +1603,21 @@ 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: -error: return ret; + +error: + return -1; } -void lttng_live_read(struct lttng_live_ctx *ctx) +int lttng_live_read(struct lttng_live_ctx *ctx) { - int ret, i; + int ret = -1; + int i; struct bt_ctf_iter *iter; const struct bt_ctf_event *event; struct bt_iter_pos begin_pos; @@ -1478,7 +1657,7 @@ void 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) { @@ -1495,8 +1674,15 @@ void lttng_live_read(struct lttng_live_ctx *ctx) for (;;) { int flags; + if (lttng_live_should_quit()) { + ret = 0; + goto end_free; + } + while (!ctx->session->stream_count) { - if (ctx->session_ids->len == 0) { + if (lttng_live_should_quit() + || ctx->session_ids->len == 0) { + ret = 0; goto end_free; } ret = ask_new_streams(ctx); @@ -1508,16 +1694,26 @@ void 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); if (!iter) { + if (lttng_live_should_quit()) { + ret = 0; + goto end; + } fprintf(stderr, "[error] Iterator creation error\n"); goto end; } for (;;) { + if (lttng_live_should_quit()) { + ret = 0; + goto end_free; + } event = bt_ctf_iter_read_event_flags(iter, &flags); if (!(flags & BT_ITER_FLAG_RETRY)) { if (!event) { @@ -1546,5 +1742,8 @@ void lttng_live_read(struct lttng_live_ctx *ctx) end_free: bt_context_put(ctx->bt_ctx); end: - return; + if (lttng_live_should_quit()) { + ret = 0; + } + return ret; }