X-Git-Url: http://git.efficios.com/?p=babeltrace.git;a=blobdiff_plain;f=formats%2Flttng-live%2Flttng-live-comm.c;h=c19cc6460b5b06ee07ed46e6e5b5f54adf17d80a;hp=d26d83eb4bbb01a9d383585d57fab2761543fb4a;hb=b727abf5b1da0904c4321b534bd6d002ce4597f1;hpb=5ea98697b9f786ac73fd5152d96aebe02906a10d diff --git a/formats/lttng-live/lttng-live-comm.c b/formats/lttng-live/lttng-live-comm.c index d26d83eb..c19cc646 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,27 +168,28 @@ int lttng_live_establish_connection(struct lttng_live_ctx *ctx) } cmd.cmd = htobe32(LTTNG_VIEWER_CONNECT); - cmd.data_size = sizeof(connect); - cmd.cmd_version = 0; + cmd.data_size = htobe64((uint64_t) sizeof(connect)); + cmd.cmd_version = htobe32(0); connect.viewer_session_id = -1ULL; /* will be set on recv */ connect.major = htobe32(LTTNG_LIVE_MAJOR); 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) { @@ -246,7 +249,7 @@ void print_session_list(GPtrArray *session_list, const char *path) for (i = 0; i < session_list->len; i++) { relay_session = g_ptr_array_index(session_list, i); - fprintf(stdout, "%s/host/%s/%s (timer = %u, " + fprintf(LTTNG_LIVE_OUTPUT_FP, "%s/host/%s/%s (timer = %u, " "%u stream(s), %u client(s) connected)\n", path, relay_session->hostname, relay_session->name, relay_session->timer, @@ -307,8 +310,8 @@ int lttng_live_list_sessions(struct lttng_live_ctx *ctx, const char *path) } cmd.cmd = htobe32(LTTNG_VIEWER_LIST_SESSIONS); - cmd.data_size = 0; - cmd.cmd_version = 0; + cmd.data_size = htobe64((uint64_t) 0); + cmd.cmd_version = htobe32(0); ret_len = lttng_live_send(ctx->control_sock, &cmd, sizeof(cmd)); if (ret_len < 0) { @@ -423,6 +426,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,8 +437,8 @@ int lttng_live_attach_session(struct lttng_live_ctx *ctx, uint64_t id) } cmd.cmd = htobe32(LTTNG_VIEWER_ATTACH_SESSION); - cmd.data_size = sizeof(rq); - cmd.cmd_version = 0; + cmd.data_size = htobe64((uint64_t) sizeof(rq)); + cmd.cmd_version = htobe32(0); memset(&rq, 0, sizeof(rq)); rq.session_id = htobe64(id); @@ -441,19 +446,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) { @@ -641,6 +647,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; @@ -649,28 +657,30 @@ retry: ret = -1; goto end; } + cmd.cmd = htobe32(LTTNG_VIEWER_GET_PACKET); - cmd.data_size = sizeof(rq); - cmd.cmd_version = 0; + cmd.data_size = htobe64((uint64_t) sizeof(rq)); + cmd.cmd_version = htobe32(0); memset(&rq, 0, sizeof(rq)); rq.stream_id = htobe64(stream->id); 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) { @@ -794,6 +804,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; @@ -802,22 +814,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 = sizeof(rq); - cmd.cmd_version = 0; + 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) { @@ -932,8 +945,10 @@ int get_new_metadata(struct lttng_live_ctx *ctx, } } while (ret > 0 || !len_read); - if (fclose(metadata_stream->metadata_fp_write)) - perror("fclose"); + if (babeltrace_close_memstream(metadata_buf, &size, + metadata_stream->metadata_fp_write)) { + perror("babeltrace_close_memstream"); + } metadata_stream->metadata_fp_write = NULL; error: @@ -973,32 +988,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 = sizeof(rq); - cmd.cmd_version = 0; + 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) { @@ -1118,7 +1135,7 @@ int handle_seek_position(size_t index, int whence, struct ctf_stream_pos *pos, struct ctf_file_stream *file_stream) { - int ret; + int ret = 0; switch (whence) { case SEEK_CUR: @@ -1279,6 +1296,17 @@ retry: 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; } @@ -1318,8 +1346,8 @@ int lttng_live_create_viewer_session(struct lttng_live_ctx *ctx) } cmd.cmd = htobe32(LTTNG_VIEWER_CREATE_SESSION); - cmd.data_size = 0; - cmd.cmd_version = 0; + cmd.data_size = htobe64((uint64_t) 0); + cmd.cmd_version = htobe32(0); ret_len = lttng_live_send(ctx->control_sock, &cmd, sizeof(cmd)); if (ret_len < 0) { @@ -1501,6 +1529,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; @@ -1508,25 +1538,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 = sizeof(rq); - cmd.cmd_version = 0; + 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) {