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