+ 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;
+}
+
+static enum relay_connection_status relay_process_data_receive_header(
+ struct relay_connection *conn)
+{
+ 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;
+
+ 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;
+ }
+ 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;
+ }
+
+ 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);
+ goto end;
+ }
+
+ /* Transition to next state: receiving the payload. */
+ conn->protocol.data.state_id = DATA_CONNECTION_STATE_RECEIVE_PAYLOAD;
+
+ 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);
+ /* Prepare stream for the reception of a new packet. */
+ ret = stream_init_packet(stream, header.data_size,
+ &conn->protocol.data.state.receive_payload.rotate_index);
+ pthread_mutex_unlock(&stream->lock);
+ if (ret) {
+ ERR("Failed to rotate stream output file");
+ status = RELAY_CONNECTION_STATUS_ERROR;
+ goto end_stream_unlock;
+ }
+
+end_stream_unlock:
+ 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, index_flushed = 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;