From: Jérémie Galarneau Date: Wed, 21 Feb 2018 05:57:26 +0000 (-0500) Subject: Backport: Fix: relayd streams can be leaked on connection error X-Git-Url: http://drtracing.org/?a=commitdiff_plain;h=44a2fbf116008c62184cc12f144fc69d29d7f606;p=lttng-tools.git Backport: Fix: relayd streams can be leaked on connection error There are cases where a connection error can cause streams to be leaked. For instance, the control connection could receive an index and close. Since a packet is in-flight, the stream corresponding to that index will not close. However, nothing guarantees that the data connection will be able to receive the packet's data. If the protocol is respected, this is not a problem. However, a buggy consumerd or network errors can cause the streams to remain in the "data in-flight" state and never close. To mitigate a case observed in the field where a consumerd would be forcibly closed (network interface brought down) and cause leaks on the relay daemon, the session is aborted whenever the control or data connection encounters an error. Aborting a session causes the streams to be closed regardless of the fact that data is in-flight. Currently, only the control connection holds an ownership of the session object. This can cause the following scenario to leak streams: 1) Control connection receives an index - Stream is put in "in-flight data" mode 2) Control connection is closed/shutdown cleanly - try_stream_close refuses to close the stream as data is in-flight, but it puts the stream in "closed" mode. When the data is received, the stream will be closed as soon as possible. 3) Data connection closes cleanly or due to an error - The stream "closing" condition will never be re-evaluated. Since the data connection has no ownership of the session, it can never clean-up the streams that are waiting for "in-flight" data to arrive before closing. This patch lazily associates the data connection to its session so that the session can be aborted whenever an error happens on either the data or control connection. Note that this leaves the relayd vulnerable to a case which will still leak. If the control connection receives an index and closes cleanly, the data connection could have never been established with the consumer daemon and result in a leak. Signed-off-by: Jonathan Rajotte Signed-off-by: Jérémie Galarneau --- diff --git a/src/bin/lttng-relayd/Makefile.am b/src/bin/lttng-relayd/Makefile.am index 6598a23ef..d820b9248 100644 --- a/src/bin/lttng-relayd/Makefile.am +++ b/src/bin/lttng-relayd/Makefile.am @@ -8,7 +8,6 @@ bin_PROGRAMS = lttng-relayd lttng_relayd_SOURCES = main.c lttng-relayd.h utils.h utils.c cmd.h \ index.c index.h live.c live.h ctf-trace.c ctf-trace.h \ - cmd-generic.c cmd-generic.h \ cmd-2-1.c cmd-2-1.h \ cmd-2-2.c cmd-2-2.h \ cmd-2-4.c cmd-2-4.h \ diff --git a/src/bin/lttng-relayd/cmd-2-1.c b/src/bin/lttng-relayd/cmd-2-1.c index 080dc173f..99aa86668 100644 --- a/src/bin/lttng-relayd/cmd-2-1.c +++ b/src/bin/lttng-relayd/cmd-2-1.c @@ -25,14 +25,13 @@ #include #include -#include "cmd-generic.h" #include "cmd-2-1.h" #include "utils.h" /* * cmd_recv_stream_2_1 allocates path_name and channel_name. */ -int cmd_recv_stream_2_1(struct relay_connection *conn, +int cmd_recv_stream_2_1(const struct lttng_buffer_view *payload, char **ret_path_name, char **ret_channel_name) { int ret; @@ -41,11 +40,13 @@ int cmd_recv_stream_2_1(struct relay_connection *conn, char *channel_name = NULL; size_t len; - ret = cmd_recv(conn->sock, &stream_info, sizeof(stream_info)); - if (ret < 0) { - ERR("Unable to recv stream version 2.1"); + if (payload->size < sizeof(stream_info)) { + ERR("Unexpected payload size in \"cmd_recv_stream_2_1\": expected >= %zu bytes, got %zu bytes", + sizeof(stream_info), payload->size); + ret = -1; goto error; } + memcpy(&stream_info, payload->data, sizeof(stream_info)); len = lttng_strnlen(stream_info.pathname, sizeof(stream_info.pathname)); /* Ensure that NULL-terminated and fits in local filename length. */ diff --git a/src/bin/lttng-relayd/cmd-2-1.h b/src/bin/lttng-relayd/cmd-2-1.h index 46283dc5a..c1eb6b327 100644 --- a/src/bin/lttng-relayd/cmd-2-1.h +++ b/src/bin/lttng-relayd/cmd-2-1.h @@ -21,8 +21,9 @@ */ #include "lttng-relayd.h" +#include -int cmd_recv_stream_2_1(struct relay_connection *conn, +int cmd_recv_stream_2_1(const struct lttng_buffer_view *payload, char **path_name, char **channel_name); #endif /* RELAYD_CMD_2_1_H */ diff --git a/src/bin/lttng-relayd/cmd-2-2.c b/src/bin/lttng-relayd/cmd-2-2.c index 4f34d8b66..5ff628050 100644 --- a/src/bin/lttng-relayd/cmd-2-2.c +++ b/src/bin/lttng-relayd/cmd-2-2.c @@ -27,14 +27,13 @@ #include #include -#include "cmd-generic.h" #include "cmd-2-1.h" #include "utils.h" /* * cmd_recv_stream_2_2 allocates path_name and channel_name. */ -int cmd_recv_stream_2_2(struct relay_connection *conn, +int cmd_recv_stream_2_2(const struct lttng_buffer_view *payload, char **ret_path_name, char **ret_channel_name, uint64_t *tracefile_size, uint64_t *tracefile_count) { @@ -44,11 +43,13 @@ int cmd_recv_stream_2_2(struct relay_connection *conn, char *channel_name = NULL; size_t len; - ret = cmd_recv(conn->sock, &stream_info, sizeof(stream_info)); - if (ret < 0) { - ERR("Unable to recv stream version 2.2"); + if (payload->size < sizeof(stream_info)) { + ERR("Unexpected payload size in \"cmd_recv_stream_2_2\": expected >= %zu bytes, got %zu bytes", + sizeof(stream_info), payload->size); + ret = -1; goto error; } + memcpy(&stream_info, payload->data, sizeof(stream_info)); len = lttng_strnlen(stream_info.pathname, sizeof(stream_info.pathname)); /* Ensure that NULL-terminated and fits in local filename length. */ diff --git a/src/bin/lttng-relayd/cmd-2-2.h b/src/bin/lttng-relayd/cmd-2-2.h index 894a63a1e..822f627b1 100644 --- a/src/bin/lttng-relayd/cmd-2-2.h +++ b/src/bin/lttng-relayd/cmd-2-2.h @@ -21,8 +21,9 @@ */ #include "lttng-relayd.h" +#include -int cmd_recv_stream_2_2(struct relay_connection *conn, +int cmd_recv_stream_2_2(const struct lttng_buffer_view *payload, char **path_name, char **channel_name, uint64_t *tracefile_size, uint64_t *tracefile_count); diff --git a/src/bin/lttng-relayd/cmd-2-4.c b/src/bin/lttng-relayd/cmd-2-4.c index b269bc68a..aeb760a04 100644 --- a/src/bin/lttng-relayd/cmd-2-4.c +++ b/src/bin/lttng-relayd/cmd-2-4.c @@ -27,10 +27,10 @@ #include #include -#include "cmd-generic.h" +#include "cmd-2-4.h" #include "lttng-relayd.h" -int cmd_create_session_2_4(struct relay_connection *conn, +int cmd_create_session_2_4(const struct lttng_buffer_view *payload, char *session_name, char *hostname, uint32_t *live_timer, bool *snapshot) { @@ -38,11 +38,14 @@ int cmd_create_session_2_4(struct relay_connection *conn, struct lttcomm_relayd_create_session_2_4 session_info; size_t len; - ret = cmd_recv(conn->sock, &session_info, sizeof(session_info)); - if (ret < 0) { - ERR("Unable to recv session info version 2.4"); + if (payload->size < sizeof(session_info)) { + ERR("Unexpected payload size in \"cmd_create_session_2_4\": expected >= %zu bytes, got %zu bytes", + sizeof(session_info), payload->size); + ret = -1; goto error; } + memcpy(&session_info, payload->data, sizeof(session_info)); + len = lttng_strnlen(session_info.session_name, sizeof(session_info.session_name)); /* Ensure that NULL-terminated and fits in local filename length. */ if (len == sizeof(session_info.session_name) || len >= LTTNG_NAME_MAX) { diff --git a/src/bin/lttng-relayd/cmd-2-4.h b/src/bin/lttng-relayd/cmd-2-4.h index ab7347897..d2cd2c6df 100644 --- a/src/bin/lttng-relayd/cmd-2-4.h +++ b/src/bin/lttng-relayd/cmd-2-4.h @@ -21,8 +21,9 @@ */ #include "lttng-relayd.h" +#include -int cmd_create_session_2_4(struct relay_connection *conn, +int cmd_create_session_2_4(const struct lttng_buffer_view *payload, char *session_name, char *hostname, uint32_t *live_timer, bool *snapshot); diff --git a/src/bin/lttng-relayd/cmd-generic.c b/src/bin/lttng-relayd/cmd-generic.c deleted file mode 100644 index 89ae69e7a..000000000 --- a/src/bin/lttng-relayd/cmd-generic.c +++ /dev/null @@ -1,49 +0,0 @@ -/* - * Copyright (C) 2013 - Julien Desfossez - * David Goulet - * 2015 - Mathieu Desnoyers - * - * This program is free software; you can redistribute it and/or modify it - * under the terms of the GNU General Public License, version 2 only, as - * published by the Free Software Foundation. - * - * This program is distributed in the hope that it will be useful, but WITHOUT - * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or - * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for - * more details. - * - * You should have received a copy of the GNU General Public License along with - * this program; if not, write to the Free Software Foundation, Inc., 51 - * Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA. - */ - -#define _LGPL_SOURCE -#include - -#include - -#include "cmd-generic.h" - -int cmd_recv(struct lttcomm_sock *sock, void *buf, size_t len) -{ - int ret; - - assert(sock); - assert(buf); - - ret = sock->ops->recvmsg(sock, buf, len, 0); - if (ret < len) { - if (ret == 0) { - /* Orderly shutdown. Not necessary to print an error. */ - DBG("Socket %d did an orderly shutdown", sock->fd); - } else { - ERR("Relay didn't receive valid add_stream struct size. " - "Expected %zu, got %d", len, ret); - } - ret = -1; - goto error; - } - -error: - return ret; -} diff --git a/src/bin/lttng-relayd/cmd-generic.h b/src/bin/lttng-relayd/cmd-generic.h deleted file mode 100644 index 4551f0aa1..000000000 --- a/src/bin/lttng-relayd/cmd-generic.h +++ /dev/null @@ -1,29 +0,0 @@ -#ifndef RELAYD_CMD_GENERIC_H -#define RELAYD_CMD_GENERIC_H - -/* - * Copyright (C) 2013 - Julien Desfossez - * David Goulet - * 2015 - Mathieu Desnoyers - * - * This program is free software; you can redistribute it and/or modify it - * under the terms of the GNU General Public License, version 2 only, as - * published by the Free Software Foundation. - * - * This program is distributed in the hope that it will be useful, but WITHOUT - * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or - * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for - * more details. - * - * You should have received a copy of the GNU General Public License along with - * this program; if not, write to the Free Software Foundation, Inc., 51 - * Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA. - */ - -#include - -#include "connection.h" - -int cmd_recv(struct lttcomm_sock *sock, void *buf, size_t len); - -#endif /* RELAYD_CMD_GENERIC_H */ diff --git a/src/bin/lttng-relayd/cmd.h b/src/bin/lttng-relayd/cmd.h index 88db09aed..468a693e5 100644 --- a/src/bin/lttng-relayd/cmd.h +++ b/src/bin/lttng-relayd/cmd.h @@ -20,7 +20,6 @@ * Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA. */ -#include "cmd-generic.h" #include "cmd-2-1.h" #include "cmd-2-2.h" #include "cmd-2-4.h" diff --git a/src/bin/lttng-relayd/connection.c b/src/bin/lttng-relayd/connection.c index f3d7e642e..6a2c27ff6 100644 --- a/src/bin/lttng-relayd/connection.c +++ b/src/bin/lttng-relayd/connection.c @@ -65,6 +65,45 @@ end: return conn; } +int connection_reset_protocol_state(struct relay_connection *connection) +{ + int ret = 0; + + switch (connection->type) { + case RELAY_DATA: + connection->protocol.data.state_id = + DATA_CONNECTION_STATE_RECEIVE_HEADER; + memset(&connection->protocol.data.state.receive_header, + 0, + sizeof(connection->protocol.data.state.receive_header)); + connection->protocol.data.state.receive_header.left_to_receive = + sizeof(struct lttcomm_relayd_data_hdr); + break; + case RELAY_CONTROL: + connection->protocol.ctrl.state_id = + CTRL_CONNECTION_STATE_RECEIVE_HEADER; + memset(&connection->protocol.ctrl.state.receive_header, + 0, + sizeof(connection->protocol.ctrl.state.receive_header)); + connection->protocol.data.state.receive_header.left_to_receive = + sizeof(struct lttcomm_relayd_hdr); + ret = lttng_dynamic_buffer_set_size( + &connection->protocol.ctrl.reception_buffer, + sizeof(struct lttcomm_relayd_hdr)); + if (ret) { + ERR("Failed to reinitialize control connection reception buffer size to %zu bytes.", sizeof(struct lttcomm_relayd_hdr)); + goto end; + } + break; + default: + goto end; + } + DBG("Reset communication state of relay connection (fd = %i)", + connection->sock->fd); +end: + return ret; +} + struct relay_connection *connection_create(struct lttcomm_sock *sock, enum connection_type type) { @@ -80,6 +119,10 @@ struct relay_connection *connection_create(struct lttcomm_sock *sock, conn->type = type; conn->sock = sock; lttng_ht_node_init_ulong(&conn->sock_n, (unsigned long) conn->sock->fd); + if (conn->type == RELAY_CONTROL) { + lttng_dynamic_buffer_init(&conn->protocol.ctrl.reception_buffer); + } + connection_reset_protocol_state(conn); end: return conn; } @@ -145,3 +188,27 @@ void connection_ht_add(struct lttng_ht *relay_connections_ht, conn->in_socket_ht = 1; conn->socket_ht = relay_connections_ht; } + +int connection_set_session(struct relay_connection *conn, + struct relay_session *session) +{ + int ret = 0; + + assert(conn); + assert(session); + assert(!conn->session); + + if (connection_get(conn)) { + if (session_get(session)) { + conn->session = session; + } else { + ERR("Failed to get session reference in connection_set_session()"); + ret = -1; + } + connection_put(conn); + } else { + ERR("Failed to get connection reference in connection_set_session()"); + ret = -1; + } + return ret; +} diff --git a/src/bin/lttng-relayd/connection.h b/src/bin/lttng-relayd/connection.h index 443863813..d0edf8214 100644 --- a/src/bin/lttng-relayd/connection.h +++ b/src/bin/lttng-relayd/connection.h @@ -29,6 +29,8 @@ #include #include +#include +#include #include "session.h" @@ -40,6 +42,36 @@ enum connection_type { RELAY_VIEWER_NOTIFICATION = 4, }; +enum data_connection_state { + DATA_CONNECTION_STATE_RECEIVE_HEADER = 0, + DATA_CONNECTION_STATE_RECEIVE_PAYLOAD = 1, +}; + +enum ctrl_connection_state { + CTRL_CONNECTION_STATE_RECEIVE_HEADER = 0, + CTRL_CONNECTION_STATE_RECEIVE_PAYLOAD = 1, +}; + +struct data_connection_state_receive_header { + uint64_t received, left_to_receive; + char header_reception_buffer[sizeof(struct lttcomm_relayd_data_hdr)]; +}; + +struct data_connection_state_receive_payload { + uint64_t received, left_to_receive; + struct lttcomm_relayd_data_hdr header; + bool rotate_index; +}; + +struct ctrl_connection_state_receive_header { + uint64_t received, left_to_receive; +}; + +struct ctrl_connection_state_receive_payload { + uint64_t received, left_to_receive; + struct lttcomm_relayd_hdr header; +}; + /* * Internal structure to map a socket with the corresponding session. * A hashtable indexed on the socket FD is used for the lookups. @@ -88,15 +120,36 @@ struct relay_connection { bool in_socket_ht; struct lttng_ht *socket_ht; /* HACK: Contained within this hash table. */ struct rcu_head rcu_node; /* For call_rcu teardown. */ + + union { + struct { + enum data_connection_state state_id; + union { + struct data_connection_state_receive_header receive_header; + struct data_connection_state_receive_payload receive_payload; + } state; + } data; + struct { + enum ctrl_connection_state state_id; + union { + struct ctrl_connection_state_receive_header receive_header; + struct ctrl_connection_state_receive_payload receive_payload; + } state; + struct lttng_dynamic_buffer reception_buffer; + } ctrl; + } protocol; }; struct relay_connection *connection_create(struct lttcomm_sock *sock, enum connection_type type); struct relay_connection *connection_get_by_sock(struct lttng_ht *relay_connections_ht, int sock); +int connection_reset_protocol_state(struct relay_connection *connection); bool connection_get(struct relay_connection *connection); void connection_put(struct relay_connection *connection); void connection_ht_add(struct lttng_ht *relay_connections_ht, struct relay_connection *conn); +int connection_set_session(struct relay_connection *conn, + struct relay_session *session); #endif /* _CONNECTION_H */ diff --git a/src/bin/lttng-relayd/main.c b/src/bin/lttng-relayd/main.c index 4f368ddac..43498017d 100644 --- a/src/bin/lttng-relayd/main.c +++ b/src/bin/lttng-relayd/main.c @@ -55,6 +55,8 @@ #include #include #include +#include +#include #include #include "cmd.h" @@ -72,6 +74,14 @@ #include "tracefile-array.h" #include "tcp_keep_alive.h" +enum relay_connection_status { + RELAY_CONNECTION_STATUS_OK, + /* An error occured while processing an event on the connection. */ + RELAY_CONNECTION_STATUS_ERROR, + /* Connection closed/shutdown cleanly. */ + RELAY_CONNECTION_STATUS_CLOSED, +}; + /* command line options */ char *opt_output_path; static int opt_daemon, opt_background; @@ -135,10 +145,6 @@ static uint64_t last_relay_stream_id; */ static struct relay_conn_queue relay_conn_queue; -/* buffer allocated at startup, used to store the trace data */ -static char *data_buffer; -static unsigned int data_buffer_size; - /* Global relay stream hash table. */ struct lttng_ht *relay_streams_ht; @@ -1049,21 +1055,18 @@ static int set_index_control_data(struct relay_index *index, struct ctf_packet_index index_data; /* - * The index on disk is encoded in big endian, so we don't need - * to convert the data received on the network. The data_offset - * value is NEVER modified here and is updated by the data - * thread. + * The index on disk is encoded in big endian. */ - index_data.packet_size = data->packet_size; - index_data.content_size = data->content_size; - index_data.timestamp_begin = data->timestamp_begin; - index_data.timestamp_end = data->timestamp_end; - index_data.events_discarded = data->events_discarded; - index_data.stream_id = data->stream_id; + index_data.packet_size = htobe64(data->packet_size); + index_data.content_size = htobe64(data->content_size); + index_data.timestamp_begin = htobe64(data->timestamp_begin); + index_data.timestamp_end = htobe64(data->timestamp_end); + index_data.events_discarded = htobe64(data->events_discarded); + index_data.stream_id = htobe64(data->stream_id); if (conn->minor >= 8) { - index->index_data.stream_instance_id = data->stream_instance_id; - index->index_data.packet_seq_num = data->packet_seq_num; + index->index_data.stream_instance_id = htobe64(data->stream_instance_id); + index->index_data.packet_seq_num = htobe64(data->packet_seq_num); } return relay_index_set_data(index, &index_data); @@ -1074,10 +1077,12 @@ static int set_index_control_data(struct relay_index *index, * * On success, send back the session id or else return a negative value. */ -static int relay_create_session(struct lttcomm_relayd_hdr *recv_hdr, - struct relay_connection *conn) +static int relay_create_session(const struct lttcomm_relayd_hdr *recv_hdr, + struct relay_connection *conn, + const struct lttng_buffer_view *payload) { - int ret = 0, send_ret; + int ret = 0; + ssize_t send_ret; struct relay_session *session; struct lttcomm_relayd_status_session reply; char session_name[LTTNG_NAME_MAX]; @@ -1097,7 +1102,7 @@ static int relay_create_session(struct lttcomm_relayd_hdr *recv_hdr, break; case 4: /* LTTng sessiond 2.4 */ default: - ret = cmd_create_session_2_4(conn, session_name, + ret = cmd_create_session_2_4(payload, session_name, hostname, &live_timer, &snapshot); } if (ret < 0) { @@ -1124,9 +1129,10 @@ send_reply: } send_ret = conn->sock->ops->sendmsg(conn->sock, &reply, sizeof(reply), 0); - if (send_ret < 0) { - ERR("Relayd sending session id"); - ret = send_ret; + if (send_ret < (ssize_t) sizeof(reply)) { + ERR("Failed to send \"create session\" command reply (ret = %zd)", + send_ret); + ret = -1; } return ret; @@ -1165,8 +1171,9 @@ static void publish_connection_local_streams(struct relay_connection *conn) /* * relay_add_stream: allocate a new stream for a session */ -static int relay_add_stream(struct lttcomm_relayd_hdr *recv_hdr, - struct relay_connection *conn) +static int relay_add_stream(const struct lttcomm_relayd_hdr *recv_hdr, + struct relay_connection *conn, + const struct lttng_buffer_view *payload) { int ret; ssize_t send_ret; @@ -1178,7 +1185,7 @@ static int relay_add_stream(struct lttcomm_relayd_hdr *recv_hdr, char *path_name = NULL, *channel_name = NULL; uint64_t tracefile_size = 0, tracefile_count = 0; - if (!session || conn->version_check_done == 0) { + if (!session || !conn->version_check_done) { ERR("Trying to add a stream before version check"); ret = -1; goto end_no_session; @@ -1186,12 +1193,12 @@ static int relay_add_stream(struct lttcomm_relayd_hdr *recv_hdr, switch (session->minor) { case 1: /* LTTng sessiond 2.1. Allocates path_name and channel_name. */ - ret = cmd_recv_stream_2_1(conn, &path_name, + ret = cmd_recv_stream_2_1(payload, &path_name, &channel_name); break; case 2: /* LTTng sessiond 2.2. Allocates path_name and channel_name. */ default: - ret = cmd_recv_stream_2_2(conn, &path_name, + ret = cmd_recv_stream_2_2(payload, &path_name, &channel_name, &tracefile_size, &tracefile_count); break; } @@ -1232,9 +1239,10 @@ send_reply: send_ret = conn->sock->ops->sendmsg(conn->sock, &reply, sizeof(struct lttcomm_relayd_status_stream), 0); - if (send_ret < 0) { - ERR("Relay sending stream id"); - ret = (int) send_ret; + if (send_ret < (ssize_t) sizeof(reply)) { + ERR("Failed to send \"add stream\" command reply (ret = %zd)", + send_ret); + ret = -1; } end_no_session: @@ -1246,10 +1254,12 @@ end_no_session: /* * relay_close_stream: close a specific stream */ -static int relay_close_stream(struct lttcomm_relayd_hdr *recv_hdr, - struct relay_connection *conn) +static int relay_close_stream(const struct lttcomm_relayd_hdr *recv_hdr, + struct relay_connection *conn, + const struct lttng_buffer_view *payload) { - int ret, send_ret; + int ret; + ssize_t send_ret; struct relay_session *session = conn->session; struct lttcomm_relayd_close_stream stream_info; struct lttcomm_relayd_generic_reply reply; @@ -1257,26 +1267,23 @@ static int relay_close_stream(struct lttcomm_relayd_hdr *recv_hdr, DBG("Close stream received"); - if (!session || conn->version_check_done == 0) { + if (!session || !conn->version_check_done) { ERR("Trying to close a stream before version check"); ret = -1; goto end_no_session; } - ret = conn->sock->ops->recvmsg(conn->sock, &stream_info, - sizeof(struct lttcomm_relayd_close_stream), 0); - if (ret < sizeof(struct lttcomm_relayd_close_stream)) { - if (ret == 0) { - /* Orderly shutdown. Not necessary to print an error. */ - DBG("Socket %d did an orderly shutdown", conn->sock->fd); - } else { - ERR("Relay didn't receive valid add_stream struct size : %d", ret); - } + if (payload->size < sizeof(stream_info)) { + ERR("Unexpected payload size in \"relay_close_stream\": expected >= %zu bytes, got %zu bytes", + sizeof(stream_info), payload->size); ret = -1; goto end_no_session; } + memcpy(&stream_info, payload->data, sizeof(stream_info)); + stream_info.stream_id = be64toh(stream_info.stream_id); + stream_info.last_net_seq_num = be64toh(stream_info.last_net_seq_num); - stream = stream_get_by_id(be64toh(stream_info.stream_id)); + stream = stream_get_by_id(stream_info.stream_id); if (!stream) { ret = -1; goto end; @@ -1287,7 +1294,7 @@ static int relay_close_stream(struct lttcomm_relayd_hdr *recv_hdr, * pending check. */ pthread_mutex_lock(&stream->lock); - stream->last_net_seq_num = be64toh(stream_info.last_net_seq_num); + stream->last_net_seq_num = stream_info.last_net_seq_num; pthread_mutex_unlock(&stream->lock); /* @@ -1318,6 +1325,7 @@ static int relay_close_stream(struct lttcomm_relayd_hdr *recv_hdr, } } stream_put(stream); + ret = 0; end: memset(&reply, 0, sizeof(reply)); @@ -1328,9 +1336,10 @@ end: } send_ret = conn->sock->ops->sendmsg(conn->sock, &reply, sizeof(struct lttcomm_relayd_generic_reply), 0); - if (send_ret < 0) { - ERR("Relay sending stream id"); - ret = send_ret; + if (send_ret < (ssize_t) sizeof(reply)) { + ERR("Failed to send \"close stream\" command reply (ret = %zd)", + send_ret); + ret = -1; } end_no_session: @@ -1341,10 +1350,12 @@ end_no_session: * relay_reset_metadata: reset a metadata stream */ static -int relay_reset_metadata(struct lttcomm_relayd_hdr *recv_hdr, - struct relay_connection *conn) +int relay_reset_metadata(const struct lttcomm_relayd_hdr *recv_hdr, + struct relay_connection *conn, + const struct lttng_buffer_view *payload) { - int ret, send_ret; + int ret; + ssize_t send_ret; struct relay_session *session = conn->session; struct lttcomm_relayd_reset_metadata stream_info; struct lttcomm_relayd_generic_reply reply; @@ -1352,26 +1363,23 @@ int relay_reset_metadata(struct lttcomm_relayd_hdr *recv_hdr, DBG("Reset metadata received"); - if (!session || conn->version_check_done == 0) { + if (!session || !conn->version_check_done) { ERR("Trying to reset a metadata stream before version check"); ret = -1; goto end_no_session; } - ret = conn->sock->ops->recvmsg(conn->sock, &stream_info, - sizeof(struct lttcomm_relayd_reset_metadata), 0); - if (ret < sizeof(struct lttcomm_relayd_reset_metadata)) { - if (ret == 0) { - /* Orderly shutdown. Not necessary to print an error. */ - DBG("Socket %d did an orderly shutdown", conn->sock->fd); - } else { - ERR("Relay didn't receive valid reset_metadata struct " - "size : %d", ret); - } + if (payload->size < sizeof(stream_info)) { + ERR("Unexpected payload size in \"relay_reset_metadata\": expected >= %zu bytes, got %zu bytes", + sizeof(stream_info), payload->size); ret = -1; goto end_no_session; } - DBG("Update metadata to version %" PRIu64, be64toh(stream_info.version)); + memcpy(&stream_info, payload->data, sizeof(stream_info)); + stream_info.stream_id = be64toh(stream_info.stream_id); + stream_info.version = be64toh(stream_info.version); + + DBG("Update metadata to version %" PRIu64, stream_info.version); /* Unsupported for live sessions for now. */ if (session->live_timer != 0) { @@ -1379,7 +1387,7 @@ int relay_reset_metadata(struct lttcomm_relayd_hdr *recv_hdr, goto end; } - stream = stream_get_by_id(be64toh(stream_info.stream_id)); + stream = stream_get_by_id(stream_info.stream_id); if (!stream) { ret = -1; goto end; @@ -1412,9 +1420,10 @@ end: } send_ret = conn->sock->ops->sendmsg(conn->sock, &reply, sizeof(struct lttcomm_relayd_generic_reply), 0); - if (send_ret < 0) { - ERR("Relay sending reset metadata reply"); - ret = send_ret; + if (send_ret < (ssize_t) sizeof(reply)) { + ERR("Failed to send \"reset metadata\" command reply (ret = %zd)", + send_ret); + ret = -1; } end_no_session: @@ -1427,14 +1436,13 @@ end_no_session: static void relay_unknown_command(struct relay_connection *conn) { struct lttcomm_relayd_generic_reply reply; - int ret; + ssize_t send_ret; memset(&reply, 0, sizeof(reply)); reply.ret_code = htobe32(LTTNG_ERR_UNK); - ret = conn->sock->ops->sendmsg(conn->sock, &reply, - sizeof(struct lttcomm_relayd_generic_reply), 0); - if (ret < 0) { - ERR("Relay sending unknown command"); + send_ret = conn->sock->ops->sendmsg(conn->sock, &reply, sizeof(reply), 0); + if (send_ret < sizeof(reply)) { + ERR("Failed to send \"unknown command\" command reply (ret = %zd)", send_ret); } } @@ -1442,10 +1450,12 @@ static void relay_unknown_command(struct relay_connection *conn) * relay_start: send an acknowledgment to the client to tell if we are * ready to receive data. We are ready if a session is established. */ -static int relay_start(struct lttcomm_relayd_hdr *recv_hdr, - struct relay_connection *conn) +static int relay_start(const struct lttcomm_relayd_hdr *recv_hdr, + struct relay_connection *conn, + const struct lttng_buffer_view *payload) { - int ret = htobe32(LTTNG_OK); + int ret = 0; + ssize_t send_ret; struct lttcomm_relayd_generic_reply reply; struct relay_session *session = conn->session; @@ -1455,11 +1465,13 @@ static int relay_start(struct lttcomm_relayd_hdr *recv_hdr, } memset(&reply, 0, sizeof(reply)); - reply.ret_code = ret; - ret = conn->sock->ops->sendmsg(conn->sock, &reply, - sizeof(struct lttcomm_relayd_generic_reply), 0); - if (ret < 0) { - ERR("Relay sending start ack"); + reply.ret_code = htobe32(LTTNG_OK); + send_ret = conn->sock->ops->sendmsg(conn->sock, &reply, + sizeof(reply), 0); + if (send_ret < (ssize_t) sizeof(reply)) { + ERR("Failed to send \"relay_start\" command reply (ret = %zd)", + send_ret); + ret = -1; } return ret; @@ -1498,15 +1510,16 @@ end: /* * relay_recv_metadata: receive the metadata for the session. */ -static int relay_recv_metadata(struct lttcomm_relayd_hdr *recv_hdr, - struct relay_connection *conn) +static int relay_recv_metadata(const struct lttcomm_relayd_hdr *recv_hdr, + struct relay_connection *conn, + const struct lttng_buffer_view *payload) { int ret = 0; ssize_t size_ret; struct relay_session *session = conn->session; - struct lttcomm_relayd_metadata_payload *metadata_struct; + struct lttcomm_relayd_metadata_payload metadata_payload_header; struct relay_stream *metadata_stream; - uint64_t data_size, payload_size; + uint64_t metadata_payload_size; if (!session) { ERR("Metadata sent before version check"); @@ -1514,44 +1527,22 @@ static int relay_recv_metadata(struct lttcomm_relayd_hdr *recv_hdr, goto end; } - data_size = payload_size = be64toh(recv_hdr->data_size); - if (data_size < sizeof(struct lttcomm_relayd_metadata_payload)) { + if (recv_hdr->data_size < sizeof(struct lttcomm_relayd_metadata_payload)) { ERR("Incorrect data size"); ret = -1; goto end; } - payload_size -= sizeof(struct lttcomm_relayd_metadata_payload); + metadata_payload_size = recv_hdr->data_size - + sizeof(struct lttcomm_relayd_metadata_payload); - if (data_buffer_size < data_size) { - /* In case the realloc fails, we can free the memory */ - char *tmp_data_ptr; + memcpy(&metadata_payload_header, payload->data, + sizeof(metadata_payload_header)); + metadata_payload_header.stream_id = be64toh( + metadata_payload_header.stream_id); + metadata_payload_header.padding_size = be32toh( + metadata_payload_header.padding_size); - tmp_data_ptr = realloc(data_buffer, data_size); - if (!tmp_data_ptr) { - ERR("Allocating data buffer"); - free(data_buffer); - ret = -1; - goto end; - } - data_buffer = tmp_data_ptr; - data_buffer_size = data_size; - } - memset(data_buffer, 0, data_size); - DBG2("Relay receiving metadata, waiting for %" PRIu64 " bytes", data_size); - size_ret = conn->sock->ops->recvmsg(conn->sock, data_buffer, data_size, 0); - if (size_ret < 0 || size_ret != data_size) { - if (size_ret == 0) { - /* Orderly shutdown. Not necessary to print an error. */ - DBG("Socket %d did an orderly shutdown", conn->sock->fd); - } else { - ERR("Relay didn't receive the whole metadata"); - } - ret = -1; - goto end; - } - metadata_struct = (struct lttcomm_relayd_metadata_payload *) data_buffer; - - metadata_stream = stream_get_by_id(be64toh(metadata_struct->stream_id)); + metadata_stream = stream_get_by_id(metadata_payload_header.stream_id); if (!metadata_stream) { ret = -1; goto end; @@ -1559,22 +1550,24 @@ static int relay_recv_metadata(struct lttcomm_relayd_hdr *recv_hdr, pthread_mutex_lock(&metadata_stream->lock); - size_ret = lttng_write(metadata_stream->stream_fd->fd, metadata_struct->payload, - payload_size); - if (size_ret < payload_size) { + size_ret = lttng_write(metadata_stream->stream_fd->fd, + payload->data + sizeof(metadata_payload_header), + metadata_payload_size); + if (size_ret < metadata_payload_size) { ERR("Relay error writing metadata on file"); ret = -1; goto end_put; } size_ret = write_padding_to_file(metadata_stream->stream_fd->fd, - be32toh(metadata_struct->padding_size)); - if (size_ret < 0) { + metadata_payload_header.padding_size); + if (size_ret < (int64_t) metadata_payload_header.padding_size) { + ret = -1; goto end_put; } metadata_stream->metadata_received += - payload_size + be32toh(metadata_struct->padding_size); + metadata_payload_size + metadata_payload_header.padding_size; DBG2("Relay metadata written. Updated metadata_received %" PRIu64, metadata_stream->metadata_received); @@ -1588,53 +1581,59 @@ end: /* * relay_send_version: send relayd version number */ -static int relay_send_version(struct lttcomm_relayd_hdr *recv_hdr, - struct relay_connection *conn) +static int relay_send_version(const struct lttcomm_relayd_hdr *recv_hdr, + struct relay_connection *conn, + const struct lttng_buffer_view *payload) { int ret; + ssize_t send_ret; struct lttcomm_relayd_version reply, msg; bool compatible = true; - conn->version_check_done = 1; + conn->version_check_done = true; /* Get version from the other side. */ - ret = conn->sock->ops->recvmsg(conn->sock, &msg, sizeof(msg), 0); - if (ret < 0 || ret != sizeof(msg)) { - if (ret == 0) { - /* Orderly shutdown. Not necessary to print an error. */ - DBG("Socket %d did an orderly shutdown", conn->sock->fd); - } else { - ERR("Relay failed to receive the version values."); - } + if (payload->size < sizeof(msg)) { + ERR("Unexpected payload size in \"relay_send_version\": expected >= %zu bytes, got %zu bytes", + sizeof(msg), payload->size); ret = -1; goto end; } + memcpy(&msg, payload->data, sizeof(msg)); + msg.major = be32toh(msg.major); + msg.minor = be32toh(msg.minor); + memset(&reply, 0, sizeof(reply)); reply.major = RELAYD_VERSION_COMM_MAJOR; reply.minor = RELAYD_VERSION_COMM_MINOR; /* Major versions must be the same */ - if (reply.major != be32toh(msg.major)) { + if (reply.major != msg.major) { DBG("Incompatible major versions (%u vs %u), deleting session", - reply.major, be32toh(msg.major)); + reply.major, msg.major); compatible = false; } conn->major = reply.major; /* We adapt to the lowest compatible version */ - if (reply.minor <= be32toh(msg.minor)) { + if (reply.minor <= msg.minor) { conn->minor = reply.minor; } else { - conn->minor = be32toh(msg.minor); + conn->minor = msg.minor; } reply.major = htobe32(reply.major); reply.minor = htobe32(reply.minor); - ret = conn->sock->ops->sendmsg(conn->sock, &reply, - sizeof(struct lttcomm_relayd_version), 0); - if (ret < 0) { - ERR("Relay sending version"); + send_ret = conn->sock->ops->sendmsg(conn->sock, &reply, + sizeof(reply), 0); + if (send_ret < (ssize_t) sizeof(reply)) { + ERR("Failed to send \"send version\" command reply (ret = %zd)", + send_ret); + ret = -1; + goto end; + } else { + ret = 0; } if (!compatible) { @@ -1652,41 +1651,36 @@ end: /* * Check for data pending for a given stream id from the session daemon. */ -static int relay_data_pending(struct lttcomm_relayd_hdr *recv_hdr, - struct relay_connection *conn) +static int relay_data_pending(const struct lttcomm_relayd_hdr *recv_hdr, + struct relay_connection *conn, + const struct lttng_buffer_view *payload) { struct relay_session *session = conn->session; struct lttcomm_relayd_data_pending msg; struct lttcomm_relayd_generic_reply reply; struct relay_stream *stream; + ssize_t send_ret; int ret; - uint64_t last_net_seq_num, stream_id; DBG("Data pending command received"); - if (!session || conn->version_check_done == 0) { + if (!session || !conn->version_check_done) { ERR("Trying to check for data before version check"); ret = -1; goto end_no_session; } - ret = conn->sock->ops->recvmsg(conn->sock, &msg, sizeof(msg), 0); - if (ret < sizeof(msg)) { - if (ret == 0) { - /* Orderly shutdown. Not necessary to print an error. */ - DBG("Socket %d did an orderly shutdown", conn->sock->fd); - } else { - ERR("Relay didn't receive valid data_pending struct size : %d", - ret); - } + if (payload->size < sizeof(msg)) { + ERR("Unexpected payload size in \"relay_data_pending\": expected >= %zu bytes, got %zu bytes", + sizeof(msg), payload->size); ret = -1; goto end_no_session; } + memcpy(&msg, payload->data, sizeof(msg)); + msg.stream_id = be64toh(msg.stream_id); + msg.last_net_seq_num = be64toh(msg.last_net_seq_num); - stream_id = be64toh(msg.stream_id); - last_net_seq_num = be64toh(msg.last_net_seq_num); - - stream = stream_get_by_id(stream_id); + stream = stream_get_by_id(msg.stream_id); if (stream == NULL) { ret = -1; goto end; @@ -1695,11 +1689,11 @@ static int relay_data_pending(struct lttcomm_relayd_hdr *recv_hdr, pthread_mutex_lock(&stream->lock); DBG("Data pending for stream id %" PRIu64 " prev_seq %" PRIu64 - " and last_seq %" PRIu64, stream_id, stream->prev_seq, - last_net_seq_num); + " and last_seq %" PRIu64, msg.stream_id, + stream->prev_seq, msg.last_net_seq_num); /* Avoid wrapping issue */ - if (((int64_t) (stream->prev_seq - last_net_seq_num)) >= 0) { + if (((int64_t) (stream->prev_seq - msg.last_net_seq_num)) >= 0) { /* Data has in fact been written and is NOT pending */ ret = 0; } else { @@ -1715,9 +1709,11 @@ end: memset(&reply, 0, sizeof(reply)); reply.ret_code = htobe32(ret); - ret = conn->sock->ops->sendmsg(conn->sock, &reply, sizeof(reply), 0); - if (ret < 0) { - ERR("Relay data pending ret code failed"); + send_ret = conn->sock->ops->sendmsg(conn->sock, &reply, sizeof(reply), 0); + if (send_ret < (ssize_t) sizeof(reply)) { + ERR("Failed to send \"data pending\" command reply (ret = %zd)", + send_ret); + ret = -1; } end_no_session: @@ -1732,52 +1728,53 @@ end_no_session: * the control socket has been handled. So, this is why we simply return * OK here. */ -static int relay_quiescent_control(struct lttcomm_relayd_hdr *recv_hdr, - struct relay_connection *conn) +static int relay_quiescent_control(const struct lttcomm_relayd_hdr *recv_hdr, + struct relay_connection *conn, + const struct lttng_buffer_view *payload) { int ret; - uint64_t stream_id; + ssize_t send_ret; struct relay_stream *stream; struct lttcomm_relayd_quiescent_control msg; struct lttcomm_relayd_generic_reply reply; DBG("Checking quiescent state on control socket"); - if (!conn->session || conn->version_check_done == 0) { + if (!conn->session || !conn->version_check_done) { ERR("Trying to check for data before version check"); ret = -1; goto end_no_session; } - ret = conn->sock->ops->recvmsg(conn->sock, &msg, sizeof(msg), 0); - if (ret < sizeof(msg)) { - if (ret == 0) { - /* Orderly shutdown. Not necessary to print an error. */ - DBG("Socket %d did an orderly shutdown", conn->sock->fd); - } else { - ERR("Relay didn't receive valid begin data_pending struct size: %d", - ret); - } + if (payload->size < sizeof(msg)) { + ERR("Unexpected payload size in \"relay_quiescent_control\": expected >= %zu bytes, got %zu bytes", + sizeof(msg), payload->size); ret = -1; goto end_no_session; } + memcpy(&msg, payload->data, sizeof(msg)); + msg.stream_id = be64toh(msg.stream_id); - stream_id = be64toh(msg.stream_id); - stream = stream_get_by_id(stream_id); + stream = stream_get_by_id(msg.stream_id); if (!stream) { goto reply; } pthread_mutex_lock(&stream->lock); stream->data_pending_check_done = true; pthread_mutex_unlock(&stream->lock); - DBG("Relay quiescent control pending flag set to %" PRIu64, stream_id); + + DBG("Relay quiescent control pending flag set to %" PRIu64, msg.stream_id); stream_put(stream); reply: memset(&reply, 0, sizeof(reply)); reply.ret_code = htobe32(LTTNG_OK); - ret = conn->sock->ops->sendmsg(conn->sock, &reply, sizeof(reply), 0); - if (ret < 0) { - ERR("Relay data quiescent control ret code failed"); + send_ret = conn->sock->ops->sendmsg(conn->sock, &reply, sizeof(reply), 0); + if (send_ret < (ssize_t) sizeof(reply)) { + ERR("Failed to send \"quiescent control\" command reply (ret = %zd)", + send_ret); + ret = -1; + } else { + ret = 0; } end_no_session: @@ -1791,41 +1788,36 @@ end_no_session: * * This command returns to the client a LTTNG_OK code. */ -static int relay_begin_data_pending(struct lttcomm_relayd_hdr *recv_hdr, - struct relay_connection *conn) +static int relay_begin_data_pending(const struct lttcomm_relayd_hdr *recv_hdr, + struct relay_connection *conn, + const struct lttng_buffer_view *payload) { int ret; + ssize_t send_ret; struct lttng_ht_iter iter; struct lttcomm_relayd_begin_data_pending msg; struct lttcomm_relayd_generic_reply reply; struct relay_stream *stream; - uint64_t session_id; assert(recv_hdr); assert(conn); DBG("Init streams for data pending"); - if (!conn->session || conn->version_check_done == 0) { + if (!conn->session || !conn->version_check_done) { ERR("Trying to check for data before version check"); ret = -1; goto end_no_session; } - ret = conn->sock->ops->recvmsg(conn->sock, &msg, sizeof(msg), 0); - if (ret < sizeof(msg)) { - if (ret == 0) { - /* Orderly shutdown. Not necessary to print an error. */ - DBG("Socket %d did an orderly shutdown", conn->sock->fd); - } else { - ERR("Relay didn't receive valid begin data_pending struct size: %d", - ret); - } + if (payload->size < sizeof(msg)) { + ERR("Unexpected payload size in \"relay_begin_data_pending\": expected >= %zu bytes, got %zu bytes", + sizeof(msg), payload->size); ret = -1; goto end_no_session; } - - session_id = be64toh(msg.session_id); + memcpy(&msg, payload->data, sizeof(msg)); + msg.session_id = be64toh(msg.session_id); /* * Iterate over all streams to set the begin data pending flag. @@ -1839,7 +1831,7 @@ static int relay_begin_data_pending(struct lttcomm_relayd_hdr *recv_hdr, if (!stream_get(stream)) { continue; } - if (stream->trace->session->id == session_id) { + if (stream->trace->session->id == msg.session_id) { pthread_mutex_lock(&stream->lock); stream->data_pending_check_done = false; pthread_mutex_unlock(&stream->lock); @@ -1854,9 +1846,13 @@ static int relay_begin_data_pending(struct lttcomm_relayd_hdr *recv_hdr, /* All good, send back reply. */ reply.ret_code = htobe32(LTTNG_OK); - ret = conn->sock->ops->sendmsg(conn->sock, &reply, sizeof(reply), 0); - if (ret < 0) { - ERR("Relay begin data pending send reply failed"); + send_ret = conn->sock->ops->sendmsg(conn->sock, &reply, sizeof(reply), 0); + if (send_ret < (ssize_t) sizeof(reply)) { + ERR("Failed to send \"begin data pending\" command reply (ret = %zd)", + send_ret); + ret = -1; + } else { + ret = 0; } end_no_session: @@ -1872,39 +1868,34 @@ end_no_session: * * Return to the client if there is data in flight or not with a ret_code. */ -static int relay_end_data_pending(struct lttcomm_relayd_hdr *recv_hdr, - struct relay_connection *conn) +static int relay_end_data_pending(const struct lttcomm_relayd_hdr *recv_hdr, + struct relay_connection *conn, + const struct lttng_buffer_view *payload) { int ret; + ssize_t send_ret; struct lttng_ht_iter iter; struct lttcomm_relayd_end_data_pending msg; struct lttcomm_relayd_generic_reply reply; struct relay_stream *stream; - uint64_t session_id; uint32_t is_data_inflight = 0; DBG("End data pending command"); - if (!conn->session || conn->version_check_done == 0) { + if (!conn->session || !conn->version_check_done) { ERR("Trying to check for data before version check"); ret = -1; goto end_no_session; } - ret = conn->sock->ops->recvmsg(conn->sock, &msg, sizeof(msg), 0); - if (ret < sizeof(msg)) { - if (ret == 0) { - /* Orderly shutdown. Not necessary to print an error. */ - DBG("Socket %d did an orderly shutdown", conn->sock->fd); - } else { - ERR("Relay didn't receive valid end data_pending struct size: %d", - ret); - } + if (payload->size < sizeof(msg)) { + ERR("Unexpected payload size in \"relay_end_data_pending\": expected >= %zu bytes, got %zu bytes", + sizeof(msg), payload->size); ret = -1; goto end_no_session; } - - session_id = be64toh(msg.session_id); + memcpy(&msg, payload->data, sizeof(msg)); + msg.session_id = be64toh(msg.session_id); /* * Iterate over all streams to see if the begin data pending @@ -1916,7 +1907,7 @@ static int relay_end_data_pending(struct lttcomm_relayd_hdr *recv_hdr, if (!stream_get(stream)) { continue; } - if (stream->trace->session->id != session_id) { + if (stream->trace->session->id != msg.session_id) { stream_put(stream); continue; } @@ -1940,9 +1931,13 @@ static int relay_end_data_pending(struct lttcomm_relayd_hdr *recv_hdr, /* All good, send back reply. */ reply.ret_code = htobe32(is_data_inflight); - ret = conn->sock->ops->sendmsg(conn->sock, &reply, sizeof(reply), 0); - if (ret < 0) { - ERR("Relay end data pending send reply failed"); + send_ret = conn->sock->ops->sendmsg(conn->sock, &reply, sizeof(reply), 0); + if (send_ret < (ssize_t) sizeof(reply)) { + ERR("Failed to send \"end data pending\" command reply (ret = %zd)", + send_ret); + ret = -1; + } else { + ret = 0; } end_no_session: @@ -1954,23 +1949,24 @@ end_no_session: * * Return 0 on success else a negative value. */ -static int relay_recv_index(struct lttcomm_relayd_hdr *recv_hdr, - struct relay_connection *conn) +static int relay_recv_index(const struct lttcomm_relayd_hdr *recv_hdr, + struct relay_connection *conn, + const struct lttng_buffer_view *payload) { - int ret, send_ret; + int ret; + ssize_t send_ret; struct relay_session *session = conn->session; struct lttcomm_relayd_index index_info; struct relay_index *index; struct lttcomm_relayd_generic_reply reply; struct relay_stream *stream; - uint64_t net_seq_num; size_t msg_len; assert(conn); DBG("Relay receiving index"); - if (!session || conn->version_check_done == 0) { + if (!session || !conn->version_check_done) { ERR("Trying to close a stream before version check"); ret = -1; goto end_no_session; @@ -1979,22 +1975,29 @@ static int relay_recv_index(struct lttcomm_relayd_hdr *recv_hdr, msg_len = lttcomm_relayd_index_len( lttng_to_index_major(conn->major, conn->minor), lttng_to_index_minor(conn->major, conn->minor)); - ret = conn->sock->ops->recvmsg(conn->sock, &index_info, - msg_len, 0); - if (ret < msg_len) { - if (ret == 0) { - /* Orderly shutdown. Not necessary to print an error. */ - DBG("Socket %d did an orderly shutdown", conn->sock->fd); - } else { - ERR("Relay didn't receive valid index struct size : %d", ret); - } + if (payload->size < msg_len) { + ERR("Unexpected payload size in \"relay_recv_index\": expected >= %zu bytes, got %zu bytes", + msg_len, payload->size); ret = -1; goto end_no_session; } + memcpy(&index_info, payload->data, msg_len); + index_info.relay_stream_id = be64toh(index_info.relay_stream_id); + index_info.net_seq_num = be64toh(index_info.net_seq_num); + index_info.packet_size = be64toh(index_info.packet_size); + index_info.content_size = be64toh(index_info.content_size); + index_info.timestamp_begin = be64toh(index_info.timestamp_begin); + index_info.timestamp_end = be64toh(index_info.timestamp_end); + index_info.events_discarded = be64toh(index_info.events_discarded); + index_info.stream_id = be64toh(index_info.stream_id); - net_seq_num = be64toh(index_info.net_seq_num); + if (conn->minor >= 8) { + index_info.stream_instance_id = + be64toh(index_info.stream_instance_id); + index_info.packet_seq_num = be64toh(index_info.packet_seq_num); + } - stream = stream_get_by_id(be64toh(index_info.relay_stream_id)); + stream = stream_get_by_id(index_info.relay_stream_id); if (!stream) { ERR("stream_get_by_id not found"); ret = -1; @@ -2013,8 +2016,7 @@ static int relay_recv_index(struct lttcomm_relayd_hdr *recv_hdr, */ if (stream->index_received_seqcount > 0 && stream->indexes_in_flight == 0) { - stream->beacon_ts_end = - be64toh(index_info.timestamp_end); + stream->beacon_ts_end = index_info.timestamp_end; } ret = 0; goto end_stream_put; @@ -2023,9 +2025,9 @@ static int relay_recv_index(struct lttcomm_relayd_hdr *recv_hdr, } if (stream->ctf_stream_id == -1ULL) { - stream->ctf_stream_id = be64toh(index_info.stream_id); + stream->ctf_stream_id = index_info.stream_id; } - index = relay_index_get_by_id_or_create(stream, net_seq_num); + index = relay_index_get_by_id_or_create(stream, index_info.net_seq_num); if (!index) { ret = -1; ERR("relay_index_get_by_id_or_create index NULL"); @@ -2063,9 +2065,9 @@ end: reply.ret_code = htobe32(LTTNG_OK); } send_ret = conn->sock->ops->sendmsg(conn->sock, &reply, sizeof(reply), 0); - if (send_ret < 0) { - ERR("Relay sending close index id reply"); - ret = send_ret; + if (send_ret < (ssize_t) sizeof(reply)) { + ERR("Failed to send \"recv index\" command reply (ret = %zd)", send_ret); + ret = -1; } end_no_session: @@ -2077,17 +2079,19 @@ end_no_session: * * Return 0 on success else a negative value. */ -static int relay_streams_sent(struct lttcomm_relayd_hdr *recv_hdr, - struct relay_connection *conn) +static int relay_streams_sent(const struct lttcomm_relayd_hdr *recv_hdr, + struct relay_connection *conn, + const struct lttng_buffer_view *payload) { - int ret, send_ret; + int ret; + ssize_t send_ret; struct lttcomm_relayd_generic_reply reply; assert(conn); DBG("Relay receiving streams_sent"); - if (!conn->session || conn->version_check_done == 0) { + if (!conn->session || !conn->version_check_done) { ERR("Trying to close a stream before version check"); ret = -1; goto end_no_session; @@ -2102,9 +2106,10 @@ static int relay_streams_sent(struct lttcomm_relayd_hdr *recv_hdr, memset(&reply, 0, sizeof(reply)); reply.ret_code = htobe32(LTTNG_OK); send_ret = conn->sock->ops->sendmsg(conn->sock, &reply, sizeof(reply), 0); - if (send_ret < 0) { - ERR("Relay sending sent_stream reply"); - ret = send_ret; + if (send_ret < (ssize_t) sizeof(reply)) { + ERR("Failed to send \"streams sent\" command reply (ret = %zd)", + send_ret); + ret = -1; } else { /* Success. */ ret = 0; @@ -2114,57 +2119,71 @@ end_no_session: return ret; } -/* - * Process the commands received on the control socket - */ -static int relay_process_control(struct lttcomm_relayd_hdr *recv_hdr, - struct relay_connection *conn) +#define DBG_CMD(cmd_name, conn) \ + DBG3("Processing \"%s\" command for socket %i", cmd_name, conn->sock->fd); + +static int relay_process_control_command(struct relay_connection *conn, + const struct lttcomm_relayd_hdr *header, + const struct lttng_buffer_view *payload) { int ret = 0; - switch (be32toh(recv_hdr->cmd)) { + switch (header->cmd) { case RELAYD_CREATE_SESSION: - ret = relay_create_session(recv_hdr, conn); + DBG_CMD("RELAYD_CREATE_SESSION", conn); + ret = relay_create_session(header, conn, payload); break; case RELAYD_ADD_STREAM: - ret = relay_add_stream(recv_hdr, conn); + DBG_CMD("RELAYD_ADD_STREAM", conn); + ret = relay_add_stream(header, conn, payload); break; case RELAYD_START_DATA: - ret = relay_start(recv_hdr, conn); + DBG_CMD("RELAYD_START_DATA", conn); + ret = relay_start(header, conn, payload); break; case RELAYD_SEND_METADATA: - ret = relay_recv_metadata(recv_hdr, conn); + DBG_CMD("RELAYD_SEND_METADATA", conn); + ret = relay_recv_metadata(header, conn, payload); break; case RELAYD_VERSION: - ret = relay_send_version(recv_hdr, conn); + DBG_CMD("RELAYD_VERSION", conn); + ret = relay_send_version(header, conn, payload); break; case RELAYD_CLOSE_STREAM: - ret = relay_close_stream(recv_hdr, conn); + DBG_CMD("RELAYD_CLOSE_STREAM", conn); + ret = relay_close_stream(header, conn, payload); break; case RELAYD_DATA_PENDING: - ret = relay_data_pending(recv_hdr, conn); + DBG_CMD("RELAYD_DATA_PENDING", conn); + ret = relay_data_pending(header, conn, payload); break; case RELAYD_QUIESCENT_CONTROL: - ret = relay_quiescent_control(recv_hdr, conn); + DBG_CMD("RELAYD_QUIESCENT_CONTROL", conn); + ret = relay_quiescent_control(header, conn, payload); break; case RELAYD_BEGIN_DATA_PENDING: - ret = relay_begin_data_pending(recv_hdr, conn); + DBG_CMD("RELAYD_BEGIN_DATA_PENDING", conn); + ret = relay_begin_data_pending(header, conn, payload); break; case RELAYD_END_DATA_PENDING: - ret = relay_end_data_pending(recv_hdr, conn); + DBG_CMD("RELAYD_END_DATA_PENDING", conn); + ret = relay_end_data_pending(header, conn, payload); break; case RELAYD_SEND_INDEX: - ret = relay_recv_index(recv_hdr, conn); + DBG_CMD("RELAYD_SEND_INDEX", conn); + ret = relay_recv_index(header, conn, payload); break; case RELAYD_STREAMS_SENT: - ret = relay_streams_sent(recv_hdr, conn); + DBG_CMD("RELAYD_STREAMS_SENT", conn); + ret = relay_streams_sent(header, conn, payload); break; case RELAYD_RESET_METADATA: - ret = relay_reset_metadata(recv_hdr, conn); + DBG_CMD("RELAYD_RESET_METADATA", conn); + ret = relay_reset_metadata(header, conn, payload); break; case RELAYD_UPDATE_SYNC_INFO: default: - ERR("Received unknown command (%u)", be32toh(recv_hdr->cmd)); + ERR("Received unknown command (%u)", header->cmd); relay_unknown_command(conn); ret = -1; goto end; @@ -2174,6 +2193,193 @@ end: return ret; } +static enum relay_connection_status relay_process_control_receive_payload( + struct relay_connection *conn) +{ + int ret = 0; + enum relay_connection_status status = RELAY_CONNECTION_STATUS_OK; + struct lttng_dynamic_buffer *reception_buffer = + &conn->protocol.ctrl.reception_buffer; + struct ctrl_connection_state_receive_payload *state = + &conn->protocol.ctrl.state.receive_payload; + struct lttng_buffer_view payload_view; + + if (state->left_to_receive == 0) { + /* Short-circuit for payload-less commands. */ + goto reception_complete; + } + ret = conn->sock->ops->recvmsg(conn->sock, + reception_buffer->data + state->received, + state->left_to_receive, MSG_DONTWAIT); + if (ret < 0) { + if (errno != EAGAIN && errno != EWOULDBLOCK) { + PERROR("Unable to receive command payload on sock %d", + conn->sock->fd); + status = RELAY_CONNECTION_STATUS_ERROR; + } + goto end; + } else if (ret == 0) { + DBG("Socket %d performed an orderly shutdown (received EOF)", conn->sock->fd); + status = RELAY_CONNECTION_STATUS_CLOSED; + goto end; + } + + assert(ret > 0); + assert(ret <= state->left_to_receive); + + state->left_to_receive -= ret; + state->received += ret; + + if (state->left_to_receive > 0) { + /* + * Can't transition to the protocol's next state, wait to + * receive the rest of the header. + */ + DBG3("Partial reception of control connection protocol payload (received %" PRIu64 " bytes, %" PRIu64 " bytes left to receive, fd = %i)", + state->received, state->left_to_receive, + conn->sock->fd); + goto end; + } + +reception_complete: + DBG("Done receiving control command payload: fd = %i, payload size = %" PRIu64 " bytes", + conn->sock->fd, state->received); + /* + * The payload required to process the command has been received. + * A view to the reception buffer is forwarded to the various + * commands and the state of the control is reset on success. + * + * Commands are responsible for sending their reply to the peer. + */ + payload_view = lttng_buffer_view_from_dynamic_buffer(reception_buffer, + 0, -1); + ret = relay_process_control_command(conn, + &state->header, &payload_view); + if (ret < 0) { + status = RELAY_CONNECTION_STATUS_ERROR; + goto end; + } + + ret = connection_reset_protocol_state(conn); + if (ret) { + status = RELAY_CONNECTION_STATUS_ERROR; + } +end: + return status; +} + +static enum relay_connection_status relay_process_control_receive_header( + struct relay_connection *conn) +{ + int ret = 0; + enum relay_connection_status status = RELAY_CONNECTION_STATUS_OK; + struct lttcomm_relayd_hdr header; + struct lttng_dynamic_buffer *reception_buffer = + &conn->protocol.ctrl.reception_buffer; + struct ctrl_connection_state_receive_header *state = + &conn->protocol.ctrl.state.receive_header; + + assert(state->left_to_receive != 0); + + ret = conn->sock->ops->recvmsg(conn->sock, + reception_buffer->data + state->received, + state->left_to_receive, MSG_DONTWAIT); + if (ret < 0) { + if (errno != EAGAIN && errno != EWOULDBLOCK) { + PERROR("Unable to receive control command header on sock %d", + conn->sock->fd); + status = RELAY_CONNECTION_STATUS_ERROR; + } + goto end; + } else if (ret == 0) { + DBG("Socket %d performed an orderly shutdown (received EOF)", conn->sock->fd); + status = RELAY_CONNECTION_STATUS_CLOSED; + goto end; + } + + assert(ret > 0); + assert(ret <= state->left_to_receive); + + state->left_to_receive -= ret; + state->received += ret; + + if (state->left_to_receive > 0) { + /* + * Can't transition to the protocol's next state, wait to + * receive the rest of the header. + */ + DBG3("Partial reception of control connection protocol header (received %" PRIu64 " bytes, %" PRIu64 " bytes left to receive, fd = %i)", + state->received, state->left_to_receive, + conn->sock->fd); + goto end; + } + + /* Transition to next state: receiving the command's payload. */ + conn->protocol.ctrl.state_id = + CTRL_CONNECTION_STATE_RECEIVE_PAYLOAD; + memcpy(&header, reception_buffer->data, sizeof(header)); + header.circuit_id = be64toh(header.circuit_id); + header.data_size = be64toh(header.data_size); + header.cmd = be32toh(header.cmd); + header.cmd_version = be32toh(header.cmd_version); + memcpy(&conn->protocol.ctrl.state.receive_payload.header, + &header, sizeof(header)); + + DBG("Done receiving control command header: fd = %i, cmd = %" PRIu32 ", cmd_version = %" PRIu32 ", payload size = %" PRIu64 " bytes", + conn->sock->fd, header.cmd, header.cmd_version, + header.data_size); + + if (header.data_size > DEFAULT_NETWORK_RELAYD_CTRL_MAX_PAYLOAD_SIZE) { + ERR("Command header indicates a payload (%" PRIu64 " bytes) that exceeds the maximal payload size allowed on a control connection.", + header.data_size); + status = RELAY_CONNECTION_STATUS_ERROR; + goto end; + } + + conn->protocol.ctrl.state.receive_payload.left_to_receive = + header.data_size; + conn->protocol.ctrl.state.receive_payload.received = 0; + ret = lttng_dynamic_buffer_set_size(reception_buffer, + header.data_size); + if (ret) { + status = RELAY_CONNECTION_STATUS_ERROR; + goto end; + } + + if (header.data_size == 0) { + /* + * Manually invoke the next state as the poll loop + * will not wake-up to allow us to proceed further. + */ + status = relay_process_control_receive_payload(conn); + } +end: + return status; +} + +/* + * Process the commands received on the control socket + */ +static enum relay_connection_status relay_process_control( + struct relay_connection *conn) +{ + enum relay_connection_status status; + + switch (conn->protocol.ctrl.state_id) { + case CTRL_CONNECTION_STATE_RECEIVE_HEADER: + status = relay_process_control_receive_header(conn); + break; + case CTRL_CONNECTION_STATE_RECEIVE_PAYLOAD: + status = relay_process_control_receive_payload(conn); + break; + default: + ERR("Unknown control connection protocol state encountered."); + abort(); + } + + return status; +} + /* * Handle index for a data stream. * @@ -2182,7 +2388,7 @@ end: * Return 0 on success else a negative value. */ static int handle_index_data(struct relay_stream *stream, uint64_t net_seq_num, - int rotate_index) + bool rotate_index) { int ret = 0; uint64_t data_offset; @@ -2255,57 +2461,87 @@ end: return ret; } -/* - * relay_process_data: Process the data received on the data socket - */ -static int relay_process_data(struct relay_connection *conn) +static enum relay_connection_status relay_process_data_receive_header( + struct relay_connection *conn) { - int ret = 0, rotate_index = 0; - ssize_t size_ret; + int ret; + enum relay_connection_status status = RELAY_CONNECTION_STATUS_OK; + struct data_connection_state_receive_header *state = + &conn->protocol.data.state.receive_header; + struct lttcomm_relayd_data_hdr header; struct relay_stream *stream; - struct lttcomm_relayd_data_hdr data_hdr; - uint64_t stream_id; - uint64_t net_seq_num; - uint32_t data_size; - struct relay_session *session; - bool new_stream = false, close_requested = false; - size_t chunk_size = RECV_DATA_BUFFER_SIZE; - size_t recv_off = 0; - char data_buffer[chunk_size]; - ret = conn->sock->ops->recvmsg(conn->sock, &data_hdr, - sizeof(struct lttcomm_relayd_data_hdr), 0); - if (ret <= 0) { - if (ret == 0) { - /* Orderly shutdown. Not necessary to print an error. */ - DBG("Socket %d did an orderly shutdown", conn->sock->fd); - } else { - ERR("Unable to receive data header on sock %d", conn->sock->fd); + assert(state->left_to_receive != 0); + + ret = conn->sock->ops->recvmsg(conn->sock, + state->header_reception_buffer + state->received, + state->left_to_receive, MSG_DONTWAIT); + if (ret < 0) { + if (errno != EAGAIN && errno != EWOULDBLOCK) { + PERROR("Unable to receive data header on sock %d", conn->sock->fd); + status = RELAY_CONNECTION_STATUS_ERROR; } - ret = -1; + goto end; + } else if (ret == 0) { + /* Orderly shutdown. Not necessary to print an error. */ + DBG("Socket %d performed an orderly shutdown (received EOF)", conn->sock->fd); + status = RELAY_CONNECTION_STATUS_CLOSED; goto end; } - stream_id = be64toh(data_hdr.stream_id); - stream = stream_get_by_id(stream_id); - if (!stream) { - ERR("relay_process_data: Cannot find stream %" PRIu64, stream_id); - ret = -1; + assert(ret > 0); + assert(ret <= state->left_to_receive); + + state->left_to_receive -= ret; + state->received += ret; + + if (state->left_to_receive > 0) { + /* + * Can't transition to the protocol's next state, wait to + * receive the rest of the header. + */ + DBG3("Partial reception of data connection header (received %" PRIu64 " bytes, %" PRIu64 " bytes left to receive, fd = %i)", + state->received, state->left_to_receive, + conn->sock->fd); + ret = 0; goto end; } - session = stream->trace->session; - data_size = be32toh(data_hdr.data_size); - net_seq_num = be64toh(data_hdr.net_seq_num); + /* Transition to next state: receiving the payload. */ + conn->protocol.data.state_id = DATA_CONNECTION_STATE_RECEIVE_PAYLOAD; - DBG3("Receiving data of size %u for stream id %" PRIu64 " seqnum %" PRIu64, - data_size, stream_id, net_seq_num); + memcpy(&header, state->header_reception_buffer, sizeof(header)); + header.circuit_id = be64toh(header.circuit_id); + header.stream_id = be64toh(header.stream_id); + header.data_size = be32toh(header.data_size); + header.net_seq_num = be64toh(header.net_seq_num); + header.padding_size = be32toh(header.padding_size); + memcpy(&conn->protocol.data.state.receive_payload.header, &header, sizeof(header)); + + conn->protocol.data.state.receive_payload.left_to_receive = + header.data_size; + conn->protocol.data.state.receive_payload.received = 0; + conn->protocol.data.state.receive_payload.rotate_index = false; + + DBG("Received data connection header on fd %i: circuit_id = %" PRIu64 ", stream_id = %" PRIu64 ", data_size = %" PRIu32 ", net_seq_num = %" PRIu64 ", padding_size = %" PRIu32, + conn->sock->fd, header.circuit_id, + header.stream_id, header.data_size, + header.net_seq_num, header.padding_size); + + stream = stream_get_by_id(header.stream_id); + if (!stream) { + DBG("relay_process_data_receive_payload: Cannot find stream %" PRIu64, + header.stream_id); + /* Protocol error. */ + status = RELAY_CONNECTION_STATUS_ERROR; + goto end; + } pthread_mutex_lock(&stream->lock); /* Check if a rotation is needed. */ if (stream->tracefile_size > 0 && - (stream->tracefile_size_current + data_size) > + (stream->tracefile_size_current + header.data_size) > stream->tracefile_size) { uint64_t old_id, new_id; @@ -2321,77 +2557,171 @@ static int relay_process_data(struct relay_connection *conn) -1, stream->stream_fd->fd, &new_id, &stream->stream_fd->fd); if (ret < 0) { - ERR("Rotating stream output file"); + ERR("Failed to rotate stream output file"); + status = RELAY_CONNECTION_STATUS_ERROR; goto end_stream_unlock; } + /* * Reset current size because we just performed a stream * rotation. */ stream->tracefile_size_current = 0; - rotate_index = 1; + conn->protocol.data.state.receive_payload.rotate_index = true; } - /* - * Index are handled in protocol version 2.4 and above. Also, - * snapshot and index are NOT supported. - */ - if (session->minor >= 4 && !session->snapshot) { - ret = handle_index_data(stream, net_seq_num, rotate_index); - if (ret < 0) { - ERR("handle_index_data: fail stream %" PRIu64 " net_seq_num %" PRIu64 " ret %d", - stream->stream_handle, net_seq_num, ret); + ret = 0; +end_stream_unlock: + pthread_mutex_unlock(&stream->lock); + stream_put(stream); +end: + return status; +} + +static enum relay_connection_status relay_process_data_receive_payload( + struct relay_connection *conn) +{ + int ret; + enum relay_connection_status status = RELAY_CONNECTION_STATUS_OK; + struct relay_stream *stream; + struct data_connection_state_receive_payload *state = + &conn->protocol.data.state.receive_payload; + const size_t chunk_size = RECV_DATA_BUFFER_SIZE; + char data_buffer[chunk_size]; + bool partial_recv = false; + bool new_stream = false, close_requested = false; + uint64_t left_to_receive = state->left_to_receive; + struct relay_session *session; + + DBG3("Receiving data for stream id %" PRIu64 " seqnum %" PRIu64 ", %" PRIu64" bytes received, %" PRIu64 " bytes left to receive", + state->header.stream_id, state->header.net_seq_num, + state->received, left_to_receive); + + stream = stream_get_by_id(state->header.stream_id); + if (!stream) { + /* Protocol error. */ + ERR("relay_process_data_receive_payload: cannot find stream %" PRIu64, + state->header.stream_id); + status = RELAY_CONNECTION_STATUS_ERROR; + goto end; + } + + pthread_mutex_lock(&stream->lock); + session = stream->trace->session; + if (!conn->session) { + ret = connection_set_session(conn, session); + if (ret) { + status = RELAY_CONNECTION_STATUS_ERROR; goto end_stream_unlock; } } - for (recv_off = 0; recv_off < data_size; recv_off += chunk_size) { - size_t recv_size = min(data_size - recv_off, chunk_size); + /* + * The size of the "chunk" received on any iteration is bounded by: + * - the data left to receive, + * - the data immediately available on the socket, + * - the on-stack data buffer + */ + while (left_to_receive > 0 && !partial_recv) { + ssize_t write_ret; + size_t recv_size = min(left_to_receive, chunk_size); - ret = conn->sock->ops->recvmsg(conn->sock, data_buffer, recv_size, 0); - if (ret <= 0) { - if (ret == 0) { - /* Orderly shutdown. Not necessary to print an error. */ - DBG("Socket %d did an orderly shutdown", conn->sock->fd); - } else { - ERR("Socket %d error %d", conn->sock->fd, ret); + ret = conn->sock->ops->recvmsg(conn->sock, data_buffer, + recv_size, MSG_DONTWAIT); + if (ret < 0) { + if (errno != EAGAIN && errno != EWOULDBLOCK) { + PERROR("Socket %d error", conn->sock->fd); + status = RELAY_CONNECTION_STATUS_ERROR; } - ret = -1; goto end_stream_unlock; + } else if (ret == 0) { + /* No more data ready to be consumed on socket. */ + DBG3("No more data ready for consumption on data socket of stream id %" PRIu64, + state->header.stream_id); + status = RELAY_CONNECTION_STATUS_CLOSED; + break; + } else if (ret < (int) recv_size) { + /* + * All the data available on the socket has been + * consumed. + */ + partial_recv = true; } + recv_size = ret; + /* Write data to stream output fd. */ - size_ret = lttng_write(stream->stream_fd->fd, data_buffer, + write_ret = lttng_write(stream->stream_fd->fd, data_buffer, recv_size); - if (size_ret < recv_size) { + if (write_ret < (ssize_t) recv_size) { ERR("Relay error writing data to file"); - ret = -1; + status = RELAY_CONNECTION_STATUS_ERROR; goto end_stream_unlock; } + left_to_receive -= recv_size; + state->received += recv_size; + state->left_to_receive = left_to_receive; + DBG2("Relay wrote %zd bytes to tracefile for stream id %" PRIu64, - size_ret, stream->stream_handle); + write_ret, stream->stream_handle); + } + + if (state->left_to_receive > 0) { + /* + * Did not receive all the data expected, wait for more data to + * become available on the socket. + */ + DBG3("Partial receive on data connection of stream id %" PRIu64 ", %" PRIu64 " bytes received, %" PRIu64 " bytes left to receive", + state->header.stream_id, state->received, + state->left_to_receive); + goto end_stream_unlock; } ret = write_padding_to_file(stream->stream_fd->fd, - be32toh(data_hdr.padding_size)); - if (ret < 0) { + state->header.padding_size); + if ((int64_t) ret < (int64_t) state->header.padding_size) { ERR("write_padding_to_file: fail stream %" PRIu64 " net_seq_num %" PRIu64 " ret %d", - stream->stream_handle, net_seq_num, ret); + stream->stream_handle, + state->header.net_seq_num, ret); + status = RELAY_CONNECTION_STATUS_ERROR; goto end_stream_unlock; } - stream->tracefile_size_current += - data_size + be32toh(data_hdr.padding_size); + + + if (session->minor >= 4 && !session->snapshot) { + ret = handle_index_data(stream, state->header.net_seq_num, + state->rotate_index); + if (ret < 0) { + ERR("handle_index_data: fail stream %" PRIu64 " net_seq_num %" PRIu64 " ret %d", + stream->stream_handle, + state->header.net_seq_num, ret); + status = RELAY_CONNECTION_STATUS_ERROR; + goto end_stream_unlock; + } + } + + stream->tracefile_size_current += state->header.data_size + + state->header.padding_size; + if (stream->prev_seq == -1ULL) { new_stream = true; } - stream->prev_seq = net_seq_num; + stream->prev_seq = state->header.net_seq_num; + + /* + * Resetting the protocol state (to RECEIVE_HEADER) will trash the + * contents of *state which are aliased (union) to the same location as + * the new state. Don't use it beyond this point. + */ + connection_reset_protocol_state(conn); + state = NULL; end_stream_unlock: close_requested = stream->close_requested; pthread_mutex_unlock(&stream->lock); - if (close_requested) { + if (close_requested && left_to_receive == 0) { try_stream_close(stream); } @@ -2400,9 +2730,33 @@ end_stream_unlock: uatomic_set(&session->new_streams, 1); pthread_mutex_unlock(&session->lock); } + stream_put(stream); end: - return ret; + return status; +} + +/* + * relay_process_data: Process the data received on the data socket + */ +static enum relay_connection_status relay_process_data( + struct relay_connection *conn) +{ + enum relay_connection_status status; + + switch (conn->protocol.data.state_id) { + case DATA_CONNECTION_STATE_RECEIVE_HEADER: + status = relay_process_data_receive_header(conn); + break; + case DATA_CONNECTION_STATE_RECEIVE_PAYLOAD: + status = relay_process_data_receive_payload(conn); + break; + default: + ERR("Unexpected data connection communication state."); + abort(); + } + + return status; } static void cleanup_connection_pollfd(struct lttng_poll_event *events, int pollfd) @@ -2453,7 +2807,6 @@ static void *relay_thread_worker(void *data) struct lttng_poll_event events; struct lttng_ht *relay_connections_ht; struct lttng_ht_iter iter; - struct lttcomm_relayd_hdr recv_hdr; struct relay_connection *destroy_conn = NULL; DBG("[thread] Relay worker started"); @@ -2575,21 +2928,36 @@ restart: assert(ctrl_conn->type == RELAY_CONTROL); if (revents & LPOLLIN) { - ret = ctrl_conn->sock->ops->recvmsg(ctrl_conn->sock, - &recv_hdr, sizeof(recv_hdr), 0); - if (ret <= 0) { - /* Connection closed */ - relay_thread_close_connection(&events, pollfd, - ctrl_conn); - } else { - ret = relay_process_control(&recv_hdr, ctrl_conn); - if (ret < 0) { - /* Clear the session on error. */ - relay_thread_close_connection(&events, - pollfd, ctrl_conn); + enum relay_connection_status status; + + status = relay_process_control(ctrl_conn); + if (status != RELAY_CONNECTION_STATUS_OK) { + /* + * On socket error flag the session as aborted to force + * the cleanup of its stream otherwise it can leak + * during the lifetime of the relayd. + * + * This prevents situations in which streams can be + * left opened because an index was received, the + * control connection is closed, and the data + * connection is closed (uncleanly) before the packet's + * data provided. + * + * Since the control connection encountered an error, + * it is okay to be conservative and close the + * session right now as we can't rely on the protocol + * being respected anymore. + */ + if (status == RELAY_CONNECTION_STATUS_ERROR) { + session_abort(ctrl_conn->session); } - seen_control = 1; + + /* Clear the connection on error or close. */ + relay_thread_close_connection(&events, + pollfd, + ctrl_conn); } + seen_control = 1; } else if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) { relay_thread_close_connection(&events, pollfd, ctrl_conn); @@ -2658,9 +3026,30 @@ restart: assert(data_conn->type == RELAY_DATA); if (revents & LPOLLIN) { - ret = relay_process_data(data_conn); - /* Connection closed */ - if (ret < 0) { + enum relay_connection_status status; + + status = relay_process_data(data_conn); + /* Connection closed or error. */ + if (status != RELAY_CONNECTION_STATUS_OK) { + /* + * On socket error flag the session as aborted to force + * the cleanup of its stream otherwise it can leak + * during the lifetime of the relayd. + * + * This prevents situations in which streams can be + * left opened because an index was received, the + * control connection is closed, and the data + * connection is closed (uncleanly) before the packet's + * data provided. + * + * Since the data connection encountered an error, + * it is okay to be conservative and close the + * session right now as we can't rely on the protocol + * being respected anymore. + */ + if (status == RELAY_CONNECTION_STATUS_ERROR) { + session_abort(data_conn->session); + } relay_thread_close_connection(&events, pollfd, data_conn); /* @@ -2699,9 +3088,7 @@ error: sock_n.node) { health_code_update(); - if (session_abort(destroy_conn->session)) { - assert(0); - } + session_abort(destroy_conn->session); /* * No need to grab another ref, because we own diff --git a/src/bin/lttng-relayd/session.c b/src/bin/lttng-relayd/session.c index f76fb4a42..bec521ab5 100644 --- a/src/bin/lttng-relayd/session.c +++ b/src/bin/lttng-relayd/session.c @@ -194,16 +194,8 @@ int session_close(struct relay_session *session) pthread_mutex_lock(&session->lock); DBG("closing session %" PRIu64 ": is conn already closed %d", session->id, session->connection_closed); - if (session->connection_closed) { - ret = -1; - goto unlock; - } session->connection_closed = true; -unlock: pthread_mutex_unlock(&session->lock); - if (ret) { - return ret; - } rcu_read_lock(); cds_lfht_for_each_entry(session->ctf_traces_ht->ht, @@ -238,13 +230,7 @@ int session_abort(struct relay_session *session) pthread_mutex_lock(&session->lock); DBG("aborting session %" PRIu64, session->id); - if (session->aborted) { - ERR("session %" PRIu64 " is already aborted", session->id); - ret = -1; - goto unlock; - } session->aborted = true; -unlock: pthread_mutex_unlock(&session->lock); return ret; } diff --git a/src/bin/lttng-relayd/stream.c b/src/bin/lttng-relayd/stream.c index e9c7ad172..da0b3fd26 100644 --- a/src/bin/lttng-relayd/stream.c +++ b/src/bin/lttng-relayd/stream.c @@ -384,6 +384,7 @@ void try_stream_close(struct relay_stream *stream) * a packet. Since those are sent in that order, we take * care of consumerd crashes. */ + DBG("relay_index_close_partial_fd"); relay_index_close_partial_fd(stream); /* * Use the highest net_seq_num we currently have pending @@ -391,6 +392,7 @@ void try_stream_close(struct relay_stream *stream) * at -1ULL if we cannot find any index. */ stream->last_net_seq_num = relay_index_find_last(stream); + DBG("Updating stream->last_net_seq_num to %" PRIu64, stream->last_net_seq_num); /* Fall-through into the next check. */ } diff --git a/src/common/defaults.h b/src/common/defaults.h index 323fb7216..d88d7b1a6 100644 --- a/src/common/defaults.h +++ b/src/common/defaults.h @@ -286,6 +286,10 @@ */ #define DEFAULT_INET_TCP_TIMEOUT 180 /* sec */ +/* Maximum payload size for a control connection */ + +#define DEFAULT_NETWORK_RELAYD_CTRL_MAX_PAYLOAD_SIZE 134217728 + /* * Default receiving and sending timeout for an application socket. */ diff --git a/src/common/sessiond-comm/inet.c b/src/common/sessiond-comm/inet.c index ed7f5dc16..f70fc93a0 100644 --- a/src/common/sessiond-comm/inet.c +++ b/src/common/sessiond-comm/inet.c @@ -390,18 +390,22 @@ ssize_t lttcomm_recvmsg_inet_sock(struct lttcomm_sock *sock, void *buf, len_last = iov[0].iov_len; ret = recvmsg(sock->fd, &msg, flags); if (ret > 0) { + if (flags & MSG_DONTWAIT) { + goto end; + } iov[0].iov_base += ret; iov[0].iov_len -= ret; assert(ret <= len_last); } } while ((ret > 0 && ret < len_last) || (ret < 0 && errno == EINTR)); + if (ret < 0) { PERROR("recvmsg inet"); } else if (ret > 0) { ret = len; } /* Else ret = 0 meaning an orderly shutdown. */ - +end: return ret; } diff --git a/src/common/sessiond-comm/inet6.c b/src/common/sessiond-comm/inet6.c index 1fd18a962..a041038eb 100644 --- a/src/common/sessiond-comm/inet6.c +++ b/src/common/sessiond-comm/inet6.c @@ -370,6 +370,9 @@ ssize_t lttcomm_recvmsg_inet6_sock(struct lttcomm_sock *sock, void *buf, len_last = iov[0].iov_len; ret = recvmsg(sock->fd, &msg, flags); if (ret > 0) { + if (flags & MSG_DONTWAIT) { + goto end; + } iov[0].iov_base += ret; iov[0].iov_len -= ret; assert(ret <= len_last); @@ -381,7 +384,7 @@ ssize_t lttcomm_recvmsg_inet6_sock(struct lttcomm_sock *sock, void *buf, ret = len; } /* Else ret = 0 meaning an orderly shutdown. */ - +end: return ret; }