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_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 **candidate_stream_iter)
 944 {
 945         struct lttng_live_stream_iterator *curr_candidate_stream_iter = NULL;
 946         enum lttng_live_iterator_status stream_iter_status;;
 947         int64_t curr_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                  * Find if there is are now current message for this stream
 969                  * iterator get it.
 970                  */
 971                 while (!stream_iter->current_msg) {
 972                         bt_message *msg = NULL;
 973                         int64_t curr_msg_ts_ns = INT64_MAX;
 974                         stream_iter_status = lttng_live_iterator_next_on_stream(
 975                                         lttng_live_msg_iter, stream_iter, &msg);
 976
 977                         BT_COMP_LOGD("live stream iterator returned status :%s",
 978                                         print_live_iterator_status(stream_iter_status));
 979                         if (stream_iter_status == LTTNG_LIVE_ITERATOR_STATUS_END) {
 980                                 stream_iter_is_ended = true;
 981                                 break;
 982                         }
 983
 984                         if (stream_iter_status != LTTNG_LIVE_ITERATOR_STATUS_OK) {
 985                                 goto end;
 986                         }
 987
 988                         BT_ASSERT(msg);
 989
 990                         /*
 991                          * Get the timestamp in nanoseconds from origin of this
 992                          * messsage.
 993                          */
 994                         live_get_msg_ts_ns(stream_iter, lttng_live_msg_iter,
 995                                 msg, lttng_live_msg_iter->last_msg_ts_ns,
 996                                 &curr_msg_ts_ns);
 997
 998                         /*
 999                          * Check if the message of the current live stream
1000                          * iterator occured at the exact same time or after the
1001                          * last message returned by this component's message
1002                          * iterator. If not, we return an error.
1003                          */
1004                         if (curr_msg_ts_ns >= lttng_live_msg_iter->last_msg_ts_ns) {
1005                                 stream_iter->current_msg = msg;
1006                                 stream_iter->current_msg_ts_ns = curr_msg_ts_ns;
1007                         } else {
1008                                 /*
1009                                  * We received a message in the past. To ensure
1010                                  * monotonicity, we can't send it forward.
1011                                  */
1012                                 BT_COMP_LOGE("Message's timestamp is less than "
1013                                         "lttng-live's message iterator's last "
1014                                         "returned timestamp: "
1015                                         "lttng-live-msg-iter-addr=%p, ts=%" PRId64 ", "
1016                                         "last-msg-ts=%" PRId64,
1017                                         lttng_live_msg_iter, curr_msg_ts_ns,
1018                                         lttng_live_msg_iter->last_msg_ts_ns);
1019                                 stream_iter_status = LTTNG_LIVE_ITERATOR_STATUS_ERROR;
1020                                 goto end;
1021                         }
1022                 }
1023
1024                 if (!stream_iter_is_ended &&
1025                                 stream_iter->current_msg_ts_ns <= curr_candidate_msg_ts) {
1026                         /*
1027                          * Update the current best candidate message for the
1028                          * stream iterator of thise live trace to be forwarded
1029                          * downstream.
1030                          */
1031                         curr_candidate_msg_ts = stream_iter->current_msg_ts_ns;
1032                         curr_candidate_stream_iter = stream_iter;
1033                 }
1034
1035                 if (stream_iter_is_ended) {
1036                         /*
1037                          * The live stream iterator is ENDed. We remove that
1038                          * iterator from the list and we restart the iteration
1039                          * at the beginning of the live stream iterator array
1040                          * to because the removal will shuffle the array.
1041                          */
1042                         g_ptr_array_remove_index_fast(live_trace->stream_iterators,
1043                                 stream_iter_idx);
1044                         stream_iter_idx = 0;
1045                 } else {
1046                         stream_iter_idx++;
1047                 }
1048         }
1049
1050         if (curr_candidate_stream_iter) {
1051                 *candidate_stream_iter = curr_candidate_stream_iter;
1052                 stream_iter_status = LTTNG_LIVE_ITERATOR_STATUS_OK;
1053         } else {
1054                 /*
1055                  * The only case where we don't have a candidate for this trace
1056                  * is if we reached the end of all the iterators.
1057                  */
1058                 BT_ASSERT(live_trace->stream_iterators->len == 0);
1059                 stream_iter_status = LTTNG_LIVE_ITERATOR_STATUS_END;
1060         }
1061
1062 end:
1063         return stream_iter_status;
1064 }
1065
1066 static
1067 enum lttng_live_iterator_status next_stream_iterator_for_session(
1068                 struct lttng_live_msg_iter *lttng_live_msg_iter,
1069                 struct lttng_live_session *session,
1070                 struct lttng_live_stream_iterator **candidate_session_stream_iter)
1071 {
1072         enum lttng_live_iterator_status stream_iter_status;
1073         uint64_t trace_idx = 0;
1074         int64_t curr_candidate_msg_ts = INT64_MAX;
1075         struct lttng_live_stream_iterator *curr_candidate_stream_iter = NULL;
1076
1077         /*
1078          * Make sure we are attached to the session and look for new streams
1079          * and metadata.
1080          */
1081         stream_iter_status = lttng_live_get_session(lttng_live_msg_iter, session);
1082         if (stream_iter_status != LTTNG_LIVE_ITERATOR_STATUS_OK &&
1083                         stream_iter_status != LTTNG_LIVE_ITERATOR_STATUS_CONTINUE &&
1084                         stream_iter_status != LTTNG_LIVE_ITERATOR_STATUS_END) {
1085                 goto end;
1086         }
1087
1088         BT_ASSERT(session->traces);
1089
1090         /*
1091          * Use while loops here rather then for loops so we can restart the
1092          * iteration if an element is removed from the array during the
1093          * looping.
1094          */
1095         while (trace_idx < session->traces->len) {
1096                 bool trace_is_ended = false;
1097                 struct lttng_live_stream_iterator *stream_iter;
1098                 struct lttng_live_trace *trace =
1099                         g_ptr_array_index(session->traces, trace_idx);
1100
1101                 stream_iter_status = next_stream_iterator_for_trace(
1102                         lttng_live_msg_iter, trace, &stream_iter);
1103                 if (stream_iter_status == LTTNG_LIVE_ITERATOR_STATUS_END) {
1104                         /*
1105                          * All the live stream iterators for this trace are
1106                          * ENDed. Remove the trace from this session.
1107                          */
1108                         trace_is_ended = true;
1109                 } else if (stream_iter_status != LTTNG_LIVE_ITERATOR_STATUS_OK) {
1110                         goto end;
1111                 }
1112
1113                 if (!trace_is_ended) {
1114                         BT_ASSERT(stream_iter);
1115
1116                         if (stream_iter->current_msg_ts_ns <= curr_candidate_msg_ts) {
1117                                 curr_candidate_msg_ts = stream_iter->current_msg_ts_ns;
1118                                 curr_candidate_stream_iter = stream_iter;
1119                         }
1120                         trace_idx++;
1121                 } else {
1122                         g_ptr_array_remove_index_fast(session->traces, trace_idx);
1123                         trace_idx = 0;
1124                 }
1125         }
1126         if (curr_candidate_stream_iter) {
1127                 *candidate_session_stream_iter = curr_candidate_stream_iter;
1128                 stream_iter_status = LTTNG_LIVE_ITERATOR_STATUS_OK;
1129         } else {
1130                 /*
1131                  * The only cases where we don't have a candidate for this
1132                  * trace is:
1133                  *  1. if we reached the end of all the iterators of all the
1134                  *  traces of this session,
1135                  *  2. if we never had live stream iterator in the first place.
1136                  *
1137                  * In either cases, we return END.
1138                  */
1139                 BT_ASSERT(session->traces->len == 0);
1140                 stream_iter_status = LTTNG_LIVE_ITERATOR_STATUS_END;
1141         }
1142 end:
1143         return stream_iter_status;
1144 }
1145
1146 static inline
1147 void put_messages(bt_message_array_const msgs, uint64_t count)
1148 {
1149         uint64_t i;
1150
1151         for (i = 0; i < count; i++) {
1152                 BT_MESSAGE_PUT_REF_AND_RESET(msgs[i]);
1153         }
1154 }
1155
1156 BT_HIDDEN
1157 bt_component_class_message_iterator_next_method_status lttng_live_msg_iter_next(
1158                 bt_self_message_iterator *self_msg_it,
1159                 bt_message_array_const msgs, uint64_t capacity,
1160                 uint64_t *count)
1161 {
1162         bt_component_class_message_iterator_next_method_status status;
1163         struct lttng_live_msg_iter *lttng_live_msg_iter =
1164                 bt_self_message_iterator_get_data(self_msg_it);
1165         struct lttng_live_component *lttng_live =
1166                 lttng_live_msg_iter->lttng_live_comp;
1167         bt_self_component *self_comp = lttng_live_msg_iter->self_comp;
1168         bt_logging_level log_level = lttng_live_msg_iter->log_level;
1169         enum lttng_live_iterator_status stream_iter_status;
1170         uint64_t session_idx;
1171
1172         *count = 0;
1173
1174         BT_ASSERT(lttng_live_msg_iter);
1175
1176         /*
1177          * Clear all the invalid message reference that might be left over in
1178          * the output array.
1179          */
1180         memset(msgs, 0, capacity * sizeof(*msgs));
1181
1182         /*
1183          * If no session are exposed on the relay found at the url provided by
1184          * the user, session count will be 0. In this case, we return status
1185          * end to return gracefully.
1186          */
1187         if (lttng_live_msg_iter->sessions->len == 0) {
1188                 if (lttng_live->params.sess_not_found_act !=
1189                                 SESSION_NOT_FOUND_ACTION_CONTINUE) {
1190                         status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_END;
1191                         goto no_session;
1192                 } else {
1193                         /*
1194                          * The are no more active session for this session
1195                          * name. Retry to create a viewer session for the
1196                          * requested session name.
1197                          */
1198                         if (lttng_live_create_viewer_session(lttng_live_msg_iter)) {
1199                                 status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_ERROR;
1200                                 goto no_session;
1201                         }
1202                 }
1203         }
1204
1205         if (lttng_live_msg_iter->active_stream_iter == 0) {
1206                 lttng_live_force_new_streams_and_metadata(lttng_live_msg_iter);
1207         }
1208
1209         /*
1210          * Here the muxing of message is done.
1211          *
1212          * We need to iterate over all the streams of all the traces of all the
1213          * viewer sessions in order to get the message with the smallest
1214          * timestamp. In this case, a session is a viewer session and there is
1215          * one viewer session per consumer daemon. (UST 32bit, UST 64bit and/or
1216          * kernel). Each viewer session can have multiple traces, for example,
1217          * 64bit UST viewer sessions could have multiple per-pid traces.
1218          *
1219          * We iterate over the streams of each traces to update and see what is
1220          * their next message's timestamp. From those timestamps, we select the
1221          * message with the smallest timestamp as the best candidate message
1222          * for that trace and do the same thing across all the sessions.
1223          *
1224          * We then compare the timestamp of best candidate message of all the
1225          * sessions to pick the message with the smallest timestamp and we
1226          * return it.
1227          */
1228         while (*count < capacity) {
1229                 struct lttng_live_stream_iterator *next_stream_iter = NULL,
1230                                                   *candidate_stream_iter = NULL;
1231                 int64_t next_msg_ts_ns = INT64_MAX;
1232
1233                 BT_ASSERT(lttng_live_msg_iter->sessions);
1234                 session_idx = 0;
1235                 /*
1236                  * Use a while loop instead of a for loop so we can restart the
1237                  * iteration if we remove an element. We can safely call
1238                  * next_stream_iterator_for_session() multiple times on the
1239                  * same session as we only fetch a new message if there is no
1240                  * current next message for each live stream iterator.
1241                  * If all live stream iterator of that session already have a
1242                  * current next message, the function will simply exit return
1243                  * the same candidate live stream iterator every time.
1244                  */
1245                 while (session_idx < lttng_live_msg_iter->sessions->len) {
1246                         struct lttng_live_session *session =
1247                                 g_ptr_array_index(lttng_live_msg_iter->sessions,
1248                                         session_idx);
1249
1250                         /* Find the best candidate message to send downstream. */
1251                         stream_iter_status = next_stream_iterator_for_session(
1252                                 lttng_live_msg_iter, session,
1253                                 &candidate_stream_iter);
1254
1255                         /* If we receive an END status, it means that either:
1256                          * - Those traces never had active streams (UST with no
1257                          *   data produced yet),
1258                          * - All live stream iterators have ENDed.*/
1259                         if (stream_iter_status == LTTNG_LIVE_ITERATOR_STATUS_END) {
1260                                 if (session->closed && session->traces->len == 0) {
1261                                         /*
1262                                          * Remove the session from the list and restart the
1263                                          * iteration at the beginning of the array since the
1264                                          * removal shuffle the elements of the array.
1265                                          */
1266                                         g_ptr_array_remove_index_fast(
1267                                                 lttng_live_msg_iter->sessions,
1268                                                 session_idx);
1269                                         session_idx = 0;
1270                                 } else {
1271                                         session_idx++;
1272                                 }
1273                                 continue;
1274                         }
1275
1276                         if (stream_iter_status != LTTNG_LIVE_ITERATOR_STATUS_OK) {
1277                                 goto end;
1278                         }
1279
1280                         if (G_UNLIKELY(next_stream_iter == NULL) ||
1281                                         candidate_stream_iter->current_msg_ts_ns <= next_msg_ts_ns) {
1282                                 next_msg_ts_ns = candidate_stream_iter->current_msg_ts_ns;
1283                                 next_stream_iter = candidate_stream_iter;
1284                         } else if (candidate_stream_iter->current_msg_ts_ns == next_msg_ts_ns) {
1285                                 /*
1286                                  * The currently selected message to be sent
1287                                  * downstream next has the exact same timestamp
1288                                  * that of the current candidate message. We
1289                                  * must break the tie in a predictable manner.
1290                                  */
1291                                 BT_COMP_LOGD_STR("Two of the next message candidates have the same timestamps, pick one deterministically.");
1292                                 /*
1293                                  * Order the messages in an arbitrary but
1294                                  * determinitic way.
1295                                  */
1296                                 int ret = common_muxing_compare_messages(candidate_stream_iter->current_msg,
1297                                         next_stream_iter->current_msg);
1298                                 if (ret < 0) {
1299                                         /*
1300                                          * The `candidate_stream_iter->current_msg`
1301                                          * should go first. Update the next
1302                                          * iterator and the current timestamp.
1303                                          */
1304                                         next_msg_ts_ns = candidate_stream_iter->current_msg_ts_ns;
1305                                         next_stream_iter = candidate_stream_iter;
1306                                 } else if (ret == 0) {
1307                                         /* Unable to pick which one should go first. */
1308                                         BT_COMP_LOGW("Cannot deterministically pick next live stream message iterator because they have identical next messages: "
1309                                                 "next-stream-iter-addr=%p" "candidate-stream-iter-addr=%p",
1310                                                 next_stream_iter, candidate_stream_iter);
1311                                 }
1312                         }
1313
1314                         session_idx++;
1315                 }
1316
1317                 if (!next_stream_iter) {
1318                         stream_iter_status = LTTNG_LIVE_ITERATOR_STATUS_AGAIN;
1319                         goto end;
1320                 }
1321
1322                 BT_ASSERT(next_stream_iter->current_msg);
1323                 /* Ensure monotonicity. */
1324                 BT_ASSERT(lttng_live_msg_iter->last_msg_ts_ns <=
1325                         next_stream_iter->current_msg_ts_ns);
1326
1327                 /*
1328                  * Insert the next message to the message batch. This will set
1329                  * stream iterator current messsage to NULL so that next time
1330                  * we fetch the next message of that stream iterator
1331                  */
1332                 BT_MESSAGE_MOVE_REF(msgs[*count], next_stream_iter->current_msg);
1333                 (*count)++;
1334
1335                 /* Update the last timestamp in nanoseconds sent downstream. */
1336                 lttng_live_msg_iter->last_msg_ts_ns = next_msg_ts_ns;
1337                 next_stream_iter->current_msg_ts_ns = INT64_MAX;
1338
1339                 stream_iter_status = LTTNG_LIVE_ITERATOR_STATUS_OK;
1340         }
1341 end:
1342         switch (stream_iter_status) {
1343         case LTTNG_LIVE_ITERATOR_STATUS_OK:
1344         case LTTNG_LIVE_ITERATOR_STATUS_AGAIN:
1345                 if (*count > 0) {
1346                         /*
1347                          * We received a again status but we have some messages
1348                          * to send downstream. We send them and return OK for
1349                          * now. On the next call we return again if there are
1350                          * still no new message to send.
1351                          */
1352                         status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK;
1353                 } else {
1354                         status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_AGAIN;
1355                 }
1356                 break;
1357         case LTTNG_LIVE_ITERATOR_STATUS_END:
1358                 status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_END;
1359                 break;
1360         case LTTNG_LIVE_ITERATOR_STATUS_NOMEM:
1361                 status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_MEMORY_ERROR;
1362                 break;
1363         case LTTNG_LIVE_ITERATOR_STATUS_ERROR:
1364         case LTTNG_LIVE_ITERATOR_STATUS_INVAL:
1365         case LTTNG_LIVE_ITERATOR_STATUS_UNSUPPORTED:
1366                 status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_ERROR;
1367                 /* Put all existing messages on error. */
1368                 put_messages(msgs, *count);
1369                 break;
1370         default:
1371                 abort();
1372         }
1373
1374 no_session:
1375         return status;
1376 }
1377
1378 BT_HIDDEN
1379 bt_component_class_message_iterator_init_method_status lttng_live_msg_iter_init(
1380                 bt_self_message_iterator *self_msg_it,
1381                 bt_self_component_source *self_comp_src,
1382                 bt_self_component_port_output *self_port)
1383 {
1384         bt_component_class_message_iterator_init_method_status ret =
1385                 BT_COMPONENT_CLASS_MESSAGE_ITERATOR_INIT_METHOD_STATUS_OK;
1386         bt_self_component *self_comp =
1387                 bt_self_component_source_as_self_component(self_comp_src);
1388         struct lttng_live_component *lttng_live;
1389         struct lttng_live_msg_iter *lttng_live_msg_iter;
1390         bt_logging_level log_level;
1391
1392         BT_ASSERT(self_msg_it);
1393
1394         lttng_live = bt_self_component_get_data(self_comp);
1395         log_level = lttng_live->log_level;
1396         self_comp = lttng_live->self_comp;
1397
1398         /* There can be only one downstream iterator at the same time. */
1399         BT_ASSERT(!lttng_live->has_msg_iter);
1400         lttng_live->has_msg_iter = true;
1401
1402         lttng_live_msg_iter = g_new0(struct lttng_live_msg_iter, 1);
1403         if (!lttng_live_msg_iter) {
1404                 ret = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_INIT_METHOD_STATUS_ERROR;
1405                 goto end;
1406         }
1407
1408         lttng_live_msg_iter->log_level = lttng_live->log_level;
1409         lttng_live_msg_iter->self_comp = lttng_live->self_comp;
1410         lttng_live_msg_iter->lttng_live_comp = lttng_live;
1411         lttng_live_msg_iter->self_msg_iter = self_msg_it;
1412
1413         lttng_live_msg_iter->active_stream_iter = 0;
1414         lttng_live_msg_iter->last_msg_ts_ns = INT64_MIN;
1415         lttng_live_msg_iter->sessions = g_ptr_array_new_with_free_func(
1416                 (GDestroyNotify) lttng_live_destroy_session);
1417         BT_ASSERT(lttng_live_msg_iter->sessions);
1418
1419         lttng_live_msg_iter->viewer_connection =
1420                 live_viewer_connection_create(lttng_live->params.url->str, false,
1421                         lttng_live_msg_iter, log_level);
1422         if (!lttng_live_msg_iter->viewer_connection) {
1423                 goto error;
1424         }
1425
1426         if (lttng_live_create_viewer_session(lttng_live_msg_iter)) {
1427                 goto error;
1428         }
1429         if (lttng_live_msg_iter->sessions->len == 0) {
1430                 switch (lttng_live->params.sess_not_found_act) {
1431                 case SESSION_NOT_FOUND_ACTION_CONTINUE:
1432                         BT_COMP_LOGI("Unable to connect to the requested live viewer "
1433                                 "session. Keep trying to connect because of "
1434                                 "%s=\"%s\" component parameter: url=\"%s\"",
1435                                 SESS_NOT_FOUND_ACTION_PARAM,
1436                                 SESS_NOT_FOUND_ACTION_CONTINUE_STR,
1437                                 lttng_live->params.url->str);
1438                         break;
1439                 case SESSION_NOT_FOUND_ACTION_FAIL:
1440                         BT_COMP_LOGE("Unable to connect to the requested live viewer "
1441                                 "session. Fail the message iterator"
1442                                 "initialization because of %s=\"%s\" "
1443                                 "component parameter: url =\"%s\"",
1444                                 SESS_NOT_FOUND_ACTION_PARAM,
1445                                 SESS_NOT_FOUND_ACTION_FAIL_STR,
1446                                 lttng_live->params.url->str);
1447                         goto error;
1448                 case SESSION_NOT_FOUND_ACTION_END:
1449                         BT_COMP_LOGI("Unable to connect to the requested live viewer "
1450                                 "session. End gracefully at the first _next() "
1451                                 "call because of %s=\"%s\" component parameter: "
1452                                 "url=\"%s\"", SESS_NOT_FOUND_ACTION_PARAM,
1453                                 SESS_NOT_FOUND_ACTION_END_STR,
1454                                 lttng_live->params.url->str);
1455                         break;
1456                 case SESSION_NOT_FOUND_ACTION_UNKNOWN:
1457                 default:
1458                         /* Fallthrough */
1459                         BT_COMP_LOGE("Unknown action for session not found"
1460                                 "error. Fail the message iterator"
1461                                 "initialization because of %s=\"%s\" "
1462                                 "component parameter: url =\"%s\"",
1463                                 SESS_NOT_FOUND_ACTION_PARAM,
1464                                 SESS_NOT_FOUND_ACTION_FAIL_STR,
1465                                 lttng_live->params.url->str);
1466                         break;
1467                 }
1468         }
1469
1470         bt_self_message_iterator_set_data(self_msg_it, lttng_live_msg_iter);
1471
1472         goto end;
1473 error:
1474         ret = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_INIT_METHOD_STATUS_ERROR;
1475         lttng_live_msg_iter_destroy(lttng_live_msg_iter);
1476 end:
1477         return ret;
1478 }
1479
1480 static
1481 bt_component_class_query_method_status lttng_live_query_list_sessions(
1482                 const bt_value *params, const bt_value **result,
1483                 bt_logging_level log_level)
1484 {
1485         bt_component_class_query_method_status status =
1486                 BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_OK;
1487         const bt_value *url_value = NULL;
1488         const char *url;
1489         struct live_viewer_connection *viewer_connection = NULL;
1490         bt_self_component *self_comp = NULL;
1491
1492         url_value = bt_value_map_borrow_entry_value_const(params, URL_PARAM);
1493         if (!url_value) {
1494                 BT_COMP_LOGE("Mandatory `%s` parameter missing", URL_PARAM);
1495                 status = BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_ERROR;
1496                 goto error;
1497         }
1498
1499         if (!bt_value_is_string(url_value)) {
1500                 BT_COMP_LOGE("`%s` parameter is required to be a string value",
1501                         URL_PARAM);
1502                 status = BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_ERROR;
1503                 goto error;
1504         }
1505
1506         url = bt_value_string_get(url_value);
1507
1508         viewer_connection = live_viewer_connection_create(url, true, NULL,
1509                 log_level);
1510         if (!viewer_connection) {
1511                 goto error;
1512         }
1513
1514         status = live_viewer_connection_list_sessions(viewer_connection,
1515                         result);
1516         if (status != BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_OK) {
1517                 goto error;
1518         }
1519
1520         goto end;
1521
1522 error:
1523         BT_VALUE_PUT_REF_AND_RESET(*result);
1524
1525         if (status >= 0) {
1526                 status = BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_ERROR;
1527         }
1528
1529 end:
1530         if (viewer_connection) {
1531                 live_viewer_connection_destroy(viewer_connection);
1532         }
1533         return status;
1534 }
1535
1536 BT_HIDDEN
1537 bt_component_class_query_method_status lttng_live_query(
1538                 bt_self_component_class_source *comp_class,
1539                 bt_private_query_executor *priv_query_exec,
1540                 const char *object, const bt_value *params,
1541                 __attribute__((unused)) void *method_data,
1542                 const bt_value **result)
1543 {
1544         bt_component_class_query_method_status status =
1545                 BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_OK;
1546         bt_self_component *self_comp = NULL;
1547         bt_logging_level log_level = bt_query_executor_get_logging_level(
1548                 bt_private_query_executor_as_query_executor_const(
1549                         priv_query_exec));
1550
1551         if (strcmp(object, "sessions") == 0) {
1552                 status = lttng_live_query_list_sessions(params, result,
1553                         log_level);
1554         } else {
1555                 BT_COMP_LOGI("Unknown query object `%s`", object);
1556                 status = BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_UNKNOWN_OBJECT;
1557                 goto end;
1558         }
1559
1560 end:
1561         return status;
1562 }
1563
1564 static
1565 void lttng_live_component_destroy_data(struct lttng_live_component *lttng_live)
1566 {
1567         if (!lttng_live) {
1568                 return;
1569         }
1570         if (lttng_live->params.url) {
1571                 g_string_free(lttng_live->params.url, TRUE);
1572         }
1573         g_free(lttng_live);
1574 }
1575
1576 BT_HIDDEN
1577 void lttng_live_component_finalize(bt_self_component_source *component)
1578 {
1579         void *data = bt_self_component_get_data(
1580                         bt_self_component_source_as_self_component(component));
1581
1582         if (!data) {
1583                 return;
1584         }
1585         lttng_live_component_destroy_data(data);
1586 }
1587
1588 static
1589 enum session_not_found_action parse_session_not_found_action_param(
1590         const bt_value *no_session_param)
1591 {
1592         enum session_not_found_action action;
1593         const char *no_session_act_str;
1594         no_session_act_str = bt_value_string_get(no_session_param);
1595         if (strcmp(no_session_act_str, SESS_NOT_FOUND_ACTION_CONTINUE_STR) == 0) {
1596                 action = SESSION_NOT_FOUND_ACTION_CONTINUE;
1597         } else if (strcmp(no_session_act_str, SESS_NOT_FOUND_ACTION_FAIL_STR) == 0) {
1598                 action = SESSION_NOT_FOUND_ACTION_FAIL;
1599         } else if (strcmp(no_session_act_str, SESS_NOT_FOUND_ACTION_END_STR) == 0) {
1600                 action = SESSION_NOT_FOUND_ACTION_END;
1601         } else {
1602                 action = SESSION_NOT_FOUND_ACTION_UNKNOWN;
1603         }
1604
1605         return action;
1606 }
1607
1608 struct lttng_live_component *lttng_live_component_create(const bt_value *params,
1609                 bt_logging_level log_level, bt_self_component *self_comp)
1610 {
1611         struct lttng_live_component *lttng_live;
1612         const bt_value *value = NULL;
1613         const char *url;
1614
1615         lttng_live = g_new0(struct lttng_live_component, 1);
1616         if (!lttng_live) {
1617                 goto end;
1618         }
1619         lttng_live->log_level = log_level;
1620         lttng_live->self_comp = self_comp;
1621         lttng_live->max_query_size = MAX_QUERY_SIZE;
1622         lttng_live->has_msg_iter = false;
1623
1624         value = bt_value_map_borrow_entry_value_const(params, URL_PARAM);
1625         if (!value || !bt_value_is_string(value)) {
1626                 BT_COMP_LOGW("Mandatory `%s` parameter missing or not a string",
1627                         URL_PARAM);
1628                 goto error;
1629         }
1630         url = bt_value_string_get(value);
1631         lttng_live->params.url = g_string_new(url);
1632         if (!lttng_live->params.url) {
1633                 goto error;
1634         }
1635
1636         value = bt_value_map_borrow_entry_value_const(params,
1637                 SESS_NOT_FOUND_ACTION_PARAM);
1638
1639         if (value && bt_value_is_string(value)) {
1640                 lttng_live->params.sess_not_found_act =
1641                         parse_session_not_found_action_param(value);
1642                 if (lttng_live->params.sess_not_found_act == SESSION_NOT_FOUND_ACTION_UNKNOWN) {
1643                         BT_COMP_LOGE("Unexpected value for `%s` parameter: "
1644                                 "value=\"%s\"", SESS_NOT_FOUND_ACTION_PARAM,
1645                                 bt_value_string_get(value));
1646                         goto error;
1647                 }
1648         } else {
1649                 BT_COMP_LOGW("Optional `%s` parameter is missing or "
1650                         "not a string value. Defaulting to %s=\"%s\".",
1651                         SESS_NOT_FOUND_ACTION_PARAM,
1652                         SESS_NOT_FOUND_ACTION_PARAM,
1653                         SESS_NOT_FOUND_ACTION_CONTINUE_STR);
1654                 lttng_live->params.sess_not_found_act =
1655                         SESSION_NOT_FOUND_ACTION_CONTINUE;
1656         }
1657
1658         goto end;
1659
1660 error:
1661         lttng_live_component_destroy_data(lttng_live);
1662         lttng_live = NULL;
1663 end:
1664         return lttng_live;
1665 }
1666
1667 BT_HIDDEN
1668 bt_component_class_init_method_status lttng_live_component_init(
1669                 bt_self_component_source *self_comp_src,
1670                 const bt_value *params, __attribute__((unused)) void *init_method_data)
1671 {
1672         struct lttng_live_component *lttng_live;
1673         bt_component_class_init_method_status ret =
1674                 BT_COMPONENT_CLASS_INIT_METHOD_STATUS_OK;
1675         bt_self_component *self_comp =
1676                 bt_self_component_source_as_self_component(self_comp_src);
1677         bt_logging_level log_level = bt_component_get_logging_level(
1678                 bt_self_component_as_component(self_comp));
1679         bt_self_component_add_port_status add_port_status;
1680
1681         lttng_live = lttng_live_component_create(params, log_level, self_comp);
1682         if (!lttng_live) {
1683                 ret = BT_COMPONENT_CLASS_INIT_METHOD_STATUS_MEMORY_ERROR;
1684                 goto error;
1685         }
1686
1687         add_port_status = bt_self_component_source_add_output_port(
1688                 self_comp_src, "out", NULL, NULL);
1689         switch (add_port_status) {
1690         case BT_SELF_COMPONENT_ADD_PORT_STATUS_ERROR:
1691                 ret = BT_COMPONENT_CLASS_INIT_METHOD_STATUS_ERROR;
1692                 goto error;
1693         case BT_SELF_COMPONENT_ADD_PORT_STATUS_MEMORY_ERROR:
1694                 ret = BT_COMPONENT_CLASS_INIT_METHOD_STATUS_MEMORY_ERROR;
1695                 goto error;
1696         default:
1697                 break;
1698         }
1699
1700         bt_self_component_set_data(self_comp, lttng_live);
1701         goto end;
1702
1703 error:
1704         lttng_live_component_destroy_data(lttng_live);
1705         lttng_live = NULL;
1706 end:
1707         return ret;
1708 }