src/plugins/ctf/lttng-live/lttng-live.cpp

   1 /*
   2  * SPDX-License-Identifier: MIT
   3  *
   4  * Copyright 2019 Francis Deslauriers <francis.deslauriers@efficios.com>
   5  * Copyright 2016 Jérémie Galarneau <jeremie.galarneau@efficios.com>
   6  * Copyright 2016 Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
   7  *
   8  * Babeltrace CTF LTTng-live Client Component
   9  */
  10
  11 #include <glib.h>
  12 #include <unistd.h>
  13
  14 #include "common/assert.h"
  15 #include "cpp-common/bt2c/fmt.hpp"
  16 #include "cpp-common/bt2c/glib-up.hpp"
  17 #include "cpp-common/bt2s/make-unique.hpp"
  18 #include "cpp-common/vendor/fmt/format.h"
  19
  20 #include "plugins/common/muxing/muxing.h"
  21 #include "plugins/common/param-validation/param-validation.h"
  22
  23 #include "data-stream.hpp"
  24 #include "lttng-live.hpp"
  25 #include "metadata.hpp"
  26
  27 #define MAX_QUERY_SIZE                     (256 * 1024)
  28 #define URL_PARAM                          "url"
  29 #define INPUTS_PARAM                       "inputs"
  30 #define SESS_NOT_FOUND_ACTION_PARAM        "session-not-found-action"
  31 #define SESS_NOT_FOUND_ACTION_CONTINUE_STR "continue"
  32 #define SESS_NOT_FOUND_ACTION_FAIL_STR     "fail"
  33 #define SESS_NOT_FOUND_ACTION_END_STR      "end"
  34
  35 void lttng_live_stream_iterator_set_state(struct lttng_live_stream_iterator *stream_iter,
  36                                           enum lttng_live_stream_state new_state)
  37 {
  38     BT_CPPLOGD_SPEC(stream_iter->logger,
  39                     "Setting live stream iterator state: viewer-stream-id={}, "
  40                     "old-state={}, new-state={}",
  41                     stream_iter->viewer_stream_id, stream_iter->state, new_state);
  42
  43     stream_iter->state = new_state;
  44 }
  45
  46 #define LTTNG_LIVE_LOGD_STREAM_ITER(live_stream_iter)                                              \
  47     do {                                                                                           \
  48         BT_CPPLOGD_SPEC((live_stream_iter)->logger,                                                \
  49                         "Live stream iterator state={}, "                                          \
  50                         "last-inact-ts-is-set={}, last-inact-ts-value={}, "                        \
  51                         "curr-inact-ts={}",                                                        \
  52                         (live_stream_iter)->state, (live_stream_iter)->last_inactivity_ts.is_set,  \
  53                         (live_stream_iter)->last_inactivity_ts.value,                              \
  54                         (live_stream_iter)->current_inactivity_ts);                                \
  55     } while (0);
  56
  57 bool lttng_live_graph_is_canceled(struct lttng_live_msg_iter *msg_iter)
  58 {
  59     bool ret;
  60
  61     if (!msg_iter) {
  62         ret = false;
  63         goto end;
  64     }
  65
  66     ret = bt_self_message_iterator_is_interrupted(msg_iter->self_msg_iter);
  67
  68 end:
  69     return ret;
  70 }
  71
  72 static struct lttng_live_trace *
  73 lttng_live_session_borrow_trace_by_id(struct lttng_live_session *session, uint64_t trace_id)
  74 {
  75     uint64_t trace_idx;
  76     struct lttng_live_trace *ret_trace = NULL;
  77
  78     for (trace_idx = 0; trace_idx < session->traces->len; trace_idx++) {
  79         struct lttng_live_trace *trace =
  80             (lttng_live_trace *) g_ptr_array_index(session->traces, trace_idx);
  81         if (trace->id == trace_id) {
  82             ret_trace = trace;
  83             goto end;
  84         }
  85     }
  86
  87 end:
  88     return ret_trace;
  89 }
  90
  91 static void lttng_live_destroy_trace(struct lttng_live_trace *trace)
  92 {
  93     BT_CPPLOGD_SPEC(trace->logger, "Destroying live trace: trace-id={}", trace->id);
  94
  95     BT_ASSERT(trace->stream_iterators);
  96     g_ptr_array_free(trace->stream_iterators, TRUE);
  97
  98     delete trace;
  99 }
 100
 101 static struct lttng_live_trace *lttng_live_create_trace(struct lttng_live_session *session,
 102                                                         uint64_t trace_id)
 103 {
 104     BT_CPPLOGD_SPEC(session->logger, "Creating live trace: session-id={}, trace-id={}", session->id,
 105                     trace_id);
 106
 107     lttng_live_trace *trace = new lttng_live_trace {session->logger};
 108     trace->session = session;
 109     trace->id = trace_id;
 110     trace->stream_iterators =
 111         g_ptr_array_new_with_free_func((GDestroyNotify) lttng_live_stream_iterator_destroy);
 112     BT_ASSERT(trace->stream_iterators);
 113     trace->metadata_stream_state = LTTNG_LIVE_METADATA_STREAM_STATE_NEEDED;
 114     g_ptr_array_add(session->traces, trace);
 115
 116     return trace;
 117 }
 118
 119 struct lttng_live_trace *
 120 lttng_live_session_borrow_or_create_trace_by_id(struct lttng_live_session *session,
 121                                                 uint64_t trace_id)
 122 {
 123     struct lttng_live_trace *trace;
 124
 125     trace = lttng_live_session_borrow_trace_by_id(session, trace_id);
 126     if (trace) {
 127         goto end;
 128     }
 129
 130     /* The session is the owner of the newly created trace. */
 131     trace = lttng_live_create_trace(session, trace_id);
 132
 133 end:
 134     return trace;
 135 }
 136
 137 int lttng_live_add_session(struct lttng_live_msg_iter *lttng_live_msg_iter, uint64_t session_id,
 138                            const char *hostname, const char *session_name)
 139 {
 140     BT_CPPLOGD_SPEC(lttng_live_msg_iter->logger,
 141                     "Adding live session: "
 142                     "session-id={}, hostname=\"{}\", session-name=\"{}\"",
 143                     session_id, hostname, session_name);
 144
 145     lttng_live_session *session = new lttng_live_session {lttng_live_msg_iter->logger};
 146     session->self_comp = lttng_live_msg_iter->self_comp;
 147     session->id = session_id;
 148     session->traces = g_ptr_array_new_with_free_func((GDestroyNotify) lttng_live_destroy_trace);
 149     BT_ASSERT(session->traces);
 150     session->lttng_live_msg_iter = lttng_live_msg_iter;
 151     session->new_streams_needed = true;
 152     session->hostname = g_string_new(hostname);
 153     BT_ASSERT(session->hostname);
 154
 155     session->session_name = g_string_new(session_name);
 156     BT_ASSERT(session->session_name);
 157
 158     g_ptr_array_add(lttng_live_msg_iter->sessions, session);
 159
 160     return 0;
 161 }
 162
 163 static void lttng_live_destroy_session(struct lttng_live_session *session)
 164 {
 165     if (!session) {
 166         goto end;
 167     }
 168
 169     BT_CPPLOGD_SPEC(session->logger,
 170                     "Destroying live session: "
 171                     "session-id={}, session-name=\"{}\"",
 172                     session->id, session->session_name->str);
 173     if (session->id != -1ULL) {
 174         if (lttng_live_session_detach(session)) {
 175             if (!lttng_live_graph_is_canceled(session->lttng_live_msg_iter)) {
 176                 /* Old relayd cannot detach sessions. */
 177                 BT_CPPLOGD_SPEC(session->logger, "Unable to detach lttng live session {}",
 178                                 session->id);
 179             }
 180         }
 181         session->id = -1ULL;
 182     }
 183
 184     if (session->traces) {
 185         g_ptr_array_free(session->traces, TRUE);
 186     }
 187
 188     if (session->hostname) {
 189         g_string_free(session->hostname, TRUE);
 190     }
 191
 192     if (session->session_name) {
 193         g_string_free(session->session_name, TRUE);
 194     }
 195
 196     delete session;
 197
 198 end:
 199     return;
 200 }
 201
 202 static void lttng_live_msg_iter_destroy(struct lttng_live_msg_iter *lttng_live_msg_iter)
 203 {
 204     if (!lttng_live_msg_iter) {
 205         goto end;
 206     }
 207
 208     if (lttng_live_msg_iter->sessions) {
 209         g_ptr_array_free(lttng_live_msg_iter->sessions, TRUE);
 210     }
 211
 212     if (lttng_live_msg_iter->viewer_connection) {
 213         live_viewer_connection_destroy(lttng_live_msg_iter->viewer_connection);
 214     }
 215     BT_ASSERT(lttng_live_msg_iter->lttng_live_comp);
 216     BT_ASSERT(lttng_live_msg_iter->lttng_live_comp->has_msg_iter);
 217
 218     /* All stream iterators must be destroyed at this point. */
 219     BT_ASSERT(lttng_live_msg_iter->active_stream_iter == 0);
 220     lttng_live_msg_iter->lttng_live_comp->has_msg_iter = false;
 221
 222     delete lttng_live_msg_iter;
 223
 224 end:
 225     return;
 226 }
 227
 228 void lttng_live_msg_iter_finalize(bt_self_message_iterator *self_msg_iter)
 229 {
 230     struct lttng_live_msg_iter *lttng_live_msg_iter;
 231
 232     BT_ASSERT(self_msg_iter);
 233
 234     lttng_live_msg_iter =
 235         (struct lttng_live_msg_iter *) bt_self_message_iterator_get_data(self_msg_iter);
 236     BT_ASSERT(lttng_live_msg_iter);
 237     lttng_live_msg_iter_destroy(lttng_live_msg_iter);
 238 }
 239
 240 static enum lttng_live_iterator_status
 241 lttng_live_iterator_next_check_stream_state(struct lttng_live_stream_iterator *lttng_live_stream)
 242 {
 243     switch (lttng_live_stream->state) {
 244     case LTTNG_LIVE_STREAM_QUIESCENT:
 245     case LTTNG_LIVE_STREAM_ACTIVE_DATA:
 246         break;
 247     case LTTNG_LIVE_STREAM_ACTIVE_NO_DATA:
 248         /* Invalid state. */
 249         BT_CPPLOGF_SPEC(lttng_live_stream->logger, "Unexpected stream state \"ACTIVE_NO_DATA\"");
 250         bt_common_abort();
 251     case LTTNG_LIVE_STREAM_QUIESCENT_NO_DATA:
 252         /* Invalid state. */
 253         BT_CPPLOGF_SPEC(lttng_live_stream->logger, "Unexpected stream state \"QUIESCENT_NO_DATA\"");
 254         bt_common_abort();
 255     case LTTNG_LIVE_STREAM_EOF:
 256         break;
 257     }
 258     return LTTNG_LIVE_ITERATOR_STATUS_OK;
 259 }
 260
 261 /*
 262  * For active no data stream, fetch next index. As a result of that it can
 263  * become either:
 264  * - quiescent: won't have events for a bit,
 265  * - have data: need to get that data and produce the event,
 266  * - have no data on this stream at this point: need to retry (AGAIN) or return
 267  *   EOF.
 268  */
 269 static enum lttng_live_iterator_status lttng_live_iterator_next_handle_one_no_data_stream(
 270     struct lttng_live_msg_iter *lttng_live_msg_iter,
 271     struct lttng_live_stream_iterator *lttng_live_stream)
 272 {
 273     enum lttng_live_iterator_status ret = LTTNG_LIVE_ITERATOR_STATUS_OK;
 274     enum lttng_live_stream_state orig_state = lttng_live_stream->state;
 275     struct packet_index index;
 276
 277     if (lttng_live_stream->trace->metadata_stream_state ==
 278         LTTNG_LIVE_METADATA_STREAM_STATE_NEEDED) {
 279         BT_CPPLOGD_SPEC(lttng_live_msg_iter->logger,
 280                         "Need to get an update for the metadata stream before proceeding "
 281                         "further with this stream: stream-name=\"{}\"",
 282                         lttng_live_stream->name);
 283         ret = LTTNG_LIVE_ITERATOR_STATUS_CONTINUE;
 284         goto end;
 285     }
 286
 287     if (lttng_live_stream->trace->session->new_streams_needed) {
 288         BT_CPPLOGD_SPEC(lttng_live_msg_iter->logger,
 289                         "Need to get an update of all streams before proceeding further "
 290                         "with this stream: stream-name=\"{}\"",
 291                         lttng_live_stream->name);
 292         ret = LTTNG_LIVE_ITERATOR_STATUS_CONTINUE;
 293         goto end;
 294     }
 295
 296     if (lttng_live_stream->state != LTTNG_LIVE_STREAM_ACTIVE_NO_DATA &&
 297         lttng_live_stream->state != LTTNG_LIVE_STREAM_QUIESCENT_NO_DATA) {
 298         goto end;
 299     }
 300     ret = lttng_live_get_next_index(lttng_live_msg_iter, lttng_live_stream, &index);
 301     if (ret != LTTNG_LIVE_ITERATOR_STATUS_OK) {
 302         goto end;
 303     }
 304
 305     BT_ASSERT_DBG(lttng_live_stream->state != LTTNG_LIVE_STREAM_EOF);
 306
 307     if (lttng_live_stream->state == LTTNG_LIVE_STREAM_QUIESCENT) {
 308         uint64_t last_inact_ts = lttng_live_stream->last_inactivity_ts.value,
 309                  curr_inact_ts = lttng_live_stream->current_inactivity_ts;
 310
 311         if (orig_state == LTTNG_LIVE_STREAM_QUIESCENT_NO_DATA && last_inact_ts == curr_inact_ts) {
 312             /*
 313              * Because the stream is in the QUIESCENT_NO_DATA
 314              * state, we can assert that the last_inactivity_ts was
 315              * set and can be safely used in the `if` above.
 316              */
 317             BT_ASSERT(lttng_live_stream->last_inactivity_ts.is_set);
 318
 319             ret = LTTNG_LIVE_ITERATOR_STATUS_AGAIN;
 320             LTTNG_LIVE_LOGD_STREAM_ITER(lttng_live_stream);
 321         } else {
 322             ret = LTTNG_LIVE_ITERATOR_STATUS_CONTINUE;
 323         }
 324         goto end;
 325     }
 326
 327     lttng_live_stream->base_offset = index.offset;
 328     lttng_live_stream->offset = index.offset;
 329     lttng_live_stream->len = index.packet_size / CHAR_BIT;
 330
 331     BT_CPPLOGD_SPEC(lttng_live_msg_iter->logger,
 332                     "Setting live stream reading info: stream-name=\"{}\", "
 333                     "viewer-stream-id={}, stream-base-offset={}, stream-offset={}, stream-len={}",
 334                     lttng_live_stream->name, lttng_live_stream->viewer_stream_id,
 335                     lttng_live_stream->base_offset, lttng_live_stream->offset,
 336                     lttng_live_stream->len);
 337
 338 end:
 339     if (ret == LTTNG_LIVE_ITERATOR_STATUS_OK) {
 340         ret = lttng_live_iterator_next_check_stream_state(lttng_live_stream);
 341     }
 342     return ret;
 343 }
 344
 345 /*
 346  * Creation of the message requires the ctf trace class to be created
 347  * beforehand, but the live protocol gives us all streams (including metadata)
 348  * at once. So we split it in three steps: getting streams, getting metadata
 349  * (which creates the ctf trace class), and then creating the per-stream
 350  * messages.
 351  */
 352 static enum lttng_live_iterator_status
 353 lttng_live_get_session(struct lttng_live_msg_iter *lttng_live_msg_iter,
 354                        struct lttng_live_session *session)
 355 {
 356     enum lttng_live_iterator_status status;
 357     uint64_t trace_idx;
 358
 359     if (!session->attached) {
 360         BT_CPPLOGD_SPEC(lttng_live_msg_iter->logger, "Attach to session: session-id={}",
 361                         session->id);
 362         enum lttng_live_viewer_status attach_status =
 363             lttng_live_session_attach(session, lttng_live_msg_iter->self_msg_iter);
 364         if (attach_status != LTTNG_LIVE_VIEWER_STATUS_OK) {
 365             if (lttng_live_graph_is_canceled(lttng_live_msg_iter)) {
 366                 /*
 367                  * Clear any causes appended in
 368                  * `lttng_live_attach_session()` as we want to
 369                  * return gracefully since the graph was
 370                  * cancelled.
 371                  */
 372                 bt_current_thread_clear_error();
 373                 status = LTTNG_LIVE_ITERATOR_STATUS_AGAIN;
 374             } else {
 375                 status = LTTNG_LIVE_ITERATOR_STATUS_ERROR;
 376                 BT_CPPLOGE_APPEND_CAUSE_SPEC(lttng_live_msg_iter->logger,
 377                                              "Error attaching to LTTng live session");
 378             }
 379             goto end;
 380         }
 381     }
 382
 383     BT_CPPLOGD_SPEC(lttng_live_msg_iter->logger,
 384                     "Updating all data streams: "
 385                     "session-id={}, session-name=\"{}\"",
 386                     session->id, session->session_name->str);
 387
 388     status = lttng_live_session_get_new_streams(session, lttng_live_msg_iter->self_msg_iter);
 389     switch (status) {
 390     case LTTNG_LIVE_ITERATOR_STATUS_OK:
 391         break;
 392     case LTTNG_LIVE_ITERATOR_STATUS_END:
 393         /*
 394          * We received a `_END` from the `_get_new_streams()` function,
 395          * which means no more data will ever be received from the data
 396          * streams of this session. But it's possible that the metadata
 397          * is incomplete.
 398          * The live protocol guarantees that we receive all the
 399          * metadata needed before we receive data streams needing it.
 400          * But it's possible to receive metadata NOT needed by
 401          * data streams after the session was closed. For example, this
 402          * could happen if a new event is registered and the session is
 403          * stopped before any tracepoint for that event is actually
 404          * fired.
 405          */
 406         BT_CPPLOGD_SPEC(
 407             lttng_live_msg_iter->logger,
 408             "Updating streams returned _END status. Override status to _OK in order fetch any remaining metadata:"
 409             "session-id={}, session-name=\"{}\"",
 410             session->id, session->session_name->str);
 411         status = LTTNG_LIVE_ITERATOR_STATUS_OK;
 412         break;
 413     default:
 414         goto end;
 415     }
 416
 417     BT_CPPLOGD_SPEC(lttng_live_msg_iter->logger,
 418                     "Updating metadata stream for session: "
 419                     "session-id={}, session-name=\"{}\"",
 420                     session->id, session->session_name->str);
 421
 422     trace_idx = 0;
 423     while (trace_idx < session->traces->len) {
 424         struct lttng_live_trace *trace =
 425             (lttng_live_trace *) g_ptr_array_index(session->traces, trace_idx);
 426
 427         status = lttng_live_metadata_update(trace);
 428         switch (status) {
 429         case LTTNG_LIVE_ITERATOR_STATUS_END:
 430         case LTTNG_LIVE_ITERATOR_STATUS_OK:
 431             trace_idx++;
 432             break;
 433         case LTTNG_LIVE_ITERATOR_STATUS_CONTINUE:
 434         case LTTNG_LIVE_ITERATOR_STATUS_AGAIN:
 435             goto end;
 436         default:
 437             BT_CPPLOGE_APPEND_CAUSE_SPEC(lttng_live_msg_iter->logger,
 438                                          "Error updating trace metadata: "
 439                                          "stream-iter-status={}, trace-id={}",
 440                                          status, trace->id);
 441             goto end;
 442         }
 443     }
 444
 445     /*
 446      * Now that we have the metadata we can initialize the downstream
 447      * iterator.
 448      */
 449     status = lttng_live_lazy_msg_init(session, lttng_live_msg_iter->self_msg_iter);
 450
 451 end:
 452     return status;
 453 }
 454
 455 static void
 456 lttng_live_force_new_streams_and_metadata(struct lttng_live_msg_iter *lttng_live_msg_iter)
 457 {
 458     uint64_t session_idx, trace_idx;
 459
 460     for (session_idx = 0; session_idx < lttng_live_msg_iter->sessions->len; session_idx++) {
 461         struct lttng_live_session *session =
 462             (lttng_live_session *) g_ptr_array_index(lttng_live_msg_iter->sessions, session_idx);
 463         BT_CPPLOGD_SPEC(lttng_live_msg_iter->logger,
 464                         "Force marking session as needing new streams: "
 465                         "session-id={}",
 466                         session->id);
 467         session->new_streams_needed = true;
 468         for (trace_idx = 0; trace_idx < session->traces->len; trace_idx++) {
 469             struct lttng_live_trace *trace =
 470                 (lttng_live_trace *) g_ptr_array_index(session->traces, trace_idx);
 471             BT_CPPLOGD_SPEC(lttng_live_msg_iter->logger,
 472                             "Force marking trace metadata state as needing an update: "
 473                             "session-id={}, trace-id={}",
 474                             session->id, trace->id);
 475
 476             BT_ASSERT(trace->metadata_stream_state != LTTNG_LIVE_METADATA_STREAM_STATE_CLOSED);
 477
 478             trace->metadata_stream_state = LTTNG_LIVE_METADATA_STREAM_STATE_NEEDED;
 479         }
 480     }
 481 }
 482
 483 static enum lttng_live_iterator_status
 484 lttng_live_iterator_handle_new_streams_and_metadata(struct lttng_live_msg_iter *lttng_live_msg_iter)
 485 {
 486     enum lttng_live_iterator_status status;
 487     enum lttng_live_viewer_status viewer_status;
 488     uint64_t session_idx = 0, nr_sessions_opened = 0;
 489     struct lttng_live_session *session;
 490     enum session_not_found_action sess_not_found_act =
 491         lttng_live_msg_iter->lttng_live_comp->params.sess_not_found_act;
 492
 493     BT_CPPLOGD_SPEC(lttng_live_msg_iter->logger,
 494                     "Update data and metadata of all sessions: "
 495                     "live-msg-iter-addr={}",
 496                     fmt::ptr(lttng_live_msg_iter));
 497     /*
 498      * In a remotely distant future, we could add a "new
 499      * session" flag to the protocol, which would tell us that we
 500      * need to query for new sessions even though we have sessions
 501      * currently ongoing.
 502      */
 503     if (lttng_live_msg_iter->sessions->len == 0) {
 504         if (sess_not_found_act != SESSION_NOT_FOUND_ACTION_CONTINUE) {
 505             BT_CPPLOGD_SPEC(
 506                 lttng_live_msg_iter->logger,
 507                 "No session found. Exiting in accordance with the `session-not-found-action` parameter");
 508             status = LTTNG_LIVE_ITERATOR_STATUS_END;
 509             goto end;
 510         } else {
 511             BT_CPPLOGD_SPEC(
 512                 lttng_live_msg_iter->logger,
 513                 "No session found. Try creating a new one in accordance with the `session-not-found-action` parameter");
 514             /*
 515              * Retry to create a viewer session for the requested
 516              * session name.
 517              */
 518             viewer_status = lttng_live_create_viewer_session(lttng_live_msg_iter);
 519             if (viewer_status != LTTNG_LIVE_VIEWER_STATUS_OK) {
 520                 if (viewer_status == LTTNG_LIVE_VIEWER_STATUS_ERROR) {
 521                     status = LTTNG_LIVE_ITERATOR_STATUS_ERROR;
 522                     BT_CPPLOGE_APPEND_CAUSE_SPEC(lttng_live_msg_iter->logger,
 523                                                  "Error creating LTTng live viewer session");
 524                 } else if (viewer_status == LTTNG_LIVE_VIEWER_STATUS_INTERRUPTED) {
 525                     status = LTTNG_LIVE_ITERATOR_STATUS_AGAIN;
 526                 } else {
 527                     bt_common_abort();
 528                 }
 529                 goto end;
 530             }
 531         }
 532     }
 533
 534     for (session_idx = 0; session_idx < lttng_live_msg_iter->sessions->len; session_idx++) {
 535         session =
 536             (lttng_live_session *) g_ptr_array_index(lttng_live_msg_iter->sessions, session_idx);
 537         status = lttng_live_get_session(lttng_live_msg_iter, session);
 538         switch (status) {
 539         case LTTNG_LIVE_ITERATOR_STATUS_OK:
 540         case LTTNG_LIVE_ITERATOR_STATUS_END:
 541             /*
 542              * A session returned `_END`. Other sessions may still
 543              * be active so we override the status and continue
 544              * looping if needed.
 545              */
 546             break;
 547         default:
 548             goto end;
 549         }
 550         if (!session->closed) {
 551             nr_sessions_opened++;
 552         }
 553     }
 554
 555     if (sess_not_found_act != SESSION_NOT_FOUND_ACTION_CONTINUE && nr_sessions_opened == 0) {
 556         status = LTTNG_LIVE_ITERATOR_STATUS_END;
 557     } else {
 558         status = LTTNG_LIVE_ITERATOR_STATUS_OK;
 559     }
 560
 561 end:
 562     return status;
 563 }
 564
 565 static enum lttng_live_iterator_status
 566 emit_inactivity_message(struct lttng_live_msg_iter *lttng_live_msg_iter,
 567                         struct lttng_live_stream_iterator *stream_iter,
 568                         bt2::ConstMessage::Shared& message, uint64_t timestamp)
 569 {
 570     BT_ASSERT(stream_iter->trace->clock_class);
 571     BT_CPPLOGD_SPEC(lttng_live_msg_iter->logger,
 572                     "Emitting inactivity message for stream: ctf-stream-id={}, "
 573                     "viewer-stream-id={}, timestamp={}",
 574                     stream_iter->ctf_stream_class_id.value, stream_iter->viewer_stream_id,
 575                     timestamp);
 576
 577     const auto msg = bt_message_message_iterator_inactivity_create(
 578         lttng_live_msg_iter->self_msg_iter, stream_iter->trace->clock_class, timestamp);
 579
 580     if (!msg) {
 581         BT_CPPLOGE_APPEND_CAUSE_SPEC(lttng_live_msg_iter->logger,
 582                                      "Error emitting message iterator inactivity message");
 583         return LTTNG_LIVE_ITERATOR_STATUS_ERROR;
 584     }
 585
 586     message = bt2::ConstMessage::Shared::createWithoutRef(msg);
 587     return LTTNG_LIVE_ITERATOR_STATUS_OK;
 588 }
 589
 590 static enum lttng_live_iterator_status lttng_live_iterator_next_handle_one_quiescent_stream(
 591     struct lttng_live_msg_iter *lttng_live_msg_iter,
 592     struct lttng_live_stream_iterator *lttng_live_stream, bt2::ConstMessage::Shared& message)
 593 {
 594     enum lttng_live_iterator_status ret = LTTNG_LIVE_ITERATOR_STATUS_OK;
 595
 596     if (lttng_live_stream->state != LTTNG_LIVE_STREAM_QUIESCENT) {
 597         return LTTNG_LIVE_ITERATOR_STATUS_OK;
 598     }
 599
 600     /*
 601      * Check if we already sent an inactivty message downstream for this
 602      * `current_inactivity_ts` value.
 603      */
 604     if (lttng_live_stream->last_inactivity_ts.is_set &&
 605         lttng_live_stream->current_inactivity_ts == lttng_live_stream->last_inactivity_ts.value) {
 606         lttng_live_stream_iterator_set_state(lttng_live_stream,
 607                                              LTTNG_LIVE_STREAM_QUIESCENT_NO_DATA);
 608
 609         ret = LTTNG_LIVE_ITERATOR_STATUS_CONTINUE;
 610         goto end;
 611     }
 612
 613     ret = emit_inactivity_message(lttng_live_msg_iter, lttng_live_stream, message,
 614                                   lttng_live_stream->current_inactivity_ts);
 615
 616     lttng_live_stream->last_inactivity_ts.value = lttng_live_stream->current_inactivity_ts;
 617     lttng_live_stream->last_inactivity_ts.is_set = true;
 618 end:
 619     return ret;
 620 }
 621
 622 static int live_get_msg_ts_ns(struct lttng_live_msg_iter *lttng_live_msg_iter,
 623                               const bt_message *msg, int64_t last_msg_ts_ns, int64_t *ts_ns)
 624 {
 625     const bt_clock_snapshot *clock_snapshot = NULL;
 626     int ret = 0;
 627
 628     BT_ASSERT_DBG(msg);
 629     BT_ASSERT_DBG(ts_ns);
 630
 631     BT_CPPLOGD_SPEC(lttng_live_msg_iter->logger,
 632                     "Getting message's timestamp: iter-data-addr={}, msg-addr={}, "
 633                     "last-msg-ts={}",
 634                     fmt::ptr(lttng_live_msg_iter), fmt::ptr(msg), last_msg_ts_ns);
 635
 636     switch (bt_message_get_type(msg)) {
 637     case BT_MESSAGE_TYPE_EVENT:
 638         clock_snapshot = bt_message_event_borrow_default_clock_snapshot_const(msg);
 639         break;
 640     case BT_MESSAGE_TYPE_PACKET_BEGINNING:
 641         clock_snapshot = bt_message_packet_beginning_borrow_default_clock_snapshot_const(msg);
 642         break;
 643     case BT_MESSAGE_TYPE_PACKET_END:
 644         clock_snapshot = bt_message_packet_end_borrow_default_clock_snapshot_const(msg);
 645         break;
 646     case BT_MESSAGE_TYPE_DISCARDED_EVENTS:
 647         clock_snapshot =
 648             bt_message_discarded_events_borrow_beginning_default_clock_snapshot_const(msg);
 649         break;
 650     case BT_MESSAGE_TYPE_DISCARDED_PACKETS:
 651         clock_snapshot =
 652             bt_message_discarded_packets_borrow_beginning_default_clock_snapshot_const(msg);
 653         break;
 654     case BT_MESSAGE_TYPE_MESSAGE_ITERATOR_INACTIVITY:
 655         clock_snapshot = bt_message_message_iterator_inactivity_borrow_clock_snapshot_const(msg);
 656         break;
 657     default:
 658         /* All the other messages have a higher priority */
 659         BT_CPPLOGD_STR_SPEC(lttng_live_msg_iter->logger,
 660                             "Message has no timestamp: using the last message timestamp.");
 661         *ts_ns = last_msg_ts_ns;
 662         goto end;
 663     }
 664
 665     ret = bt_clock_snapshot_get_ns_from_origin(clock_snapshot, ts_ns);
 666     if (ret) {
 667         BT_CPPLOGE_APPEND_CAUSE_SPEC(lttng_live_msg_iter->logger,
 668                                      "Cannot get nanoseconds from Epoch of clock snapshot: "
 669                                      "clock-snapshot-addr={}",
 670                                      fmt::ptr(clock_snapshot));
 671         goto error;
 672     }
 673
 674     goto end;
 675
 676 error:
 677     ret = -1;
 678
 679 end:
 680     if (ret == 0) {
 681         BT_CPPLOGD_SPEC(lttng_live_msg_iter->logger,
 682                         "Found message's timestamp: iter-data-addr={}, msg-addr={}, "
 683                         "last-msg-ts={}, ts={}",
 684                         fmt::ptr(lttng_live_msg_iter), fmt::ptr(msg), last_msg_ts_ns, *ts_ns);
 685     }
 686
 687     return ret;
 688 }
 689
 690 static enum lttng_live_iterator_status lttng_live_iterator_next_handle_one_active_data_stream(
 691     struct lttng_live_msg_iter *lttng_live_msg_iter,
 692     struct lttng_live_stream_iterator *lttng_live_stream, bt2::ConstMessage::Shared& message)
 693 {
 694     enum lttng_live_iterator_status ret = LTTNG_LIVE_ITERATOR_STATUS_OK;
 695     enum ctf_msg_iter_status status;
 696     uint64_t session_idx, trace_idx;
 697
 698     for (session_idx = 0; session_idx < lttng_live_msg_iter->sessions->len; session_idx++) {
 699         struct lttng_live_session *session =
 700             (lttng_live_session *) g_ptr_array_index(lttng_live_msg_iter->sessions, session_idx);
 701
 702         if (session->new_streams_needed) {
 703             BT_CPPLOGD_SPEC(lttng_live_msg_iter->logger,
 704                             "Need an update for streams: "
 705                             "session-id={}",
 706                             session->id);
 707             ret = LTTNG_LIVE_ITERATOR_STATUS_CONTINUE;
 708             goto end;
 709         }
 710         for (trace_idx = 0; trace_idx < session->traces->len; trace_idx++) {
 711             struct lttng_live_trace *trace =
 712                 (lttng_live_trace *) g_ptr_array_index(session->traces, trace_idx);
 713             if (trace->metadata_stream_state == LTTNG_LIVE_METADATA_STREAM_STATE_NEEDED) {
 714                 BT_CPPLOGD_SPEC(lttng_live_msg_iter->logger,
 715                                 "Need an update for metadata stream: "
 716                                 "session-id={}, trace-id={}",
 717                                 session->id, trace->id);
 718                 ret = LTTNG_LIVE_ITERATOR_STATUS_CONTINUE;
 719                 goto end;
 720             }
 721         }
 722     }
 723
 724     if (lttng_live_stream->state != LTTNG_LIVE_STREAM_ACTIVE_DATA) {
 725         ret = LTTNG_LIVE_ITERATOR_STATUS_ERROR;
 726         BT_CPPLOGE_APPEND_CAUSE_SPEC(lttng_live_msg_iter->logger,
 727                                      "Invalid state of live stream iterator"
 728                                      "stream-iter-status={}",
 729                                      lttng_live_stream->state);
 730         goto end;
 731     }
 732
 733     const bt_message *msg;
 734     status = ctf_msg_iter_get_next_message(lttng_live_stream->msg_iter.get(), &msg);
 735     switch (status) {
 736     case CTF_MSG_ITER_STATUS_EOF:
 737         ret = LTTNG_LIVE_ITERATOR_STATUS_END;
 738         break;
 739     case CTF_MSG_ITER_STATUS_OK:
 740         message = bt2::ConstMessage::Shared::createWithoutRef(msg);
 741         ret = LTTNG_LIVE_ITERATOR_STATUS_OK;
 742         break;
 743     case CTF_MSG_ITER_STATUS_AGAIN:
 744         /*
 745          * Continue immediately (end of packet). The next
 746          * get_index may return AGAIN to delay the following
 747          * attempt.
 748          */
 749         ret = LTTNG_LIVE_ITERATOR_STATUS_CONTINUE;
 750         break;
 751     case CTF_MSG_ITER_STATUS_ERROR:
 752     default:
 753         ret = LTTNG_LIVE_ITERATOR_STATUS_ERROR;
 754         BT_CPPLOGE_APPEND_CAUSE_SPEC(lttng_live_msg_iter->logger,
 755                                      "CTF message iterator failed to get next message: "
 756                                      "msg-iter={}, msg-iter-status={}",
 757                                      fmt::ptr(lttng_live_stream->msg_iter), status);
 758         break;
 759     }
 760
 761 end:
 762     return ret;
 763 }
 764
 765 static enum lttng_live_iterator_status
 766 lttng_live_iterator_close_stream(struct lttng_live_msg_iter *lttng_live_msg_iter,
 767                                  struct lttng_live_stream_iterator *stream_iter,
 768                                  bt2::ConstMessage::Shared& curr_msg)
 769 {
 770     enum lttng_live_iterator_status live_status = LTTNG_LIVE_ITERATOR_STATUS_OK;
 771
 772     BT_CPPLOGD_SPEC(lttng_live_msg_iter->logger,
 773                     "Closing live stream iterator: stream-name=\"{}\", "
 774                     "viewer-stream-id={}",
 775                     stream_iter->name, stream_iter->viewer_stream_id);
 776
 777     /*
 778      * The viewer has hung up on us so we are closing the stream. The
 779      * `ctf_msg_iter` should simply realize that it needs to close the
 780      * stream properly by emitting the necessary stream end message.
 781      */
 782     const bt_message *msg;
 783     enum ctf_msg_iter_status status =
 784         ctf_msg_iter_get_next_message(stream_iter->msg_iter.get(), &msg);
 785
 786     if (status == CTF_MSG_ITER_STATUS_ERROR) {
 787         BT_CPPLOGE_APPEND_CAUSE_SPEC(lttng_live_msg_iter->logger,
 788                                      "Error getting the next message from CTF message iterator");
 789         live_status = LTTNG_LIVE_ITERATOR_STATUS_ERROR;
 790         goto end;
 791     } else if (status == CTF_MSG_ITER_STATUS_EOF) {
 792         BT_CPPLOGI_SPEC(lttng_live_msg_iter->logger,
 793                         "Reached the end of the live stream iterator.");
 794         live_status = LTTNG_LIVE_ITERATOR_STATUS_END;
 795         goto end;
 796     }
 797
 798     BT_ASSERT(status == CTF_MSG_ITER_STATUS_OK);
 799
 800     curr_msg = bt2::ConstMessage::Shared::createWithoutRef(msg);
 801
 802 end:
 803     return live_status;
 804 }
 805
 806 /*
 807  * helper function:
 808  *            handle_no_data_streams()
 809  *              retry:
 810  *                - for each ACTIVE_NO_DATA stream:
 811  *                  - query relayd for stream data, or quiescence info.
 812  *                    - if need metadata, get metadata, goto retry.
 813  *                    - if new stream, get new stream as ACTIVE_NO_DATA, goto retry
 814  *                  - if quiescent, move to QUIESCENT streams
 815  *                  - if fetched data, move to ACTIVE_DATA streams
 816  *                (at this point each stream either has data, or is quiescent)
 817  *
 818  *
 819  * iterator_next:
 820  *            handle_new_streams_and_metadata()
 821  *                  - query relayd for known streams, add them as ACTIVE_NO_DATA
 822  *                  - query relayd for metadata
 823  *
 824  *            call handle_active_no_data_streams()
 825  *
 826  *            handle_quiescent_streams()
 827  *                - if at least one stream is ACTIVE_DATA:
 828  *                  - peek stream event with lowest timestamp -> next_ts
 829  *                  - for each quiescent stream
 830  *                    - if next_ts >= quiescent end
 831  *                      - set state to ACTIVE_NO_DATA
 832  *                - else
 833  *                  - for each quiescent stream
 834  *                      - set state to ACTIVE_NO_DATA
 835  *
 836  *            call handle_active_no_data_streams()
 837  *
 838  *            handle_active_data_streams()
 839  *                - if at least one stream is ACTIVE_DATA:
 840  *                    - get stream event with lowest timestamp from heap
 841  *                    - make that stream event the current message.
 842  *                    - move this stream heap position to its next event
 843  *                      - if we need to fetch data from relayd, move
 844  *                        stream to ACTIVE_NO_DATA.
 845  *                    - return OK
 846  *                - return AGAIN
 847  *
 848  * end criterion: ctrl-c on client. If relayd exits or the session
 849  * closes on the relay daemon side, we keep on waiting for streams.
 850  * Eventually handle --end timestamp (also an end criterion).
 851  *
 852  * When disconnected from relayd: try to re-connect endlessly.
 853  */
 854 static enum lttng_live_iterator_status
 855 lttng_live_iterator_next_msg_on_stream(struct lttng_live_msg_iter *lttng_live_msg_iter,
 856                                        struct lttng_live_stream_iterator *stream_iter,
 857                                        bt2::ConstMessage::Shared& curr_msg)
 858 {
 859     enum lttng_live_iterator_status live_status;
 860
 861     BT_CPPLOGD_SPEC(lttng_live_msg_iter->logger,
 862                     "Advancing live stream iterator until next message if possible: "
 863                     "stream-name=\"{}\", viewer-stream-id={}",
 864                     stream_iter->name, stream_iter->viewer_stream_id);
 865
 866     if (stream_iter->has_stream_hung_up) {
 867         /*
 868          * The stream has hung up and the stream was properly closed
 869          * during the last call to the current function. Return _END
 870          * status now so that this stream iterator is removed for the
 871          * stream iterator list.
 872          */
 873         live_status = LTTNG_LIVE_ITERATOR_STATUS_END;
 874         goto end;
 875     }
 876
 877 retry:
 878     LTTNG_LIVE_LOGD_STREAM_ITER(stream_iter);
 879
 880     /*
 881      * Make sure we have the most recent metadata and possibly some new
 882      * streams.
 883      */
 884     live_status = lttng_live_iterator_handle_new_streams_and_metadata(lttng_live_msg_iter);
 885     if (live_status != LTTNG_LIVE_ITERATOR_STATUS_OK) {
 886         goto end;
 887     }
 888
 889     live_status =
 890         lttng_live_iterator_next_handle_one_no_data_stream(lttng_live_msg_iter, stream_iter);
 891     if (live_status != LTTNG_LIVE_ITERATOR_STATUS_OK) {
 892         if (live_status == LTTNG_LIVE_ITERATOR_STATUS_END) {
 893             /*
 894              * We overwrite `live_status` since `curr_msg` is
 895              * likely set to a valid message in this function.
 896              */
 897             live_status =
 898                 lttng_live_iterator_close_stream(lttng_live_msg_iter, stream_iter, curr_msg);
 899         }
 900         goto end;
 901     }
 902
 903     live_status = lttng_live_iterator_next_handle_one_quiescent_stream(lttng_live_msg_iter,
 904                                                                        stream_iter, curr_msg);
 905     if (live_status != LTTNG_LIVE_ITERATOR_STATUS_OK) {
 906         BT_ASSERT(!curr_msg);
 907         goto end;
 908     }
 909     if (curr_msg) {
 910         goto end;
 911     }
 912     live_status = lttng_live_iterator_next_handle_one_active_data_stream(lttng_live_msg_iter,
 913                                                                          stream_iter, curr_msg);
 914     if (live_status != LTTNG_LIVE_ITERATOR_STATUS_OK) {
 915         BT_ASSERT(!curr_msg);
 916     }
 917
 918 end:
 919     if (live_status == LTTNG_LIVE_ITERATOR_STATUS_CONTINUE) {
 920         BT_CPPLOGD_SPEC(
 921             lttng_live_msg_iter->logger,
 922             "Ask the relay daemon for an updated view of the data and metadata streams");
 923         goto retry;
 924     }
 925
 926     BT_CPPLOGD_SPEC(lttng_live_msg_iter->logger,
 927                     "Returning from advancing live stream iterator: status={}, "
 928                     "stream-name=\"{}\", viewer-stream-id={}",
 929                     live_status, stream_iter->name, stream_iter->viewer_stream_id);
 930
 931     return live_status;
 932 }
 933
 934 static bool is_discarded_packet_or_event_message(const bt2::ConstMessage msg)
 935 {
 936     return msg.type() == bt2::MessageType::DiscardedEvents ||
 937            msg.type() == bt2::MessageType::DiscardedPackets;
 938 }
 939
 940 static enum lttng_live_iterator_status
 941 adjust_discarded_packets_message(bt_self_message_iterator *iter, const bt_stream *stream,
 942                                  const bt_message *msg_in, bt2::ConstMessage::Shared& msg_out,
 943                                  uint64_t new_begin_ts)
 944 {
 945     enum bt_property_availability availability;
 946     const bt_clock_snapshot *clock_snapshot;
 947     uint64_t end_ts;
 948     uint64_t count;
 949
 950     clock_snapshot = bt_message_discarded_packets_borrow_end_default_clock_snapshot_const(msg_in);
 951     end_ts = bt_clock_snapshot_get_value(clock_snapshot);
 952
 953     availability = bt_message_discarded_packets_get_count(msg_in, &count);
 954     BT_ASSERT_DBG(availability == BT_PROPERTY_AVAILABILITY_AVAILABLE);
 955
 956     const auto msg = bt_message_discarded_packets_create_with_default_clock_snapshots(
 957         iter, stream, new_begin_ts, end_ts);
 958
 959     if (!msg) {
 960         return LTTNG_LIVE_ITERATOR_STATUS_NOMEM;
 961     }
 962
 963     bt_message_discarded_packets_set_count(msg, count);
 964     msg_out = bt2::ConstMessage::Shared::createWithoutRef(msg);
 965     return LTTNG_LIVE_ITERATOR_STATUS_OK;
 966 }
 967
 968 static enum lttng_live_iterator_status
 969 adjust_discarded_events_message(bt_self_message_iterator *iter, const bt_stream *stream,
 970                                 const bt_message *msg_in, bt2::ConstMessage::Shared& msg_out,
 971                                 uint64_t new_begin_ts)
 972 {
 973     enum bt_property_availability availability;
 974     const bt_clock_snapshot *clock_snapshot;
 975     uint64_t end_ts;
 976     uint64_t count;
 977
 978     clock_snapshot = bt_message_discarded_events_borrow_end_default_clock_snapshot_const(msg_in);
 979     end_ts = bt_clock_snapshot_get_value(clock_snapshot);
 980
 981     availability = bt_message_discarded_events_get_count(msg_in, &count);
 982     BT_ASSERT_DBG(availability == BT_PROPERTY_AVAILABILITY_AVAILABLE);
 983
 984     const auto msg = bt_message_discarded_events_create_with_default_clock_snapshots(
 985         iter, stream, new_begin_ts, end_ts);
 986
 987     if (!msg) {
 988         return LTTNG_LIVE_ITERATOR_STATUS_NOMEM;
 989     }
 990
 991     bt_message_discarded_events_set_count(msg, count);
 992     msg_out = bt2::ConstMessage::Shared::createWithoutRef(msg);
 993     return LTTNG_LIVE_ITERATOR_STATUS_OK;
 994 }
 995
 996 static enum lttng_live_iterator_status
 997 handle_late_message(struct lttng_live_msg_iter *lttng_live_msg_iter,
 998                     struct lttng_live_stream_iterator *stream_iter, int64_t late_msg_ts_ns,
 999                     const bt2::ConstMessage& late_msg)
1000 {
1001     const bt_clock_class *clock_class;
1002     const bt_stream_class *stream_class;
1003     enum bt_clock_class_cycles_to_ns_from_origin_status ts_ns_status;
1004     int64_t last_inactivity_ts_ns;
1005     enum lttng_live_iterator_status stream_iter_status = LTTNG_LIVE_ITERATOR_STATUS_OK;
1006     enum lttng_live_iterator_status adjust_status;
1007     bt2::ConstMessage::Shared adjusted_message;
1008
1009     /*
1010      * The timestamp of the current message is before the last message sent
1011      * by this component. We CANNOT send it as is.
1012      *
1013      * The only expected scenario in which that could happen is the
1014      * following, everything else is a bug in this component, relay daemon,
1015      * or CTF parser.
1016      *
1017      * Expected scenario: The CTF message iterator emitted discarded
1018      * packets and discarded events with synthesized beginning and end
1019      * timestamps from the bounds of the last known packet and the newly
1020      * decoded packet header. The CTF message iterator is not aware of
1021      * stream inactivity beacons. Hence, we have to adjust the beginning
1022      * timestamp of those types of messages if a stream signalled its
1023      * inactivity up until _after_ the last known packet's beginning
1024      * timestamp.
1025      *
1026      * Otherwise, the monotonicity guarantee of message timestamps would
1027      * not be preserved.
1028      *
1029      * In short, the only scenario in which it's okay and fixable to
1030      * received a late message is when:
1031      *  1. the late message is a discarded packets or discarded events
1032      *     message,
1033      *  2. this stream produced an inactivity message downstream, and
1034      *  3. the timestamp of the late message is within the inactivity
1035      *     timespan we sent downstream through the inactivity message.
1036      */
1037
1038     BT_CPPLOGD_SPEC(lttng_live_msg_iter->logger,
1039                     "Handling late message on live stream iterator: "
1040                     "stream-name=\"{}\", viewer-stream-id={}",
1041                     stream_iter->name, stream_iter->viewer_stream_id);
1042
1043     if (!stream_iter->last_inactivity_ts.is_set) {
1044         BT_CPPLOGE_APPEND_CAUSE_SPEC(lttng_live_msg_iter->logger,
1045                                      "Invalid live stream state: "
1046                                      "have a late message when no inactivity message "
1047                                      "was ever sent for that stream.");
1048         stream_iter_status = LTTNG_LIVE_ITERATOR_STATUS_ERROR;
1049         goto end;
1050     }
1051
1052     if (!is_discarded_packet_or_event_message(late_msg)) {
1053         BT_CPPLOGE_APPEND_CAUSE_SPEC(lttng_live_msg_iter->logger,
1054                                      "Invalid live stream state: "
1055                                      "have a late message that is not a packet discarded or "
1056                                      "event discarded message: late-msg-type={}",
1057                                      late_msg.type());
1058         stream_iter_status = LTTNG_LIVE_ITERATOR_STATUS_ERROR;
1059         goto end;
1060     }
1061
1062     stream_class = bt_stream_borrow_class_const(stream_iter->stream->libObjPtr());
1063     clock_class = bt_stream_class_borrow_default_clock_class_const(stream_class);
1064
1065     ts_ns_status = bt_clock_class_cycles_to_ns_from_origin(
1066         clock_class, stream_iter->last_inactivity_ts.value, &last_inactivity_ts_ns);
1067     if (ts_ns_status != BT_CLOCK_CLASS_CYCLES_TO_NS_FROM_ORIGIN_STATUS_OK) {
1068         BT_CPPLOGE_APPEND_CAUSE_SPEC(lttng_live_msg_iter->logger,
1069                                      "Error converting last "
1070                                      "inactivity message timestamp to nanoseconds");
1071         stream_iter_status = LTTNG_LIVE_ITERATOR_STATUS_ERROR;
1072         goto end;
1073     }
1074
1075     if (last_inactivity_ts_ns <= late_msg_ts_ns) {
1076         BT_CPPLOGE_APPEND_CAUSE_SPEC(lttng_live_msg_iter->logger,
1077                                      "Invalid live stream state: "
1078                                      "have a late message that is none included in a stream "
1079                                      "inactivity timespan: last-inactivity-ts-ns={}, "
1080                                      "late-msg-ts-ns={}",
1081                                      last_inactivity_ts_ns, late_msg_ts_ns);
1082         stream_iter_status = LTTNG_LIVE_ITERATOR_STATUS_ERROR;
1083         goto end;
1084     }
1085
1086     /*
1087      * We now know that it's okay for this message to be late, we can now
1088      * adjust its timestamp to ensure monotonicity.
1089      */
1090     BT_CPPLOGD_SPEC(lttng_live_msg_iter->logger,
1091                     "Adjusting the timestamp of late message: late-msg-type={}, "
1092                     "msg-new-ts-ns={}",
1093                     late_msg.type(), stream_iter->last_inactivity_ts.value);
1094     switch (late_msg.type()) {
1095     case bt2::MessageType::DiscardedEvents:
1096         adjust_status = adjust_discarded_events_message(
1097             lttng_live_msg_iter->self_msg_iter, stream_iter->stream->libObjPtr(),
1098             late_msg.libObjPtr(), adjusted_message, stream_iter->last_inactivity_ts.value);
1099         break;
1100     case bt2::MessageType::DiscardedPackets:
1101         adjust_status = adjust_discarded_packets_message(
1102             lttng_live_msg_iter->self_msg_iter, stream_iter->stream->libObjPtr(),
1103             late_msg.libObjPtr(), adjusted_message, stream_iter->last_inactivity_ts.value);
1104         break;
1105     default:
1106         bt_common_abort();
1107     }
1108
1109     if (adjust_status != LTTNG_LIVE_ITERATOR_STATUS_OK) {
1110         stream_iter_status = adjust_status;
1111         goto end;
1112     }
1113
1114     BT_ASSERT_DBG(adjusted_message);
1115     stream_iter->current_msg = adjusted_message;
1116     stream_iter->current_msg_ts_ns = last_inactivity_ts_ns;
1117
1118 end:
1119     return stream_iter_status;
1120 }
1121
1122 static enum lttng_live_iterator_status
1123 next_stream_iterator_for_trace(struct lttng_live_msg_iter *lttng_live_msg_iter,
1124                                struct lttng_live_trace *live_trace,
1125                                struct lttng_live_stream_iterator **youngest_trace_stream_iter)
1126 {
1127     struct lttng_live_stream_iterator *youngest_candidate_stream_iter = NULL;
1128     enum lttng_live_iterator_status stream_iter_status;
1129     int64_t youngest_candidate_msg_ts = INT64_MAX;
1130     uint64_t stream_iter_idx;
1131
1132     BT_ASSERT_DBG(live_trace);
1133     BT_ASSERT_DBG(live_trace->stream_iterators);
1134
1135     BT_CPPLOGD_SPEC(lttng_live_msg_iter->logger,
1136                     "Finding the next stream iterator for trace: "
1137                     "trace-id={}",
1138                     live_trace->id);
1139     /*
1140      * Update the current message of every stream iterators of this trace.
1141      * The current msg of every stream must have a timestamp equal or
1142      * larger than the last message returned by this iterator. We must
1143      * ensure monotonicity.
1144      */
1145     stream_iter_idx = 0;
1146     while (stream_iter_idx < live_trace->stream_iterators->len) {
1147         bool stream_iter_is_ended = false;
1148         struct lttng_live_stream_iterator *stream_iter =
1149             (lttng_live_stream_iterator *) g_ptr_array_index(live_trace->stream_iterators,
1150                                                              stream_iter_idx);
1151
1152         /*
1153          * If there is no current message for this stream, go fetch
1154          * one.
1155          */
1156         while (!stream_iter->current_msg) {
1157             bt2::ConstMessage::Shared msg;
1158             int64_t curr_msg_ts_ns = INT64_MAX;
1159
1160             stream_iter_status =
1161                 lttng_live_iterator_next_msg_on_stream(lttng_live_msg_iter, stream_iter, msg);
1162
1163             if (stream_iter_status == LTTNG_LIVE_ITERATOR_STATUS_END) {
1164                 stream_iter_is_ended = true;
1165                 break;
1166             }
1167
1168             if (stream_iter_status != LTTNG_LIVE_ITERATOR_STATUS_OK) {
1169                 goto end;
1170             }
1171
1172             BT_ASSERT_DBG(msg);
1173
1174             BT_CPPLOGD_SPEC(lttng_live_msg_iter->logger,
1175                             "Live stream iterator returned message: msg-type={}, "
1176                             "stream-name=\"{}\", viewer-stream-id={}",
1177                             msg->type(), stream_iter->name, stream_iter->viewer_stream_id);
1178
1179             /*
1180              * Get the timestamp in nanoseconds from origin of this
1181              * message.
1182              */
1183             live_get_msg_ts_ns(lttng_live_msg_iter, msg->libObjPtr(),
1184                                lttng_live_msg_iter->last_msg_ts_ns, &curr_msg_ts_ns);
1185
1186             /*
1187              * Check if the message of the current live stream
1188              * iterator occurred at the exact same time or after the
1189              * last message returned by this component's message
1190              * iterator. If not, we need to handle it with care.
1191              */
1192             if (curr_msg_ts_ns >= lttng_live_msg_iter->last_msg_ts_ns) {
1193                 stream_iter->current_msg = std::move(msg);
1194                 stream_iter->current_msg_ts_ns = curr_msg_ts_ns;
1195             } else {
1196                 /*
1197                  * We received a message from the past. This
1198                  * may be fixable but it can also be an error.
1199                  */
1200                 stream_iter_status =
1201                     handle_late_message(lttng_live_msg_iter, stream_iter, curr_msg_ts_ns, *msg);
1202                 if (stream_iter_status != LTTNG_LIVE_ITERATOR_STATUS_OK) {
1203                     BT_CPPLOGE_APPEND_CAUSE_SPEC(lttng_live_msg_iter->logger,
1204                                                  "Late message could not be handled correctly: "
1205                                                  "lttng-live-msg-iter-addr={}, "
1206                                                  "stream-name=\"{}\", "
1207                                                  "curr-msg-ts={}, last-msg-ts={}",
1208                                                  fmt::ptr(lttng_live_msg_iter), stream_iter->name,
1209                                                  curr_msg_ts_ns,
1210                                                  lttng_live_msg_iter->last_msg_ts_ns);
1211                     stream_iter_status = LTTNG_LIVE_ITERATOR_STATUS_ERROR;
1212                     goto end;
1213                 }
1214             }
1215         }
1216
1217         BT_ASSERT_DBG(stream_iter != youngest_candidate_stream_iter);
1218
1219         if (!stream_iter_is_ended) {
1220             if (G_UNLIKELY(youngest_candidate_stream_iter == NULL) ||
1221                 stream_iter->current_msg_ts_ns < youngest_candidate_msg_ts) {
1222                 /*
1223                  * Update the current best candidate message
1224                  * for the stream iterator of this live trace
1225                  * to be forwarded downstream.
1226                  */
1227                 youngest_candidate_msg_ts = stream_iter->current_msg_ts_ns;
1228                 youngest_candidate_stream_iter = stream_iter;
1229             } else if (stream_iter->current_msg_ts_ns == youngest_candidate_msg_ts) {
1230                 /*
1231                  * Order the messages in an arbitrary but
1232                  * deterministic way.
1233                  */
1234                 BT_ASSERT_DBG(stream_iter != youngest_candidate_stream_iter);
1235                 int ret = common_muxing_compare_messages(
1236                     stream_iter->current_msg->libObjPtr(),
1237                     youngest_candidate_stream_iter->current_msg->libObjPtr());
1238                 if (ret < 0) {
1239                     /*
1240                      * The `youngest_candidate_stream_iter->current_msg`
1241                      * should go first. Update the next
1242                      * iterator and the current timestamp.
1243                      */
1244                     youngest_candidate_msg_ts = stream_iter->current_msg_ts_ns;
1245                     youngest_candidate_stream_iter = stream_iter;
1246                 } else if (ret == 0) {
1247                     /*
1248                      * Unable to pick which one should go
1249                      * first.
1250                      */
1251                     BT_CPPLOGW_SPEC(
1252                         lttng_live_msg_iter->logger,
1253                         "Cannot deterministically pick next live stream message iterator because they have identical next messages: "
1254                         "stream-iter-addr={}"
1255                         "stream-iter-addr={}",
1256                         fmt::ptr(stream_iter), fmt::ptr(youngest_candidate_stream_iter));
1257                 }
1258             }
1259
1260             stream_iter_idx++;
1261         } else {
1262             /*
1263              * The live stream iterator has ended. That
1264              * iterator is removed from the array, but
1265              * there is no need to increment
1266              * stream_iter_idx as
1267              * g_ptr_array_remove_index_fast replaces the
1268              * removed element with the array's last
1269              * element.
1270              */
1271             g_ptr_array_remove_index_fast(live_trace->stream_iterators, stream_iter_idx);
1272         }
1273     }
1274
1275     if (youngest_candidate_stream_iter) {
1276         *youngest_trace_stream_iter = youngest_candidate_stream_iter;
1277         stream_iter_status = LTTNG_LIVE_ITERATOR_STATUS_OK;
1278     } else {
1279         /*
1280          * The only case where we don't have a candidate for this trace
1281          * is if we reached the end of all the iterators.
1282          */
1283         BT_ASSERT(live_trace->stream_iterators->len == 0);
1284         stream_iter_status = LTTNG_LIVE_ITERATOR_STATUS_END;
1285     }
1286
1287 end:
1288     return stream_iter_status;
1289 }
1290
1291 static enum lttng_live_iterator_status
1292 next_stream_iterator_for_session(struct lttng_live_msg_iter *lttng_live_msg_iter,
1293                                  struct lttng_live_session *session,
1294                                  struct lttng_live_stream_iterator **youngest_session_stream_iter)
1295 {
1296     enum lttng_live_iterator_status stream_iter_status;
1297     uint64_t trace_idx = 0;
1298     int64_t youngest_candidate_msg_ts = INT64_MAX;
1299     struct lttng_live_stream_iterator *youngest_candidate_stream_iter = NULL;
1300
1301     BT_CPPLOGD_SPEC(lttng_live_msg_iter->logger,
1302                     "Finding the next stream iterator for session: "
1303                     "session-id={}",
1304                     session->id);
1305     /*
1306      * Make sure we are attached to the session and look for new streams
1307      * and metadata.
1308      */
1309     stream_iter_status = lttng_live_get_session(lttng_live_msg_iter, session);
1310     if (stream_iter_status != LTTNG_LIVE_ITERATOR_STATUS_OK &&
1311         stream_iter_status != LTTNG_LIVE_ITERATOR_STATUS_CONTINUE &&
1312         stream_iter_status != LTTNG_LIVE_ITERATOR_STATUS_END) {
1313         goto end;
1314     }
1315
1316     BT_ASSERT_DBG(session->traces);
1317
1318     while (trace_idx < session->traces->len) {
1319         bool trace_is_ended = false;
1320         struct lttng_live_stream_iterator *stream_iter;
1321         struct lttng_live_trace *trace =
1322             (lttng_live_trace *) g_ptr_array_index(session->traces, trace_idx);
1323
1324         stream_iter_status =
1325             next_stream_iterator_for_trace(lttng_live_msg_iter, trace, &stream_iter);
1326         if (stream_iter_status == LTTNG_LIVE_ITERATOR_STATUS_END) {
1327             /*
1328              * All the live stream iterators for this trace are
1329              * ENDed. Remove the trace from this session.
1330              */
1331             trace_is_ended = true;
1332         } else if (stream_iter_status != LTTNG_LIVE_ITERATOR_STATUS_OK) {
1333             goto end;
1334         }
1335
1336         if (!trace_is_ended) {
1337             BT_ASSERT_DBG(stream_iter);
1338
1339             if (G_UNLIKELY(youngest_candidate_stream_iter == NULL) ||
1340                 stream_iter->current_msg_ts_ns < youngest_candidate_msg_ts) {
1341                 youngest_candidate_msg_ts = stream_iter->current_msg_ts_ns;
1342                 youngest_candidate_stream_iter = stream_iter;
1343             } else if (stream_iter->current_msg_ts_ns == youngest_candidate_msg_ts) {
1344                 /*
1345                  * Order the messages in an arbitrary but
1346                  * deterministic way.
1347                  */
1348                 int ret = common_muxing_compare_messages(
1349                     stream_iter->current_msg->libObjPtr(),
1350                     youngest_candidate_stream_iter->current_msg->libObjPtr());
1351                 if (ret < 0) {
1352                     /*
1353                      * The `youngest_candidate_stream_iter->current_msg`
1354                      * should go first. Update the next iterator
1355                      * and the current timestamp.
1356                      */
1357                     youngest_candidate_msg_ts = stream_iter->current_msg_ts_ns;
1358                     youngest_candidate_stream_iter = stream_iter;
1359                 } else if (ret == 0) {
1360                     /* Unable to pick which one should go first. */
1361                     BT_CPPLOGW_SPEC(
1362                         lttng_live_msg_iter->logger,
1363                         "Cannot deterministically pick next live stream message iterator because they have identical next messages: "
1364                         "stream-iter-addr={}"
1365                         "youngest-candidate-stream-iter-addr={}",
1366                         fmt::ptr(stream_iter), fmt::ptr(youngest_candidate_stream_iter));
1367                 }
1368             }
1369             trace_idx++;
1370         } else {
1371             /*
1372              * trace_idx is not incremented since
1373              * g_ptr_array_remove_index_fast replaces the
1374              * element at trace_idx with the array's last element.
1375              */
1376             g_ptr_array_remove_index_fast(session->traces, trace_idx);
1377         }
1378     }
1379     if (youngest_candidate_stream_iter) {
1380         *youngest_session_stream_iter = youngest_candidate_stream_iter;
1381         stream_iter_status = LTTNG_LIVE_ITERATOR_STATUS_OK;
1382     } else {
1383         /*
1384          * The only cases where we don't have a candidate for this
1385          * trace is:
1386          *  1. if we reached the end of all the iterators of all the
1387          *  traces of this session,
1388          *  2. if we never had live stream iterator in the first place.
1389          *
1390          * In either cases, we return END.
1391          */
1392         BT_ASSERT(session->traces->len == 0);
1393         stream_iter_status = LTTNG_LIVE_ITERATOR_STATUS_END;
1394     }
1395 end:
1396     return stream_iter_status;
1397 }
1398
1399 static inline void put_messages(bt_message_array_const msgs, uint64_t count)
1400 {
1401     uint64_t i;
1402
1403     for (i = 0; i < count; i++) {
1404         BT_MESSAGE_PUT_REF_AND_RESET(msgs[i]);
1405     }
1406 }
1407
1408 bt_message_iterator_class_next_method_status
1409 lttng_live_msg_iter_next(bt_self_message_iterator *self_msg_it, bt_message_array_const msgs,
1410                          uint64_t capacity, uint64_t *count)
1411 {
1412     try {
1413         bt_message_iterator_class_next_method_status status;
1414         enum lttng_live_viewer_status viewer_status;
1415         struct lttng_live_msg_iter *lttng_live_msg_iter =
1416             (struct lttng_live_msg_iter *) bt_self_message_iterator_get_data(self_msg_it);
1417         struct lttng_live_component *lttng_live = lttng_live_msg_iter->lttng_live_comp;
1418         enum lttng_live_iterator_status stream_iter_status;
1419         uint64_t session_idx;
1420
1421         *count = 0;
1422
1423         BT_ASSERT_DBG(lttng_live_msg_iter);
1424
1425         if (G_UNLIKELY(lttng_live_msg_iter->was_interrupted)) {
1426             /*
1427              * The iterator was interrupted in a previous call to the
1428              * `_next()` method. We currently do not support generating
1429              * messages after such event. The babeltrace2 CLI should never
1430              * be running the graph after being interrupted. So this check
1431              * is to prevent other graph users from using this live
1432              * iterator in an messed up internal state.
1433              */
1434             status = BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_ERROR;
1435             BT_CPPLOGE_APPEND_CAUSE_SPEC(
1436                 lttng_live_msg_iter->logger,
1437                 "Message iterator was interrupted during a previous call to the `next()` and currently does not support continuing after such event.");
1438             goto end;
1439         }
1440
1441         /*
1442          * Clear all the invalid message reference that might be left over in
1443          * the output array.
1444          */
1445         memset(msgs, 0, capacity * sizeof(*msgs));
1446
1447         /*
1448          * If no session are exposed on the relay found at the url provided by
1449          * the user, session count will be 0. In this case, we return status
1450          * end to return gracefully.
1451          */
1452         if (lttng_live_msg_iter->sessions->len == 0) {
1453             if (lttng_live->params.sess_not_found_act != SESSION_NOT_FOUND_ACTION_CONTINUE) {
1454                 status = BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_END;
1455                 goto end;
1456             } else {
1457                 /*
1458                  * The are no more active session for this session
1459                  * name. Retry to create a viewer session for the
1460                  * requested session name.
1461                  */
1462                 viewer_status = lttng_live_create_viewer_session(lttng_live_msg_iter);
1463                 if (viewer_status != LTTNG_LIVE_VIEWER_STATUS_OK) {
1464                     if (viewer_status == LTTNG_LIVE_VIEWER_STATUS_ERROR) {
1465                         status = BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_ERROR;
1466                         BT_CPPLOGE_APPEND_CAUSE_SPEC(lttng_live_msg_iter->logger,
1467                                                      "Error creating LTTng live viewer session");
1468                     } else if (viewer_status == LTTNG_LIVE_VIEWER_STATUS_INTERRUPTED) {
1469                         status = BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_AGAIN;
1470                     } else {
1471                         bt_common_abort();
1472                     }
1473                     goto end;
1474                 }
1475             }
1476         }
1477
1478         if (lttng_live_msg_iter->active_stream_iter == 0) {
1479             lttng_live_force_new_streams_and_metadata(lttng_live_msg_iter);
1480         }
1481
1482         /*
1483          * Here the muxing of message is done.
1484          *
1485          * We need to iterate over all the streams of all the traces of all the
1486          * viewer sessions in order to get the message with the smallest
1487          * timestamp. In this case, a session is a viewer session and there is
1488          * one viewer session per consumer daemon. (UST 32bit, UST 64bit and/or
1489          * kernel). Each viewer session can have multiple traces, for example,
1490          * 64bit UST viewer sessions could have multiple per-pid traces.
1491          *
1492          * We iterate over the streams of each traces to update and see what is
1493          * their next message's timestamp. From those timestamps, we select the
1494          * message with the smallest timestamp as the best candidate message
1495          * for that trace and do the same thing across all the sessions.
1496          *
1497          * We then compare the timestamp of best candidate message of all the
1498          * sessions to pick the message with the smallest timestamp and we
1499          * return it.
1500          */
1501         while (*count < capacity) {
1502             struct lttng_live_stream_iterator *youngest_stream_iter = NULL,
1503                                               *candidate_stream_iter = NULL;
1504             int64_t youngest_msg_ts_ns = INT64_MAX;
1505
1506             BT_ASSERT_DBG(lttng_live_msg_iter->sessions);
1507             session_idx = 0;
1508             while (session_idx < lttng_live_msg_iter->sessions->len) {
1509                 struct lttng_live_session *session = (lttng_live_session *) g_ptr_array_index(
1510                     lttng_live_msg_iter->sessions, session_idx);
1511
1512                 /* Find the best candidate message to send downstream. */
1513                 stream_iter_status = next_stream_iterator_for_session(lttng_live_msg_iter, session,
1514                                                                       &candidate_stream_iter);
1515
1516                 /* If we receive an END status, it means that either:
1517                  * - Those traces never had active streams (UST with no
1518                  *   data produced yet),
1519                  * - All live stream iterators have ENDed.
1520                  */
1521                 if (stream_iter_status == LTTNG_LIVE_ITERATOR_STATUS_END) {
1522                     if (session->closed && session->traces->len == 0) {
1523                         /*
1524                          * Remove the session from the list.
1525                          * session_idx is not modified since
1526                          * g_ptr_array_remove_index_fast
1527                          * replaces the the removed element with
1528                          * the array's last element.
1529                          */
1530                         g_ptr_array_remove_index_fast(lttng_live_msg_iter->sessions, session_idx);
1531                     } else {
1532                         session_idx++;
1533                     }
1534                     continue;
1535                 }
1536
1537                 if (stream_iter_status != LTTNG_LIVE_ITERATOR_STATUS_OK) {
1538                     goto return_status;
1539                 }
1540
1541                 if (G_UNLIKELY(youngest_stream_iter == NULL) ||
1542                     candidate_stream_iter->current_msg_ts_ns < youngest_msg_ts_ns) {
1543                     youngest_msg_ts_ns = candidate_stream_iter->current_msg_ts_ns;
1544                     youngest_stream_iter = candidate_stream_iter;
1545                 } else if (candidate_stream_iter->current_msg_ts_ns == youngest_msg_ts_ns) {
1546                     /*
1547                      * The currently selected message to be sent
1548                      * downstream next has the exact same timestamp
1549                      * that of the current candidate message. We
1550                      * must break the tie in a predictable manner.
1551                      */
1552                     BT_CPPLOGD_STR_SPEC(
1553                         lttng_live_msg_iter->logger,
1554                         "Two of the next message candidates have the same timestamps, pick one deterministically.");
1555                     /*
1556                      * Order the messages in an arbitrary but
1557                      * deterministic way.
1558                      */
1559                     int ret = common_muxing_compare_messages(
1560                         candidate_stream_iter->current_msg->libObjPtr(),
1561                         youngest_stream_iter->current_msg->libObjPtr());
1562                     if (ret < 0) {
1563                         /*
1564                          * The `candidate_stream_iter->current_msg`
1565                          * should go first. Update the next
1566                          * iterator and the current timestamp.
1567                          */
1568                         youngest_msg_ts_ns = candidate_stream_iter->current_msg_ts_ns;
1569                         youngest_stream_iter = candidate_stream_iter;
1570                     } else if (ret == 0) {
1571                         /* Unable to pick which one should go first. */
1572                         BT_CPPLOGW_SPEC(
1573                             lttng_live_msg_iter->logger,
1574                             "Cannot deterministically pick next live stream message iterator because they have identical next messages: "
1575                             "next-stream-iter-addr={}"
1576                             "candidate-stream-iter-addr={}",
1577                             fmt::ptr(youngest_stream_iter), fmt::ptr(candidate_stream_iter));
1578                     }
1579                 }
1580
1581                 session_idx++;
1582             }
1583
1584             if (!youngest_stream_iter) {
1585                 stream_iter_status = LTTNG_LIVE_ITERATOR_STATUS_AGAIN;
1586                 goto return_status;
1587             }
1588
1589             BT_ASSERT_DBG(youngest_stream_iter->current_msg);
1590             /* Ensure monotonicity. */
1591             BT_ASSERT_DBG(lttng_live_msg_iter->last_msg_ts_ns <=
1592                           youngest_stream_iter->current_msg_ts_ns);
1593
1594             /*
1595              * Insert the next message to the message batch. This will set
1596              * stream iterator current message to NULL so that next time
1597              * we fetch the next message of that stream iterator
1598              */
1599             msgs[*count] = youngest_stream_iter->current_msg.release().libObjPtr();
1600             (*count)++;
1601
1602             /* Update the last timestamp in nanoseconds sent downstream. */
1603             lttng_live_msg_iter->last_msg_ts_ns = youngest_msg_ts_ns;
1604             youngest_stream_iter->current_msg_ts_ns = INT64_MAX;
1605
1606             stream_iter_status = LTTNG_LIVE_ITERATOR_STATUS_OK;
1607         }
1608
1609 return_status:
1610         switch (stream_iter_status) {
1611         case LTTNG_LIVE_ITERATOR_STATUS_OK:
1612         case LTTNG_LIVE_ITERATOR_STATUS_AGAIN:
1613             /*
1614              * If we gathered messages, return _OK even if the graph was
1615              * interrupted. This allows for the components downstream to at
1616              * least get the those messages. If the graph was indeed
1617              * interrupted there should not be another _next() call as the
1618              * application will tear down the graph. This component class
1619              * doesn't support restarting after an interruption.
1620              */
1621             if (*count > 0) {
1622                 status = BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_OK;
1623             } else {
1624                 status = BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_AGAIN;
1625             }
1626             break;
1627         case LTTNG_LIVE_ITERATOR_STATUS_END:
1628             status = BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_END;
1629             break;
1630         case LTTNG_LIVE_ITERATOR_STATUS_NOMEM:
1631             BT_CPPLOGE_APPEND_CAUSE_SPEC(lttng_live_msg_iter->logger,
1632                                          "Memory error preparing the next batch of messages: "
1633                                          "live-iter-status={}",
1634                                          stream_iter_status);
1635             status = BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_MEMORY_ERROR;
1636             break;
1637         case LTTNG_LIVE_ITERATOR_STATUS_ERROR:
1638         case LTTNG_LIVE_ITERATOR_STATUS_INVAL:
1639         case LTTNG_LIVE_ITERATOR_STATUS_UNSUPPORTED:
1640             BT_CPPLOGE_APPEND_CAUSE_SPEC(lttng_live_msg_iter->logger,
1641                                          "Error preparing the next batch of messages: "
1642                                          "live-iter-status={}",
1643                                          stream_iter_status);
1644
1645             status = BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_ERROR;
1646             /* Put all existing messages on error. */
1647             put_messages(msgs, *count);
1648             break;
1649         default:
1650             bt_common_abort();
1651         }
1652
1653 end:
1654         return status;
1655     } catch (const std::bad_alloc&) {
1656         return BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_MEMORY_ERROR;
1657     } catch (const bt2::Error&) {
1658         return BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_ERROR;
1659     }
1660 }
1661
1662 static struct lttng_live_msg_iter *
1663 lttng_live_msg_iter_create(struct lttng_live_component *lttng_live_comp,
1664                            bt_self_message_iterator *self_msg_it)
1665 {
1666     lttng_live_msg_iter *msg_iter = new lttng_live_msg_iter {lttng_live_comp->logger};
1667     msg_iter->self_comp = lttng_live_comp->self_comp;
1668     msg_iter->lttng_live_comp = lttng_live_comp;
1669     msg_iter->self_msg_iter = self_msg_it;
1670
1671     msg_iter->active_stream_iter = 0;
1672     msg_iter->last_msg_ts_ns = INT64_MIN;
1673     msg_iter->was_interrupted = false;
1674
1675     msg_iter->sessions =
1676         g_ptr_array_new_with_free_func((GDestroyNotify) lttng_live_destroy_session);
1677     BT_ASSERT(msg_iter->sessions);
1678
1679     return msg_iter;
1680 }
1681
1682 bt_message_iterator_class_initialize_method_status
1683 lttng_live_msg_iter_init(bt_self_message_iterator *self_msg_it,
1684                          bt_self_message_iterator_configuration *, bt_self_component_port_output *)
1685 {
1686     try {
1687         bt_message_iterator_class_initialize_method_status status;
1688         struct lttng_live_component *lttng_live;
1689         struct lttng_live_msg_iter *lttng_live_msg_iter;
1690         enum lttng_live_viewer_status viewer_status;
1691         bt_self_component *self_comp = bt_self_message_iterator_borrow_component(self_msg_it);
1692
1693         lttng_live = (lttng_live_component *) bt_self_component_get_data(self_comp);
1694
1695         /* There can be only one downstream iterator at the same time. */
1696         BT_ASSERT(!lttng_live->has_msg_iter);
1697         lttng_live->has_msg_iter = true;
1698
1699         lttng_live_msg_iter = lttng_live_msg_iter_create(lttng_live, self_msg_it);
1700         if (!lttng_live_msg_iter) {
1701             BT_CPPLOGE_APPEND_CAUSE_SPEC(lttng_live->logger,
1702                                          "Failed to create lttng_live_msg_iter");
1703             status = BT_MESSAGE_ITERATOR_CLASS_INITIALIZE_METHOD_STATUS_MEMORY_ERROR;
1704             goto error;
1705         }
1706
1707         viewer_status = live_viewer_connection_create(
1708             lttng_live->params.url.c_str(), false, lttng_live_msg_iter, lttng_live_msg_iter->logger,
1709             &lttng_live_msg_iter->viewer_connection);
1710         if (viewer_status != LTTNG_LIVE_VIEWER_STATUS_OK) {
1711             if (viewer_status == LTTNG_LIVE_VIEWER_STATUS_ERROR) {
1712                 BT_CPPLOGE_APPEND_CAUSE_SPEC(lttng_live_msg_iter->logger,
1713                                              "Failed to create viewer connection");
1714             } else if (viewer_status == LTTNG_LIVE_VIEWER_STATUS_INTERRUPTED) {
1715                 /*
1716                  * Interruption in the _iter_init() method is not
1717                  * supported. Return an error.
1718                  */
1719                 BT_CPPLOGE_APPEND_CAUSE_SPEC(lttng_live_msg_iter->logger,
1720                                              "Interrupted while creating viewer connection");
1721             }
1722
1723             status = BT_MESSAGE_ITERATOR_CLASS_INITIALIZE_METHOD_STATUS_ERROR;
1724             goto error;
1725         }
1726
1727         viewer_status = lttng_live_create_viewer_session(lttng_live_msg_iter);
1728         if (viewer_status != LTTNG_LIVE_VIEWER_STATUS_OK) {
1729             if (viewer_status == LTTNG_LIVE_VIEWER_STATUS_ERROR) {
1730                 BT_CPPLOGE_APPEND_CAUSE_SPEC(lttng_live_msg_iter->logger,
1731                                              "Failed to create viewer session");
1732             } else if (viewer_status == LTTNG_LIVE_VIEWER_STATUS_INTERRUPTED) {
1733                 /*
1734                  * Interruption in the _iter_init() method is not
1735                  * supported. Return an error.
1736                  */
1737                 BT_CPPLOGE_APPEND_CAUSE_SPEC(lttng_live_msg_iter->logger,
1738                                              "Interrupted when creating viewer session");
1739             }
1740
1741             status = BT_MESSAGE_ITERATOR_CLASS_INITIALIZE_METHOD_STATUS_ERROR;
1742             goto error;
1743         }
1744
1745         if (lttng_live_msg_iter->sessions->len == 0) {
1746             switch (lttng_live->params.sess_not_found_act) {
1747             case SESSION_NOT_FOUND_ACTION_CONTINUE:
1748                 BT_CPPLOGI_SPEC(
1749                     lttng_live_msg_iter->logger,
1750                     "Unable to connect to the requested live viewer session. "
1751                     "Keep trying to connect because of {}=\"{}\" component parameter: url=\"{}\"",
1752                     SESS_NOT_FOUND_ACTION_PARAM, SESS_NOT_FOUND_ACTION_CONTINUE_STR,
1753                     lttng_live->params.url);
1754                 break;
1755             case SESSION_NOT_FOUND_ACTION_FAIL:
1756                 BT_CPPLOGE_APPEND_CAUSE_SPEC(
1757                     lttng_live_msg_iter->logger,
1758                     "Unable to connect to the requested live viewer session. "
1759                     "Fail the message iterator initialization because of {}=\"{}\" "
1760                     "component parameter: url =\"{}\"",
1761                     SESS_NOT_FOUND_ACTION_PARAM, SESS_NOT_FOUND_ACTION_FAIL_STR,
1762                     lttng_live->params.url);
1763                 status = BT_MESSAGE_ITERATOR_CLASS_INITIALIZE_METHOD_STATUS_ERROR;
1764                 goto error;
1765             case SESSION_NOT_FOUND_ACTION_END:
1766                 BT_CPPLOGI_SPEC(lttng_live_msg_iter->logger,
1767                                 "Unable to connect to the requested live viewer session. "
1768                                 "End gracefully at the first _next() call because of {}=\"{}\""
1769                                 " component parameter: url=\"{}\"",
1770                                 SESS_NOT_FOUND_ACTION_PARAM, SESS_NOT_FOUND_ACTION_END_STR,
1771                                 lttng_live->params.url);
1772                 break;
1773             default:
1774                 bt_common_abort();
1775             }
1776         }
1777
1778         bt_self_message_iterator_set_data(self_msg_it, lttng_live_msg_iter);
1779         status = BT_MESSAGE_ITERATOR_CLASS_INITIALIZE_METHOD_STATUS_OK;
1780         goto end;
1781
1782 error:
1783         lttng_live_msg_iter_destroy(lttng_live_msg_iter);
1784 end:
1785         return status;
1786     } catch (const std::bad_alloc&) {
1787         return BT_MESSAGE_ITERATOR_CLASS_INITIALIZE_METHOD_STATUS_MEMORY_ERROR;
1788     } catch (const bt2::Error&) {
1789         return BT_MESSAGE_ITERATOR_CLASS_INITIALIZE_METHOD_STATUS_ERROR;
1790     }
1791 }
1792
1793 static struct bt_param_validation_map_value_entry_descr list_sessions_params[] = {
1794     {URL_PARAM, BT_PARAM_VALIDATION_MAP_VALUE_ENTRY_MANDATORY,
1795      bt_param_validation_value_descr::makeString()},
1796     BT_PARAM_VALIDATION_MAP_VALUE_ENTRY_END};
1797
1798 static bt_component_class_query_method_status
1799 lttng_live_query_list_sessions(const bt_value *params, const bt_value **result,
1800                                const bt2c::Logger& logger)
1801 {
1802     bt_component_class_query_method_status status;
1803     const bt_value *url_value = NULL;
1804     const char *url;
1805     struct live_viewer_connection *viewer_connection = NULL;
1806     enum lttng_live_viewer_status viewer_status;
1807     enum bt_param_validation_status validation_status;
1808     gchar *validate_error = NULL;
1809
1810     validation_status = bt_param_validation_validate(params, list_sessions_params, &validate_error);
1811     if (validation_status == BT_PARAM_VALIDATION_STATUS_MEMORY_ERROR) {
1812         status = BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_MEMORY_ERROR;
1813         goto error;
1814     } else if (validation_status == BT_PARAM_VALIDATION_STATUS_VALIDATION_ERROR) {
1815         status = BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_ERROR;
1816         BT_CPPLOGE_APPEND_CAUSE_SPEC(logger, "{}", validate_error);
1817         goto error;
1818     }
1819
1820     url_value = bt_value_map_borrow_entry_value_const(params, URL_PARAM);
1821     url = bt_value_string_get(url_value);
1822
1823     viewer_status = live_viewer_connection_create(url, true, NULL, logger, &viewer_connection);
1824     if (viewer_status != LTTNG_LIVE_VIEWER_STATUS_OK) {
1825         if (viewer_status == LTTNG_LIVE_VIEWER_STATUS_ERROR) {
1826             BT_CPPLOGE_APPEND_CAUSE_SPEC(logger, "Failed to create viewer connection");
1827             status = BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_ERROR;
1828         } else if (viewer_status == LTTNG_LIVE_VIEWER_STATUS_INTERRUPTED) {
1829             status = BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_AGAIN;
1830         } else {
1831             bt_common_abort();
1832         }
1833         goto error;
1834     }
1835
1836     status = live_viewer_connection_list_sessions(viewer_connection, result);
1837     if (status != BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_OK) {
1838         BT_CPPLOGE_APPEND_CAUSE_SPEC(logger, "Failed to list viewer sessions");
1839         goto error;
1840     }
1841
1842     goto end;
1843
1844 error:
1845     BT_VALUE_PUT_REF_AND_RESET(*result);
1846
1847     if (status >= 0) {
1848         status = BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_ERROR;
1849     }
1850
1851 end:
1852     if (viewer_connection) {
1853         live_viewer_connection_destroy(viewer_connection);
1854     }
1855
1856     g_free(validate_error);
1857
1858     return status;
1859 }
1860
1861 static bt_component_class_query_method_status
1862 lttng_live_query_support_info(const bt_value *params, const bt_value **result,
1863                               const bt2c::Logger& logger)
1864 {
1865     bt_component_class_query_method_status status = BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_OK;
1866     const bt_value *input_type_value;
1867     const bt_value *input_value;
1868     double weight = 0;
1869     struct bt_common_lttng_live_url_parts parts = {};
1870
1871     /* Used by the logging macros */
1872     __attribute__((unused)) bt_self_component *self_comp = NULL;
1873
1874     *result = NULL;
1875     input_type_value = bt_value_map_borrow_entry_value_const(params, "type");
1876     if (!input_type_value) {
1877         BT_CPPLOGE_APPEND_CAUSE_SPEC(logger, "Missing expected `type` parameter.");
1878         goto error;
1879     }
1880
1881     if (!bt_value_is_string(input_type_value)) {
1882         BT_CPPLOGE_APPEND_CAUSE_SPEC(logger, "`type` parameter is not a string value.");
1883         goto error;
1884     }
1885
1886     if (strcmp(bt_value_string_get(input_type_value), "string") != 0) {
1887         /* We don't handle file system paths */
1888         goto create_result;
1889     }
1890
1891     input_value = bt_value_map_borrow_entry_value_const(params, "input");
1892     if (!input_value) {
1893         BT_CPPLOGE_APPEND_CAUSE_SPEC(logger, "Missing expected `input` parameter.");
1894         goto error;
1895     }
1896
1897     if (!bt_value_is_string(input_value)) {
1898         BT_CPPLOGE_APPEND_CAUSE_SPEC(logger, "`input` parameter is not a string value.");
1899         goto error;
1900     }
1901
1902     parts = bt_common_parse_lttng_live_url(bt_value_string_get(input_value), NULL, 0);
1903     if (parts.session_name) {
1904         /*
1905          * Looks pretty much like an LTTng live URL: we got the
1906          * session name part, which forms a complete URL.
1907          */
1908         weight = .75;
1909     }
1910
1911 create_result:
1912     *result = bt_value_real_create_init(weight);
1913     if (!*result) {
1914         status = BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_MEMORY_ERROR;
1915         goto error;
1916     }
1917
1918     goto end;
1919
1920 error:
1921     if (status >= 0) {
1922         status = BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_ERROR;
1923     }
1924
1925     BT_ASSERT(!*result);
1926
1927 end:
1928     bt_common_destroy_lttng_live_url_parts(&parts);
1929     return status;
1930 }
1931
1932 bt_component_class_query_method_status lttng_live_query(bt_self_component_class_source *comp_class,
1933                                                         bt_private_query_executor *priv_query_exec,
1934                                                         const char *object, const bt_value *params,
1935                                                         __attribute__((unused)) void *method_data,
1936                                                         const bt_value **result)
1937 {
1938     try {
1939         bt_component_class_query_method_status status = BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_OK;
1940         bt2c::Logger logger {bt2::SelfComponentClass {comp_class},
1941                              bt2::PrivateQueryExecutor {priv_query_exec},
1942                              "PLUGIN/SRC.CTF.LTTNG-LIVE/QUERY"};
1943
1944         if (strcmp(object, "sessions") == 0) {
1945             status = lttng_live_query_list_sessions(params, result, logger);
1946         } else if (strcmp(object, "babeltrace.support-info") == 0) {
1947             status = lttng_live_query_support_info(params, result, logger);
1948         } else {
1949             BT_CPPLOGI_SPEC(logger, "Unknown query object `{}`", object);
1950             status = BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_UNKNOWN_OBJECT;
1951             goto end;
1952         }
1953
1954 end:
1955         return status;
1956     } catch (const std::bad_alloc&) {
1957         return BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_MEMORY_ERROR;
1958     } catch (const bt2::Error&) {
1959         return BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_ERROR;
1960     }
1961 }
1962
1963 void lttng_live_component_finalize(bt_self_component_source *component)
1964 {
1965     lttng_live_component::UP {static_cast<lttng_live_component *>(
1966         bt_self_component_get_data(bt_self_component_source_as_self_component(component)))};
1967 }
1968
1969 static enum session_not_found_action
1970 parse_session_not_found_action_param(const bt_value *no_session_param)
1971 {
1972     enum session_not_found_action action;
1973     const char *no_session_act_str = bt_value_string_get(no_session_param);
1974
1975     if (strcmp(no_session_act_str, SESS_NOT_FOUND_ACTION_CONTINUE_STR) == 0) {
1976         action = SESSION_NOT_FOUND_ACTION_CONTINUE;
1977     } else if (strcmp(no_session_act_str, SESS_NOT_FOUND_ACTION_FAIL_STR) == 0) {
1978         action = SESSION_NOT_FOUND_ACTION_FAIL;
1979     } else {
1980         BT_ASSERT(strcmp(no_session_act_str, SESS_NOT_FOUND_ACTION_END_STR) == 0);
1981         action = SESSION_NOT_FOUND_ACTION_END;
1982     }
1983
1984     return action;
1985 }
1986
1987 static bt_param_validation_value_descr inputs_elem_descr =
1988     bt_param_validation_value_descr::makeString();
1989
1990 static const char *sess_not_found_action_choices[] = {
1991     SESS_NOT_FOUND_ACTION_CONTINUE_STR,
1992     SESS_NOT_FOUND_ACTION_FAIL_STR,
1993     SESS_NOT_FOUND_ACTION_END_STR,
1994 };
1995
1996 static struct bt_param_validation_map_value_entry_descr params_descr[] = {
1997     {INPUTS_PARAM, BT_PARAM_VALIDATION_MAP_VALUE_ENTRY_MANDATORY,
1998      bt_param_validation_value_descr::makeArray(1, 1, inputs_elem_descr)},
1999     {SESS_NOT_FOUND_ACTION_PARAM, BT_PARAM_VALIDATION_MAP_VALUE_ENTRY_OPTIONAL,
2000      bt_param_validation_value_descr::makeString(sess_not_found_action_choices)},
2001     BT_PARAM_VALIDATION_MAP_VALUE_ENTRY_END};
2002
2003 static bt_component_class_initialize_method_status
2004 lttng_live_component_create(const bt_value *params, bt_self_component_source *self_comp,
2005                             lttng_live_component::UP& component)
2006 {
2007     const bt_value *inputs_value;
2008     const bt_value *url_value;
2009     const bt_value *value;
2010     enum bt_param_validation_status validation_status;
2011     gchar *validation_error = NULL;
2012     bt2c::Logger logger {bt2::SelfSourceComponent {self_comp}, "PLUGIN/SRC.CTF.LTTNG-LIVE/COMP"};
2013
2014     validation_status = bt_param_validation_validate(params, params_descr, &validation_error);
2015     if (validation_status == BT_PARAM_VALIDATION_STATUS_MEMORY_ERROR) {
2016         return BT_COMPONENT_CLASS_INITIALIZE_METHOD_STATUS_MEMORY_ERROR;
2017     } else if (validation_status == BT_PARAM_VALIDATION_STATUS_VALIDATION_ERROR) {
2018         bt2c::GCharUP errorFreer {validation_error};
2019         BT_CPPLOGE_APPEND_CAUSE_SPEC(logger, "{}", validation_error);
2020         return BT_COMPONENT_CLASS_INITIALIZE_METHOD_STATUS_ERROR;
2021     }
2022
2023     auto lttng_live = bt2s::make_unique<lttng_live_component>(std::move(logger));
2024     lttng_live->self_comp = bt_self_component_source_as_self_component(self_comp);
2025     lttng_live->max_query_size = MAX_QUERY_SIZE;
2026     lttng_live->has_msg_iter = false;
2027
2028     inputs_value = bt_value_map_borrow_entry_value_const(params, INPUTS_PARAM);
2029     url_value = bt_value_array_borrow_element_by_index_const(inputs_value, 0);
2030     lttng_live->params.url = bt_value_string_get(url_value);
2031
2032     value = bt_value_map_borrow_entry_value_const(params, SESS_NOT_FOUND_ACTION_PARAM);
2033     if (value) {
2034         lttng_live->params.sess_not_found_act = parse_session_not_found_action_param(value);
2035     } else {
2036         BT_CPPLOGI_SPEC(lttng_live->logger,
2037                         "Optional `{}` parameter is missing: defaulting to `{}`.",
2038                         SESS_NOT_FOUND_ACTION_PARAM, SESS_NOT_FOUND_ACTION_CONTINUE_STR);
2039         lttng_live->params.sess_not_found_act = SESSION_NOT_FOUND_ACTION_CONTINUE;
2040     }
2041
2042     component = std::move(lttng_live);
2043     return BT_COMPONENT_CLASS_INITIALIZE_METHOD_STATUS_OK;
2044 }
2045
2046 bt_component_class_initialize_method_status
2047 lttng_live_component_init(bt_self_component_source *self_comp_src,
2048                           bt_self_component_source_configuration *, const bt_value *params, void *)
2049 {
2050     try {
2051         lttng_live_component::UP lttng_live;
2052         bt_component_class_initialize_method_status ret;
2053         bt_self_component *self_comp = bt_self_component_source_as_self_component(self_comp_src);
2054         bt_self_component_add_port_status add_port_status;
2055
2056         ret = lttng_live_component_create(params, self_comp_src, lttng_live);
2057         if (ret != BT_COMPONENT_CLASS_INITIALIZE_METHOD_STATUS_OK) {
2058             return ret;
2059         }
2060
2061         add_port_status =
2062             bt_self_component_source_add_output_port(self_comp_src, "out", NULL, NULL);
2063         if (add_port_status != BT_SELF_COMPONENT_ADD_PORT_STATUS_OK) {
2064             ret = (bt_component_class_initialize_method_status) add_port_status;
2065             return ret;
2066         }
2067
2068         bt_self_component_set_data(self_comp, lttng_live.release());
2069
2070         return BT_COMPONENT_CLASS_INITIALIZE_METHOD_STATUS_OK;
2071     } catch (const std::bad_alloc&) {
2072         return BT_COMPONENT_CLASS_INITIALIZE_METHOD_STATUS_MEMORY_ERROR;
2073     } catch (const bt2::Error&) {
2074         return BT_COMPONENT_CLASS_INITIALIZE_METHOD_STATUS_ERROR;
2075     }
2076 }