+ session_id = be64toh(msg.session_id);
+
+ /*
+ * Iterate over all streams to set the begin data pending flag.
+ * For now, the streams are indexed by stream handle so we have
+ * to iterate over all streams to find the one associated with
+ * the right session_id.
+ */
+ rcu_read_lock();
+ cds_lfht_for_each_entry(relay_streams_ht->ht, &iter.iter, stream,
+ node.node) {
+ if (!stream_get(stream)) {
+ continue;
+ }
+ if (stream->trace->session->id == session_id) {
+ pthread_mutex_lock(&stream->lock);
+ stream->data_pending_check_done = false;
+ pthread_mutex_unlock(&stream->lock);
+ DBG("Set begin data pending flag to stream %" PRIu64,
+ stream->stream_handle);
+ }
+ stream_put(stream);
+ }
+ rcu_read_unlock();
+
+ memset(&reply, 0, sizeof(reply));
+ /* 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");
+ }
+
+end_no_session:
+ return ret;
+}
+
+/*
+ * End data pending command. This will check, for a given session id, if
+ * each stream associated with it has its data_pending_check_done flag
+ * set. If not, this means that the client lost track of the stream but
+ * the data is still being streamed on our side. In this case, we inform
+ * the client that data is in flight.
+ *
+ * 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)
+{
+ int 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) {
+ 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);
+ }
+ ret = -1;
+ goto end_no_session;
+ }
+
+ session_id = be64toh(msg.session_id);
+
+ /*
+ * Iterate over all streams to see if the begin data pending
+ * flag is set.
+ */
+ rcu_read_lock();
+ cds_lfht_for_each_entry(relay_streams_ht->ht, &iter.iter, stream,
+ node.node) {
+ if (!stream_get(stream)) {
+ continue;
+ }
+ if (stream->trace->session->id != session_id) {
+ stream_put(stream);
+ continue;
+ }
+ pthread_mutex_lock(&stream->lock);
+ if (!stream->data_pending_check_done) {
+ if (!stream->closed || !(((int64_t) (stream->prev_seq - stream->last_net_seq_num)) >= 0)) {
+ is_data_inflight = 1;
+ DBG("Data is still in flight for stream %" PRIu64,
+ stream->stream_handle);
+ pthread_mutex_unlock(&stream->lock);
+ stream_put(stream);
+ break;
+ }
+ }
+ pthread_mutex_unlock(&stream->lock);
+ stream_put(stream);
+ }
+ rcu_read_unlock();
+
+ memset(&reply, 0, sizeof(reply));
+ /* 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");
+ }
+
+end_no_session:
+ return ret;
+}
+
+/*
+ * Receive an index for a specific stream.
+ *
+ * Return 0 on success else a negative value.
+ */
+static int relay_recv_index(struct lttcomm_relayd_hdr *recv_hdr,
+ struct relay_connection *conn)
+{
+ int ret, 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;
+
+ assert(conn);
+
+ DBG("Relay receiving index");
+
+ if (!session || conn->version_check_done == 0) {
+ ERR("Trying to close a stream before version check");
+ ret = -1;
+ goto end_no_session;
+ }
+
+ ret = conn->sock->ops->recvmsg(conn->sock, &index_info,
+ sizeof(index_info), 0);
+ if (ret < sizeof(index_info)) {
+ 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);
+ }
+ ret = -1;
+ goto end_no_session;
+ }
+
+ net_seq_num = be64toh(index_info.net_seq_num);
+
+ stream = stream_get_by_id(be64toh(index_info.relay_stream_id));
+ if (!stream) {
+ ERR("stream_get_by_id not found");
+ ret = -1;
+ goto end;
+ }
+ pthread_mutex_lock(&stream->lock);
+
+ /* Live beacon handling */
+ if (index_info.packet_size == 0) {
+ DBG("Received live beacon for stream %" PRIu64,
+ stream->stream_handle);
+
+ /*
+ * Only flag a stream inactive when it has already
+ * received data and no indexes are in flight.
+ */
+ if (stream->total_index_received > 0
+ && stream->indexes_in_flight == 0) {
+ stream->beacon_ts_end =
+ be64toh(index_info.timestamp_end);
+ }
+ ret = 0;
+ goto end_stream_put;
+ } else {
+ stream->beacon_ts_end = -1ULL;
+ }
+
+ if (stream->ctf_stream_id == -1ULL) {
+ stream->ctf_stream_id = be64toh(index_info.stream_id);
+ }
+ index = relay_index_get_by_id_or_create(stream, net_seq_num);
+ if (!index) {
+ ret = -1;
+ ERR("relay_index_get_by_id_or_create index NULL");
+ goto end_stream_put;
+ }
+ if (set_index_control_data(index, &index_info)) {
+ ERR("set_index_control_data error");
+ relay_index_put(index);
+ ret = -1;
+ goto end_stream_put;
+ }
+ ret = relay_index_try_flush(index);
+ if (ret == 0) {
+ stream->total_index_received++;
+ } else if (ret > 0) {
+ /* no flush. */
+ ret = 0;
+ } else {
+ ERR("relay_index_try_flush error %d", ret);
+ relay_index_put(index);
+ ret = -1;
+ }
+
+end_stream_put:
+ pthread_mutex_unlock(&stream->lock);
+ stream_put(stream);
+
+end:
+
+ memset(&reply, 0, sizeof(reply));
+ 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;
+ }
+
+ /*
+ * Publish every pending stream in the connection recv list which are
+ * now ready to be used by the viewer.
+ */
+ publish_connection_local_streams(conn);
+
+ 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;
+ } 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.
+ *
+ * Called with the stream lock held.
+ *
+ * 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;
+ uint64_t data_offset;
+ struct relay_index *index;
+
+ /* Get data offset because we are about to update the index. */
+ data_offset = htobe64(stream->tracefile_size_current);
+
+ DBG("handle_index_data: stream %" PRIu64 " data offset %" PRIu64,
+ stream->stream_handle, stream->tracefile_size_current);
+
+ /*
+ * Lookup for an existing index for that stream id/sequence
+ * number. If it exists, the control thread has already received the
+ * data for it, thus we need to write it to disk.
+ */
+ index = relay_index_get_by_id_or_create(stream, net_seq_num);
+ if (!index) {
+ ret = -1;
+ goto end;
+ }
+
+ if (rotate_index || !stream->index_fd) {
+ int fd;
+
+ /* Put ref on previous index_fd. */
+ if (stream->index_fd) {
+ stream_fd_put(stream->index_fd);
+ stream->index_fd = NULL;
+ }
+
+ fd = index_create_file(stream->path_name, stream->channel_name,
+ -1, -1, stream->tracefile_size,
+ stream->current_tracefile_id);
+ if (fd < 0) {
+ ret = -1;
+ /* Put self-ref for this index due to error. */
+ relay_index_put(index);
+ goto end;
+ }
+ stream->index_fd = stream_fd_create(fd);
+ if (!stream->index_fd) {
+ ret = -1;
+ if (close(fd)) {
+ PERROR("Error closing FD %d", fd);
+ }
+ /* Put self-ref for this index due to error. */
+ relay_index_put(index);
+ /* Will put the local ref. */
+ goto end;
+ }
+ }
+
+ if (relay_index_set_fd(index, stream->index_fd, data_offset)) {
+ ret = -1;
+ /* Put self-ref for this index due to error. */
+ relay_index_put(index);
+ goto end;
+ }
+
+ ret = relay_index_try_flush(index);
+ if (ret == 0) {
+ stream->total_index_received++;
+ } else if (ret > 0) {
+ /* No flush. */
+ ret = 0;
+ } else {
+ /* Put self-ref for this index due to error. */
+ relay_index_put(index);
+ ret = -1;
+ }
+end:
+ 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;