+
+end_rcu_unlock:
+ rcu_read_unlock();
+
+ if (ret < 0) {
+ reply.ret_code = htobe32(LTTNG_ERR_UNK);
+ } else {
+ 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;
+ }
+
+end_no_session:
+ return ret;
+}
+
+/*
+ * Receive the streams_sent message.
+ *
+ * Return 0 on success else a negative value.
+ */
+static
+int relay_streams_sent(struct lttcomm_relayd_hdr *recv_hdr,
+ struct relay_connection *conn)
+{
+ int ret, send_ret;
+ struct lttcomm_relayd_generic_reply reply;
+
+ assert(conn);
+
+ DBG("Relay receiving streams_sent");
+
+ if (!conn->session || conn->version_check_done == 0) {
+ ERR("Trying to close a stream before version check");
+ ret = -1;
+ goto end_no_session;
+ }
+
+ /*
+ * Flag every pending stream in the connection recv list that they are
+ * ready to be used by the viewer.
+ */
+ set_viewer_ready_flag(conn);
+
+ /*
+ * Inform the viewer that there are new streams in the session.
+ */
+ if (conn->session->viewer_refcount) {
+ uatomic_set(&conn->session->new_streams, 1);
+ }
+
+ 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;
+ } else {
+ /* Success. */
+ ret = 0;
+ }
+
+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)
+{
+ int ret = 0;
+
+ switch (be32toh(recv_hdr->cmd)) {
+ case RELAYD_CREATE_SESSION:
+ ret = relay_create_session(recv_hdr, conn);
+ break;
+ case RELAYD_ADD_STREAM:
+ ret = relay_add_stream(recv_hdr, conn);
+ break;
+ case RELAYD_START_DATA:
+ ret = relay_start(recv_hdr, conn);
+ break;
+ case RELAYD_SEND_METADATA:
+ ret = relay_recv_metadata(recv_hdr, conn);
+ break;
+ case RELAYD_VERSION:
+ ret = relay_send_version(recv_hdr, conn);
+ break;
+ case RELAYD_CLOSE_STREAM:
+ ret = relay_close_stream(recv_hdr, conn);
+ break;
+ case RELAYD_DATA_PENDING:
+ ret = relay_data_pending(recv_hdr, conn);
+ break;
+ case RELAYD_QUIESCENT_CONTROL:
+ ret = relay_quiescent_control(recv_hdr, conn);
+ break;
+ case RELAYD_BEGIN_DATA_PENDING:
+ ret = relay_begin_data_pending(recv_hdr, conn);
+ break;
+ case RELAYD_END_DATA_PENDING:
+ ret = relay_end_data_pending(recv_hdr, conn);
+ break;
+ case RELAYD_SEND_INDEX:
+ ret = relay_recv_index(recv_hdr, conn);
+ break;
+ case RELAYD_STREAMS_SENT:
+ ret = relay_streams_sent(recv_hdr, conn);
+ break;
+ case RELAYD_UPDATE_SYNC_INFO:
+ default:
+ ERR("Received unknown command (%u)", be32toh(recv_hdr->cmd));
+ relay_unknown_command(conn);
+ ret = -1;
+ goto end;
+ }
+
+end:
+ return ret;
+}
+
+/*
+ * Handle index for a data stream.
+ *
+ * RCU read side lock MUST be acquired.
+ *
+ * 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)
+{
+ int ret = 0, index_created = 0;
+ uint64_t stream_id, data_offset;
+ struct relay_index *index, *wr_index = NULL;
+
+ assert(stream);
+
+ stream_id = stream->stream_handle;
+ /* Get data offset because we are about to update the index. */
+ data_offset = htobe64(stream->tracefile_size_current);
+
+ /*
+ * Lookup for an existing index for that stream id/sequence number. If on
+ * exists, the control thread already received the data for it thus we need
+ * to write it on disk.
+ */
+ index = relay_index_find(stream_id, net_seq_num);
+ if (!index) {
+ /* A successful creation will add the object to the HT. */
+ index = relay_index_create(stream_id, net_seq_num);
+ if (!index) {
+ ret = -1;
+ goto error;
+ }
+ index_created = 1;
+ }
+
+ if (rotate_index || stream->index_fd < 0) {
+ index->to_close_fd = stream->index_fd;
+ ret = index_create_file(stream->path_name, stream->channel_name,
+ relayd_uid, relayd_gid, stream->tracefile_size,
+ stream->tracefile_count_current);
+ if (ret < 0) {
+ /* This will close the stream's index fd if one. */
+ relay_index_free_safe(index);
+ goto error;
+ }
+ stream->index_fd = ret;
+ }
+ index->fd = stream->index_fd;
+ index->index_data.offset = data_offset;
+
+ if (index_created) {
+ /*
+ * Try to add the relay index object to the hash table. If an object
+ * already exist, destroy back the index created and set the data.
+ */
+ relay_index_add(index, &wr_index);
+ if (wr_index) {
+ /* Copy back data from the created index. */
+ wr_index->fd = index->fd;
+ wr_index->to_close_fd = index->to_close_fd;
+ wr_index->index_data.offset = data_offset;
+ free(index);
+ }
+ } else {
+ /* The index already exists so write it on disk. */
+ wr_index = index;
+ }
+
+ /* Do we have a writable ready index to write on disk. */
+ if (wr_index) {
+ ret = relay_index_write(wr_index->fd, wr_index);
+ if (ret < 0) {
+ goto error;
+ }
+ stream->total_index_received++;
+ }
+
+error:
+ return ret;
+}
+
+/*
+ * relay_process_data: Process the data received on the data socket
+ */
+static
+int relay_process_data(struct relay_connection *conn)
+{
+ int ret = 0, rotate_index = 0;
+ ssize_t size_ret;
+ 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;
+
+ assert(conn);
+
+ 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);
+ }
+ ret = -1;
+ goto end;
+ }
+
+ stream_id = be64toh(data_hdr.stream_id);
+
+ rcu_read_lock();
+ stream = stream_find_by_id(relay_streams_ht, stream_id);
+ if (!stream) {
+ ret = -1;
+ goto end_rcu_unlock;
+ }
+
+ session = session_find_by_id(conn->sessions_ht, stream->session_id);
+ assert(session);
+
+ data_size = be32toh(data_hdr.data_size);
+ if (data_buffer_size < data_size) {
+ char *tmp_data_ptr;
+
+ tmp_data_ptr = realloc(data_buffer, data_size);
+ if (!tmp_data_ptr) {
+ ERR("Allocating data buffer");
+ free(data_buffer);
+ ret = -1;
+ goto end_rcu_unlock;
+ }
+ data_buffer = tmp_data_ptr;
+ data_buffer_size = data_size;
+ }
+ memset(data_buffer, 0, data_size);
+
+ net_seq_num = be64toh(data_hdr.net_seq_num);
+
+ DBG3("Receiving data of size %u for stream id %" PRIu64 " seqnum %" PRIu64,
+ data_size, stream_id, net_seq_num);
+ ret = conn->sock->ops->recvmsg(conn->sock, data_buffer, data_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);
+ }
+ ret = -1;
+ goto end_rcu_unlock;
+ }
+
+ /* Check if a rotation is needed. */
+ if (stream->tracefile_size > 0 &&
+ (stream->tracefile_size_current + data_size) >
+ stream->tracefile_size) {
+ struct relay_viewer_stream *vstream;
+ uint64_t new_id;
+
+ new_id = (stream->tracefile_count_current + 1) %
+ stream->tracefile_count;
+ /*
+ * When we wrap-around back to 0, we start overwriting old
+ * trace data.
+ */
+ if (!stream->tracefile_overwrite && new_id == 0) {
+ stream->tracefile_overwrite = 1;
+ }
+ pthread_mutex_lock(&stream->viewer_stream_rotation_lock);
+ if (stream->tracefile_overwrite) {
+ stream->oldest_tracefile_id =
+ (stream->oldest_tracefile_id + 1) %
+ stream->tracefile_count;
+ }
+ vstream = viewer_stream_find_by_id(stream->stream_handle);
+ if (vstream) {
+ /*
+ * The viewer is reading a file about to be
+ * overwritten. Close the FDs it is
+ * currently using and let it handle the fault.
+ */
+ if (vstream->tracefile_count_current == new_id) {
+ pthread_mutex_lock(&vstream->overwrite_lock);
+ vstream->abort_flag = 1;
+ pthread_mutex_unlock(&vstream->overwrite_lock);
+ DBG("Streaming side setting abort_flag on stream %s_%lu\n",
+ stream->channel_name, new_id);
+ } else if (vstream->tracefile_count_current ==
+ stream->tracefile_count_current) {
+ /*
+ * The reader and writer were in the
+ * same trace file, inform the viewer
+ * that no new index will ever be added
+ * to this file.
+ */
+ vstream->close_write_flag = 1;
+ }
+ }
+ ret = utils_rotate_stream_file(stream->path_name, stream->channel_name,
+ stream->tracefile_size, stream->tracefile_count,
+ relayd_uid, relayd_gid, stream->fd,
+ &(stream->tracefile_count_current), &stream->fd);
+ stream->total_index_received = 0;
+ pthread_mutex_unlock(&stream->viewer_stream_rotation_lock);
+ if (ret < 0) {
+ ERR("Rotating stream output file");
+ goto end_rcu_unlock;
+ }
+ /* Reset current size because we just perform a stream rotation. */
+ stream->tracefile_size_current = 0;
+ rotate_index = 1;
+ }
+
+ /*
+ * 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) {
+ goto end_rcu_unlock;
+ }
+ }
+
+ /* Write data to stream output fd. */
+ size_ret = lttng_write(stream->fd, data_buffer, data_size);
+ if (size_ret < data_size) {
+ ERR("Relay error writing data to file");
+ ret = -1;
+ goto end_rcu_unlock;
+ }
+
+ DBG2("Relay wrote %d bytes to tracefile for stream id %" PRIu64,
+ ret, stream->stream_handle);
+
+ ret = write_padding_to_file(stream->fd, be32toh(data_hdr.padding_size));
+ if (ret < 0) {
+ goto end_rcu_unlock;
+ }
+ stream->tracefile_size_current += data_size + be32toh(data_hdr.padding_size);
+
+ stream->prev_seq = net_seq_num;
+
+ try_close_stream(session, stream);
+
+end_rcu_unlock:
+ rcu_read_unlock();
+end:
+ return ret;
+}
+
+static
+void cleanup_connection_pollfd(struct lttng_poll_event *events, int pollfd)
+{
+ int ret;
+
+ assert(events);
+
+ (void) lttng_poll_del(events, pollfd);
+
+ ret = close(pollfd);
+ if (ret < 0) {
+ ERR("Closing pollfd %d", pollfd);
+ }
+}
+
+static void destroy_connection(struct lttng_ht *relay_connections_ht,
+ struct relay_connection *conn)
+{
+ assert(relay_connections_ht);
+ assert(conn);
+
+ connection_delete(relay_connections_ht, conn);
+
+ /* For the control socket, we try to destroy the session. */
+ if (conn->type == RELAY_CONTROL) {
+ destroy_session(conn->session, conn->sessions_ht);
+ }
+
+ connection_destroy(conn);