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 "plugins/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 "data-stream.h"
  46 #include "metadata.h"
  47 #include "lttng-live.h"
  48
  49 #define MAX_QUERY_SIZE                      (256*1024)
  50 #define URL_PARAM                           "url"
  51 #define SESS_NOT_FOUND_ACTION_PARAM         "session-not-found-action"
  52 #define SESS_NOT_FOUND_ACTION_CONTINUE_STR  "continue"
  53 #define SESS_NOT_FOUND_ACTION_FAIL_STR      "fail"
  54 #define SESS_NOT_FOUND_ACTION_END_STR       "end"
  55
  56 #define print_dbg(fmt, ...)     BT_COMP_LOGD(fmt, ## __VA_ARGS__)
  57
  58 static
  59 const char *print_live_iterator_status(enum lttng_live_iterator_status status)
  60 {
  61         switch (status) {
  62         case LTTNG_LIVE_ITERATOR_STATUS_CONTINUE:
  63                 return "LTTNG_LIVE_ITERATOR_STATUS_CONTINUE";
  64         case LTTNG_LIVE_ITERATOR_STATUS_AGAIN:
  65                 return "LTTNG_LIVE_ITERATOR_STATUS_AGAIN";
  66         case LTTNG_LIVE_ITERATOR_STATUS_END:
  67                 return "LTTNG_LIVE_ITERATOR_STATUS_END";
  68         case LTTNG_LIVE_ITERATOR_STATUS_OK:
  69                 return "LTTNG_LIVE_ITERATOR_STATUS_OK";
  70         case LTTNG_LIVE_ITERATOR_STATUS_INVAL:
  71                 return "LTTNG_LIVE_ITERATOR_STATUS_INVAL";
  72         case LTTNG_LIVE_ITERATOR_STATUS_ERROR:
  73                 return "LTTNG_LIVE_ITERATOR_STATUS_ERROR";
  74         case LTTNG_LIVE_ITERATOR_STATUS_NOMEM:
  75                 return "LTTNG_LIVE_ITERATOR_STATUS_NOMEM";
  76         case LTTNG_LIVE_ITERATOR_STATUS_UNSUPPORTED:
  77                 return "LTTNG_LIVE_ITERATOR_STATUS_UNSUPPORTED";
  78         default:
  79                 abort();
  80         }
  81 }
  82
  83 static
  84 const char *print_state(struct lttng_live_stream_iterator *s)
  85 {
  86         switch (s->state) {
  87         case LTTNG_LIVE_STREAM_ACTIVE_NO_DATA:
  88                 return "ACTIVE_NO_DATA";
  89         case LTTNG_LIVE_STREAM_QUIESCENT_NO_DATA:
  90                 return "QUIESCENT_NO_DATA";
  91         case LTTNG_LIVE_STREAM_QUIESCENT:
  92                 return "QUIESCENT";
  93         case LTTNG_LIVE_STREAM_ACTIVE_DATA:
  94                 return "ACTIVE_DATA";
  95         case LTTNG_LIVE_STREAM_EOF:
  96                 return "EOF";
  97         default:
  98                 return "ERROR";
  99         }
 100 }
 101
 102 #define print_stream_state(live_stream_iter) \
 103         do { \
 104                 BT_COMP_LOGD("stream state %s last_inact_ts %" PRId64  \
 105                         ", curr_inact_ts %" PRId64, \
 106                         print_state(live_stream_iter), \
 107                         live_stream_iter->last_inactivity_ts, \
 108                         live_stream_iter->current_inactivity_ts); \
 109         } while (0);
 110
 111 BT_HIDDEN
 112 bool lttng_live_graph_is_canceled(struct lttng_live_component *lttng_live)
 113 {
 114         const bt_component *component;
 115         bool ret;
 116
 117         if (!lttng_live) {
 118                 ret = false;
 119                 goto end;
 120         }
 121
 122         component = bt_self_component_as_component(lttng_live->self_comp);
 123         ret = bt_component_graph_is_canceled(component);
 124
 125 end:
 126         return ret;
 127 }
 128
 129 static
 130 struct lttng_live_trace *lttng_live_find_trace(struct lttng_live_session *session,
 131                 uint64_t trace_id)
 132 {
 133         uint64_t trace_idx;
 134         struct lttng_live_trace *ret_trace = NULL;
 135
 136         for (trace_idx = 0; trace_idx < session->traces->len; trace_idx++) {
 137                 struct lttng_live_trace *trace =
 138                         g_ptr_array_index(session->traces, trace_idx);
 139                 if (trace->id == trace_id) {
 140                         ret_trace = trace;
 141                         goto end;
 142                 }
 143         }
 144
 145 end:
 146         return ret_trace;
 147 }
 148
 149 static
 150 void lttng_live_destroy_trace(struct lttng_live_trace *trace)
 151 {
 152         bt_logging_level log_level = trace->log_level;
 153         bt_self_component *self_comp = trace->self_comp;
 154
 155         BT_COMP_LOGD("Destroy lttng_live_trace");
 156
 157         BT_ASSERT(trace->stream_iterators);
 158         g_ptr_array_free(trace->stream_iterators, TRUE);
 159
 160         BT_TRACE_PUT_REF_AND_RESET(trace->trace);
 161         BT_TRACE_CLASS_PUT_REF_AND_RESET(trace->trace_class);
 162
 163         lttng_live_metadata_fini(trace);
 164         g_free(trace);
 165 }
 166
 167 static
 168 struct lttng_live_trace *lttng_live_create_trace(struct lttng_live_session *session,
 169                 uint64_t trace_id)
 170 {
 171         struct lttng_live_trace *trace = NULL;
 172         bt_logging_level log_level = session->log_level;
 173         bt_self_component *self_comp = session->self_comp;
 174
 175         trace = g_new0(struct lttng_live_trace, 1);
 176         if (!trace) {
 177                 goto error;
 178         }
 179         trace->log_level = session->log_level;
 180         trace->self_comp = session->self_comp;
 181         trace->session = session;
 182         trace->id = trace_id;
 183         trace->trace_class = NULL;
 184         trace->trace = NULL;
 185         trace->stream_iterators = g_ptr_array_new_with_free_func(
 186                         (GDestroyNotify) lttng_live_stream_iterator_destroy);
 187         BT_ASSERT(trace->stream_iterators);
 188         trace->new_metadata_needed = true;
 189         g_ptr_array_add(session->traces, trace);
 190
 191         BT_COMP_LOGI("Create trace");
 192         goto end;
 193 error:
 194         g_free(trace);
 195         trace = NULL;
 196 end:
 197         return trace;
 198 }
 199
 200 BT_HIDDEN
 201 struct lttng_live_trace *lttng_live_borrow_trace(
 202                 struct lttng_live_session *session, uint64_t trace_id)
 203 {
 204         struct lttng_live_trace *trace;
 205
 206         trace = lttng_live_find_trace(session, trace_id);
 207         if (trace) {
 208                 goto end;
 209         }
 210
 211         /* The session is the owner of the newly created trace. */
 212         trace = lttng_live_create_trace(session, trace_id);
 213
 214 end:
 215         return trace;
 216 }
 217
 218 BT_HIDDEN
 219 int lttng_live_add_session(struct lttng_live_msg_iter *lttng_live_msg_iter,
 220                 uint64_t session_id, const char *hostname,
 221                 const char *session_name)
 222 {
 223         int ret = 0;
 224         struct lttng_live_session *session;
 225         bt_logging_level log_level = lttng_live_msg_iter->log_level;
 226         bt_self_component *self_comp = lttng_live_msg_iter->self_comp;
 227
 228         session = g_new0(struct lttng_live_session, 1);
 229         if (!session) {
 230                 goto error;
 231         }
 232
 233         session->log_level = lttng_live_msg_iter->log_level;
 234         session->self_comp = lttng_live_msg_iter->self_comp;
 235         session->id = session_id;
 236         session->traces = g_ptr_array_new_with_free_func(
 237                         (GDestroyNotify) lttng_live_destroy_trace);
 238         BT_ASSERT(session->traces);
 239         session->lttng_live_msg_iter = lttng_live_msg_iter;
 240         session->new_streams_needed = true;
 241         session->hostname = g_string_new(hostname);
 242         BT_ASSERT(session->hostname);
 243
 244         session->session_name = g_string_new(session_name);
 245         BT_ASSERT(session->session_name);
 246
 247         BT_COMP_LOGI("Reading from session: %" PRIu64 " hostname: %s session_name: %s",
 248                 session->id, hostname, session_name);
 249         g_ptr_array_add(lttng_live_msg_iter->sessions, session);
 250         goto end;
 251 error:
 252         BT_COMP_LOGE("Error adding session");
 253         g_free(session);
 254         ret = -1;
 255 end:
 256         return ret;
 257 }
 258
 259 static
 260 void lttng_live_destroy_session(struct lttng_live_session *session)
 261 {
 262         struct lttng_live_component *live_comp;
 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                         live_comp = session->lttng_live_msg_iter->lttng_live_comp;
 276                         if (session->lttng_live_msg_iter &&
 277                                         !lttng_live_graph_is_canceled(live_comp)) {
 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->lttng_live_comp)) {
 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 /*
 799  * helper function:
 800  *            handle_no_data_streams()
 801  *              retry:
 802  *                - for each ACTIVE_NO_DATA stream:
 803  *                  - query relayd for stream data, or quiescence info.
 804  *                    - if need metadata, get metadata, goto retry.
 805  *                    - if new stream, get new stream as ACTIVE_NO_DATA, goto retry
 806  *                  - if quiescent, move to QUIESCENT streams
 807  *                  - if fetched data, move to ACTIVE_DATA streams
 808  *                (at this point each stream either has data, or is quiescent)
 809  *
 810  *
 811  * iterator_next:
 812  *            handle_new_streams_and_metadata()
 813  *                  - query relayd for known streams, add them as ACTIVE_NO_DATA
 814  *                  - query relayd for metadata
 815  *
 816  *            call handle_active_no_data_streams()
 817  *
 818  *            handle_quiescent_streams()
 819  *                - if at least one stream is ACTIVE_DATA:
 820  *                  - peek stream event with lowest timestamp -> next_ts
 821  *                  - for each quiescent stream
 822  *                    - if next_ts >= quiescent end
 823  *                      - set state to ACTIVE_NO_DATA
 824  *                - else
 825  *                  - for each quiescent stream
 826  *                      - set state to ACTIVE_NO_DATA
 827  *
 828  *            call handle_active_no_data_streams()
 829  *
 830  *            handle_active_data_streams()
 831  *                - if at least one stream is ACTIVE_DATA:
 832  *                    - get stream event with lowest timestamp from heap
 833  *                    - make that stream event the current message.
 834  *                    - move this stream heap position to its next event
 835  *                      - if we need to fetch data from relayd, move
 836  *                        stream to ACTIVE_NO_DATA.
 837  *                    - return OK
 838  *                - return AGAIN
 839  *
 840  * end criterion: ctrl-c on client. If relayd exits or the session
 841  * closes on the relay daemon side, we keep on waiting for streams.
 842  * Eventually handle --end timestamp (also an end criterion).
 843  *
 844  * When disconnected from relayd: try to re-connect endlessly.
 845  */
 846 static
 847 enum lttng_live_iterator_status lttng_live_iterator_next_on_stream(
 848                 struct lttng_live_msg_iter *lttng_live_msg_iter,
 849                 struct lttng_live_stream_iterator *stream_iter,
 850                 bt_message **curr_msg)
 851 {
 852         bt_logging_level log_level = lttng_live_msg_iter->log_level;
 853         bt_self_component *self_comp = lttng_live_msg_iter->self_comp;
 854         enum lttng_live_iterator_status live_status;
 855
 856 retry:
 857         print_stream_state(stream_iter);
 858         live_status = lttng_live_iterator_handle_new_streams_and_metadata(
 859                         lttng_live_msg_iter);
 860         if (live_status != LTTNG_LIVE_ITERATOR_STATUS_OK) {
 861                 goto end;
 862         }
 863         live_status = lttng_live_iterator_next_handle_one_no_data_stream(
 864                         lttng_live_msg_iter, stream_iter);
 865         if (live_status != LTTNG_LIVE_ITERATOR_STATUS_OK) {
 866                 goto end;
 867         }
 868         live_status = lttng_live_iterator_next_handle_one_quiescent_stream(
 869                         lttng_live_msg_iter, stream_iter, curr_msg);
 870         if (live_status != LTTNG_LIVE_ITERATOR_STATUS_OK) {
 871                 BT_ASSERT(!*curr_msg);
 872                 goto end;
 873         }
 874         if (*curr_msg) {
 875                 goto end;
 876         }
 877         live_status = lttng_live_iterator_next_handle_one_active_data_stream(
 878                         lttng_live_msg_iter, stream_iter, curr_msg);
 879         if (live_status != LTTNG_LIVE_ITERATOR_STATUS_OK) {
 880                 BT_ASSERT(!*curr_msg);
 881         }
 882
 883 end:
 884         if (live_status == LTTNG_LIVE_ITERATOR_STATUS_CONTINUE) {
 885                 goto retry;
 886         }
 887
 888         return live_status;
 889 }
 890
 891 static
 892 enum lttng_live_iterator_status next_stream_iterator_for_trace(
 893                 struct lttng_live_msg_iter *lttng_live_msg_iter,
 894                 struct lttng_live_trace *live_trace,
 895                 struct lttng_live_stream_iterator **candidate_stream_iter)
 896 {
 897         struct lttng_live_stream_iterator *curr_candidate_stream_iter = NULL;
 898         enum lttng_live_iterator_status stream_iter_status;;
 899         int64_t curr_candidate_msg_ts = INT64_MAX;
 900         uint64_t stream_iter_idx;
 901         bt_logging_level log_level = lttng_live_msg_iter->log_level;
 902         bt_self_component *self_comp = lttng_live_msg_iter->self_comp;
 903
 904         BT_ASSERT(live_trace);
 905         BT_ASSERT(live_trace->stream_iterators);
 906         /*
 907          * Update the current message of every stream iterators of this trace.
 908          * The current msg of every stream must have a timestamp equal or
 909          * larger than the last message returned by this iterator. We must
 910          * ensure monotonicity.
 911          */
 912         stream_iter_idx = 0;
 913         while (stream_iter_idx < live_trace->stream_iterators->len) {
 914                 bool stream_iter_is_ended = false;
 915                 struct lttng_live_stream_iterator *stream_iter =
 916                         g_ptr_array_index(live_trace->stream_iterators,
 917                                         stream_iter_idx);
 918
 919                 /*
 920                  * Find if there is are now current message for this stream
 921                  * iterator get it.
 922                  */
 923                 while (!stream_iter->current_msg) {
 924                         bt_message *msg = NULL;
 925                         int64_t curr_msg_ts_ns = INT64_MAX;
 926                         stream_iter_status = lttng_live_iterator_next_on_stream(
 927                                         lttng_live_msg_iter, stream_iter, &msg);
 928
 929                         BT_COMP_LOGD("live stream iterator returned status :%s",
 930                                         print_live_iterator_status(stream_iter_status));
 931                         if (stream_iter_status == LTTNG_LIVE_ITERATOR_STATUS_END) {
 932                                 stream_iter_is_ended = true;
 933                                 break;
 934                         }
 935
 936                         if (stream_iter_status != LTTNG_LIVE_ITERATOR_STATUS_OK) {
 937                                 goto end;
 938                         }
 939
 940                         BT_ASSERT(msg);
 941
 942                         /*
 943                          * Get the timestamp in nanoseconds from origin of this
 944                          * messsage.
 945                          */
 946                         live_get_msg_ts_ns(stream_iter, lttng_live_msg_iter,
 947                                 msg, lttng_live_msg_iter->last_msg_ts_ns,
 948                                 &curr_msg_ts_ns);
 949
 950                         /*
 951                          * Check if the message of the current live stream
 952                          * iterator occured at the exact same time or after the
 953                          * last message returned by this component's message
 954                          * iterator. If not, we return an error.
 955                          */
 956                         if (curr_msg_ts_ns >= lttng_live_msg_iter->last_msg_ts_ns) {
 957                                 stream_iter->current_msg = msg;
 958                                 stream_iter->current_msg_ts_ns = curr_msg_ts_ns;
 959                         } else {
 960                                 /*
 961                                  * We received a message in the past. To ensure
 962                                  * monotonicity, we can't send it forward.
 963                                  */
 964                                 BT_COMP_LOGE("Message's timestamp is less than "
 965                                         "lttng-live's message iterator's last "
 966                                         "returned timestamp: "
 967                                         "lttng-live-msg-iter-addr=%p, ts=%" PRId64 ", "
 968                                         "last-msg-ts=%" PRId64,
 969                                         lttng_live_msg_iter, curr_msg_ts_ns,
 970                                         lttng_live_msg_iter->last_msg_ts_ns);
 971                                 stream_iter_status = LTTNG_LIVE_ITERATOR_STATUS_ERROR;
 972                                 goto end;
 973                         }
 974                 }
 975
 976                 if (!stream_iter_is_ended &&
 977                                 stream_iter->current_msg_ts_ns <= curr_candidate_msg_ts) {
 978                         /*
 979                          * Update the current best candidate message for the
 980                          * stream iterator of thise live trace to be forwarded
 981                          * downstream.
 982                          */
 983                         curr_candidate_msg_ts = stream_iter->current_msg_ts_ns;
 984                         curr_candidate_stream_iter = stream_iter;
 985                 }
 986
 987                 if (stream_iter_is_ended) {
 988                         /*
 989                          * The live stream iterator is ENDed. We remove that
 990                          * iterator from the list and we restart the iteration
 991                          * at the beginning of the live stream iterator array
 992                          * to because the removal will shuffle the array.
 993                          */
 994                         g_ptr_array_remove_index_fast(live_trace->stream_iterators,
 995                                 stream_iter_idx);
 996                         stream_iter_idx = 0;
 997                 } else {
 998                         stream_iter_idx++;
 999                 }
1000         }
1001
1002         if (curr_candidate_stream_iter) {
1003                 *candidate_stream_iter = curr_candidate_stream_iter;
1004                 stream_iter_status = LTTNG_LIVE_ITERATOR_STATUS_OK;
1005         } else {
1006                 /*
1007                  * The only case where we don't have a candidate for this trace
1008                  * is if we reached the end of all the iterators.
1009                  */
1010                 BT_ASSERT(live_trace->stream_iterators->len == 0);
1011                 stream_iter_status = LTTNG_LIVE_ITERATOR_STATUS_END;
1012         }
1013
1014 end:
1015         return stream_iter_status;
1016 }
1017
1018 static
1019 enum lttng_live_iterator_status next_stream_iterator_for_session(
1020                 struct lttng_live_msg_iter *lttng_live_msg_iter,
1021                 struct lttng_live_session *session,
1022                 struct lttng_live_stream_iterator **candidate_session_stream_iter)
1023 {
1024         enum lttng_live_iterator_status stream_iter_status;
1025         uint64_t trace_idx = 0;
1026         int64_t curr_candidate_msg_ts = INT64_MAX;
1027         struct lttng_live_stream_iterator *curr_candidate_stream_iter = NULL;
1028
1029         /*
1030          * Make sure we are attached to the session and look for new streams
1031          * and metadata.
1032          */
1033         stream_iter_status = lttng_live_get_session(lttng_live_msg_iter, session);
1034         if (stream_iter_status != LTTNG_LIVE_ITERATOR_STATUS_OK &&
1035                         stream_iter_status != LTTNG_LIVE_ITERATOR_STATUS_CONTINUE &&
1036                         stream_iter_status != LTTNG_LIVE_ITERATOR_STATUS_END) {
1037                 goto end;
1038         }
1039
1040         BT_ASSERT(session->traces);
1041
1042         /*
1043          * Use while loops here rather then for loops so we can restart the
1044          * iteration if an element is removed from the array during the
1045          * looping.
1046          */
1047         while (trace_idx < session->traces->len) {
1048                 bool trace_is_ended = false;
1049                 struct lttng_live_stream_iterator *stream_iter;
1050                 struct lttng_live_trace *trace =
1051                         g_ptr_array_index(session->traces, trace_idx);
1052
1053                 stream_iter_status = next_stream_iterator_for_trace(
1054                         lttng_live_msg_iter, trace, &stream_iter);
1055                 if (stream_iter_status == LTTNG_LIVE_ITERATOR_STATUS_END) {
1056                         /*
1057                          * All the live stream iterators for this trace are
1058                          * ENDed. Remove the trace from this session.
1059                          */
1060                         trace_is_ended = true;
1061                 } else if (stream_iter_status != LTTNG_LIVE_ITERATOR_STATUS_OK) {
1062                         goto end;
1063                 }
1064
1065                 if (!trace_is_ended) {
1066                         BT_ASSERT(stream_iter);
1067
1068                         if (stream_iter->current_msg_ts_ns <= curr_candidate_msg_ts) {
1069                                 curr_candidate_msg_ts = stream_iter->current_msg_ts_ns;
1070                                 curr_candidate_stream_iter = stream_iter;
1071                         }
1072                         trace_idx++;
1073                 } else {
1074                         g_ptr_array_remove_index_fast(session->traces, trace_idx);
1075                         trace_idx = 0;
1076                 }
1077         }
1078         if (curr_candidate_stream_iter) {
1079                 *candidate_session_stream_iter = curr_candidate_stream_iter;
1080                 stream_iter_status = LTTNG_LIVE_ITERATOR_STATUS_OK;
1081         } else {
1082                 /*
1083                  * The only cases where we don't have a candidate for this
1084                  * trace is:
1085                  *  1. if we reached the end of all the iterators of all the
1086                  *  traces of this session,
1087                  *  2. if we never had live stream iterator in the first place.
1088                  *
1089                  * In either cases, we return END.
1090                  */
1091                 BT_ASSERT(session->traces->len == 0);
1092                 stream_iter_status = LTTNG_LIVE_ITERATOR_STATUS_END;
1093         }
1094 end:
1095         return stream_iter_status;
1096 }
1097
1098 static inline
1099 void put_messages(bt_message_array_const msgs, uint64_t count)
1100 {
1101         uint64_t i;
1102
1103         for (i = 0; i < count; i++) {
1104                 BT_MESSAGE_PUT_REF_AND_RESET(msgs[i]);
1105         }
1106 }
1107
1108 BT_HIDDEN
1109 bt_component_class_message_iterator_next_method_status lttng_live_msg_iter_next(
1110                 bt_self_message_iterator *self_msg_it,
1111                 bt_message_array_const msgs, uint64_t capacity,
1112                 uint64_t *count)
1113 {
1114         bt_component_class_message_iterator_next_method_status status;
1115         struct lttng_live_msg_iter *lttng_live_msg_iter =
1116                 bt_self_message_iterator_get_data(self_msg_it);
1117         struct lttng_live_component *lttng_live =
1118                 lttng_live_msg_iter->lttng_live_comp;
1119         enum lttng_live_iterator_status stream_iter_status;
1120         uint64_t session_idx;
1121
1122         *count = 0;
1123
1124         BT_ASSERT(lttng_live_msg_iter);
1125
1126         /*
1127          * Clear all the invalid message reference that might be left over in
1128          * the output array.
1129          */
1130         memset(msgs, 0, capacity * sizeof(*msgs));
1131
1132         /*
1133          * If no session are exposed on the relay found at the url provided by
1134          * the user, session count will be 0. In this case, we return status
1135          * end to return gracefully.
1136          */
1137         if (lttng_live_msg_iter->sessions->len == 0) {
1138                 if (lttng_live->params.sess_not_found_act !=
1139                                 SESSION_NOT_FOUND_ACTION_CONTINUE) {
1140                         status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_END;
1141                         goto no_session;
1142                 } else {
1143                         /*
1144                          * The are no more active session for this session
1145                          * name. Retry to create a viewer session for the
1146                          * requested session name.
1147                          */
1148                         if (lttng_live_create_viewer_session(lttng_live_msg_iter)) {
1149                                 status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_ERROR;
1150                                 goto no_session;
1151                         }
1152                 }
1153         }
1154
1155         if (lttng_live_msg_iter->active_stream_iter == 0) {
1156                 lttng_live_force_new_streams_and_metadata(lttng_live_msg_iter);
1157         }
1158
1159         /*
1160          * Here the muxing of message is done.
1161          *
1162          * We need to iterate over all the streams of all the traces of all the
1163          * viewer sessions in order to get the message with the smallest
1164          * timestamp. In this case, a session is a viewer session and there is
1165          * one viewer session per consumer daemon. (UST 32bit, UST 64bit and/or
1166          * kernel). Each viewer session can have multiple traces, for example,
1167          * 64bit UST viewer sessions could have multiple per-pid traces.
1168          *
1169          * We iterate over the streams of each traces to update and see what is
1170          * their next message's timestamp. From those timestamps, we select the
1171          * message with the smallest timestamp as the best candidate message
1172          * for that trace and do the same thing across all the sessions.
1173          *
1174          * We then compare the timestamp of best candidate message of all the
1175          * sessions to pick the message with the smallest timestamp and we
1176          * return it.
1177          */
1178         while (*count < capacity) {
1179                 struct lttng_live_stream_iterator *next_stream_iter = NULL,
1180                                                   *candidate_stream_iter = NULL;
1181                 int64_t next_msg_ts_ns = INT64_MAX;
1182
1183                 BT_ASSERT(lttng_live_msg_iter->sessions);
1184                 session_idx = 0;
1185                 /*
1186                  * Use a while loop instead of a for loop so we can restart the
1187                  * iteration if we remove an element. We can safely call
1188                  * next_stream_iterator_for_session() multiple times on the
1189                  * same session as we only fetch a new message if there is no
1190                  * current next message for each live stream iterator.
1191                  * If all live stream iterator of that session already have a
1192                  * current next message, the function will simply exit return
1193                  * the same candidate live stream iterator every time.
1194                  */
1195                 while (session_idx < lttng_live_msg_iter->sessions->len) {
1196                         struct lttng_live_session *session =
1197                                 g_ptr_array_index(lttng_live_msg_iter->sessions,
1198                                         session_idx);
1199
1200                         /* Find the best candidate message to send downstream. */
1201                         stream_iter_status = next_stream_iterator_for_session(
1202                                 lttng_live_msg_iter, session,
1203                                 &candidate_stream_iter);
1204
1205                         /* If we receive an END status, it means that either:
1206                          * - Those traces never had active streams (UST with no
1207                          *   data produced yet),
1208                          * - All live stream iterators have ENDed.*/
1209                         if (stream_iter_status == LTTNG_LIVE_ITERATOR_STATUS_END) {
1210                                 if (session->closed && session->traces->len == 0) {
1211                                         /*
1212                                          * Remove the session from the list and restart the
1213                                          * iteration at the beginning of the array since the
1214                                          * removal shuffle the elements of the array.
1215                                          */
1216                                         g_ptr_array_remove_index_fast(
1217                                                 lttng_live_msg_iter->sessions,
1218                                                 session_idx);
1219                                         session_idx = 0;
1220                                 } else {
1221                                         session_idx++;
1222                                 }
1223                                 continue;
1224                         }
1225
1226                         if (stream_iter_status != LTTNG_LIVE_ITERATOR_STATUS_OK) {
1227                                 goto end;
1228                         }
1229
1230                         if (candidate_stream_iter->current_msg_ts_ns <= next_msg_ts_ns) {
1231                                 next_msg_ts_ns = candidate_stream_iter->current_msg_ts_ns;
1232                                 next_stream_iter = candidate_stream_iter;
1233                         }
1234
1235                         session_idx++;
1236                 }
1237
1238                 if (!next_stream_iter) {
1239                         stream_iter_status = LTTNG_LIVE_ITERATOR_STATUS_AGAIN;
1240                         goto end;
1241                 }
1242
1243                 BT_ASSERT(next_stream_iter->current_msg);
1244                 /* Ensure monotonicity. */
1245                 BT_ASSERT(lttng_live_msg_iter->last_msg_ts_ns <=
1246                         next_stream_iter->current_msg_ts_ns);
1247
1248                 /*
1249                  * Insert the next message to the message batch. This will set
1250                  * stream iterator current messsage to NULL so that next time
1251                  * we fetch the next message of that stream iterator
1252                  */
1253                 BT_MESSAGE_MOVE_REF(msgs[*count], next_stream_iter->current_msg);
1254                 (*count)++;
1255
1256                 /* Update the last timestamp in nanoseconds sent downstream. */
1257                 lttng_live_msg_iter->last_msg_ts_ns = next_msg_ts_ns;
1258                 next_stream_iter->current_msg_ts_ns = INT64_MAX;
1259
1260                 stream_iter_status = LTTNG_LIVE_ITERATOR_STATUS_OK;
1261         }
1262 end:
1263         switch (stream_iter_status) {
1264         case LTTNG_LIVE_ITERATOR_STATUS_OK:
1265         case LTTNG_LIVE_ITERATOR_STATUS_AGAIN:
1266                 if (*count > 0) {
1267                         /*
1268                          * We received a again status but we have some messages
1269                          * to send downstream. We send them and return OK for
1270                          * now. On the next call we return again if there are
1271                          * still no new message to send.
1272                          */
1273                         status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK;
1274                 } else {
1275                         status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_AGAIN;
1276                 }
1277                 break;
1278         case LTTNG_LIVE_ITERATOR_STATUS_END:
1279                 status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_END;
1280                 break;
1281         case LTTNG_LIVE_ITERATOR_STATUS_NOMEM:
1282                 status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_MEMORY_ERROR;
1283                 break;
1284         case LTTNG_LIVE_ITERATOR_STATUS_ERROR:
1285         case LTTNG_LIVE_ITERATOR_STATUS_INVAL:
1286         case LTTNG_LIVE_ITERATOR_STATUS_UNSUPPORTED:
1287                 status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_ERROR;
1288                 /* Put all existing messages on error. */
1289                 put_messages(msgs, *count);
1290                 break;
1291         default:
1292                 abort();
1293         }
1294
1295 no_session:
1296         return status;
1297 }
1298
1299 BT_HIDDEN
1300 bt_component_class_message_iterator_init_method_status lttng_live_msg_iter_init(
1301                 bt_self_message_iterator *self_msg_it,
1302                 bt_self_component_source *self_comp_src,
1303                 bt_self_component_port_output *self_port)
1304 {
1305         bt_component_class_message_iterator_init_method_status ret =
1306                 BT_COMPONENT_CLASS_MESSAGE_ITERATOR_INIT_METHOD_STATUS_OK;
1307         bt_self_component *self_comp =
1308                 bt_self_component_source_as_self_component(self_comp_src);
1309         struct lttng_live_component *lttng_live;
1310         struct lttng_live_msg_iter *lttng_live_msg_iter;
1311         bt_logging_level log_level;
1312
1313         BT_ASSERT(self_msg_it);
1314
1315         lttng_live = bt_self_component_get_data(self_comp);
1316         log_level = lttng_live->log_level;
1317         self_comp = lttng_live->self_comp;
1318
1319         /* There can be only one downstream iterator at the same time. */
1320         BT_ASSERT(!lttng_live->has_msg_iter);
1321         lttng_live->has_msg_iter = true;
1322
1323         lttng_live_msg_iter = g_new0(struct lttng_live_msg_iter, 1);
1324         if (!lttng_live_msg_iter) {
1325                 ret = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_INIT_METHOD_STATUS_ERROR;
1326                 goto end;
1327         }
1328
1329         lttng_live_msg_iter->log_level = lttng_live->log_level;
1330         lttng_live_msg_iter->self_comp = lttng_live->self_comp;
1331         lttng_live_msg_iter->lttng_live_comp = lttng_live;
1332         lttng_live_msg_iter->self_msg_iter = self_msg_it;
1333
1334         lttng_live_msg_iter->active_stream_iter = 0;
1335         lttng_live_msg_iter->last_msg_ts_ns = INT64_MIN;
1336         lttng_live_msg_iter->sessions = g_ptr_array_new_with_free_func(
1337                 (GDestroyNotify) lttng_live_destroy_session);
1338         BT_ASSERT(lttng_live_msg_iter->sessions);
1339
1340         lttng_live_msg_iter->viewer_connection =
1341                 live_viewer_connection_create(lttng_live->params.url->str, false,
1342                         lttng_live_msg_iter);
1343         if (!lttng_live_msg_iter->viewer_connection) {
1344                 goto error;
1345         }
1346
1347         if (lttng_live_create_viewer_session(lttng_live_msg_iter)) {
1348                 goto error;
1349         }
1350         if (lttng_live_msg_iter->sessions->len == 0) {
1351                 switch (lttng_live->params.sess_not_found_act) {
1352                 case SESSION_NOT_FOUND_ACTION_CONTINUE:
1353                         BT_COMP_LOGI("Unable to connect to the requested live viewer "
1354                                 "session. Keep trying to connect because of "
1355                                 "%s=\"%s\" component parameter: url=\"%s\"",
1356                                 SESS_NOT_FOUND_ACTION_PARAM,
1357                                 SESS_NOT_FOUND_ACTION_CONTINUE_STR,
1358                                 lttng_live->params.url->str);
1359                         break;
1360                 case SESSION_NOT_FOUND_ACTION_FAIL:
1361                         BT_COMP_LOGE("Unable to connect to the requested live viewer "
1362                                 "session. Fail the message iterator"
1363                                 "initialization because of %s=\"%s\" "
1364                                 "component parameter: url =\"%s\"",
1365                                 SESS_NOT_FOUND_ACTION_PARAM,
1366                                 SESS_NOT_FOUND_ACTION_FAIL_STR,
1367                                 lttng_live->params.url->str);
1368                         goto error;
1369                 case SESSION_NOT_FOUND_ACTION_END:
1370                         BT_COMP_LOGI("Unable to connect to the requested live viewer "
1371                                 "session. End gracefully at the first _next() "
1372                                 "call because of %s=\"%s\" component parameter: "
1373                                 "url=\"%s\"", SESS_NOT_FOUND_ACTION_PARAM,
1374                                 SESS_NOT_FOUND_ACTION_END_STR,
1375                                 lttng_live->params.url->str);
1376                         break;
1377                 }
1378         }
1379
1380         bt_self_message_iterator_set_data(self_msg_it, lttng_live_msg_iter);
1381
1382         goto end;
1383 error:
1384         ret = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_INIT_METHOD_STATUS_ERROR;
1385         lttng_live_msg_iter_destroy(lttng_live_msg_iter);
1386 end:
1387         return ret;
1388 }
1389
1390 static
1391 bt_component_class_query_method_status lttng_live_query_list_sessions(
1392                 const bt_value *params, const bt_value **result,
1393                 bt_logging_level log_level)
1394 {
1395         bt_component_class_query_method_status status =
1396                 BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_OK;
1397         const bt_value *url_value = NULL;
1398         const char *url;
1399         struct live_viewer_connection *viewer_connection = NULL;
1400         bt_self_component *self_comp = NULL;
1401
1402         url_value = bt_value_map_borrow_entry_value_const(params, URL_PARAM);
1403         if (!url_value) {
1404                 BT_COMP_LOGW("Mandatory `%s` parameter missing", URL_PARAM);
1405                 status = BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_INVALID_PARAMS;
1406                 goto error;
1407         }
1408
1409         if (!bt_value_is_string(url_value)) {
1410                 BT_COMP_LOGW("`%s` parameter is required to be a string value",
1411                         URL_PARAM);
1412                 status = BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_INVALID_PARAMS;
1413                 goto error;
1414         }
1415
1416         url = bt_value_string_get(url_value);
1417
1418         viewer_connection = live_viewer_connection_create(url, true, NULL);
1419         if (!viewer_connection) {
1420                 goto error;
1421         }
1422
1423         status = live_viewer_connection_list_sessions(viewer_connection,
1424                         result);
1425         if (status != BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_OK) {
1426                 goto error;
1427         }
1428
1429         goto end;
1430
1431 error:
1432         BT_VALUE_PUT_REF_AND_RESET(*result);
1433
1434         if (status >= 0) {
1435                 status = BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_ERROR;
1436         }
1437
1438 end:
1439         if (viewer_connection) {
1440                 live_viewer_connection_destroy(viewer_connection);
1441         }
1442         return status;
1443 }
1444
1445 BT_HIDDEN
1446 bt_component_class_query_method_status lttng_live_query(
1447                 bt_self_component_class_source *comp_class,
1448                 const bt_query_executor *query_exec,
1449                 const char *object, const bt_value *params,
1450                 bt_logging_level log_level, const bt_value **result)
1451 {
1452         bt_component_class_query_method_status status =
1453                 BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_OK;
1454         bt_self_component *self_comp = NULL;
1455
1456         if (strcmp(object, "sessions") == 0) {
1457                 status = lttng_live_query_list_sessions(params, result,
1458                         log_level);
1459         } else {
1460                 BT_COMP_LOGI("Unknown query object `%s`", object);
1461                 status = BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_INVALID_OBJECT;
1462                 goto end;
1463         }
1464
1465 end:
1466         return status;
1467 }
1468
1469 static
1470 void lttng_live_component_destroy_data(struct lttng_live_component *lttng_live)
1471 {
1472         if (!lttng_live) {
1473                 return;
1474         }
1475         if (lttng_live->params.url) {
1476                 g_string_free(lttng_live->params.url, TRUE);
1477         }
1478         g_free(lttng_live);
1479 }
1480
1481 BT_HIDDEN
1482 void lttng_live_component_finalize(bt_self_component_source *component)
1483 {
1484         void *data = bt_self_component_get_data(
1485                         bt_self_component_source_as_self_component(component));
1486
1487         if (!data) {
1488                 return;
1489         }
1490         lttng_live_component_destroy_data(data);
1491 }
1492
1493 static
1494 enum session_not_found_action parse_session_not_found_action_param(
1495         const bt_value *no_session_param)
1496 {
1497         enum session_not_found_action action;
1498         const char *no_session_act_str;
1499         no_session_act_str = bt_value_string_get(no_session_param);
1500         if (strcmp(no_session_act_str, SESS_NOT_FOUND_ACTION_CONTINUE_STR) == 0) {
1501                 action = SESSION_NOT_FOUND_ACTION_CONTINUE;
1502         } else if (strcmp(no_session_act_str, SESS_NOT_FOUND_ACTION_FAIL_STR) == 0) {
1503                 action = SESSION_NOT_FOUND_ACTION_FAIL;
1504         } else if (strcmp(no_session_act_str, SESS_NOT_FOUND_ACTION_END_STR) == 0) {
1505                 action = SESSION_NOT_FOUND_ACTION_END;
1506         } else {
1507                 action = -1;
1508         }
1509
1510         return action;
1511 }
1512
1513 struct lttng_live_component *lttng_live_component_create(const bt_value *params,
1514                 bt_logging_level log_level, bt_self_component *self_comp)
1515 {
1516         struct lttng_live_component *lttng_live;
1517         const bt_value *value = NULL;
1518         const char *url;
1519
1520         lttng_live = g_new0(struct lttng_live_component, 1);
1521         if (!lttng_live) {
1522                 goto end;
1523         }
1524         lttng_live->log_level = log_level;
1525         lttng_live->self_comp = self_comp;
1526         lttng_live->max_query_size = MAX_QUERY_SIZE;
1527         lttng_live->has_msg_iter = false;
1528
1529         value = bt_value_map_borrow_entry_value_const(params, URL_PARAM);
1530         if (!value || !bt_value_is_string(value)) {
1531                 BT_COMP_LOGW("Mandatory `%s` parameter missing or not a string",
1532                         URL_PARAM);
1533                 goto error;
1534         }
1535         url = bt_value_string_get(value);
1536         lttng_live->params.url = g_string_new(url);
1537         if (!lttng_live->params.url) {
1538                 goto error;
1539         }
1540
1541         value = bt_value_map_borrow_entry_value_const(params,
1542                 SESS_NOT_FOUND_ACTION_PARAM);
1543
1544         if (value && bt_value_is_string(value)) {
1545                 lttng_live->params.sess_not_found_act =
1546                         parse_session_not_found_action_param(value);
1547                 if (lttng_live->params.sess_not_found_act == -1) {
1548                         BT_COMP_LOGE("Unexpected value for `%s` parameter: "
1549                                 "value=\"%s\"", SESS_NOT_FOUND_ACTION_PARAM,
1550                                 bt_value_string_get(value));
1551                         goto error;
1552                 }
1553         } else {
1554                 BT_COMP_LOGW("Optional `%s` parameter is missing or "
1555                         "not a string value. Defaulting to %s=\"%s\".",
1556                         SESS_NOT_FOUND_ACTION_PARAM,
1557                         SESS_NOT_FOUND_ACTION_PARAM,
1558                         SESS_NOT_FOUND_ACTION_CONTINUE_STR);
1559                 lttng_live->params.sess_not_found_act =
1560                         SESSION_NOT_FOUND_ACTION_CONTINUE;
1561         }
1562
1563         goto end;
1564
1565 error:
1566         lttng_live_component_destroy_data(lttng_live);
1567         lttng_live = NULL;
1568 end:
1569         return lttng_live;
1570 }
1571
1572 BT_HIDDEN
1573 bt_component_class_init_method_status lttng_live_component_init(
1574                 bt_self_component_source *self_comp_src,
1575                 const bt_value *params, __attribute__((unused)) void *init_method_data)
1576 {
1577         struct lttng_live_component *lttng_live;
1578         bt_component_class_init_method_status ret =
1579                 BT_COMPONENT_CLASS_INIT_METHOD_STATUS_OK;
1580         bt_self_component *self_comp =
1581                 bt_self_component_source_as_self_component(self_comp_src);
1582         bt_logging_level log_level = bt_component_get_logging_level(
1583                 bt_self_component_as_component(self_comp));
1584         bt_self_component_add_port_status add_port_status;
1585
1586         lttng_live = lttng_live_component_create(params, log_level, self_comp);
1587         if (!lttng_live) {
1588                 ret = BT_COMPONENT_CLASS_INIT_METHOD_STATUS_MEMORY_ERROR;
1589                 goto error;
1590         }
1591
1592         if (lttng_live_graph_is_canceled(lttng_live)) {
1593                 ret = BT_COMPONENT_CLASS_INIT_METHOD_STATUS_ERROR;
1594                 goto error;
1595         }
1596
1597         add_port_status = bt_self_component_source_add_output_port(
1598                 self_comp_src, "out", NULL, NULL);
1599         switch (add_port_status) {
1600         case BT_SELF_COMPONENT_ADD_PORT_STATUS_ERROR:
1601                 ret = BT_COMPONENT_CLASS_INIT_METHOD_STATUS_ERROR;
1602                 goto error;
1603         case BT_SELF_COMPONENT_ADD_PORT_STATUS_MEMORY_ERROR:
1604                 ret = BT_COMPONENT_CLASS_INIT_METHOD_STATUS_MEMORY_ERROR;
1605                 goto error;
1606         default:
1607                 break;
1608         }
1609
1610         bt_self_component_set_data(self_comp, lttng_live);
1611         goto end;
1612
1613 error:
1614         lttng_live_component_destroy_data(lttng_live);
1615         lttng_live = NULL;
1616 end:
1617         return ret;
1618 }