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