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

   1 /*
   2  * lttng-live.c
   3  *
   4  * Babeltrace CTF LTTng-live Client Component
   5  *
   6  * Copyright 2019 Francis Deslauriers <francis.deslauriers@efficios.com>
   7  * Copyright 2016 Jérémie Galarneau <jeremie.galarneau@efficios.com>
   8  * Copyright 2016 Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
   9  *
  10  * Author: Jérémie Galarneau <jeremie.galarneau@efficios.com>
  11  *
  12  * Permission is hereby granted, free of charge, to any person obtaining a copy
  13  * of this software and associated documentation files (the "Software"), to deal
  14  * in the Software without restriction, including without limitation the rights
  15  * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
  16  * copies of the Software, and to permit persons to whom the Software is
  17  * furnished to do so, subject to the following conditions:
  18  *
  19  * The above copyright notice and this permission notice shall be included in
  20  * all copies or substantial portions of the Software.
  21  *
  22  * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
  23  * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
  24  * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
  25  * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
  26  * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
  27  * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
  28  * SOFTWARE.
  29  */
  30
  31 #define BT_COMP_LOG_SELF_COMP self_comp
  32 #define BT_LOG_OUTPUT_LEVEL log_level
  33 #define BT_LOG_TAG "PLUGIN/SRC.CTF.LTTNG-LIVE"
  34 #include "logging/comp-logging.h"
  35
  36 #include <glib.h>
  37 #include <inttypes.h>
  38 #include <unistd.h>
  39
  40 #include "common/assert.h"
  41 #include <babeltrace2/babeltrace.h>
  42 #include "compat/compiler.h"
  43 #include <babeltrace2/types.h>
  44
  45 #include "plugins/common/muxing/muxing.h"
  46
  47 #include "data-stream.h"
  48 #include "metadata.h"
  49 #include "lttng-live.h"
  50
  51 #define MAX_QUERY_SIZE                      (256*1024)
  52 #define URL_PARAM                           "url"
  53 #define SESS_NOT_FOUND_ACTION_PARAM         "session-not-found-action"
  54 #define SESS_NOT_FOUND_ACTION_CONTINUE_STR  "continue"
  55 #define SESS_NOT_FOUND_ACTION_FAIL_STR      "fail"
  56 #define SESS_NOT_FOUND_ACTION_END_STR       "end"
  57
  58 #define print_dbg(fmt, ...)     BT_COMP_LOGD(fmt, ## __VA_ARGS__)
  59
  60 static
  61 const char *print_live_iterator_status(enum lttng_live_iterator_status status)
  62 {
  63         switch (status) {
  64         case LTTNG_LIVE_ITERATOR_STATUS_CONTINUE:
  65                 return "LTTNG_LIVE_ITERATOR_STATUS_CONTINUE";
  66         case LTTNG_LIVE_ITERATOR_STATUS_AGAIN:
  67                 return "LTTNG_LIVE_ITERATOR_STATUS_AGAIN";
  68         case LTTNG_LIVE_ITERATOR_STATUS_END:
  69                 return "LTTNG_LIVE_ITERATOR_STATUS_END";
  70         case LTTNG_LIVE_ITERATOR_STATUS_OK:
  71                 return "LTTNG_LIVE_ITERATOR_STATUS_OK";
  72         case LTTNG_LIVE_ITERATOR_STATUS_INVAL:
  73                 return "LTTNG_LIVE_ITERATOR_STATUS_INVAL";
  74         case LTTNG_LIVE_ITERATOR_STATUS_ERROR:
  75                 return "LTTNG_LIVE_ITERATOR_STATUS_ERROR";
  76         case LTTNG_LIVE_ITERATOR_STATUS_NOMEM:
  77                 return "LTTNG_LIVE_ITERATOR_STATUS_NOMEM";
  78         case LTTNG_LIVE_ITERATOR_STATUS_UNSUPPORTED:
  79                 return "LTTNG_LIVE_ITERATOR_STATUS_UNSUPPORTED";
  80         default:
  81                 abort();
  82         }
  83 }
  84
  85 static
  86 const char *print_state(struct lttng_live_stream_iterator *s)
  87 {
  88         switch (s->state) {
  89         case LTTNG_LIVE_STREAM_ACTIVE_NO_DATA:
  90                 return "ACTIVE_NO_DATA";
  91         case LTTNG_LIVE_STREAM_QUIESCENT_NO_DATA:
  92                 return "QUIESCENT_NO_DATA";
  93         case LTTNG_LIVE_STREAM_QUIESCENT:
  94                 return "QUIESCENT";
  95         case LTTNG_LIVE_STREAM_ACTIVE_DATA:
  96                 return "ACTIVE_DATA";
  97         case LTTNG_LIVE_STREAM_EOF:
  98                 return "EOF";
  99         default:
 100                 return "ERROR";
 101         }
 102 }
 103
 104 #define print_stream_state(live_stream_iter) \
 105         do { \
 106                 BT_COMP_LOGD("stream state %s last_inact_ts %" PRId64  \
 107                         ", curr_inact_ts %" PRId64, \
 108                         print_state(live_stream_iter), \
 109                         live_stream_iter->last_inactivity_ts, \
 110                         live_stream_iter->current_inactivity_ts); \
 111         } while (0);
 112
 113 BT_HIDDEN
 114 bool lttng_live_graph_is_canceled(struct lttng_live_msg_iter *msg_iter)
 115 {
 116         bool ret;
 117
 118         if (!msg_iter) {
 119                 ret = false;
 120                 goto end;
 121         }
 122
 123         ret = bt_self_message_iterator_is_interrupted(
 124                 msg_iter->self_msg_iter);
 125
 126 end:
 127         return ret;
 128 }
 129
 130 static
 131 struct lttng_live_trace *lttng_live_find_trace(struct lttng_live_session *session,
 132                 uint64_t trace_id)
 133 {
 134         uint64_t trace_idx;
 135         struct lttng_live_trace *ret_trace = NULL;
 136
 137         for (trace_idx = 0; trace_idx < session->traces->len; trace_idx++) {
 138                 struct lttng_live_trace *trace =
 139                         g_ptr_array_index(session->traces, trace_idx);
 140                 if (trace->id == trace_id) {
 141                         ret_trace = trace;
 142                         goto end;
 143                 }
 144         }
 145
 146 end:
 147         return ret_trace;
 148 }
 149
 150 static
 151 void lttng_live_destroy_trace(struct lttng_live_trace *trace)
 152 {
 153         bt_logging_level log_level = trace->log_level;
 154         bt_self_component *self_comp = trace->self_comp;
 155
 156         BT_COMP_LOGD("Destroy lttng_live_trace");
 157
 158         BT_ASSERT(trace->stream_iterators);
 159         g_ptr_array_free(trace->stream_iterators, TRUE);
 160
 161         BT_TRACE_PUT_REF_AND_RESET(trace->trace);
 162         BT_TRACE_CLASS_PUT_REF_AND_RESET(trace->trace_class);
 163
 164         lttng_live_metadata_fini(trace);
 165         g_free(trace);
 166 }
 167
 168 static
 169 struct lttng_live_trace *lttng_live_create_trace(struct lttng_live_session *session,
 170                 uint64_t trace_id)
 171 {
 172         struct lttng_live_trace *trace = NULL;
 173         bt_logging_level log_level = session->log_level;
 174         bt_self_component *self_comp = session->self_comp;
 175
 176         trace = g_new0(struct lttng_live_trace, 1);
 177         if (!trace) {
 178                 goto error;
 179         }
 180         trace->log_level = session->log_level;
 181         trace->self_comp = session->self_comp;
 182         trace->session = session;
 183         trace->id = trace_id;
 184         trace->trace_class = NULL;
 185         trace->trace = NULL;
 186         trace->stream_iterators = g_ptr_array_new_with_free_func(
 187                         (GDestroyNotify) lttng_live_stream_iterator_destroy);
 188         BT_ASSERT(trace->stream_iterators);
 189         trace->new_metadata_needed = true;
 190         g_ptr_array_add(session->traces, trace);
 191
 192         BT_COMP_LOGI("Create trace");
 193         goto end;
 194 error:
 195         g_free(trace);
 196         trace = NULL;
 197 end:
 198         return trace;
 199 }
 200
 201 BT_HIDDEN
 202 struct lttng_live_trace *lttng_live_borrow_trace(
 203                 struct lttng_live_session *session, uint64_t trace_id)
 204 {
 205         struct lttng_live_trace *trace;
 206
 207         trace = lttng_live_find_trace(session, trace_id);
 208         if (trace) {
 209                 goto end;
 210         }
 211
 212         /* The session is the owner of the newly created trace. */
 213         trace = lttng_live_create_trace(session, trace_id);
 214
 215 end:
 216         return trace;
 217 }
 218
 219 BT_HIDDEN
 220 int lttng_live_add_session(struct lttng_live_msg_iter *lttng_live_msg_iter,
 221                 uint64_t session_id, const char *hostname,
 222                 const char *session_name)
 223 {
 224         int ret = 0;
 225         struct lttng_live_session *session;
 226         bt_logging_level log_level = lttng_live_msg_iter->log_level;
 227         bt_self_component *self_comp = lttng_live_msg_iter->self_comp;
 228
 229         session = g_new0(struct lttng_live_session, 1);
 230         if (!session) {
 231                 goto error;
 232         }
 233
 234         session->log_level = lttng_live_msg_iter->log_level;
 235         session->self_comp = lttng_live_msg_iter->self_comp;
 236         session->id = session_id;
 237         session->traces = g_ptr_array_new_with_free_func(
 238                         (GDestroyNotify) lttng_live_destroy_trace);
 239         BT_ASSERT(session->traces);
 240         session->lttng_live_msg_iter = lttng_live_msg_iter;
 241         session->new_streams_needed = true;
 242         session->hostname = g_string_new(hostname);
 243         BT_ASSERT(session->hostname);
 244
 245         session->session_name = g_string_new(session_name);
 246         BT_ASSERT(session->session_name);
 247
 248         BT_COMP_LOGI("Reading from session: %" PRIu64 " hostname: %s session_name: %s",
 249                 session->id, hostname, session_name);
 250         g_ptr_array_add(lttng_live_msg_iter->sessions, session);
 251         goto end;
 252 error:
 253         BT_COMP_LOGE("Error adding session");
 254         g_free(session);
 255         ret = -1;
 256 end:
 257         return ret;
 258 }
 259
 260 static
 261 void lttng_live_destroy_session(struct lttng_live_session *session)
 262 {
 263         bt_logging_level log_level;
 264         bt_self_component *self_comp;
 265
 266         if (!session) {
 267                 goto end;
 268         }
 269
 270         log_level = session->log_level;
 271         self_comp = session->self_comp;
 272         BT_COMP_LOGD("Destroy lttng live session");
 273         if (session->id != -1ULL) {
 274                 if (lttng_live_detach_session(session)) {
 275                         if (session->lttng_live_msg_iter &&
 276                                         !lttng_live_graph_is_canceled(
 277                                                 session->lttng_live_msg_iter)) {
 278                                 /* Old relayd cannot detach sessions. */
 279                                 BT_COMP_LOGD("Unable to detach lttng live session %" PRIu64,
 280                                         session->id);
 281                         }
 282                 }
 283                 session->id = -1ULL;
 284         }
 285
 286         if (session->traces) {
 287                 g_ptr_array_free(session->traces, TRUE);
 288         }
 289
 290         if (session->hostname) {
 291                 g_string_free(session->hostname, TRUE);
 292         }
 293         if (session->session_name) {
 294                 g_string_free(session->session_name, TRUE);
 295         }
 296         g_free(session);
 297
 298 end:
 299         return;
 300 }
 301
 302 static
 303 void lttng_live_msg_iter_destroy(struct lttng_live_msg_iter *lttng_live_msg_iter)
 304 {
 305         if (!lttng_live_msg_iter) {
 306                 goto end;
 307         }
 308
 309         if (lttng_live_msg_iter->sessions) {
 310                 g_ptr_array_free(lttng_live_msg_iter->sessions, TRUE);
 311         }
 312
 313         BT_OBJECT_PUT_REF_AND_RESET(lttng_live_msg_iter->viewer_connection);
 314         BT_ASSERT(lttng_live_msg_iter->lttng_live_comp);
 315         BT_ASSERT(lttng_live_msg_iter->lttng_live_comp->has_msg_iter);
 316
 317         /* All stream iterators must be destroyed at this point. */
 318         BT_ASSERT(lttng_live_msg_iter->active_stream_iter == 0);
 319         lttng_live_msg_iter->lttng_live_comp->has_msg_iter = false;
 320
 321         g_free(lttng_live_msg_iter);
 322
 323 end:
 324         return;
 325 }
 326
 327 BT_HIDDEN
 328 void lttng_live_msg_iter_finalize(bt_self_message_iterator *self_msg_iter)
 329 {
 330         struct lttng_live_msg_iter *lttng_live_msg_iter;
 331
 332         BT_ASSERT(self_msg_iter);
 333
 334         lttng_live_msg_iter = bt_self_message_iterator_get_data(self_msg_iter);
 335         BT_ASSERT(lttng_live_msg_iter);
 336         lttng_live_msg_iter_destroy(lttng_live_msg_iter);
 337 }
 338
 339 static
 340 enum lttng_live_iterator_status lttng_live_iterator_next_check_stream_state(
 341                 struct lttng_live_stream_iterator *lttng_live_stream)
 342 {
 343         bt_logging_level log_level = lttng_live_stream->log_level;
 344         bt_self_component *self_comp = lttng_live_stream->self_comp;
 345
 346         switch (lttng_live_stream->state) {
 347         case LTTNG_LIVE_STREAM_QUIESCENT:
 348         case LTTNG_LIVE_STREAM_ACTIVE_DATA:
 349                 break;
 350         case LTTNG_LIVE_STREAM_ACTIVE_NO_DATA:
 351                 /* Invalid state. */
 352                 BT_COMP_LOGF("Unexpected stream state \"ACTIVE_NO_DATA\"");
 353                 abort();
 354         case LTTNG_LIVE_STREAM_QUIESCENT_NO_DATA:
 355                 /* Invalid state. */
 356                 BT_COMP_LOGF("Unexpected stream state \"QUIESCENT_NO_DATA\"");
 357                 abort();
 358         case LTTNG_LIVE_STREAM_EOF:
 359                 break;
 360         }
 361         return LTTNG_LIVE_ITERATOR_STATUS_OK;
 362 }
 363
 364 /*
 365  * For active no data stream, fetch next data. It can be either:
 366  * - quiescent: need to put it in the prio heap at quiescent end
 367  *   timestamp,
 368  * - have data: need to wire up first event into the prio heap,
 369  * - have no data on this stream at this point: need to retry (AGAIN) or
 370  *   return EOF.
 371  */
 372 static
 373 enum lttng_live_iterator_status lttng_live_iterator_next_handle_one_no_data_stream(
 374                 struct lttng_live_msg_iter *lttng_live_msg_iter,
 375                 struct lttng_live_stream_iterator *lttng_live_stream)
 376 {
 377         bt_logging_level log_level = lttng_live_msg_iter->log_level;
 378         bt_self_component *self_comp = lttng_live_msg_iter->self_comp;
 379         enum lttng_live_iterator_status ret =
 380                         LTTNG_LIVE_ITERATOR_STATUS_OK;
 381         struct packet_index index;
 382         enum lttng_live_stream_state orig_state = lttng_live_stream->state;
 383
 384         if (lttng_live_stream->trace->new_metadata_needed) {
 385                 ret = LTTNG_LIVE_ITERATOR_STATUS_CONTINUE;
 386                 goto end;
 387         }
 388         if (lttng_live_stream->trace->session->new_streams_needed) {
 389                 ret = LTTNG_LIVE_ITERATOR_STATUS_CONTINUE;
 390                 goto end;
 391         }
 392         if (lttng_live_stream->state != LTTNG_LIVE_STREAM_ACTIVE_NO_DATA &&
 393                         lttng_live_stream->state != LTTNG_LIVE_STREAM_QUIESCENT_NO_DATA) {
 394                 goto end;
 395         }
 396         ret = lttng_live_get_next_index(lttng_live_msg_iter, lttng_live_stream,
 397                 &index);
 398         if (ret != LTTNG_LIVE_ITERATOR_STATUS_OK) {
 399                 goto end;
 400         }
 401         BT_ASSERT(lttng_live_stream->state != LTTNG_LIVE_STREAM_EOF);
 402         if (lttng_live_stream->state == LTTNG_LIVE_STREAM_QUIESCENT) {
 403                 uint64_t last_inact_ts = lttng_live_stream->last_inactivity_ts,
 404                          curr_inact_ts = lttng_live_stream->current_inactivity_ts;
 405
 406                 if (orig_state == LTTNG_LIVE_STREAM_QUIESCENT_NO_DATA &&
 407                                 last_inact_ts == curr_inact_ts) {
 408                         ret = LTTNG_LIVE_ITERATOR_STATUS_AGAIN;
 409                         print_stream_state(lttng_live_stream);
 410                 } else {
 411                         ret = LTTNG_LIVE_ITERATOR_STATUS_CONTINUE;
 412                 }
 413                 goto end;
 414         }
 415         lttng_live_stream->base_offset = index.offset;
 416         lttng_live_stream->offset = index.offset;
 417         lttng_live_stream->len = index.packet_size / CHAR_BIT;
 418 end:
 419         if (ret == LTTNG_LIVE_ITERATOR_STATUS_OK) {
 420                 ret = lttng_live_iterator_next_check_stream_state(lttng_live_stream);
 421         }
 422         return ret;
 423 }
 424
 425 /*
 426  * Creation of the message requires the ctf trace class to be created
 427  * beforehand, but the live protocol gives us all streams (including metadata)
 428  * at once. So we split it in three steps: getting streams, getting metadata
 429  * (which creates the ctf trace class), and then creating the per-stream
 430  * messages.
 431  */
 432 static
 433 enum lttng_live_iterator_status lttng_live_get_session(
 434                 struct lttng_live_msg_iter *lttng_live_msg_iter,
 435                 struct lttng_live_session *session)
 436 {
 437         enum lttng_live_iterator_status status;
 438         uint64_t trace_idx;
 439         int ret = 0;
 440
 441         if (!session->attached) {
 442                 ret = lttng_live_attach_session(session);
 443                 if (ret) {
 444                         if (lttng_live_msg_iter && lttng_live_graph_is_canceled(
 445                                         lttng_live_msg_iter)) {
 446                                 status = LTTNG_LIVE_ITERATOR_STATUS_AGAIN;
 447                         } else {
 448                                 status = LTTNG_LIVE_ITERATOR_STATUS_ERROR;
 449                         }
 450                         goto end;
 451                 }
 452         }
 453
 454         status = lttng_live_get_new_streams(session);
 455         if (status != LTTNG_LIVE_ITERATOR_STATUS_OK &&
 456                         status != LTTNG_LIVE_ITERATOR_STATUS_END) {
 457                 goto end;
 458         }
 459         for (trace_idx = 0; trace_idx < session->traces->len; trace_idx++) {
 460                 struct lttng_live_trace *trace =
 461                         g_ptr_array_index(session->traces, trace_idx);
 462
 463                 status = lttng_live_metadata_update(trace);
 464                 if (status != LTTNG_LIVE_ITERATOR_STATUS_OK &&
 465                                 status != LTTNG_LIVE_ITERATOR_STATUS_END) {
 466                         goto end;
 467                 }
 468         }
 469         status = lttng_live_lazy_msg_init(session);
 470
 471 end:
 472         return status;
 473 }
 474
 475 BT_HIDDEN
 476 void lttng_live_need_new_streams(struct lttng_live_msg_iter *lttng_live_msg_iter)
 477 {
 478         uint64_t session_idx;
 479
 480         for (session_idx = 0; session_idx < lttng_live_msg_iter->sessions->len;
 481                         session_idx++) {
 482                 struct lttng_live_session *session =
 483                         g_ptr_array_index(lttng_live_msg_iter->sessions, session_idx);
 484                 session->new_streams_needed = true;
 485         }
 486 }
 487
 488 static
 489 void lttng_live_force_new_streams_and_metadata(struct lttng_live_msg_iter *lttng_live_msg_iter)
 490 {
 491         uint64_t session_idx, trace_idx;
 492
 493         for (session_idx = 0; session_idx < lttng_live_msg_iter->sessions->len;
 494                         session_idx++) {
 495                 struct lttng_live_session *session =
 496                         g_ptr_array_index(lttng_live_msg_iter->sessions, session_idx);
 497                 session->new_streams_needed = true;
 498                 for (trace_idx = 0; trace_idx < session->traces->len;
 499                                 trace_idx++) {
 500                         struct lttng_live_trace *trace =
 501                                 g_ptr_array_index(session->traces, trace_idx);
 502                         trace->new_metadata_needed = true;
 503                 }
 504         }
 505 }
 506
 507 static
 508 enum lttng_live_iterator_status
 509 lttng_live_iterator_handle_new_streams_and_metadata(
 510                 struct lttng_live_msg_iter *lttng_live_msg_iter)
 511 {
 512         enum lttng_live_iterator_status ret =
 513                         LTTNG_LIVE_ITERATOR_STATUS_OK;
 514         uint64_t session_idx = 0, nr_sessions_opened = 0;
 515         struct lttng_live_session *session;
 516         enum session_not_found_action sess_not_found_act =
 517                 lttng_live_msg_iter->lttng_live_comp->params.sess_not_found_act;
 518
 519         /*
 520          * In a remotely distant future, we could add a "new
 521          * session" flag to the protocol, which would tell us that we
 522          * need to query for new sessions even though we have sessions
 523          * currently ongoing.
 524          */
 525         if (lttng_live_msg_iter->sessions->len == 0) {
 526                 if (sess_not_found_act != SESSION_NOT_FOUND_ACTION_CONTINUE) {
 527                         ret = LTTNG_LIVE_ITERATOR_STATUS_END;
 528                         goto end;
 529                 } else {
 530                         /*
 531                          * Retry to create a viewer session for the requested
 532                          * session name.
 533                          */
 534                         if (lttng_live_create_viewer_session(lttng_live_msg_iter)) {
 535                                 ret = LTTNG_LIVE_ITERATOR_STATUS_ERROR;
 536                                 goto end;
 537                         }
 538                 }
 539         }
 540
 541         for (session_idx = 0; session_idx < lttng_live_msg_iter->sessions->len;
 542                         session_idx++) {
 543                 session = g_ptr_array_index(lttng_live_msg_iter->sessions,
 544                                 session_idx);
 545                 ret = lttng_live_get_session(lttng_live_msg_iter, session);
 546                 switch (ret) {
 547                 case LTTNG_LIVE_ITERATOR_STATUS_OK:
 548                         break;
 549                 case LTTNG_LIVE_ITERATOR_STATUS_END:
 550                         ret = LTTNG_LIVE_ITERATOR_STATUS_OK;
 551                         break;
 552                 default:
 553                         goto end;
 554                 }
 555                 if (!session->closed) {
 556                         nr_sessions_opened++;
 557                 }
 558         }
 559 end:
 560         if (ret == LTTNG_LIVE_ITERATOR_STATUS_OK &&
 561                         sess_not_found_act != SESSION_NOT_FOUND_ACTION_CONTINUE &&
 562                         nr_sessions_opened == 0) {
 563                 ret = LTTNG_LIVE_ITERATOR_STATUS_END;
 564         }
 565         return ret;
 566 }
 567
 568 static
 569 enum lttng_live_iterator_status emit_inactivity_message(
 570                 struct lttng_live_msg_iter *lttng_live_msg_iter,
 571                 struct lttng_live_stream_iterator *stream_iter,
 572                 bt_message **message, uint64_t timestamp)
 573 {
 574         enum lttng_live_iterator_status ret =
 575                         LTTNG_LIVE_ITERATOR_STATUS_OK;
 576         bt_message *msg = NULL;
 577
 578         BT_ASSERT(stream_iter->trace->clock_class);
 579
 580         msg = bt_message_message_iterator_inactivity_create(
 581                         lttng_live_msg_iter->self_msg_iter,
 582                         stream_iter->trace->clock_class,
 583                         timestamp);
 584         if (!msg) {
 585                 goto error;
 586         }
 587
 588         *message = msg;
 589 end:
 590         return ret;
 591
 592 error:
 593         ret = LTTNG_LIVE_ITERATOR_STATUS_ERROR;
 594         bt_message_put_ref(msg);
 595         goto end;
 596 }
 597
 598 static
 599 enum lttng_live_iterator_status lttng_live_iterator_next_handle_one_quiescent_stream(
 600                 struct lttng_live_msg_iter *lttng_live_msg_iter,
 601                 struct lttng_live_stream_iterator *lttng_live_stream,
 602                 bt_message **message)
 603 {
 604         enum lttng_live_iterator_status ret =
 605                         LTTNG_LIVE_ITERATOR_STATUS_OK;
 606
 607         if (lttng_live_stream->state != LTTNG_LIVE_STREAM_QUIESCENT) {
 608                 return LTTNG_LIVE_ITERATOR_STATUS_OK;
 609         }
 610
 611         if (lttng_live_stream->current_inactivity_ts ==
 612                         lttng_live_stream->last_inactivity_ts) {
 613                 lttng_live_stream->state = LTTNG_LIVE_STREAM_QUIESCENT_NO_DATA;
 614                 ret = LTTNG_LIVE_ITERATOR_STATUS_CONTINUE;
 615                 goto end;
 616         }
 617
 618         ret = emit_inactivity_message(lttng_live_msg_iter, lttng_live_stream,
 619                         message, lttng_live_stream->current_inactivity_ts);
 620
 621         lttng_live_stream->last_inactivity_ts =
 622                         lttng_live_stream->current_inactivity_ts;
 623 end:
 624         return ret;
 625 }
 626
 627 static
 628 int live_get_msg_ts_ns(struct lttng_live_stream_iterator *stream_iter,
 629                 struct lttng_live_msg_iter *lttng_live_msg_iter,
 630                 const bt_message *msg, int64_t last_msg_ts_ns,
 631                 int64_t *ts_ns)
 632 {
 633         const bt_clock_class *clock_class = NULL;
 634         const bt_clock_snapshot *clock_snapshot = NULL;
 635         int ret = 0;
 636         bt_logging_level log_level = lttng_live_msg_iter->log_level;
 637         bt_self_component *self_comp = lttng_live_msg_iter->self_comp;
 638
 639         BT_ASSERT(msg);
 640         BT_ASSERT(ts_ns);
 641
 642         BT_COMP_LOGD("Getting message's timestamp: iter-data-addr=%p, msg-addr=%p, "
 643                 "last-msg-ts=%" PRId64, lttng_live_msg_iter, msg,
 644                 last_msg_ts_ns);
 645
 646         switch (bt_message_get_type(msg)) {
 647         case BT_MESSAGE_TYPE_EVENT:
 648                 clock_class =
 649                         bt_message_event_borrow_stream_class_default_clock_class_const(
 650                                 msg);
 651                 BT_ASSERT(clock_class);
 652
 653                 clock_snapshot = bt_message_event_borrow_default_clock_snapshot_const(
 654                         msg);
 655                 break;
 656         case BT_MESSAGE_TYPE_PACKET_BEGINNING:
 657                 clock_class =
 658                         bt_message_packet_beginning_borrow_stream_class_default_clock_class_const(
 659                         msg);
 660                 BT_ASSERT(clock_class);
 661
 662                 clock_snapshot = bt_message_packet_beginning_borrow_default_clock_snapshot_const(
 663                         msg);
 664                 break;
 665         case BT_MESSAGE_TYPE_PACKET_END:
 666                 clock_class =
 667                         bt_message_packet_end_borrow_stream_class_default_clock_class_const(
 668                         msg);
 669                 BT_ASSERT(clock_class);
 670
 671                 clock_snapshot = bt_message_packet_end_borrow_default_clock_snapshot_const(
 672                         msg);
 673                 break;
 674         case BT_MESSAGE_TYPE_DISCARDED_EVENTS:
 675                 clock_class =
 676                         bt_message_discarded_events_borrow_stream_class_default_clock_class_const(
 677                         msg);
 678                 BT_ASSERT(clock_class);
 679
 680                 clock_snapshot = bt_message_discarded_events_borrow_beginning_default_clock_snapshot_const(
 681                         msg);
 682                 break;
 683         case BT_MESSAGE_TYPE_DISCARDED_PACKETS:
 684                 clock_class =
 685                         bt_message_discarded_packets_borrow_stream_class_default_clock_class_const(
 686                         msg);
 687                 BT_ASSERT(clock_class);
 688
 689                 clock_snapshot = bt_message_discarded_packets_borrow_beginning_default_clock_snapshot_const(
 690                         msg);
 691                 break;
 692         case BT_MESSAGE_TYPE_MESSAGE_ITERATOR_INACTIVITY:
 693                 clock_snapshot =
 694                         bt_message_message_iterator_inactivity_borrow_default_clock_snapshot_const(
 695                                 msg);
 696                 break;
 697         default:
 698                 /* All the other messages have a higher priority */
 699                 BT_COMP_LOGD_STR("Message has no timestamp: using the last message timestamp.");
 700                 *ts_ns = last_msg_ts_ns;
 701                 goto end;
 702         }
 703
 704         clock_class = bt_clock_snapshot_borrow_clock_class_const(clock_snapshot);
 705         BT_ASSERT(clock_class);
 706
 707         ret = bt_clock_snapshot_get_ns_from_origin(clock_snapshot, ts_ns);
 708         if (ret) {
 709                 BT_COMP_LOGE("Cannot get nanoseconds from Epoch of clock snapshot: "
 710                         "clock-snapshot-addr=%p", clock_snapshot);
 711                 goto error;
 712         }
 713
 714         goto end;
 715
 716 error:
 717         ret = -1;
 718
 719 end:
 720         if (ret == 0) {
 721                 BT_COMP_LOGD("Found message's timestamp: "
 722                         "iter-data-addr=%p, msg-addr=%p, "
 723                         "last-msg-ts=%" PRId64 ", ts=%" PRId64,
 724                         lttng_live_msg_iter, msg, last_msg_ts_ns, *ts_ns);
 725         }
 726
 727         return ret;
 728 }
 729
 730 static
 731 enum lttng_live_iterator_status lttng_live_iterator_next_handle_one_active_data_stream(
 732                 struct lttng_live_msg_iter *lttng_live_msg_iter,
 733                 struct lttng_live_stream_iterator *lttng_live_stream,
 734                 bt_message **message)
 735 {
 736         enum lttng_live_iterator_status ret = LTTNG_LIVE_ITERATOR_STATUS_OK;
 737         enum bt_msg_iter_status status;
 738         uint64_t session_idx, trace_idx;
 739         bt_logging_level log_level = lttng_live_msg_iter->log_level;
 740         bt_self_component *self_comp = lttng_live_msg_iter->self_comp;
 741
 742         for (session_idx = 0; session_idx < lttng_live_msg_iter->sessions->len;
 743                         session_idx++) {
 744                 struct lttng_live_session *session =
 745                         g_ptr_array_index(lttng_live_msg_iter->sessions, session_idx);
 746
 747                 if (session->new_streams_needed) {
 748                         ret = LTTNG_LIVE_ITERATOR_STATUS_CONTINUE;
 749                         goto end;
 750                 }
 751                 for (trace_idx = 0; trace_idx < session->traces->len;
 752                                 trace_idx++) {
 753                         struct lttng_live_trace *trace =
 754                                 g_ptr_array_index(session->traces, trace_idx);
 755                         if (trace->new_metadata_needed) {
 756                                 ret = LTTNG_LIVE_ITERATOR_STATUS_CONTINUE;
 757                                 goto end;
 758                         }
 759                 }
 760         }
 761
 762         if (lttng_live_stream->state != LTTNG_LIVE_STREAM_ACTIVE_DATA) {
 763                 ret = LTTNG_LIVE_ITERATOR_STATUS_ERROR;
 764                 goto end;
 765         }
 766
 767         status = bt_msg_iter_get_next_message(lttng_live_stream->msg_iter,
 768                 lttng_live_msg_iter->self_msg_iter, message);
 769         switch (status) {
 770         case BT_MSG_ITER_STATUS_EOF:
 771                 ret = LTTNG_LIVE_ITERATOR_STATUS_END;
 772                 break;
 773         case BT_MSG_ITER_STATUS_OK:
 774                 ret = LTTNG_LIVE_ITERATOR_STATUS_OK;
 775                 break;
 776         case BT_MSG_ITER_STATUS_AGAIN:
 777                 /*
 778                  * Continue immediately (end of packet). The next
 779                  * get_index may return AGAIN to delay the following
 780                  * attempt.
 781                  */
 782                 ret = LTTNG_LIVE_ITERATOR_STATUS_CONTINUE;
 783                 break;
 784         case BT_MSG_ITER_STATUS_INVAL:
 785                 /* No argument provided by the user, so don't return INVAL. */
 786         case BT_MSG_ITER_STATUS_ERROR:
 787         default:
 788                 ret = LTTNG_LIVE_ITERATOR_STATUS_ERROR;
 789                 BT_COMP_LOGW("CTF msg iterator return an error or failed msg_iter=%p",
 790                         lttng_live_stream->msg_iter);
 791                 break;
 792         }
 793
 794 end:
 795         return ret;
 796 }
 797
 798 static
 799 enum lttng_live_iterator_status lttng_live_iterator_close_stream(
 800                 struct lttng_live_msg_iter *lttng_live_msg_iter,
 801                 struct lttng_live_stream_iterator *stream_iter,
 802                 bt_message **curr_msg)
 803 {
 804         enum lttng_live_iterator_status live_status =
 805                 LTTNG_LIVE_ITERATOR_STATUS_OK;
 806         /*
 807          * The viewer has hung up on us so we are closing the stream. The
 808          * `bt_msg_iter` should simply realize that it needs to close the
 809          * stream properly by emitting the necessary stream end message.
 810          */
 811         enum bt_msg_iter_status status =
 812                 bt_msg_iter_get_next_message(stream_iter->msg_iter,
 813                         lttng_live_msg_iter->self_msg_iter, curr_msg);
 814
 815         if (status == BT_MSG_ITER_STATUS_ERROR) {
 816                 live_status = LTTNG_LIVE_ITERATOR_STATUS_ERROR;
 817                 goto end;
 818         }
 819
 820         BT_ASSERT(status == BT_MSG_ITER_STATUS_OK);
 821
 822 end:
 823         return live_status;
 824 }
 825
 826 /*
 827  * helper function:
 828  *            handle_no_data_streams()
 829  *              retry:
 830  *                - for each ACTIVE_NO_DATA stream:
 831  *                  - query relayd for stream data, or quiescence info.
 832  *                    - if need metadata, get metadata, goto retry.
 833  *                    - if new stream, get new stream as ACTIVE_NO_DATA, goto retry
 834  *                  - if quiescent, move to QUIESCENT streams
 835  *                  - if fetched data, move to ACTIVE_DATA streams
 836  *                (at this point each stream either has data, or is quiescent)
 837  *
 838  *
 839  * iterator_next:
 840  *            handle_new_streams_and_metadata()
 841  *                  - query relayd for known streams, add them as ACTIVE_NO_DATA
 842  *                  - query relayd for metadata
 843  *
 844  *            call handle_active_no_data_streams()
 845  *
 846  *            handle_quiescent_streams()
 847  *                - if at least one stream is ACTIVE_DATA:
 848  *                  - peek stream event with lowest timestamp -> next_ts
 849  *                  - for each quiescent stream
 850  *                    - if next_ts >= quiescent end
 851  *                      - set state to ACTIVE_NO_DATA
 852  *                - else
 853  *                  - for each quiescent stream
 854  *                      - set state to ACTIVE_NO_DATA
 855  *
 856  *            call handle_active_no_data_streams()
 857  *
 858  *            handle_active_data_streams()
 859  *                - if at least one stream is ACTIVE_DATA:
 860  *                    - get stream event with lowest timestamp from heap
 861  *                    - make that stream event the current message.
 862  *                    - move this stream heap position to its next event
 863  *                      - if we need to fetch data from relayd, move
 864  *                        stream to ACTIVE_NO_DATA.
 865  *                    - return OK
 866  *                - return AGAIN
 867  *
 868  * end criterion: ctrl-c on client. If relayd exits or the session
 869  * closes on the relay daemon side, we keep on waiting for streams.
 870  * Eventually handle --end timestamp (also an end criterion).
 871  *
 872  * When disconnected from relayd: try to re-connect endlessly.
 873  */
 874 static
 875 enum lttng_live_iterator_status lttng_live_iterator_next_msg_on_stream(
 876                 struct lttng_live_msg_iter *lttng_live_msg_iter,
 877                 struct lttng_live_stream_iterator *stream_iter,
 878                 bt_message **curr_msg)
 879 {
 880         bt_logging_level log_level = lttng_live_msg_iter->log_level;
 881         bt_self_component *self_comp = lttng_live_msg_iter->self_comp;
 882         enum lttng_live_iterator_status live_status;
 883
 884         if (stream_iter->has_stream_hung_up) {
 885                 /*
 886                  * The stream has hung up and the stream was properly closed
 887                  * during the last call to the current function. Return _END
 888                  * status now so that this stream iterator is removed for the
 889                  * stream iterator list.
 890                  */
 891                 live_status = LTTNG_LIVE_ITERATOR_STATUS_END;
 892                 goto end;
 893         }
 894
 895 retry:
 896         print_stream_state(stream_iter);
 897         live_status = lttng_live_iterator_handle_new_streams_and_metadata(
 898                         lttng_live_msg_iter);
 899         if (live_status != LTTNG_LIVE_ITERATOR_STATUS_OK) {
 900                 goto end;
 901         }
 902         live_status = lttng_live_iterator_next_handle_one_no_data_stream(
 903                         lttng_live_msg_iter, stream_iter);
 904
 905         if (live_status != LTTNG_LIVE_ITERATOR_STATUS_OK) {
 906                 if (live_status == LTTNG_LIVE_ITERATOR_STATUS_END) {
 907                         /*
 908                          * We overwrite `live_status` since `curr_msg` is
 909                          * likely set to a valid message in this function.
 910                          */
 911                         live_status = lttng_live_iterator_close_stream(
 912                                 lttng_live_msg_iter, stream_iter, curr_msg);
 913                 }
 914                 goto end;
 915         }
 916         live_status = lttng_live_iterator_next_handle_one_quiescent_stream(
 917                         lttng_live_msg_iter, stream_iter, curr_msg);
 918         if (live_status != LTTNG_LIVE_ITERATOR_STATUS_OK) {
 919                 BT_ASSERT(!*curr_msg);
 920                 goto end;
 921         }
 922         if (*curr_msg) {
 923                 goto end;
 924         }
 925         live_status = lttng_live_iterator_next_handle_one_active_data_stream(
 926                         lttng_live_msg_iter, stream_iter, curr_msg);
 927         if (live_status != LTTNG_LIVE_ITERATOR_STATUS_OK) {
 928                 BT_ASSERT(!*curr_msg);
 929         }
 930
 931 end:
 932         if (live_status == LTTNG_LIVE_ITERATOR_STATUS_CONTINUE) {
 933                 goto retry;
 934         }
 935
 936         return live_status;
 937 }
 938
 939 static
 940 enum lttng_live_iterator_status next_stream_iterator_for_trace(
 941                 struct lttng_live_msg_iter *lttng_live_msg_iter,
 942                 struct lttng_live_trace *live_trace,
 943                 struct lttng_live_stream_iterator **youngest_trace_stream_iter)
 944 {
 945         struct lttng_live_stream_iterator *youngest_candidate_stream_iter = NULL;
 946         enum lttng_live_iterator_status stream_iter_status;;
 947         int64_t youngest_candidate_msg_ts = INT64_MAX;
 948         uint64_t stream_iter_idx;
 949         bt_logging_level log_level = lttng_live_msg_iter->log_level;
 950         bt_self_component *self_comp = lttng_live_msg_iter->self_comp;
 951
 952         BT_ASSERT(live_trace);
 953         BT_ASSERT(live_trace->stream_iterators);
 954         /*
 955          * Update the current message of every stream iterators of this trace.
 956          * The current msg of every stream must have a timestamp equal or
 957          * larger than the last message returned by this iterator. We must
 958          * ensure monotonicity.
 959          */
 960         stream_iter_idx = 0;
 961         while (stream_iter_idx < live_trace->stream_iterators->len) {
 962                 bool stream_iter_is_ended = false;
 963                 struct lttng_live_stream_iterator *stream_iter =
 964                         g_ptr_array_index(live_trace->stream_iterators,
 965                                         stream_iter_idx);
 966
 967                 /*
 968                  * Since we may remove elements from the GPtrArray as we
 969                  * iterate over it, it's possible to see the same element more
 970                  * than once.
 971                  */
 972                 if (stream_iter == youngest_candidate_stream_iter) {
 973                         stream_iter_idx++;
 974                         continue;
 975                 }
 976
 977                 /*
 978                  * Find if there is are now current message for this stream
 979                  * iterator get it.
 980                  */
 981                 while (!stream_iter->current_msg) {
 982                         bt_message *msg = NULL;
 983                         int64_t curr_msg_ts_ns = INT64_MAX;
 984                         stream_iter_status = lttng_live_iterator_next_msg_on_stream(
 985                                         lttng_live_msg_iter, stream_iter, &msg);
 986
 987                         BT_COMP_LOGD("live stream iterator returned status :%s",
 988                                         print_live_iterator_status(stream_iter_status));
 989                         if (stream_iter_status == LTTNG_LIVE_ITERATOR_STATUS_END) {
 990                                 stream_iter_is_ended = true;
 991                                 break;
 992                         }
 993
 994                         if (stream_iter_status != LTTNG_LIVE_ITERATOR_STATUS_OK) {
 995                                 goto end;
 996                         }
 997
 998                         BT_ASSERT(msg);
 999
1000                         /*
1001                          * Get the timestamp in nanoseconds from origin of this
1002                          * messsage.
1003                          */
1004                         live_get_msg_ts_ns(stream_iter, lttng_live_msg_iter,
1005                                 msg, lttng_live_msg_iter->last_msg_ts_ns,
1006                                 &curr_msg_ts_ns);
1007
1008                         /*
1009                          * Check if the message of the current live stream
1010                          * iterator occured at the exact same time or after the
1011                          * last message returned by this component's message
1012                          * iterator. If not, we return an error.
1013                          */
1014                         if (curr_msg_ts_ns >= lttng_live_msg_iter->last_msg_ts_ns) {
1015                                 stream_iter->current_msg = msg;
1016                                 stream_iter->current_msg_ts_ns = curr_msg_ts_ns;
1017                         } else {
1018                                 /*
1019                                  * We received a message in the past. To ensure
1020                                  * monotonicity, we can't send it forward.
1021                                  */
1022                                 BT_COMP_LOGE("Message's timestamp is less than "
1023                                         "lttng-live's message iterator's last "
1024                                         "returned timestamp: "
1025                                         "lttng-live-msg-iter-addr=%p, ts=%" PRId64 ", "
1026                                         "last-msg-ts=%" PRId64,
1027                                         lttng_live_msg_iter, curr_msg_ts_ns,
1028                                         lttng_live_msg_iter->last_msg_ts_ns);
1029                                 stream_iter_status = LTTNG_LIVE_ITERATOR_STATUS_ERROR;
1030                                 goto end;
1031                         }
1032                 }
1033
1034                 BT_ASSERT(stream_iter != youngest_candidate_stream_iter);
1035
1036                 if (!stream_iter_is_ended) {
1037                         if (G_UNLIKELY(youngest_candidate_stream_iter == NULL) ||
1038                                         stream_iter->current_msg_ts_ns < youngest_candidate_msg_ts) {
1039                                 /*
1040                                  * Update the current best candidate message
1041                                  * for the stream iterator of this live trace
1042                                  * to be forwarded downstream.
1043                                  */
1044                                 youngest_candidate_msg_ts = stream_iter->current_msg_ts_ns;
1045                                 youngest_candidate_stream_iter = stream_iter;
1046                         } else if (stream_iter->current_msg_ts_ns == youngest_candidate_msg_ts) {
1047                                 /*
1048                                  * Order the messages in an arbitrary but
1049                                  * deterministic way.
1050                                  */
1051                                 BT_ASSERT(stream_iter != youngest_candidate_stream_iter);
1052                                 int ret = common_muxing_compare_messages(
1053                                         stream_iter->current_msg,
1054                                         youngest_candidate_stream_iter->current_msg);
1055                                 if (ret < 0) {
1056                                         /*
1057                                          * The `youngest_candidate_stream_iter->current_msg`
1058                                          * should go first. Update the next
1059                                          * iterator and the current timestamp.
1060                                          */
1061                                         youngest_candidate_msg_ts = stream_iter->current_msg_ts_ns;
1062                                         youngest_candidate_stream_iter = stream_iter;
1063                                 } else if (ret == 0) {
1064                                         /*
1065                                          * Unable to pick which one should go
1066                                          * first.
1067                                          */
1068                                         BT_COMP_LOGW("Cannot deterministically pick next live stream message iterator because they have identical next messages: "
1069                                                 "stream-iter-addr=%p"
1070                                                 "stream-iter-addr=%p",
1071                                                 stream_iter,
1072                                                 youngest_candidate_stream_iter);
1073                                 }
1074                         }
1075
1076                         stream_iter_idx++;
1077                 } else {
1078                         /*
1079                          * The live stream iterator is ENDed. We remove that
1080                          * iterator from the list and we restart the iteration
1081                          * at the beginning of the live stream iterator array
1082                          * to because the removal will shuffle the array.
1083                          */
1084                         g_ptr_array_remove_index_fast(live_trace->stream_iterators,
1085                                 stream_iter_idx);
1086                         stream_iter_idx = 0;
1087                 }
1088         }
1089
1090         if (youngest_candidate_stream_iter) {
1091                 *youngest_trace_stream_iter = youngest_candidate_stream_iter;
1092                 stream_iter_status = LTTNG_LIVE_ITERATOR_STATUS_OK;
1093         } else {
1094                 /*
1095                  * The only case where we don't have a candidate for this trace
1096                  * is if we reached the end of all the iterators.
1097                  */
1098                 BT_ASSERT(live_trace->stream_iterators->len == 0);
1099                 stream_iter_status = LTTNG_LIVE_ITERATOR_STATUS_END;
1100         }
1101
1102 end:
1103         return stream_iter_status;
1104 }
1105
1106 static
1107 enum lttng_live_iterator_status next_stream_iterator_for_session(
1108                 struct lttng_live_msg_iter *lttng_live_msg_iter,
1109                 struct lttng_live_session *session,
1110                 struct lttng_live_stream_iterator **youngest_session_stream_iter)
1111 {
1112         bt_self_component *self_comp = lttng_live_msg_iter->self_comp;
1113         bt_logging_level log_level = lttng_live_msg_iter->log_level;
1114         enum lttng_live_iterator_status stream_iter_status;
1115         uint64_t trace_idx = 0;
1116         int64_t youngest_candidate_msg_ts = INT64_MAX;
1117         struct lttng_live_stream_iterator *youngest_candidate_stream_iter = NULL;
1118
1119         /*
1120          * Make sure we are attached to the session and look for new streams
1121          * and metadata.
1122          */
1123         stream_iter_status = lttng_live_get_session(lttng_live_msg_iter, session);
1124         if (stream_iter_status != LTTNG_LIVE_ITERATOR_STATUS_OK &&
1125                         stream_iter_status != LTTNG_LIVE_ITERATOR_STATUS_CONTINUE &&
1126                         stream_iter_status != LTTNG_LIVE_ITERATOR_STATUS_END) {
1127                 goto end;
1128         }
1129
1130         BT_ASSERT(session->traces);
1131
1132         /*
1133          * Use while loops here rather then for loops so we can restart the
1134          * iteration if an element is removed from the array during the
1135          * looping.
1136          */
1137         while (trace_idx < session->traces->len) {
1138                 bool trace_is_ended = false;
1139                 struct lttng_live_stream_iterator *stream_iter;
1140                 struct lttng_live_trace *trace =
1141                         g_ptr_array_index(session->traces, trace_idx);
1142
1143                 stream_iter_status = next_stream_iterator_for_trace(
1144                         lttng_live_msg_iter, trace, &stream_iter);
1145                 if (stream_iter_status == LTTNG_LIVE_ITERATOR_STATUS_END) {
1146                         /*
1147                          * All the live stream iterators for this trace are
1148                          * ENDed. Remove the trace from this session.
1149                          */
1150                         trace_is_ended = true;
1151                 } else if (stream_iter_status != LTTNG_LIVE_ITERATOR_STATUS_OK) {
1152                         goto end;
1153                 }
1154
1155                 if (!trace_is_ended) {
1156                         BT_ASSERT(stream_iter);
1157
1158                         if (G_UNLIKELY(youngest_candidate_stream_iter == NULL) ||
1159                                         stream_iter->current_msg_ts_ns < youngest_candidate_msg_ts) {
1160                                 youngest_candidate_msg_ts = stream_iter->current_msg_ts_ns;
1161                                 youngest_candidate_stream_iter = stream_iter;
1162                         } else if (stream_iter->current_msg_ts_ns == youngest_candidate_msg_ts) {
1163                                 /*
1164                                  * Order the messages in an arbitrary but
1165                                  * deterministic way.
1166                                  */
1167                                 int ret = common_muxing_compare_messages(
1168                                         stream_iter->current_msg,
1169                                         youngest_candidate_stream_iter->current_msg);
1170                                 if (ret < 0) {
1171                                         /*
1172                                          * The `youngest_candidate_stream_iter->current_msg`
1173                                          * should go first. Update the next iterator
1174                                          * and the current timestamp.
1175                                          */
1176                                         youngest_candidate_msg_ts = stream_iter->current_msg_ts_ns;
1177                                         youngest_candidate_stream_iter = stream_iter;
1178                                 } else if (ret == 0) {
1179                                         /* Unable to pick which one should go first. */
1180                                         BT_COMP_LOGW("Cannot deterministically pick next live stream message iterator because they have identical next messages: "
1181                                                 "stream-iter-addr=%p" "stream-iter-addr=%p",
1182                                                 stream_iter, youngest_candidate_stream_iter);
1183                                 }
1184                         }
1185                         trace_idx++;
1186                 } else {
1187                         g_ptr_array_remove_index_fast(session->traces, trace_idx);
1188                         trace_idx = 0;
1189                 }
1190         }
1191         if (youngest_candidate_stream_iter) {
1192                 *youngest_session_stream_iter = youngest_candidate_stream_iter;
1193                 stream_iter_status = LTTNG_LIVE_ITERATOR_STATUS_OK;
1194         } else {
1195                 /*
1196                  * The only cases where we don't have a candidate for this
1197                  * trace is:
1198                  *  1. if we reached the end of all the iterators of all the
1199                  *  traces of this session,
1200                  *  2. if we never had live stream iterator in the first place.
1201                  *
1202                  * In either cases, we return END.
1203                  */
1204                 BT_ASSERT(session->traces->len == 0);
1205                 stream_iter_status = LTTNG_LIVE_ITERATOR_STATUS_END;
1206         }
1207 end:
1208         return stream_iter_status;
1209 }
1210
1211 static inline
1212 void put_messages(bt_message_array_const msgs, uint64_t count)
1213 {
1214         uint64_t i;
1215
1216         for (i = 0; i < count; i++) {
1217                 BT_MESSAGE_PUT_REF_AND_RESET(msgs[i]);
1218         }
1219 }
1220
1221 BT_HIDDEN
1222 bt_component_class_message_iterator_next_method_status lttng_live_msg_iter_next(
1223                 bt_self_message_iterator *self_msg_it,
1224                 bt_message_array_const msgs, uint64_t capacity,
1225                 uint64_t *count)
1226 {
1227         bt_component_class_message_iterator_next_method_status status;
1228         struct lttng_live_msg_iter *lttng_live_msg_iter =
1229                 bt_self_message_iterator_get_data(self_msg_it);
1230         struct lttng_live_component *lttng_live =
1231                 lttng_live_msg_iter->lttng_live_comp;
1232         bt_self_component *self_comp = lttng_live_msg_iter->self_comp;
1233         bt_logging_level log_level = lttng_live_msg_iter->log_level;
1234         enum lttng_live_iterator_status stream_iter_status;
1235         uint64_t session_idx;
1236
1237         *count = 0;
1238
1239         BT_ASSERT(lttng_live_msg_iter);
1240
1241         /*
1242          * Clear all the invalid message reference that might be left over in
1243          * the output array.
1244          */
1245         memset(msgs, 0, capacity * sizeof(*msgs));
1246
1247         /*
1248          * If no session are exposed on the relay found at the url provided by
1249          * the user, session count will be 0. In this case, we return status
1250          * end to return gracefully.
1251          */
1252         if (lttng_live_msg_iter->sessions->len == 0) {
1253                 if (lttng_live->params.sess_not_found_act !=
1254                                 SESSION_NOT_FOUND_ACTION_CONTINUE) {
1255                         status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_END;
1256                         goto no_session;
1257                 } else {
1258                         /*
1259                          * The are no more active session for this session
1260                          * name. Retry to create a viewer session for the
1261                          * requested session name.
1262                          */
1263                         if (lttng_live_create_viewer_session(lttng_live_msg_iter)) {
1264                                 status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_ERROR;
1265                                 goto no_session;
1266                         }
1267                 }
1268         }
1269
1270         if (lttng_live_msg_iter->active_stream_iter == 0) {
1271                 lttng_live_force_new_streams_and_metadata(lttng_live_msg_iter);
1272         }
1273
1274         /*
1275          * Here the muxing of message is done.
1276          *
1277          * We need to iterate over all the streams of all the traces of all the
1278          * viewer sessions in order to get the message with the smallest
1279          * timestamp. In this case, a session is a viewer session and there is
1280          * one viewer session per consumer daemon. (UST 32bit, UST 64bit and/or
1281          * kernel). Each viewer session can have multiple traces, for example,
1282          * 64bit UST viewer sessions could have multiple per-pid traces.
1283          *
1284          * We iterate over the streams of each traces to update and see what is
1285          * their next message's timestamp. From those timestamps, we select the
1286          * message with the smallest timestamp as the best candidate message
1287          * for that trace and do the same thing across all the sessions.
1288          *
1289          * We then compare the timestamp of best candidate message of all the
1290          * sessions to pick the message with the smallest timestamp and we
1291          * return it.
1292          */
1293         while (*count < capacity) {
1294                 struct lttng_live_stream_iterator *youngest_stream_iter = NULL,
1295                           *candidate_stream_iter = NULL;
1296                 int64_t youngest_msg_ts_ns = INT64_MAX;
1297
1298                 BT_ASSERT(lttng_live_msg_iter->sessions);
1299                 session_idx = 0;
1300                 /*
1301                  * Use a while loop instead of a for loop so we can restart the
1302                  * iteration if we remove an element. We can safely call
1303                  * next_stream_iterator_for_session() multiple times on the
1304                  * same session as we only fetch a new message if there is no
1305                  * current next message for each live stream iterator.
1306                  * If all live stream iterator of that session already have a
1307                  * current next message, the function will simply exit return
1308                  * the same candidate live stream iterator every time.
1309                  */
1310                 while (session_idx < lttng_live_msg_iter->sessions->len) {
1311                         struct lttng_live_session *session =
1312                                 g_ptr_array_index(lttng_live_msg_iter->sessions,
1313                                         session_idx);
1314
1315                         /* Find the best candidate message to send downstream. */
1316                         stream_iter_status = next_stream_iterator_for_session(
1317                                 lttng_live_msg_iter, session,
1318                                 &candidate_stream_iter);
1319
1320                         /* If we receive an END status, it means that either:
1321                          * - Those traces never had active streams (UST with no
1322                          *   data produced yet),
1323                          * - All live stream iterators have ENDed.*/
1324                         if (stream_iter_status == LTTNG_LIVE_ITERATOR_STATUS_END) {
1325                                 if (session->closed && session->traces->len == 0) {
1326                                         /*
1327                                          * Remove the session from the list and restart the
1328                                          * iteration at the beginning of the array since the
1329                                          * removal shuffle the elements of the array.
1330                                          */
1331                                         g_ptr_array_remove_index_fast(
1332                                                 lttng_live_msg_iter->sessions,
1333                                                 session_idx);
1334                                         session_idx = 0;
1335                                 } else {
1336                                         session_idx++;
1337                                 }
1338                                 continue;
1339                         }
1340
1341                         if (stream_iter_status != LTTNG_LIVE_ITERATOR_STATUS_OK) {
1342                                 goto end;
1343                         }
1344
1345                         if (G_UNLIKELY(youngest_stream_iter == NULL) ||
1346                                         candidate_stream_iter->current_msg_ts_ns <= youngest_msg_ts_ns) {
1347                                 youngest_msg_ts_ns = candidate_stream_iter->current_msg_ts_ns;
1348                                 youngest_stream_iter = candidate_stream_iter;
1349                         } else if (candidate_stream_iter->current_msg_ts_ns == youngest_msg_ts_ns) {
1350                                 /*
1351                                  * The currently selected message to be sent
1352                                  * downstream next has the exact same timestamp
1353                                  * that of the current candidate message. We
1354                                  * must break the tie in a predictable manner.
1355                                  */
1356                                 BT_COMP_LOGD_STR("Two of the next message candidates have the same timestamps, pick one deterministically.");
1357                                 /*
1358                                  * Order the messages in an arbitrary but
1359                                  * deterministic way.
1360                                  */
1361                                 int ret = common_muxing_compare_messages(
1362                                         candidate_stream_iter->current_msg,
1363                                         youngest_stream_iter->current_msg);
1364                                 if (ret < 0) {
1365                                         /*
1366                                          * The `candidate_stream_iter->current_msg`
1367                                          * should go first. Update the next
1368                                          * iterator and the current timestamp.
1369                                          */
1370                                         youngest_msg_ts_ns = candidate_stream_iter->current_msg_ts_ns;
1371                                         youngest_stream_iter = candidate_stream_iter;
1372                                 } else if (ret == 0) {
1373                                         /* Unable to pick which one should go first. */
1374                                         BT_COMP_LOGW("Cannot deterministically pick next live stream message iterator because they have identical next messages: "
1375                                                 "next-stream-iter-addr=%p" "candidate-stream-iter-addr=%p",
1376                                                 youngest_stream_iter, candidate_stream_iter);
1377                                 }
1378                         }
1379
1380                         session_idx++;
1381                 }
1382
1383                 if (!youngest_stream_iter) {
1384                         stream_iter_status = LTTNG_LIVE_ITERATOR_STATUS_AGAIN;
1385                         goto end;
1386                 }
1387
1388                 BT_ASSERT(youngest_stream_iter->current_msg);
1389                 /* Ensure monotonicity. */
1390                 BT_ASSERT(lttng_live_msg_iter->last_msg_ts_ns <=
1391                         youngest_stream_iter->current_msg_ts_ns);
1392
1393                 /*
1394                  * Insert the next message to the message batch. This will set
1395                  * stream iterator current messsage to NULL so that next time
1396                  * we fetch the next message of that stream iterator
1397                  */
1398                 BT_MESSAGE_MOVE_REF(msgs[*count], youngest_stream_iter->current_msg);
1399                 (*count)++;
1400
1401                 /* Update the last timestamp in nanoseconds sent downstream. */
1402                 lttng_live_msg_iter->last_msg_ts_ns = youngest_msg_ts_ns;
1403                 youngest_stream_iter->current_msg_ts_ns = INT64_MAX;
1404
1405                 stream_iter_status = LTTNG_LIVE_ITERATOR_STATUS_OK;
1406         }
1407 end:
1408         switch (stream_iter_status) {
1409         case LTTNG_LIVE_ITERATOR_STATUS_OK:
1410         case LTTNG_LIVE_ITERATOR_STATUS_AGAIN:
1411                 if (*count > 0) {
1412                         /*
1413                          * We received a again status but we have some messages
1414                          * to send downstream. We send them and return OK for
1415                          * now. On the next call we return again if there are
1416                          * still no new message to send.
1417                          */
1418                         status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK;
1419                 } else {
1420                         status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_AGAIN;
1421                 }
1422                 break;
1423         case LTTNG_LIVE_ITERATOR_STATUS_END:
1424                 status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_END;
1425                 break;
1426         case LTTNG_LIVE_ITERATOR_STATUS_NOMEM:
1427                 status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_MEMORY_ERROR;
1428                 break;
1429         case LTTNG_LIVE_ITERATOR_STATUS_ERROR:
1430         case LTTNG_LIVE_ITERATOR_STATUS_INVAL:
1431         case LTTNG_LIVE_ITERATOR_STATUS_UNSUPPORTED:
1432                 status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_ERROR;
1433                 /* Put all existing messages on error. */
1434                 put_messages(msgs, *count);
1435                 break;
1436         default:
1437                 abort();
1438         }
1439
1440 no_session:
1441         return status;
1442 }
1443
1444 BT_HIDDEN
1445 bt_component_class_message_iterator_init_method_status lttng_live_msg_iter_init(
1446                 bt_self_message_iterator *self_msg_it,
1447                 bt_self_component_source *self_comp_src,
1448                 bt_self_component_port_output *self_port)
1449 {
1450         bt_component_class_message_iterator_init_method_status ret =
1451                 BT_COMPONENT_CLASS_MESSAGE_ITERATOR_INIT_METHOD_STATUS_OK;
1452         bt_self_component *self_comp =
1453                 bt_self_component_source_as_self_component(self_comp_src);
1454         struct lttng_live_component *lttng_live;
1455         struct lttng_live_msg_iter *lttng_live_msg_iter;
1456         bt_logging_level log_level;
1457
1458         BT_ASSERT(self_msg_it);
1459
1460         lttng_live = bt_self_component_get_data(self_comp);
1461         log_level = lttng_live->log_level;
1462         self_comp = lttng_live->self_comp;
1463
1464         /* There can be only one downstream iterator at the same time. */
1465         BT_ASSERT(!lttng_live->has_msg_iter);
1466         lttng_live->has_msg_iter = true;
1467
1468         lttng_live_msg_iter = g_new0(struct lttng_live_msg_iter, 1);
1469         if (!lttng_live_msg_iter) {
1470                 ret = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_INIT_METHOD_STATUS_ERROR;
1471                 goto end;
1472         }
1473
1474         lttng_live_msg_iter->log_level = lttng_live->log_level;
1475         lttng_live_msg_iter->self_comp = lttng_live->self_comp;
1476         lttng_live_msg_iter->lttng_live_comp = lttng_live;
1477         lttng_live_msg_iter->self_msg_iter = self_msg_it;
1478
1479         lttng_live_msg_iter->active_stream_iter = 0;
1480         lttng_live_msg_iter->last_msg_ts_ns = INT64_MIN;
1481         lttng_live_msg_iter->sessions = g_ptr_array_new_with_free_func(
1482                 (GDestroyNotify) lttng_live_destroy_session);
1483         BT_ASSERT(lttng_live_msg_iter->sessions);
1484
1485         lttng_live_msg_iter->viewer_connection =
1486                 live_viewer_connection_create(lttng_live->params.url->str, false,
1487                         lttng_live_msg_iter, log_level);
1488         if (!lttng_live_msg_iter->viewer_connection) {
1489                 goto error;
1490         }
1491
1492         if (lttng_live_create_viewer_session(lttng_live_msg_iter)) {
1493                 goto error;
1494         }
1495         if (lttng_live_msg_iter->sessions->len == 0) {
1496                 switch (lttng_live->params.sess_not_found_act) {
1497                 case SESSION_NOT_FOUND_ACTION_CONTINUE:
1498                         BT_COMP_LOGI("Unable to connect to the requested live viewer "
1499                                 "session. Keep trying to connect because of "
1500                                 "%s=\"%s\" component parameter: url=\"%s\"",
1501                                 SESS_NOT_FOUND_ACTION_PARAM,
1502                                 SESS_NOT_FOUND_ACTION_CONTINUE_STR,
1503                                 lttng_live->params.url->str);
1504                         break;
1505                 case SESSION_NOT_FOUND_ACTION_FAIL:
1506                         BT_COMP_LOGE("Unable to connect to the requested live viewer "
1507                                 "session. Fail the message iterator"
1508                                 "initialization because of %s=\"%s\" "
1509                                 "component parameter: url =\"%s\"",
1510                                 SESS_NOT_FOUND_ACTION_PARAM,
1511                                 SESS_NOT_FOUND_ACTION_FAIL_STR,
1512                                 lttng_live->params.url->str);
1513                         goto error;
1514                 case SESSION_NOT_FOUND_ACTION_END:
1515                         BT_COMP_LOGI("Unable to connect to the requested live viewer "
1516                                 "session. End gracefully at the first _next() "
1517                                 "call because of %s=\"%s\" component parameter: "
1518                                 "url=\"%s\"", SESS_NOT_FOUND_ACTION_PARAM,
1519                                 SESS_NOT_FOUND_ACTION_END_STR,
1520                                 lttng_live->params.url->str);
1521                         break;
1522                 case SESSION_NOT_FOUND_ACTION_UNKNOWN:
1523                 default:
1524                         /* Fallthrough */
1525                         BT_COMP_LOGE("Unknown action for session not found"
1526                                 "error. Fail the message iterator"
1527                                 "initialization because of %s=\"%s\" "
1528                                 "component parameter: url =\"%s\"",
1529                                 SESS_NOT_FOUND_ACTION_PARAM,
1530                                 SESS_NOT_FOUND_ACTION_FAIL_STR,
1531                                 lttng_live->params.url->str);
1532                         break;
1533                 }
1534         }
1535
1536         bt_self_message_iterator_set_data(self_msg_it, lttng_live_msg_iter);
1537
1538         goto end;
1539 error:
1540         ret = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_INIT_METHOD_STATUS_ERROR;
1541         lttng_live_msg_iter_destroy(lttng_live_msg_iter);
1542 end:
1543         return ret;
1544 }
1545
1546 static
1547 bt_component_class_query_method_status lttng_live_query_list_sessions(
1548                 const bt_value *params, const bt_value **result,
1549                 bt_logging_level log_level)
1550 {
1551         bt_component_class_query_method_status status =
1552                 BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_OK;
1553         const bt_value *url_value = NULL;
1554         const char *url;
1555         struct live_viewer_connection *viewer_connection = NULL;
1556         bt_self_component *self_comp = NULL;
1557
1558         url_value = bt_value_map_borrow_entry_value_const(params, URL_PARAM);
1559         if (!url_value) {
1560                 BT_COMP_LOGE("Mandatory `%s` parameter missing", URL_PARAM);
1561                 status = BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_ERROR;
1562                 goto error;
1563         }
1564
1565         if (!bt_value_is_string(url_value)) {
1566                 BT_COMP_LOGE("`%s` parameter is required to be a string value",
1567                         URL_PARAM);
1568                 status = BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_ERROR;
1569                 goto error;
1570         }
1571
1572         url = bt_value_string_get(url_value);
1573
1574         viewer_connection = live_viewer_connection_create(url, true, NULL,
1575                 log_level);
1576         if (!viewer_connection) {
1577                 goto error;
1578         }
1579
1580         status = live_viewer_connection_list_sessions(viewer_connection,
1581                         result);
1582         if (status != BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_OK) {
1583                 goto error;
1584         }
1585
1586         goto end;
1587
1588 error:
1589         BT_VALUE_PUT_REF_AND_RESET(*result);
1590
1591         if (status >= 0) {
1592                 status = BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_ERROR;
1593         }
1594
1595 end:
1596         if (viewer_connection) {
1597                 live_viewer_connection_destroy(viewer_connection);
1598         }
1599         return status;
1600 }
1601
1602 BT_HIDDEN
1603 bt_component_class_query_method_status lttng_live_query(
1604                 bt_self_component_class_source *comp_class,
1605                 bt_private_query_executor *priv_query_exec,
1606                 const char *object, const bt_value *params,
1607                 __attribute__((unused)) void *method_data,
1608                 const bt_value **result)
1609 {
1610         bt_component_class_query_method_status status =
1611                 BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_OK;
1612         bt_self_component *self_comp = NULL;
1613         bt_logging_level log_level = bt_query_executor_get_logging_level(
1614                 bt_private_query_executor_as_query_executor_const(
1615                         priv_query_exec));
1616
1617         if (strcmp(object, "sessions") == 0) {
1618                 status = lttng_live_query_list_sessions(params, result,
1619                         log_level);
1620         } else {
1621                 BT_COMP_LOGI("Unknown query object `%s`", object);
1622                 status = BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_UNKNOWN_OBJECT;
1623                 goto end;
1624         }
1625
1626 end:
1627         return status;
1628 }
1629
1630 static
1631 void lttng_live_component_destroy_data(struct lttng_live_component *lttng_live)
1632 {
1633         if (!lttng_live) {
1634                 return;
1635         }
1636         if (lttng_live->params.url) {
1637                 g_string_free(lttng_live->params.url, TRUE);
1638         }
1639         g_free(lttng_live);
1640 }
1641
1642 BT_HIDDEN
1643 void lttng_live_component_finalize(bt_self_component_source *component)
1644 {
1645         void *data = bt_self_component_get_data(
1646                         bt_self_component_source_as_self_component(component));
1647
1648         if (!data) {
1649                 return;
1650         }
1651         lttng_live_component_destroy_data(data);
1652 }
1653
1654 static
1655 enum session_not_found_action parse_session_not_found_action_param(
1656         const bt_value *no_session_param)
1657 {
1658         enum session_not_found_action action;
1659         const char *no_session_act_str;
1660         no_session_act_str = bt_value_string_get(no_session_param);
1661         if (strcmp(no_session_act_str, SESS_NOT_FOUND_ACTION_CONTINUE_STR) == 0) {
1662                 action = SESSION_NOT_FOUND_ACTION_CONTINUE;
1663         } else if (strcmp(no_session_act_str, SESS_NOT_FOUND_ACTION_FAIL_STR) == 0) {
1664                 action = SESSION_NOT_FOUND_ACTION_FAIL;
1665         } else if (strcmp(no_session_act_str, SESS_NOT_FOUND_ACTION_END_STR) == 0) {
1666                 action = SESSION_NOT_FOUND_ACTION_END;
1667         } else {
1668                 action = SESSION_NOT_FOUND_ACTION_UNKNOWN;
1669         }
1670
1671         return action;
1672 }
1673
1674 struct lttng_live_component *lttng_live_component_create(const bt_value *params,
1675                 bt_logging_level log_level, bt_self_component *self_comp)
1676 {
1677         struct lttng_live_component *lttng_live;
1678         const bt_value *value = NULL;
1679         const char *url;
1680
1681         lttng_live = g_new0(struct lttng_live_component, 1);
1682         if (!lttng_live) {
1683                 goto end;
1684         }
1685         lttng_live->log_level = log_level;
1686         lttng_live->self_comp = self_comp;
1687         lttng_live->max_query_size = MAX_QUERY_SIZE;
1688         lttng_live->has_msg_iter = false;
1689
1690         value = bt_value_map_borrow_entry_value_const(params, URL_PARAM);
1691         if (!value || !bt_value_is_string(value)) {
1692                 BT_COMP_LOGW("Mandatory `%s` parameter missing or not a string",
1693                         URL_PARAM);
1694                 goto error;
1695         }
1696         url = bt_value_string_get(value);
1697         lttng_live->params.url = g_string_new(url);
1698         if (!lttng_live->params.url) {
1699                 goto error;
1700         }
1701
1702         value = bt_value_map_borrow_entry_value_const(params,
1703                 SESS_NOT_FOUND_ACTION_PARAM);
1704
1705         if (value && bt_value_is_string(value)) {
1706                 lttng_live->params.sess_not_found_act =
1707                         parse_session_not_found_action_param(value);
1708                 if (lttng_live->params.sess_not_found_act == SESSION_NOT_FOUND_ACTION_UNKNOWN) {
1709                         BT_COMP_LOGE("Unexpected value for `%s` parameter: "
1710                                 "value=\"%s\"", SESS_NOT_FOUND_ACTION_PARAM,
1711                                 bt_value_string_get(value));
1712                         goto error;
1713                 }
1714         } else {
1715                 BT_COMP_LOGW("Optional `%s` parameter is missing or "
1716                         "not a string value. Defaulting to %s=\"%s\".",
1717                         SESS_NOT_FOUND_ACTION_PARAM,
1718                         SESS_NOT_FOUND_ACTION_PARAM,
1719                         SESS_NOT_FOUND_ACTION_CONTINUE_STR);
1720                 lttng_live->params.sess_not_found_act =
1721                         SESSION_NOT_FOUND_ACTION_CONTINUE;
1722         }
1723
1724         goto end;
1725
1726 error:
1727         lttng_live_component_destroy_data(lttng_live);
1728         lttng_live = NULL;
1729 end:
1730         return lttng_live;
1731 }
1732
1733 BT_HIDDEN
1734 bt_component_class_init_method_status lttng_live_component_init(
1735                 bt_self_component_source *self_comp_src,
1736                 const bt_value *params, __attribute__((unused)) void *init_method_data)
1737 {
1738         struct lttng_live_component *lttng_live;
1739         bt_component_class_init_method_status ret =
1740                 BT_COMPONENT_CLASS_INIT_METHOD_STATUS_OK;
1741         bt_self_component *self_comp =
1742                 bt_self_component_source_as_self_component(self_comp_src);
1743         bt_logging_level log_level = bt_component_get_logging_level(
1744                 bt_self_component_as_component(self_comp));
1745         bt_self_component_add_port_status add_port_status;
1746
1747         lttng_live = lttng_live_component_create(params, log_level, self_comp);
1748         if (!lttng_live) {
1749                 ret = BT_COMPONENT_CLASS_INIT_METHOD_STATUS_MEMORY_ERROR;
1750                 goto error;
1751         }
1752
1753         add_port_status = bt_self_component_source_add_output_port(
1754                 self_comp_src, "out", NULL, NULL);
1755         switch (add_port_status) {
1756         case BT_SELF_COMPONENT_ADD_PORT_STATUS_ERROR:
1757                 ret = BT_COMPONENT_CLASS_INIT_METHOD_STATUS_ERROR;
1758                 goto error;
1759         case BT_SELF_COMPONENT_ADD_PORT_STATUS_MEMORY_ERROR:
1760                 ret = BT_COMPONENT_CLASS_INIT_METHOD_STATUS_MEMORY_ERROR;
1761                 goto error;
1762         default:
1763                 break;
1764         }
1765
1766         bt_self_component_set_data(self_comp, lttng_live);
1767         goto end;
1768
1769 error:
1770         lttng_live_component_destroy_data(lttng_live);
1771         lttng_live = NULL;
1772 end:
1773         return ret;
1774 }