+ stream_put(stream);
+ stream = NULL;
+ }
+ return stream;
+
+error_no_alloc:
+ /*
+ * path_name and channel_name need to be freed explicitly here
+ * because we cannot rely on stream_put().
+ */
+ free(path_name);
+ free(channel_name);
+ return NULL;
+}
+
+/*
+ * Called with the session lock held.
+ */
+void stream_publish(struct relay_stream *stream)
+{
+ struct relay_session *session;
+
+ pthread_mutex_lock(&stream->lock);
+ if (stream->published) {
+ goto unlock;
+ }
+
+ session = stream->trace->session;
+
+ pthread_mutex_lock(&session->recv_list_lock);
+ if (stream->in_recv_list) {
+ cds_list_del_rcu(&stream->recv_node);
+ stream->in_recv_list = false;
+ }
+ pthread_mutex_unlock(&session->recv_list_lock);
+
+ pthread_mutex_lock(&stream->trace->stream_list_lock);
+ cds_list_add_rcu(&stream->stream_node, &stream->trace->stream_list);
+ pthread_mutex_unlock(&stream->trace->stream_list_lock);
+
+ stream->published = true;
+unlock:
+ pthread_mutex_unlock(&stream->lock);
+}
+
+/*
+ * Stream must be protected by holding the stream lock or by virtue of being
+ * called from stream_destroy, in which case it is guaranteed to be accessed
+ * from a single thread by the reflock.
+ */
+static void stream_unpublish(struct relay_stream *stream)
+{
+ if (stream->in_stream_ht) {
+ struct lttng_ht_iter iter;
+ int ret;
+
+ iter.iter.node = &stream->node.node;
+ ret = lttng_ht_del(relay_streams_ht, &iter);
+ assert(!ret);
+ stream->in_stream_ht = false;
+ }
+ if (stream->published) {
+ pthread_mutex_lock(&stream->trace->stream_list_lock);
+ cds_list_del_rcu(&stream->stream_node);
+ pthread_mutex_unlock(&stream->trace->stream_list_lock);
+ stream->published = false;