+ * 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_command *cmd)
+{
+ int ret, send_ret, index_created = 0;
+ struct relay_session *session = cmd->session;
+ struct lttcomm_relayd_index index_info;
+ struct relay_index *index, *wr_index = NULL;
+ struct lttcomm_relayd_generic_reply reply;
+ struct relay_stream *stream;
+ uint64_t net_seq_num;
+
+ assert(cmd);
+
+ DBG("Relay receiving index");
+
+ if (!session || cmd->version_check_done == 0) {
+ ERR("Trying to close a stream before version check");
+ ret = -1;
+ goto end_no_session;
+ }
+
+ ret = cmd->sock->ops->recvmsg(cmd->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", cmd->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);
+
+ rcu_read_lock();
+ stream = relay_stream_find_by_id(be64toh(index_info.relay_stream_id));
+ if (!stream) {
+ ret = -1;
+ goto end_rcu_unlock;
+ }
+
+ /* 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.
+ */
+ if (stream->total_index_received > 0) {
+ stream->beacon_ts_end = be64toh(index_info.timestamp_end);
+ }
+ ret = 0;
+ goto end_rcu_unlock;
+ } else {
+ stream->beacon_ts_end = -1ULL;
+ }
+
+ index = relay_index_find(stream->stream_handle, net_seq_num);
+ if (!index) {
+ /* A successful creation will add the object to the HT. */
+ index = relay_index_create(stream->stream_handle, net_seq_num);
+ if (!index) {
+ goto end_rcu_unlock;
+ }
+ index_created = 1;
+ }
+
+ copy_index_control_data(index, &index_info);
+
+ if (index_created) {
+ /*
+ * Try to add the relay index object to the hash table. If an object
+ * already exist, destroy back the index created, set the data in this
+ * object and write it on disk.
+ */
+ relay_index_add(index, &wr_index);
+ if (wr_index) {
+ copy_index_control_data(wr_index, &index_info);
+ 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) {
+ /* Starting at 2.4, create the index file if none available. */
+ if (cmd->minor >= 4 && stream->index_fd < 0) {
+ ret = index_create_file(stream->path_name, stream->channel_name,
+ relayd_uid, relayd_gid, stream->tracefile_size,
+ stream->tracefile_count_current);
+ if (ret < 0) {
+ goto end_rcu_unlock;
+ }
+ stream->index_fd = ret;
+ }
+
+ ret = relay_index_write(wr_index->fd, wr_index);
+ if (ret < 0) {
+ goto end_rcu_unlock;
+ }
+ stream->total_index_received++;
+ }
+
+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 = cmd->sock->ops->sendmsg(cmd->sock, &reply, sizeof(reply), 0);
+ if (send_ret < 0) {
+ ERR("Relay sending close index id reply");
+ ret = send_ret;
+ }
+
+end_no_session:
+ return ret;
+}
+
+/*
+ * Process the commands received on the control socket