+ bt_self_message_iterator_status status;
+ struct lttng_live_msg_iter *lttng_live_msg_iter =
+ bt_self_message_iterator_get_data(self_msg_it);
+ struct lttng_live_component *lttng_live =
+ lttng_live_msg_iter->lttng_live_comp;
+ enum lttng_live_iterator_status stream_iter_status;
+ uint64_t session_idx;
+
+ *count = 0;
+
+ BT_ASSERT(lttng_live_msg_iter);
+
+ /*
+ * Clear all the invalid message reference that might be left over in
+ * the output array.
+ */
+ memset(msgs, 0, capacity * sizeof(*msgs));
+
+ /*
+ * If no session are exposed on the relay found at the url provided by
+ * the user, session count will be 0. In this case, we return status
+ * end to return gracefully.
+ */
+ if (lttng_live_msg_iter->sessions->len == 0) {
+ if (lttng_live->params.sess_not_found_act !=
+ SESSION_NOT_FOUND_ACTION_CONTINUE) {
+ status = BT_SELF_MESSAGE_ITERATOR_STATUS_END;
+ goto no_session;
+ } else {
+ /*
+ * The are no more active session for this session
+ * name. Retry to create a viewer session for the
+ * requested session name.
+ */
+ if (lttng_live_create_viewer_session(lttng_live_msg_iter)) {
+ status = BT_SELF_MESSAGE_ITERATOR_STATUS_ERROR;
+ goto no_session;
+ }
+ }
+ }
+
+ if (lttng_live_msg_iter->active_stream_iter == 0) {
+ lttng_live_force_new_streams_and_metadata(lttng_live_msg_iter);
+ }
+
+ /*
+ * Here the muxing of message is done.
+ *
+ * We need to iterate over all the streams of all the traces of all the
+ * viewer sessions in order to get the message with the smallest
+ * timestamp. In this case, a session is a viewer session and there is
+ * one viewer session per consumer daemon. (UST 32bit, UST 64bit and/or
+ * kernel). Each viewer session can have multiple traces, for example,
+ * 64bit UST viewer sessions could have multiple per-pid traces.
+ *
+ * We iterate over the streams of each traces to update and see what is
+ * their next message's timestamp. From those timestamps, we select the
+ * message with the smallest timestamp as the best candidate message
+ * for that trace and do the same thing across all the sessions.
+ *
+ * We then compare the timestamp of best candidate message of all the
+ * sessions to pick the message with the smallest timestamp and we
+ * return it.
+ */
+ while (*count < capacity) {
+ struct lttng_live_stream_iterator *next_stream_iter = NULL,
+ *candidate_stream_iter = NULL;
+ int64_t next_msg_ts_ns = INT64_MAX;
+
+ BT_ASSERT(lttng_live_msg_iter->sessions);
+ session_idx = 0;
+ /*
+ * Use a while loop instead of a for loop so we can restart the
+ * iteration if we remove an element. We can safely call
+ * next_stream_iterator_for_session() multiple times on the
+ * same session as we only fetch a new message if there is no
+ * current next message for each live stream iterator.
+ * If all live stream iterator of that session already have a
+ * current next message, the function will simply exit return
+ * the same candidate live stream iterator every time.
+ */
+ while (session_idx < lttng_live_msg_iter->sessions->len) {
+ struct lttng_live_session *session =
+ g_ptr_array_index(lttng_live_msg_iter->sessions,
+ session_idx);
+
+ /* Find the best candidate message to send downstream. */
+ stream_iter_status = next_stream_iterator_for_session(
+ lttng_live_msg_iter, session,
+ &candidate_stream_iter);
+
+ /* If we receive an END status, it means that either:
+ * - Those traces never had active streams (UST with no
+ * data produced yet),
+ * - All live stream iterators have ENDed.*/
+ if (stream_iter_status == LTTNG_LIVE_ITERATOR_STATUS_END) {
+ if (session->closed && session->traces->len == 0) {
+ /*
+ * Remove the session from the list and restart the
+ * iteration at the beginning of the array since the
+ * removal shuffle the elements of the array.
+ */
+ g_ptr_array_remove_index_fast(
+ lttng_live_msg_iter->sessions,
+ session_idx);
+ session_idx = 0;
+ } else {
+ session_idx++;
+ }
+ continue;
+ }
+
+ if (stream_iter_status != LTTNG_LIVE_ITERATOR_STATUS_OK) {
+ goto end;
+ }
+
+ if (candidate_stream_iter->current_msg_ts_ns <= next_msg_ts_ns) {
+ next_msg_ts_ns = candidate_stream_iter->current_msg_ts_ns;
+ next_stream_iter = candidate_stream_iter;
+ }
+
+ session_idx++;
+ }
+
+ if (!next_stream_iter) {
+ stream_iter_status = LTTNG_LIVE_ITERATOR_STATUS_AGAIN;
+ goto end;
+ }
+
+ BT_ASSERT(next_stream_iter->current_msg);
+ /* Ensure monotonicity. */
+ BT_ASSERT(lttng_live_msg_iter->last_msg_ts_ns <=
+ next_stream_iter->current_msg_ts_ns);
+
+ /*
+ * Insert the next message to the message batch. This will set
+ * stream iterator current messsage to NULL so that next time
+ * we fetch the next message of that stream iterator
+ */
+ BT_MESSAGE_MOVE_REF(msgs[*count], next_stream_iter->current_msg);
+ (*count)++;
+
+ /* Update the last timestamp in nanoseconds sent downstream. */
+ lttng_live_msg_iter->last_msg_ts_ns = next_msg_ts_ns;
+ next_stream_iter->current_msg_ts_ns = INT64_MAX;
+
+ stream_iter_status = LTTNG_LIVE_ITERATOR_STATUS_OK;
+ }
+end:
+ switch (stream_iter_status) {
+ case LTTNG_LIVE_ITERATOR_STATUS_OK:
+ case LTTNG_LIVE_ITERATOR_STATUS_AGAIN:
+ if (*count > 0) {
+ /*
+ * We received a again status but we have some messages
+ * to send downstream. We send them and return OK for
+ * now. On the next call we return again if there are
+ * still no new message to send.
+ */
+ status = BT_SELF_MESSAGE_ITERATOR_STATUS_OK;
+ } else {
+ status = BT_SELF_MESSAGE_ITERATOR_STATUS_AGAIN;
+ }