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