projects / babeltrace.git / blame
? search:
summary | shortlog | log | commit | commitdiff | tree
blob | blame (incremental) | history | HEAD
Visibility hidden by default
[babeltrace.git] / src / plugins / ctf / lttng-live / lttng-live.cpp
CommitLineData
f3bc2010 1/*
0235b0db 2 * SPDX-License-Identifier: MIT
f3bc2010 3 *
14f28187 4 * Copyright 2019 Francis Deslauriers <francis.deslauriers@efficios.com>
f3bc2010 5 * Copyright 2016 Jérémie Galarneau <jeremie.galarneau@efficios.com>
7cdc2bab 6 * Copyright 2016 Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
f3bc2010 7 *
0235b0db 8 * Babeltrace CTF LTTng-live Client Component
f3bc2010
JG		 9 */
10
2ece7dd0 11#define BT_COMP_LOG_SELF_COMP self_comp
4164020e
SM		 12#define BT_LOG_OUTPUT_LEVEL log_level
13#define BT_LOG_TAG "PLUGIN/SRC.CTF.LTTNG-LIVE"
d9c39b0a 14#include "logging/comp-logging.h"
020bc26f 15
14f28187 16#include <inttypes.h>
c4f23e30 17#include <stdbool.h>
14f28187
FD		 18#include <unistd.h>
19
3c22a242
FD		 20#include <glib.h>
21
578e048b 22#include "common/assert.h"
3fadfbc0 23#include <babeltrace2/babeltrace.h>
578e048b 24#include "compat/compiler.h"
3fadfbc0 25#include <babeltrace2/types.h>
f3bc2010 26
376fc2bd 27#include "plugins/common/muxing/muxing.h"
80aff5ef 28#include "plugins/common/param-validation/param-validation.h"
376fc2bd 29
087cd0f5
SM		 30#include "data-stream.hpp"
31#include "metadata.hpp"
32#include "lttng-live.hpp"
7cdc2bab 33
4164020e
SM		 34#define MAX_QUERY_SIZE (256 * 1024)
35#define URL_PARAM "url"
36#define INPUTS_PARAM "inputs"
37#define SESS_NOT_FOUND_ACTION_PARAM "session-not-found-action"
38#define SESS_NOT_FOUND_ACTION_CONTINUE_STR "continue"
39#define SESS_NOT_FOUND_ACTION_FAIL_STR "fail"
40#define SESS_NOT_FOUND_ACTION_END_STR "end"
7cdc2bab 41
4164020e 42#define print_dbg(fmt, ...) BT_COMP_LOGD(fmt, ##__VA_ARGS__)
7cdc2bab 43
4164020e 44static const char *lttng_live_iterator_status_string(enum lttng_live_iterator_status status)
14f28187 45{
4164020e
SM		 46 switch (status) {
47 case LTTNG_LIVE_ITERATOR_STATUS_CONTINUE:
48 return "LTTNG_LIVE_ITERATOR_STATUS_CONTINUE";
49 case LTTNG_LIVE_ITERATOR_STATUS_AGAIN:
50 return "LTTNG_LIVE_ITERATOR_STATUS_AGAIN";
51 case LTTNG_LIVE_ITERATOR_STATUS_END:
52 return "LTTNG_LIVE_ITERATOR_STATUS_END";
53 case LTTNG_LIVE_ITERATOR_STATUS_OK:
54 return "LTTNG_LIVE_ITERATOR_STATUS_OK";
55 case LTTNG_LIVE_ITERATOR_STATUS_INVAL:
56 return "LTTNG_LIVE_ITERATOR_STATUS_INVAL";
57 case LTTNG_LIVE_ITERATOR_STATUS_ERROR:
58 return "LTTNG_LIVE_ITERATOR_STATUS_ERROR";
59 case LTTNG_LIVE_ITERATOR_STATUS_NOMEM:
60 return "LTTNG_LIVE_ITERATOR_STATUS_NOMEM";
61 case LTTNG_LIVE_ITERATOR_STATUS_UNSUPPORTED:
62 return "LTTNG_LIVE_ITERATOR_STATUS_UNSUPPORTED";
63 default:
64 bt_common_abort();
65 }
14f28187
FD		 66}
67
4164020e 68static const char *lttng_live_stream_state_string(enum lttng_live_stream_state state)
7cdc2bab 69{
4164020e
SM		 70 switch (state) {
71 case LTTNG_LIVE_STREAM_ACTIVE_NO_DATA:
72 return "ACTIVE_NO_DATA";
73 case LTTNG_LIVE_STREAM_QUIESCENT_NO_DATA:
74 return "QUIESCENT_NO_DATA";
75 case LTTNG_LIVE_STREAM_QUIESCENT:
76 return "QUIESCENT";
77 case LTTNG_LIVE_STREAM_ACTIVE_DATA:
78 return "ACTIVE_DATA";
79 case LTTNG_LIVE_STREAM_EOF:
80 return "EOF";
81 default:
82 return "ERROR";
83 }
7cdc2bab 84}
7cdc2bab 85
34533ae0 86void lttng_live_stream_iterator_set_state(struct lttng_live_stream_iterator *stream_iter,
4164020e 87 enum lttng_live_stream_state new_state)
34533ae0 88{
4164020e
SM		 89 bt_self_component *self_comp = stream_iter->self_comp;
90 bt_logging_level log_level = stream_iter->log_level;
34533ae0 91
4164020e
SM		 92 BT_COMP_LOGD("Setting live stream iterator state: viewer-stream-id=%" PRIu64
93 ", old-state=%s, new-state=%s",
94 stream_iter->viewer_stream_id, lttng_live_stream_state_string(stream_iter->state),
95 lttng_live_stream_state_string(new_state));
34533ae0 96
4164020e 97 stream_iter->state = new_state;
34533ae0
FD		 98}
99
4164020e
SM		 100#define LTTNG_LIVE_LOGD_STREAM_ITER(live_stream_iter) \
101 do { \
102 BT_COMP_LOGD("Live stream iterator state=%s, " \
103 "last-inact-ts-is-set=%d, last-inact-ts-value=%" PRId64 ", " \
104 "curr-inact-ts=%" PRId64, \
105 lttng_live_stream_state_string(live_stream_iter->state), \
106 live_stream_iter->last_inactivity_ts.is_set, \
107 live_stream_iter->last_inactivity_ts.value, \
108 live_stream_iter->current_inactivity_ts); \
109 } while (0);
6f79a7cf 110
9b4f9b42 111bool lttng_live_graph_is_canceled(struct lttng_live_msg_iter *msg_iter)
6f79a7cf 112{
4164020e 113 bool ret;
6f79a7cf 114
4164020e
SM		 115 if (!msg_iter) {
116 ret = false;
117 goto end;
118 }
7cdc2bab 119
4164020e 120 ret = bt_self_message_iterator_is_interrupted(msg_iter->self_msg_iter);
4bf0e537 121
14f28187 122end:
4164020e 123 return ret;
7cdc2bab
MD		 124}
125
4164020e
SM		 126static struct lttng_live_trace *
127lttng_live_session_borrow_trace_by_id(struct lttng_live_session *session, uint64_t trace_id)
d3e4dcd8 128{
4164020e
SM		 129 uint64_t trace_idx;
130 struct lttng_live_trace *ret_trace = NULL;
131
132 for (trace_idx = 0; trace_idx < session->traces->len; trace_idx++) {
133 struct lttng_live_trace *trace =
134 (lttng_live_trace *) g_ptr_array_index(session->traces, trace_idx);
135 if (trace->id == trace_id) {
136 ret_trace = trace;
137 goto end;
138 }
139 }
14f28187
FD		 140
141end:
4164020e 142 return ret_trace;
d3eb6e8f
PP		 143}
144
4164020e 145static void lttng_live_destroy_trace(struct lttng_live_trace *trace)
7cdc2bab 146{
4164020e
SM		 147 bt_logging_level log_level = trace->log_level;
148 bt_self_component *self_comp = trace->self_comp;
c01594de 149
4164020e 150 BT_COMP_LOGD("Destroying live trace: trace-id=%" PRIu64, trace->id);
7cdc2bab 151
4164020e
SM		 152 BT_ASSERT(trace->stream_iterators);
153 g_ptr_array_free(trace->stream_iterators, TRUE);
5bd230f4 154
4164020e
SM		 155 BT_TRACE_PUT_REF_AND_RESET(trace->trace);
156 BT_TRACE_CLASS_PUT_REF_AND_RESET(trace->trace_class);
5bd230f4 157
4164020e
SM		 158 lttng_live_metadata_fini(trace);
159 g_free(trace);
7cdc2bab
MD		 160}
161
4164020e
SM		 162static struct lttng_live_trace *lttng_live_create_trace(struct lttng_live_session *session,
163 uint64_t trace_id)
7cdc2bab 164{
4164020e
SM		 165 struct lttng_live_trace *trace = NULL;
166 bt_logging_level log_level = session->log_level;
167 bt_self_component *self_comp = session->self_comp;
168
169 BT_COMP_LOGD("Creating live trace: "
170 "session-id=%" PRIu64 ", trace-id=%" PRIu64,
171 session->id, trace_id);
172 trace = g_new0(struct lttng_live_trace, 1);
173 if (!trace) {
174 BT_COMP_LOGE_APPEND_CAUSE(self_comp, "Failed to allocate live trace");
175 goto error;
176 }
177 trace->log_level = session->log_level;
178 trace->self_comp = session->self_comp;
179 trace->session = session;
180 trace->id = trace_id;
181 trace->trace_class = NULL;
182 trace->trace = NULL;
183 trace->stream_iterators =
184 g_ptr_array_new_with_free_func((GDestroyNotify) lttng_live_stream_iterator_destroy);
185 BT_ASSERT(trace->stream_iterators);
186 trace->metadata_stream_state = LTTNG_LIVE_METADATA_STREAM_STATE_NEEDED;
187 g_ptr_array_add(session->traces, trace);
188
189 goto end;
7cdc2bab 190error:
4164020e
SM		 191 g_free(trace);
192 trace = NULL;
7cdc2bab 193end:
4164020e 194 return trace;
7cdc2bab
MD		 195}
196
4164020e
SM		 197struct lttng_live_trace *
198lttng_live_session_borrow_or_create_trace_by_id(struct lttng_live_session *session,
199 uint64_t trace_id)
7cdc2bab 200{
4164020e 201 struct lttng_live_trace *trace;
7cdc2bab 202
4164020e
SM		 203 trace = lttng_live_session_borrow_trace_by_id(session, trace_id);
204 if (trace) {
205 goto end;
206 }
7cdc2bab 207
4164020e
SM		 208 /* The session is the owner of the newly created trace. */
209 trace = lttng_live_create_trace(session, trace_id);
7cdc2bab 210
14f28187 211end:
4164020e 212 return trace;
7cdc2bab
MD		 213}
214
4164020e
SM		 215int lttng_live_add_session(struct lttng_live_msg_iter *lttng_live_msg_iter, uint64_t session_id,
216 const char *hostname, const char *session_name)
7cdc2bab 217{
4164020e
SM		 218 int ret = 0;
219 struct lttng_live_session *session;
220 bt_logging_level log_level = lttng_live_msg_iter->log_level;
221 bt_self_component *self_comp = lttng_live_msg_iter->self_comp;
222
223 BT_COMP_LOGD("Adding live session: "
224 "session-id=%" PRIu64 ", hostname=\"%s\" session-name=\"%s\"",
225 session_id, hostname, session_name);
226
227 session = g_new0(struct lttng_live_session, 1);
228 if (!session) {
229 BT_COMP_LOGE_APPEND_CAUSE(self_comp, "Failed to allocate live session");
230 goto error;
231 }
232
233 session->log_level = lttng_live_msg_iter->log_level;
234 session->self_comp = lttng_live_msg_iter->self_comp;
235 session->id = session_id;
236 session->traces = g_ptr_array_new_with_free_func((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 g_ptr_array_add(lttng_live_msg_iter->sessions, session);
247 goto end;
7cdc2bab 248error:
4164020e
SM		 249 g_free(session);
250 ret = -1;
7cdc2bab 251end:
4164020e 252 return ret;
7cdc2bab
MD		 253}
254
4164020e 255static void lttng_live_destroy_session(struct lttng_live_session *session)
7cdc2bab 256{
4164020e
SM		 257 bt_logging_level log_level;
258 bt_self_component *self_comp;
259
260 if (!session) {
261 goto end;
262 }
263
264 log_level = session->log_level;
265 self_comp = session->self_comp;
266 BT_COMP_LOGD("Destroying live session: "
267 "session-id=%" PRIu64 ", session-name=\"%s\"",
268 session->id, session->session_name->str);
269 if (session->id != -1ULL) {
270 if (lttng_live_session_detach(session)) {
271 if (!lttng_live_graph_is_canceled(session->lttng_live_msg_iter)) {
272 /* Old relayd cannot detach sessions. */
273 BT_COMP_LOGD("Unable to detach lttng live session %" PRIu64, session->id);
274 }
275 }
276 session->id = -1ULL;
277 }
278
279 if (session->traces) {
280 g_ptr_array_free(session->traces, TRUE);
281 }
282
283 if (session->hostname) {
284 g_string_free(session->hostname, TRUE);
285 }
286 if (session->session_name) {
287 g_string_free(session->session_name, TRUE);
288 }
289 g_free(session);
14f28187
FD		 290
291end:
4164020e 292 return;
7cdc2bab
MD		 293}
294
4164020e 295static void lttng_live_msg_iter_destroy(struct lttng_live_msg_iter *lttng_live_msg_iter)
7cdc2bab 296{
4164020e
SM		 297 if (!lttng_live_msg_iter) {
298 goto end;
299 }
7cdc2bab 300
4164020e
SM		 301 if (lttng_live_msg_iter->sessions) {
302 g_ptr_array_free(lttng_live_msg_iter->sessions, TRUE);
303 }
14f28187 304
4164020e
SM		 305 if (lttng_live_msg_iter->viewer_connection) {
306 live_viewer_connection_destroy(lttng_live_msg_iter->viewer_connection);
307 }
308 BT_ASSERT(lttng_live_msg_iter->lttng_live_comp);
309 BT_ASSERT(lttng_live_msg_iter->lttng_live_comp->has_msg_iter);
14f28187 310
4164020e
SM		 311 /* All stream iterators must be destroyed at this point. */
312 BT_ASSERT(lttng_live_msg_iter->active_stream_iter == 0);
313 lttng_live_msg_iter->lttng_live_comp->has_msg_iter = false;
14f28187 314
4164020e 315 g_free(lttng_live_msg_iter);
14f28187
FD		 316
317end:
4164020e 318 return;
14f28187
FD		 319}
320
14f28187
FD		 321void lttng_live_msg_iter_finalize(bt_self_message_iterator *self_msg_iter)
322{
4164020e 323 struct lttng_live_msg_iter *lttng_live_msg_iter;
14f28187 324
4164020e 325 BT_ASSERT(self_msg_iter);
14f28187 326
4164020e
SM		 327 lttng_live_msg_iter =
328 (struct lttng_live_msg_iter *) bt_self_message_iterator_get_data(self_msg_iter);
329 BT_ASSERT(lttng_live_msg_iter);
330 lttng_live_msg_iter_destroy(lttng_live_msg_iter);
7cdc2bab
MD		 331}
332
4164020e
SM		 333static enum lttng_live_iterator_status
334lttng_live_iterator_next_check_stream_state(struct lttng_live_stream_iterator *lttng_live_stream)
7cdc2bab 335{
4164020e
SM		 336 bt_logging_level log_level = lttng_live_stream->log_level;
337 bt_self_component *self_comp = lttng_live_stream->self_comp;
338
339 switch (lttng_live_stream->state) {
340 case LTTNG_LIVE_STREAM_QUIESCENT:
341 case LTTNG_LIVE_STREAM_ACTIVE_DATA:
342 break;
343 case LTTNG_LIVE_STREAM_ACTIVE_NO_DATA:
344 /* Invalid state. */
345 BT_COMP_LOGF("Unexpected stream state \"ACTIVE_NO_DATA\"");
346 bt_common_abort();
347 case LTTNG_LIVE_STREAM_QUIESCENT_NO_DATA:
348 /* Invalid state. */
349 BT_COMP_LOGF("Unexpected stream state \"QUIESCENT_NO_DATA\"");
350 bt_common_abort();
351 case LTTNG_LIVE_STREAM_EOF:
352 break;
353 }
354 return LTTNG_LIVE_ITERATOR_STATUS_OK;
7cdc2bab
MD		 355}
356
357/*
f93afbf9
FD		 358 * For active no data stream, fetch next index. As a result of that it can
359 * become either:
360 * - quiescent: won't have events for a bit,
361 * - have data: need to get that data and produce the event,
362 * - have no data on this stream at this point: need to retry (AGAIN) or return
363 * EOF.
7cdc2bab 364 */
4164020e
SM		 365static enum lttng_live_iterator_status lttng_live_iterator_next_handle_one_no_data_stream(
366 struct lttng_live_msg_iter *lttng_live_msg_iter,
367 struct lttng_live_stream_iterator *lttng_live_stream)
7cdc2bab 368{
4164020e
SM		 369 bt_logging_level log_level = lttng_live_msg_iter->log_level;
370 bt_self_component *self_comp = lttng_live_msg_iter->self_comp;
371 enum lttng_live_iterator_status ret = LTTNG_LIVE_ITERATOR_STATUS_OK;
372 enum lttng_live_stream_state orig_state = lttng_live_stream->state;
373 struct packet_index index;
374
375 if (lttng_live_stream->trace->metadata_stream_state ==
376 LTTNG_LIVE_METADATA_STREAM_STATE_NEEDED) {
377 BT_COMP_LOGD(
378 "Need to get an update for the metadata stream before proceeding further with this stream: "
379 "stream-name=\"%s\"",
380 lttng_live_stream->name->str);
381 ret = LTTNG_LIVE_ITERATOR_STATUS_CONTINUE;
382 goto end;
383 }
384
385 if (lttng_live_stream->trace->session->new_streams_needed) {
386 BT_COMP_LOGD(
387 "Need to get an update of all streams before proceeding further with this stream: "
388 "stream-name=\"%s\"",
389 lttng_live_stream->name->str);
390 ret = LTTNG_LIVE_ITERATOR_STATUS_CONTINUE;
391 goto end;
392 }
393
394 if (lttng_live_stream->state != LTTNG_LIVE_STREAM_ACTIVE_NO_DATA &&
395 lttng_live_stream->state != LTTNG_LIVE_STREAM_QUIESCENT_NO_DATA) {
396 goto end;
397 }
398 ret = lttng_live_get_next_index(lttng_live_msg_iter, lttng_live_stream, &index);
399 if (ret != LTTNG_LIVE_ITERATOR_STATUS_OK) {
400 goto end;
401 }
402
403 BT_ASSERT_DBG(lttng_live_stream->state != LTTNG_LIVE_STREAM_EOF);
404
405 if (lttng_live_stream->state == LTTNG_LIVE_STREAM_QUIESCENT) {
406 uint64_t last_inact_ts = lttng_live_stream->last_inactivity_ts.value,
407 curr_inact_ts = lttng_live_stream->current_inactivity_ts;
408
409 if (orig_state == LTTNG_LIVE_STREAM_QUIESCENT_NO_DATA && last_inact_ts == curr_inact_ts) {
410 /*
5cb27175
JG		 411 * Because the stream is in the QUIESCENT_NO_DATA
412 * state, we can assert that the last_inactivity_ts was
413 * set and can be safely used in the `if` above.
414 */
4164020e
SM		 415 BT_ASSERT(lttng_live_stream->last_inactivity_ts.is_set);
416
417 ret = LTTNG_LIVE_ITERATOR_STATUS_AGAIN;
418 LTTNG_LIVE_LOGD_STREAM_ITER(lttng_live_stream);
419 } else {
420 ret = LTTNG_LIVE_ITERATOR_STATUS_CONTINUE;
421 }
422 goto end;
423 }
424
425 lttng_live_stream->base_offset = index.offset;
426 lttng_live_stream->offset = index.offset;
427 lttng_live_stream->len = index.packet_size / CHAR_BIT;
428
429 BT_COMP_LOGD("Setting live stream reading info: stream-name=\"%s\", "
430 "viewer-stream-id=%" PRIu64 ", stream-base-offset=%" PRIu64
431 ", stream-offset=%" PRIu64 ", stream-len=%" PRIu64,
432 lttng_live_stream->name->str, lttng_live_stream->viewer_stream_id,
433 lttng_live_stream->base_offset, lttng_live_stream->offset, lttng_live_stream->len);
f93afbf9 434
7cdc2bab 435end:
4164020e
SM		 436 if (ret == LTTNG_LIVE_ITERATOR_STATUS_OK) {
437 ret = lttng_live_iterator_next_check_stream_state(lttng_live_stream);
438 }
439 return ret;
7cdc2bab
MD		 440}
441
442/*
14f28187
FD		 443 * Creation of the message requires the ctf trace class to be created
444 * beforehand, but the live protocol gives us all streams (including metadata)
445 * at once. So we split it in three steps: getting streams, getting metadata
446 * (which creates the ctf trace class), and then creating the per-stream
447 * messages.
7cdc2bab 448 */
4164020e
SM		 449static enum lttng_live_iterator_status
450lttng_live_get_session(struct lttng_live_msg_iter *lttng_live_msg_iter,
451 struct lttng_live_session *session)
7cdc2bab 452{
4164020e
SM		 453 bt_logging_level log_level = lttng_live_msg_iter->log_level;
454 bt_self_component *self_comp = lttng_live_msg_iter->self_comp;
455 enum lttng_live_iterator_status status;
456 uint64_t trace_idx;
457
458 if (!session->attached) {
459 BT_COMP_LOGD("Attach to session: session-id=%" PRIu64, session->id);
460 enum lttng_live_viewer_status attach_status =
461 lttng_live_session_attach(session, lttng_live_msg_iter->self_msg_iter);
462 if (attach_status != LTTNG_LIVE_VIEWER_STATUS_OK) {
463 if (lttng_live_graph_is_canceled(lttng_live_msg_iter)) {
464 /*
465 * Clear any causes appended in
466 * `lttng_live_attach_session()` as we want to
467 * return gracefully since the graph was
468 * cancelled.
469 */
470 bt_current_thread_clear_error();
471 status = LTTNG_LIVE_ITERATOR_STATUS_AGAIN;
472 } else {
473 status = LTTNG_LIVE_ITERATOR_STATUS_ERROR;
474 BT_COMP_LOGE_APPEND_CAUSE(self_comp, "Error attaching to LTTng live session");
475 }
476 goto end;
477 }
478 }
479
a4f118a3 480 BT_COMP_LOGD("Updating all data streams: "
4164020e
SM		 481 "session-id=%" PRIu64 ", session-name=\"%s\"",
482 session->id, session->session_name->str);
483
484 status = lttng_live_session_get_new_streams(session, lttng_live_msg_iter->self_msg_iter);
a4f118a3
FD		 485 switch (status) {
486 case LTTNG_LIVE_ITERATOR_STATUS_OK:
487 break;
488 case LTTNG_LIVE_ITERATOR_STATUS_END:
489 /*
490 * We received a `_END` from the `_get_new_streams()` function,
491 * which means no more data will ever be received from the data
492 * streams of this session. But it's possible that the metadata
493 * is incomplete.
494 * The live protocol guarantees that we receive all the
495 * metadata needed before we receive data streams needing it.
496 * But it's possible to receive metadata NOT needed by
497 * data streams after the session was closed. For example, this
498 * could happen if a new event is registered and the session is
499 * stopped before any tracepoint for that event is actually
500 * fired.
501 */
502 BT_COMP_LOGD(
503 "Updating streams returned _END status. Override status to _OK in order fetch any remaining metadata:"
504 "session-id=%" PRIu64 ", session-name=\"%s\"",
505 session->id, session->session_name->str);
506 status = LTTNG_LIVE_ITERATOR_STATUS_OK;
507 break;
508 default:
4164020e
SM		 509 goto end;
510 }
511
a4f118a3
FD		 512 BT_COMP_LOGD("Updating metadata stream for session: "
513 "session-id=%" PRIu64 ", session-name=\"%s\"",
514 session->id, session->session_name->str);
515
4164020e
SM		 516 trace_idx = 0;
517 while (trace_idx < session->traces->len) {
518 struct lttng_live_trace *trace =
519 (lttng_live_trace *) g_ptr_array_index(session->traces, trace_idx);
520
521 status = lttng_live_metadata_update(trace);
522 switch (status) {
523 case LTTNG_LIVE_ITERATOR_STATUS_END:
524 case LTTNG_LIVE_ITERATOR_STATUS_OK:
525 trace_idx++;
526 break;
527 case LTTNG_LIVE_ITERATOR_STATUS_CONTINUE:
528 case LTTNG_LIVE_ITERATOR_STATUS_AGAIN:
529 goto end;
530 default:
531 BT_COMP_LOGE_APPEND_CAUSE(self_comp,
532 "Error updating trace metadata: "
533 "stream-iter-status=%s, trace-id=%" PRIu64,
534 lttng_live_iterator_status_string(status), trace->id);
535 goto end;
536 }
537 }
538
539 /*
540 * Now that we have the metadata we can initialize the downstream
541 * iterator.
542 */
543 status = lttng_live_lazy_msg_init(session, lttng_live_msg_iter->self_msg_iter);
14f28187
FD		 544
545end:
4164020e 546 return status;
7cdc2bab
MD		 547}
548
4164020e
SM		 549static void
550lttng_live_force_new_streams_and_metadata(struct lttng_live_msg_iter *lttng_live_msg_iter)
7cdc2bab 551{
4164020e
SM		 552 uint64_t session_idx, trace_idx;
553 bt_logging_level log_level = lttng_live_msg_iter->log_level;
554 bt_self_component *self_comp = lttng_live_msg_iter->self_comp;
555
556 for (session_idx = 0; session_idx < lttng_live_msg_iter->sessions->len; session_idx++) {
557 struct lttng_live_session *session =
558 (lttng_live_session *) g_ptr_array_index(lttng_live_msg_iter->sessions, session_idx);
559 BT_COMP_LOGD("Force marking session as needing new streams: "
560 "session-id=%" PRIu64,
561 session->id);
562 session->new_streams_needed = true;
563 for (trace_idx = 0; trace_idx < session->traces->len; trace_idx++) {
564 struct lttng_live_trace *trace =
565 (lttng_live_trace *) g_ptr_array_index(session->traces, trace_idx);
566 BT_COMP_LOGD("Force marking trace metadata state as needing an update: "
567 "session-id=%" PRIu64 ", trace-id=%" PRIu64,
568 session->id, trace->id);
569
570 BT_ASSERT(trace->metadata_stream_state != LTTNG_LIVE_METADATA_STREAM_STATE_CLOSED);
571
572 trace->metadata_stream_state = LTTNG_LIVE_METADATA_STREAM_STATE_NEEDED;
573 }
574 }
7cdc2bab
MD		 575}
576
4164020e
SM		 577static enum lttng_live_iterator_status
578lttng_live_iterator_handle_new_streams_and_metadata(struct lttng_live_msg_iter *lttng_live_msg_iter)
7cdc2bab 579{
4164020e
SM		 580 enum lttng_live_iterator_status status;
581 enum lttng_live_viewer_status viewer_status;
582 bt_logging_level log_level = lttng_live_msg_iter->log_level;
583 bt_self_component *self_comp = lttng_live_msg_iter->self_comp;
584 uint64_t session_idx = 0, nr_sessions_opened = 0;
585 struct lttng_live_session *session;
586 enum session_not_found_action sess_not_found_act =
587 lttng_live_msg_iter->lttng_live_comp->params.sess_not_found_act;
588
589 BT_COMP_LOGD("Update data and metadata of all sessions: "
590 "live-msg-iter-addr=%p",
591 lttng_live_msg_iter);
592 /*
593 * In a remotely distant future, we could add a "new
594 * session" flag to the protocol, which would tell us that we
595 * need to query for new sessions even though we have sessions
596 * currently ongoing.
597 */
598 if (lttng_live_msg_iter->sessions->len == 0) {
599 if (sess_not_found_act != SESSION_NOT_FOUND_ACTION_CONTINUE) {
600 BT_COMP_LOGD(
601 "No session found. Exiting in accordance with the `session-not-found-action` parameter");
602 status = LTTNG_LIVE_ITERATOR_STATUS_END;
603 goto end;
604 } else {
605 BT_COMP_LOGD(
606 "No session found. Try creating a new one in accordance with the `session-not-found-action` parameter");
607 /*
608 * Retry to create a viewer session for the requested
609 * session name.
610 */
611 viewer_status = lttng_live_create_viewer_session(lttng_live_msg_iter);
612 if (viewer_status != LTTNG_LIVE_VIEWER_STATUS_OK) {
613 if (viewer_status == LTTNG_LIVE_VIEWER_STATUS_ERROR) {
614 status = LTTNG_LIVE_ITERATOR_STATUS_ERROR;
615 BT_COMP_LOGE_APPEND_CAUSE(self_comp,
616 "Error creating LTTng live viewer session");
617 } else if (viewer_status == LTTNG_LIVE_VIEWER_STATUS_INTERRUPTED) {
618 status = LTTNG_LIVE_ITERATOR_STATUS_AGAIN;
619 } else {
620 bt_common_abort();
621 }
622 goto end;
623 }
624 }
625 }
626
627 for (session_idx = 0; session_idx < lttng_live_msg_iter->sessions->len; session_idx++) {
628 session =
629 (lttng_live_session *) g_ptr_array_index(lttng_live_msg_iter->sessions, session_idx);
630 status = lttng_live_get_session(lttng_live_msg_iter, session);
631 switch (status) {
632 case LTTNG_LIVE_ITERATOR_STATUS_OK:
4164020e 633 case LTTNG_LIVE_ITERATOR_STATUS_END:
a4f118a3
FD		 634 /*
635 * A session returned `_END`. Other sessions may still
636 * be active so we override the status and continue
637 * looping if needed.
638 */
4164020e
SM		 639 break;
640 default:
641 goto end;
642 }
643 if (!session->closed) {
644 nr_sessions_opened++;
645 }
646 }
647
648 if (sess_not_found_act != SESSION_NOT_FOUND_ACTION_CONTINUE && nr_sessions_opened == 0) {
649 status = LTTNG_LIVE_ITERATOR_STATUS_END;
650 } else {
651 status = LTTNG_LIVE_ITERATOR_STATUS_OK;
652 }
f79c2d7a
FD		 653
654end:
4164020e 655 return status;
7cdc2bab
MD		 656}
657
4164020e
SM		 658static enum lttng_live_iterator_status
659emit_inactivity_message(struct lttng_live_msg_iter *lttng_live_msg_iter,
660 struct lttng_live_stream_iterator *stream_iter, const bt_message **message,
661 uint64_t timestamp)
7cdc2bab 662{
4164020e
SM		 663 enum lttng_live_iterator_status ret = LTTNG_LIVE_ITERATOR_STATUS_OK;
664 bt_logging_level log_level = lttng_live_msg_iter->log_level;
665 bt_self_component *self_comp = lttng_live_msg_iter->self_comp;
666 bt_message *msg = NULL;
667
668 BT_ASSERT(stream_iter->trace->clock_class);
669
670 BT_COMP_LOGD("Emitting inactivity message for stream: ctf-stream-id=%" PRIu64
671 ", viewer-stream-id=%" PRIu64 ", timestamp=%" PRIu64,
672 stream_iter->ctf_stream_class_id.value, stream_iter->viewer_stream_id, timestamp);
673
674 msg = bt_message_message_iterator_inactivity_create(lttng_live_msg_iter->self_msg_iter,
675 stream_iter->trace->clock_class, timestamp);
676 if (!msg) {
677 BT_COMP_LOGE_APPEND_CAUSE(self_comp, "Error emitting message iterator inactivity message");
678 goto error;
679 }
680
681 *message = msg;
7cdc2bab 682end:
4164020e 683 return ret;
7cdc2bab
MD		 684
685error:
4164020e
SM		 686 ret = LTTNG_LIVE_ITERATOR_STATUS_ERROR;
687 bt_message_put_ref(msg);
688 goto end;
7cdc2bab
MD		 689}
690
4164020e
SM		 691static enum lttng_live_iterator_status lttng_live_iterator_next_handle_one_quiescent_stream(
692 struct lttng_live_msg_iter *lttng_live_msg_iter,
693 struct lttng_live_stream_iterator *lttng_live_stream, const bt_message **message)
7cdc2bab 694{
4164020e
SM		 695 enum lttng_live_iterator_status ret = LTTNG_LIVE_ITERATOR_STATUS_OK;
696
697 if (lttng_live_stream->state != LTTNG_LIVE_STREAM_QUIESCENT) {
698 return LTTNG_LIVE_ITERATOR_STATUS_OK;
699 }
700
701 /*
702 * Check if we already sent an inactivty message downstream for this
703 * `current_inactivity_ts` value.
704 */
705 if (lttng_live_stream->last_inactivity_ts.is_set &&
706 lttng_live_stream->current_inactivity_ts == lttng_live_stream->last_inactivity_ts.value) {
707 lttng_live_stream_iterator_set_state(lttng_live_stream,
708 LTTNG_LIVE_STREAM_QUIESCENT_NO_DATA);
709
710 ret = LTTNG_LIVE_ITERATOR_STATUS_CONTINUE;
711 goto end;
712 }
713
714 ret = emit_inactivity_message(lttng_live_msg_iter, lttng_live_stream, message,
715 lttng_live_stream->current_inactivity_ts);
716
717 lttng_live_stream->last_inactivity_ts.value = lttng_live_stream->current_inactivity_ts;
718 lttng_live_stream->last_inactivity_ts.is_set = true;
14f28187 719end:
4164020e 720 return ret;
14f28187
FD		 721}
722
4164020e
SM		 723static int live_get_msg_ts_ns(struct lttng_live_stream_iterator *stream_iter,
724 struct lttng_live_msg_iter *lttng_live_msg_iter,
725 const bt_message *msg, int64_t last_msg_ts_ns, int64_t *ts_ns)
14f28187 726{
4164020e
SM		 727 const bt_clock_snapshot *clock_snapshot = NULL;
728 int ret = 0;
729 bt_logging_level log_level = lttng_live_msg_iter->log_level;
730 bt_self_component *self_comp = lttng_live_msg_iter->self_comp;
731
732 BT_ASSERT_DBG(msg);
733 BT_ASSERT_DBG(ts_ns);
734
735 BT_COMP_LOGD("Getting message's timestamp: iter-data-addr=%p, msg-addr=%p, "
736 "last-msg-ts=%" PRId64,
737 lttng_live_msg_iter, msg, last_msg_ts_ns);
738
739 switch (bt_message_get_type(msg)) {
740 case BT_MESSAGE_TYPE_EVENT:
4164020e
SM		 741 clock_snapshot = bt_message_event_borrow_default_clock_snapshot_const(msg);
742 break;
743 case BT_MESSAGE_TYPE_PACKET_BEGINNING:
4164020e
SM		 744 clock_snapshot = bt_message_packet_beginning_borrow_default_clock_snapshot_const(msg);
745 break;
746 case BT_MESSAGE_TYPE_PACKET_END:
4164020e
SM		 747 clock_snapshot = bt_message_packet_end_borrow_default_clock_snapshot_const(msg);
748 break;
749 case BT_MESSAGE_TYPE_DISCARDED_EVENTS:
4164020e
SM		 750 clock_snapshot =
751 bt_message_discarded_events_borrow_beginning_default_clock_snapshot_const(msg);
752 break;
753 case BT_MESSAGE_TYPE_DISCARDED_PACKETS:
4164020e
SM		 754 clock_snapshot =
755 bt_message_discarded_packets_borrow_beginning_default_clock_snapshot_const(msg);
756 break;
757 case BT_MESSAGE_TYPE_MESSAGE_ITERATOR_INACTIVITY:
758 clock_snapshot = bt_message_message_iterator_inactivity_borrow_clock_snapshot_const(msg);
759 break;
760 default:
761 /* All the other messages have a higher priority */
762 BT_COMP_LOGD_STR("Message has no timestamp: using the last message timestamp.");
763 *ts_ns = last_msg_ts_ns;
764 goto end;
765 }
766
4164020e
SM		 767 ret = bt_clock_snapshot_get_ns_from_origin(clock_snapshot, ts_ns);
768 if (ret) {
769 BT_COMP_LOGE_APPEND_CAUSE(self_comp,
770 "Cannot get nanoseconds from Epoch of clock snapshot: "
771 "clock-snapshot-addr=%p",
772 clock_snapshot);
773 goto error;
774 }
775
776 goto end;
14f28187 777
14f28187 778error:
4164020e 779 ret = -1;
7cdc2bab 780
7cdc2bab 781end:
4164020e
SM		 782 if (ret == 0) {
783 BT_COMP_LOGD("Found message's timestamp: "
784 "iter-data-addr=%p, msg-addr=%p, "
785 "last-msg-ts=%" PRId64 ", ts=%" PRId64,
786 lttng_live_msg_iter, msg, last_msg_ts_ns, *ts_ns);
787 }
788
789 return ret;
7cdc2bab
MD		 790}
791
4164020e
SM		 792static enum lttng_live_iterator_status lttng_live_iterator_next_handle_one_active_data_stream(
793 struct lttng_live_msg_iter *lttng_live_msg_iter,
794 struct lttng_live_stream_iterator *lttng_live_stream, const bt_message **message)
7cdc2bab 795{
4164020e
SM		 796 enum lttng_live_iterator_status ret = LTTNG_LIVE_ITERATOR_STATUS_OK;
797 bt_logging_level log_level = lttng_live_msg_iter->log_level;
798 bt_self_component *self_comp = lttng_live_msg_iter->self_comp;
799 enum ctf_msg_iter_status status;
800 uint64_t session_idx, trace_idx;
801
802 for (session_idx = 0; session_idx < lttng_live_msg_iter->sessions->len; session_idx++) {
803 struct lttng_live_session *session =
804 (lttng_live_session *) g_ptr_array_index(lttng_live_msg_iter->sessions, session_idx);
805
806 if (session->new_streams_needed) {
807 BT_COMP_LOGD("Need an update for streams: "
808 "session-id=%" PRIu64,
809 session->id);
810 ret = LTTNG_LIVE_ITERATOR_STATUS_CONTINUE;
811 goto end;
812 }
813 for (trace_idx = 0; trace_idx < session->traces->len; trace_idx++) {
814 struct lttng_live_trace *trace =
815 (lttng_live_trace *) g_ptr_array_index(session->traces, trace_idx);
816 if (trace->metadata_stream_state == LTTNG_LIVE_METADATA_STREAM_STATE_NEEDED) {
817 BT_COMP_LOGD("Need an update for metadata stream: "
818 "session-id=%" PRIu64 ", trace-id=%" PRIu64,
819 session->id, trace->id);
820 ret = LTTNG_LIVE_ITERATOR_STATUS_CONTINUE;
821 goto end;
822 }
823 }
824 }
825
826 if (lttng_live_stream->state != LTTNG_LIVE_STREAM_ACTIVE_DATA) {
827 ret = LTTNG_LIVE_ITERATOR_STATUS_ERROR;
828 BT_COMP_LOGE_APPEND_CAUSE(self_comp,
829 "Invalid state of live stream iterator"
830 "stream-iter-status=%s",
831 lttng_live_stream_state_string(lttng_live_stream->state));
832 goto end;
833 }
834
835 status = ctf_msg_iter_get_next_message(lttng_live_stream->msg_iter, message);
836 switch (status) {
837 case CTF_MSG_ITER_STATUS_EOF:
838 ret = LTTNG_LIVE_ITERATOR_STATUS_END;
839 break;
840 case CTF_MSG_ITER_STATUS_OK:
841 ret = LTTNG_LIVE_ITERATOR_STATUS_OK;
842 break;
843 case CTF_MSG_ITER_STATUS_AGAIN:
844 /*
845 * Continue immediately (end of packet). The next
846 * get_index may return AGAIN to delay the following
847 * attempt.
848 */
849 ret = LTTNG_LIVE_ITERATOR_STATUS_CONTINUE;
850 break;
851 case CTF_MSG_ITER_STATUS_ERROR:
852 default:
853 ret = LTTNG_LIVE_ITERATOR_STATUS_ERROR;
854 BT_COMP_LOGE_APPEND_CAUSE(self_comp,
855 "CTF message iterator failed to get next message: "
856 "msg-iter=%p, msg-iter-status=%s",
857 lttng_live_stream->msg_iter, ctf_msg_iter_status_string(status));
858 break;
859 }
14f28187
FD		 860
861end:
4164020e 862 return ret;
7cdc2bab
MD		 863}
864
4164020e
SM		 865static enum lttng_live_iterator_status
866lttng_live_iterator_close_stream(struct lttng_live_msg_iter *lttng_live_msg_iter,
867 struct lttng_live_stream_iterator *stream_iter,
868 const bt_message **curr_msg)
4a39caef 869{
4164020e
SM		 870 enum lttng_live_iterator_status live_status = LTTNG_LIVE_ITERATOR_STATUS_OK;
871 bt_logging_level log_level = lttng_live_msg_iter->log_level;
872 bt_self_component *self_comp = lttng_live_msg_iter->self_comp;
873
874 BT_COMP_LOGD("Closing live stream iterator: stream-name=\"%s\", "
875 "viewer-stream-id=%" PRIu64,
876 stream_iter->name->str, stream_iter->viewer_stream_id);
877
878 /*
879 * The viewer has hung up on us so we are closing the stream. The
880 * `ctf_msg_iter` should simply realize that it needs to close the
881 * stream properly by emitting the necessary stream end message.
882 */
883 enum ctf_msg_iter_status status =
884 ctf_msg_iter_get_next_message(stream_iter->msg_iter, curr_msg);
885
886 if (status == CTF_MSG_ITER_STATUS_ERROR) {
887 BT_COMP_LOGE_APPEND_CAUSE(self_comp,
888 "Error getting the next message from CTF message iterator");
889 live_status = LTTNG_LIVE_ITERATOR_STATUS_ERROR;
890 goto end;
891 } else if (status == CTF_MSG_ITER_STATUS_EOF) {
892 BT_COMP_LOGI("Reached the end of the live stream iterator.");
893 live_status = LTTNG_LIVE_ITERATOR_STATUS_END;
894 goto end;
895 }
896
897 BT_ASSERT(status == CTF_MSG_ITER_STATUS_OK);
4a39caef
FD		 898
899end:
4164020e 900 return live_status;
4a39caef
FD		 901}
902
7cdc2bab
MD		 903/*
904 * helper function:
905 * handle_no_data_streams()
906 * retry:
907 * - for each ACTIVE_NO_DATA stream:
908 * - query relayd for stream data, or quiescence info.
909 * - if need metadata, get metadata, goto retry.
910 * - if new stream, get new stream as ACTIVE_NO_DATA, goto retry
911 * - if quiescent, move to QUIESCENT streams
912 * - if fetched data, move to ACTIVE_DATA streams
913 * (at this point each stream either has data, or is quiescent)
914 *
915 *
916 * iterator_next:
917 * handle_new_streams_and_metadata()
918 * - query relayd for known streams, add them as ACTIVE_NO_DATA
919 * - query relayd for metadata
920 *
921 * call handle_active_no_data_streams()
922 *
923 * handle_quiescent_streams()
924 * - if at least one stream is ACTIVE_DATA:
925 * - peek stream event with lowest timestamp -> next_ts
926 * - for each quiescent stream
927 * - if next_ts >= quiescent end
928 * - set state to ACTIVE_NO_DATA
929 * - else
930 * - for each quiescent stream
931 * - set state to ACTIVE_NO_DATA
932 *
933 * call handle_active_no_data_streams()
934 *
935 * handle_active_data_streams()
936 * - if at least one stream is ACTIVE_DATA:
937 * - get stream event with lowest timestamp from heap
d6e69534 938 * - make that stream event the current message.
7cdc2bab
MD		 939 * - move this stream heap position to its next event
940 * - if we need to fetch data from relayd, move
941 * stream to ACTIVE_NO_DATA.
942 * - return OK
943 * - return AGAIN
944 *
945 * end criterion: ctrl-c on client. If relayd exits or the session
946 * closes on the relay daemon side, we keep on waiting for streams.
947 * Eventually handle --end timestamp (also an end criterion).
948 *
949 * When disconnected from relayd: try to re-connect endlessly.
950 */
4164020e
SM		 951static enum lttng_live_iterator_status
952lttng_live_iterator_next_msg_on_stream(struct lttng_live_msg_iter *lttng_live_msg_iter,
953 struct lttng_live_stream_iterator *stream_iter,
954 const bt_message **curr_msg)
7cdc2bab 955{
4164020e
SM		 956 bt_logging_level log_level = lttng_live_msg_iter->log_level;
957 bt_self_component *self_comp = lttng_live_msg_iter->self_comp;
958 enum lttng_live_iterator_status live_status;
959
960 BT_COMP_LOGD("Advancing live stream iterator until next message if possible: "
961 "stream-name=\"%s\", viewer-stream-id=%" PRIu64,
962 stream_iter->name->str, stream_iter->viewer_stream_id);
963
964 if (stream_iter->has_stream_hung_up) {
965 /*
966 * The stream has hung up and the stream was properly closed
967 * during the last call to the current function. Return _END
968 * status now so that this stream iterator is removed for the
969 * stream iterator list.
970 */
971 live_status = LTTNG_LIVE_ITERATOR_STATUS_END;
972 goto end;
973 }
4a39caef 974
7cdc2bab 975retry:
4164020e
SM		 976 LTTNG_LIVE_LOGD_STREAM_ITER(stream_iter);
977
978 /*
979 * Make sure we have the most recent metadata and possibly some new
980 * streams.
981 */
982 live_status = lttng_live_iterator_handle_new_streams_and_metadata(lttng_live_msg_iter);
983 if (live_status != LTTNG_LIVE_ITERATOR_STATUS_OK) {
984 goto end;
985 }
986
987 live_status =
988 lttng_live_iterator_next_handle_one_no_data_stream(lttng_live_msg_iter, stream_iter);
989 if (live_status != LTTNG_LIVE_ITERATOR_STATUS_OK) {
990 if (live_status == LTTNG_LIVE_ITERATOR_STATUS_END) {
991 /*
992 * We overwrite `live_status` since `curr_msg` is
993 * likely set to a valid message in this function.
994 */
995 live_status =
996 lttng_live_iterator_close_stream(lttng_live_msg_iter, stream_iter, curr_msg);
997 }
998 goto end;
999 }
1000
1001 live_status = lttng_live_iterator_next_handle_one_quiescent_stream(lttng_live_msg_iter,
1002 stream_iter, curr_msg);
1003 if (live_status != LTTNG_LIVE_ITERATOR_STATUS_OK) {
1004 BT_ASSERT(!*curr_msg);
1005 goto end;
1006 }
1007 if (*curr_msg) {
1008 goto end;
1009 }
1010 live_status = lttng_live_iterator_next_handle_one_active_data_stream(lttng_live_msg_iter,
1011 stream_iter, curr_msg);
1012 if (live_status != LTTNG_LIVE_ITERATOR_STATUS_OK) {
1013 BT_ASSERT(!*curr_msg);
1014 }
7cdc2bab
MD		 1015
1016end:
4164020e
SM		 1017 if (live_status == LTTNG_LIVE_ITERATOR_STATUS_CONTINUE) {
1018 BT_COMP_LOGD("Ask the relay daemon for an updated view of the data and metadata streams");
1019 goto retry;
1020 }
14f28187 1021
4164020e
SM		 1022 BT_COMP_LOGD("Returning from advancing live stream iterator: status=%s, "
1023 "stream-name=\"%s\", viewer-stream-id=%" PRIu64,
1024 lttng_live_iterator_status_string(live_status), stream_iter->name->str,
1025 stream_iter->viewer_stream_id);
f93afbf9 1026
4164020e 1027 return live_status;
7cdc2bab
MD		 1028}
1029
4164020e 1030static bool is_discarded_packet_or_event_message(const bt_message *msg)
8ec4d5ff 1031{
4164020e 1032 const enum bt_message_type msg_type = bt_message_get_type(msg);
8ec4d5ff 1033
4164020e
SM		 1034 return msg_type == BT_MESSAGE_TYPE_DISCARDED_EVENTS ||
1035 msg_type == BT_MESSAGE_TYPE_DISCARDED_PACKETS;
8ec4d5ff
JG		 1036}
1037
4164020e
SM		 1038static enum lttng_live_iterator_status
1039adjust_discarded_packets_message(bt_self_message_iterator *iter, const bt_stream *stream,
1040 const bt_message *msg_in, bt_message **msg_out,
1041 uint64_t new_begin_ts)
8ec4d5ff 1042{
4164020e
SM		 1043 enum lttng_live_iterator_status status = LTTNG_LIVE_ITERATOR_STATUS_OK;
1044 enum bt_property_availability availability;
1045 const bt_clock_snapshot *clock_snapshot;
1046 uint64_t end_ts;
1047 uint64_t count;
1048
1049 clock_snapshot = bt_message_discarded_packets_borrow_end_default_clock_snapshot_const(msg_in);
1050 end_ts = bt_clock_snapshot_get_value(clock_snapshot);
1051
1052 availability = bt_message_discarded_packets_get_count(msg_in, &count);
1053 BT_ASSERT_DBG(availability == BT_PROPERTY_AVAILABILITY_AVAILABLE);
1054
1055 *msg_out = bt_message_discarded_packets_create_with_default_clock_snapshots(
1056 iter, stream, new_begin_ts, end_ts);
1057 if (!*msg_out) {
1058 status = LTTNG_LIVE_ITERATOR_STATUS_NOMEM;
1059 goto end;
1060 }
1061
1062 bt_message_discarded_packets_set_count(*msg_out, count);
8ec4d5ff 1063end:
4164020e 1064 return status;
8ec4d5ff
JG		 1065}
1066
4164020e
SM		 1067static enum lttng_live_iterator_status
1068adjust_discarded_events_message(bt_self_message_iterator *iter, const bt_stream *stream,
1069 const bt_message *msg_in, bt_message **msg_out,
1070 uint64_t new_begin_ts)
8ec4d5ff 1071{
4164020e
SM		 1072 enum lttng_live_iterator_status status = LTTNG_LIVE_ITERATOR_STATUS_OK;
1073 enum bt_property_availability availability;
1074 const bt_clock_snapshot *clock_snapshot;
1075 uint64_t end_ts;
1076 uint64_t count;
1077
1078 clock_snapshot = bt_message_discarded_events_borrow_end_default_clock_snapshot_const(msg_in);
1079 end_ts = bt_clock_snapshot_get_value(clock_snapshot);
1080
1081 availability = bt_message_discarded_events_get_count(msg_in, &count);
1082 BT_ASSERT_DBG(availability == BT_PROPERTY_AVAILABILITY_AVAILABLE);
1083
1084 *msg_out = bt_message_discarded_events_create_with_default_clock_snapshots(
1085 iter, stream, new_begin_ts, end_ts);
1086 if (!*msg_out) {
1087 status = LTTNG_LIVE_ITERATOR_STATUS_NOMEM;
1088 goto end;
1089 }
1090
1091 bt_message_discarded_events_set_count(*msg_out, count);
8ec4d5ff 1092end:
4164020e 1093 return status;
8ec4d5ff
JG		 1094}
1095
4164020e
SM		 1096static enum lttng_live_iterator_status
1097handle_late_message(struct lttng_live_msg_iter *lttng_live_msg_iter,
1098 struct lttng_live_stream_iterator *stream_iter, int64_t late_msg_ts_ns,
1099 const bt_message *late_msg)
285951be 1100{
4164020e
SM		 1101 bt_self_component *self_comp = lttng_live_msg_iter->self_comp;
1102 bt_logging_level log_level = lttng_live_msg_iter->log_level;
1103 const bt_clock_class *clock_class;
1104 const bt_stream_class *stream_class;
1105 enum bt_clock_class_cycles_to_ns_from_origin_status ts_ns_status;
1106 int64_t last_inactivity_ts_ns;
1107 enum lttng_live_iterator_status stream_iter_status = LTTNG_LIVE_ITERATOR_STATUS_OK;
1108 enum lttng_live_iterator_status adjust_status;
1109 bt_message *adjusted_message;
1110
1111 /*
1112 * The timestamp of the current message is before the last message sent
1113 * by this component. We CANNOT send it as is.
1114 *
1115 * The only expected scenario in which that could happen is the
1116 * following, everything else is a bug in this component, relay deamon,
1117 * or CTF parser.
1118 *
1119 * Expected scenario: The CTF message iterator emitted discarded
1120 * packets and discarded events with synthesized beginning and end
1121 * timestamps from the bounds of the last known packet and the newly
1122 * decoded packet header. The CTF message iterator is not aware of
1123 * stream inactivity beacons. Hence, we have to adjust the beginning
1124 * timestamp of those types of messages if a stream signalled its
1125 * inactivity up until _after_ the last known packet's beginning
1126 * timestamp.
1127 *
1128 * Otherwise, the monotonicity guarantee of message timestamps would
1129 * not be preserved.
1130 *
1131 * In short, the only scenario in which it's okay and fixable to
1132 * received a late message is when:
1133 * 1. the late message is a discarded packets or discarded events
1134 * message,
1135 * 2. this stream produced an inactivity message downstream, and
1136 * 3. the timestamp of the late message is within the inactivity
1137 * timespan we sent downstream through the inactivity message.
1138 */
1139
1140 BT_COMP_LOGD("Handling late message on live stream iterator: "
1141 "stream-name=\"%s\", viewer-stream-id=%" PRIu64,
1142 stream_iter->name->str, stream_iter->viewer_stream_id);
1143
1144 if (!stream_iter->last_inactivity_ts.is_set) {
1145 BT_COMP_LOGE_APPEND_CAUSE(self_comp, "Invalid live stream state: "
1146 "have a late message when no inactivity message "
1147 "was ever sent for that stream.");
1148 stream_iter_status = LTTNG_LIVE_ITERATOR_STATUS_ERROR;
1149 goto end;
1150 }
1151
1152 if (!is_discarded_packet_or_event_message(late_msg)) {
1153 BT_COMP_LOGE_APPEND_CAUSE(self_comp,
1154 "Invalid live stream state: "
1155 "have a late message that is not a packet discarded or "
1156 "event discarded message: late-msg-type=%s",
1157 bt_common_message_type_string(bt_message_get_type(late_msg)));
1158 stream_iter_status = LTTNG_LIVE_ITERATOR_STATUS_ERROR;
1159 goto end;
1160 }
1161
1162 stream_class = bt_stream_borrow_class_const(stream_iter->stream);
1163 clock_class = bt_stream_class_borrow_default_clock_class_const(stream_class);
1164
1165 ts_ns_status = bt_clock_class_cycles_to_ns_from_origin(
1166 clock_class, stream_iter->last_inactivity_ts.value, &last_inactivity_ts_ns);
1167 if (ts_ns_status != BT_CLOCK_CLASS_CYCLES_TO_NS_FROM_ORIGIN_STATUS_OK) {
1168 BT_COMP_LOGE_APPEND_CAUSE(self_comp, "Error converting last "
1169 "inactivity message timestamp to nanoseconds");
1170 stream_iter_status = LTTNG_LIVE_ITERATOR_STATUS_ERROR;
1171 goto end;
1172 }
1173
1174 if (last_inactivity_ts_ns <= late_msg_ts_ns) {
1175 BT_COMP_LOGE_APPEND_CAUSE(self_comp,
1176 "Invalid live stream state: "
1177 "have a late message that is none included in a stream "
1178 "inactivity timespan: last-inactivity-ts-ns=%" PRIu64
1179 "late-msg-ts-ns=%" PRIu64,
1180 last_inactivity_ts_ns, late_msg_ts_ns);
1181 stream_iter_status = LTTNG_LIVE_ITERATOR_STATUS_ERROR;
1182 goto end;
1183 }
1184
1185 /*
1186 * We now know that it's okay for this message to be late, we can now
1187 * adjust its timestamp to ensure monotonicity.
1188 */
1189 BT_COMP_LOGD("Adjusting the timestamp of late message: late-msg-type=%s, "
1190 "msg-new-ts-ns=%" PRIu64,
1191 bt_common_message_type_string(bt_message_get_type(late_msg)),
1192 stream_iter->last_inactivity_ts.value);
1193 switch (bt_message_get_type(late_msg)) {
1194 case BT_MESSAGE_TYPE_DISCARDED_EVENTS:
1195 adjust_status = adjust_discarded_events_message(
1196 lttng_live_msg_iter->self_msg_iter, stream_iter->stream, late_msg, &adjusted_message,
1197 stream_iter->last_inactivity_ts.value);
1198 break;
1199 case BT_MESSAGE_TYPE_DISCARDED_PACKETS:
1200 adjust_status = adjust_discarded_packets_message(
1201 lttng_live_msg_iter->self_msg_iter, stream_iter->stream, late_msg, &adjusted_message,
1202 stream_iter->last_inactivity_ts.value);
1203 break;
1204 default:
1205 bt_common_abort();
1206 }
1207
1208 if (adjust_status != LTTNG_LIVE_ITERATOR_STATUS_OK) {
1209 stream_iter_status = adjust_status;
1210 goto end;
1211 }
1212
1213 BT_ASSERT_DBG(adjusted_message);
1214 stream_iter->current_msg = adjusted_message;
1215 stream_iter->current_msg_ts_ns = last_inactivity_ts_ns;
1216 bt_message_put_ref(late_msg);
cefd84f8 1217
285951be 1218end:
4164020e 1219 return stream_iter_status;
285951be
FD		 1220}
1221
4164020e
SM		 1222static enum lttng_live_iterator_status
1223next_stream_iterator_for_trace(struct lttng_live_msg_iter *lttng_live_msg_iter,
1224 struct lttng_live_trace *live_trace,
1225 struct lttng_live_stream_iterator **youngest_trace_stream_iter)
7cdc2bab 1226{
4164020e
SM		 1227 struct lttng_live_stream_iterator *youngest_candidate_stream_iter = NULL;
1228 bt_logging_level log_level = lttng_live_msg_iter->log_level;
1229 bt_self_component *self_comp = lttng_live_msg_iter->self_comp;
1230 enum lttng_live_iterator_status stream_iter_status;
1231 ;
1232 int64_t youngest_candidate_msg_ts = INT64_MAX;
1233 uint64_t stream_iter_idx;
1234
1235 BT_ASSERT_DBG(live_trace);
1236 BT_ASSERT_DBG(live_trace->stream_iterators);
1237
1238 BT_COMP_LOGD("Finding the next stream iterator for trace: "
1239 "trace-id=%" PRIu64,
1240 live_trace->id);
1241 /*
1242 * Update the current message of every stream iterators of this trace.
1243 * The current msg of every stream must have a timestamp equal or
1244 * larger than the last message returned by this iterator. We must
1245 * ensure monotonicity.
1246 */
1247 stream_iter_idx = 0;
1248 while (stream_iter_idx < live_trace->stream_iterators->len) {
1249 bool stream_iter_is_ended = false;
1250 struct lttng_live_stream_iterator *stream_iter =
1251 (lttng_live_stream_iterator *) g_ptr_array_index(live_trace->stream_iterators,
1252 stream_iter_idx);
1253
1254 /*
1255 * If there is no current message for this stream, go fetch
1256 * one.
1257 */
1258 while (!stream_iter->current_msg) {
1259 const bt_message *msg = NULL;
1260 int64_t curr_msg_ts_ns = INT64_MAX;
1261
1262 stream_iter_status =
1263 lttng_live_iterator_next_msg_on_stream(lttng_live_msg_iter, stream_iter, &msg);
1264
1265 if (stream_iter_status == LTTNG_LIVE_ITERATOR_STATUS_END) {
1266 stream_iter_is_ended = true;
1267 break;
1268 }
1269
1270 if (stream_iter_status != LTTNG_LIVE_ITERATOR_STATUS_OK) {
1271 goto end;
1272 }
1273
1274 BT_ASSERT_DBG(msg);
1275
1276 BT_COMP_LOGD("Live stream iterator returned message: msg-type=%s, "
1277 "stream-name=\"%s\", viewer-stream-id=%" PRIu64,
1278 bt_common_message_type_string(bt_message_get_type(msg)),
1279 stream_iter->name->str, stream_iter->viewer_stream_id);
1280
1281 /*
1282 * Get the timestamp in nanoseconds from origin of this
1283 * messsage.
1284 */
1285 live_get_msg_ts_ns(stream_iter, lttng_live_msg_iter, msg,
1286 lttng_live_msg_iter->last_msg_ts_ns, &curr_msg_ts_ns);
1287
1288 /*
1289 * Check if the message of the current live stream
1290 * iterator occurred at the exact same time or after the
1291 * last message returned by this component's message
1292 * iterator. If not, we need to handle it with care.
1293 */
1294 if (curr_msg_ts_ns >= lttng_live_msg_iter->last_msg_ts_ns) {
1295 stream_iter->current_msg = msg;
1296 stream_iter->current_msg_ts_ns = curr_msg_ts_ns;
1297 } else {
1298 /*
1299 * We received a message from the past. This
1300 * may be fixable but it can also be an error.
1301 */
1302 stream_iter_status =
1303 handle_late_message(lttng_live_msg_iter, stream_iter, curr_msg_ts_ns, msg);
1304 if (stream_iter_status != LTTNG_LIVE_ITERATOR_STATUS_OK) {
1305 BT_COMP_LOGE_APPEND_CAUSE(self_comp,
1306 "Late message could not be handled correctly: "
1307 "lttng-live-msg-iter-addr=%p, "
1308 "stream-name=\"%s\", "
1309 "curr-msg-ts=%" PRId64 ", last-msg-ts=%" PRId64,
1310 lttng_live_msg_iter, stream_iter->name->str,
1311 curr_msg_ts_ns, lttng_live_msg_iter->last_msg_ts_ns);
1312 stream_iter_status = LTTNG_LIVE_ITERATOR_STATUS_ERROR;
1313 goto end;
1314 }
1315 }
1316 }
1317
1318 BT_ASSERT_DBG(stream_iter != youngest_candidate_stream_iter);
1319
1320 if (!stream_iter_is_ended) {
1321 if (G_UNLIKELY(youngest_candidate_stream_iter == NULL) ||
1322 stream_iter->current_msg_ts_ns < youngest_candidate_msg_ts) {
1323 /*
1324 * Update the current best candidate message
1325 * for the stream iterator of this live trace
1326 * to be forwarded downstream.
1327 */
1328 youngest_candidate_msg_ts = stream_iter->current_msg_ts_ns;
1329 youngest_candidate_stream_iter = stream_iter;
1330 } else if (stream_iter->current_msg_ts_ns == youngest_candidate_msg_ts) {
1331 /*
1332 * Order the messages in an arbitrary but
1333 * deterministic way.
1334 */
1335 BT_ASSERT_DBG(stream_iter != youngest_candidate_stream_iter);
1336 int ret = common_muxing_compare_messages(
1337 stream_iter->current_msg, youngest_candidate_stream_iter->current_msg);
1338 if (ret < 0) {
1339 /*
1340 * The `youngest_candidate_stream_iter->current_msg`
1341 * should go first. Update the next
1342 * iterator and the current timestamp.
1343 */
1344 youngest_candidate_msg_ts = stream_iter->current_msg_ts_ns;
1345 youngest_candidate_stream_iter = stream_iter;
1346 } else if (ret == 0) {
1347 /*
1348 * Unable to pick which one should go
1349 * first.
1350 */
1351 BT_COMP_LOGW(
1352 "Cannot deterministically pick next live stream message iterator because they have identical next messages: "
1353 "stream-iter-addr=%p"
1354 "stream-iter-addr=%p",
1355 stream_iter, youngest_candidate_stream_iter);
1356 }
1357 }
1358
1359 stream_iter_idx++;
1360 } else {
1361 /*
1362 * The live stream iterator has ended. That
1363 * iterator is removed from the array, but
1364 * there is no need to increment
1365 * stream_iter_idx as
1366 * g_ptr_array_remove_index_fast replaces the
1367 * removed element with the array's last
1368 * element.
1369 */
1370 g_ptr_array_remove_index_fast(live_trace->stream_iterators, stream_iter_idx);
1371 }
1372 }
1373
1374 if (youngest_candidate_stream_iter) {
1375 *youngest_trace_stream_iter = youngest_candidate_stream_iter;
1376 stream_iter_status = LTTNG_LIVE_ITERATOR_STATUS_OK;
1377 } else {
1378 /*
1379 * The only case where we don't have a candidate for this trace
1380 * is if we reached the end of all the iterators.
1381 */
1382 BT_ASSERT(live_trace->stream_iterators->len == 0);
1383 stream_iter_status = LTTNG_LIVE_ITERATOR_STATUS_END;
1384 }
14f28187
FD		 1385
1386end:
4164020e 1387 return stream_iter_status;
14f28187
FD		 1388}
1389
4164020e
SM		 1390static enum lttng_live_iterator_status
1391next_stream_iterator_for_session(struct lttng_live_msg_iter *lttng_live_msg_iter,
1392 struct lttng_live_session *session,
1393 struct lttng_live_stream_iterator **youngest_session_stream_iter)
14f28187 1394{
4164020e
SM		 1395 bt_self_component *self_comp = lttng_live_msg_iter->self_comp;
1396 bt_logging_level log_level = lttng_live_msg_iter->log_level;
1397 enum lttng_live_iterator_status stream_iter_status;
1398 uint64_t trace_idx = 0;
1399 int64_t youngest_candidate_msg_ts = INT64_MAX;
1400 struct lttng_live_stream_iterator *youngest_candidate_stream_iter = NULL;
1401
1402 BT_COMP_LOGD("Finding the next stream iterator for session: "
1403 "session-id=%" PRIu64,
1404 session->id);
1405 /*
1406 * Make sure we are attached to the session and look for new streams
1407 * and metadata.
1408 */
1409 stream_iter_status = lttng_live_get_session(lttng_live_msg_iter, session);
1410 if (stream_iter_status != LTTNG_LIVE_ITERATOR_STATUS_OK &&
1411 stream_iter_status != LTTNG_LIVE_ITERATOR_STATUS_CONTINUE &&
1412 stream_iter_status != LTTNG_LIVE_ITERATOR_STATUS_END) {
1413 goto end;
1414 }
1415
1416 BT_ASSERT_DBG(session->traces);
1417
1418 while (trace_idx < session->traces->len) {
1419 bool trace_is_ended = false;
1420 struct lttng_live_stream_iterator *stream_iter;
1421 struct lttng_live_trace *trace =
1422 (lttng_live_trace *) g_ptr_array_index(session->traces, trace_idx);
1423
1424 stream_iter_status =
1425 next_stream_iterator_for_trace(lttng_live_msg_iter, trace, &stream_iter);
1426 if (stream_iter_status == LTTNG_LIVE_ITERATOR_STATUS_END) {
1427 /*
1428 * All the live stream iterators for this trace are
1429 * ENDed. Remove the trace from this session.
1430 */
1431 trace_is_ended = true;
1432 } else if (stream_iter_status != LTTNG_LIVE_ITERATOR_STATUS_OK) {
1433 goto end;
1434 }
1435
1436 if (!trace_is_ended) {
1437 BT_ASSERT_DBG(stream_iter);
1438
1439 if (G_UNLIKELY(youngest_candidate_stream_iter == NULL) ||
1440 stream_iter->current_msg_ts_ns < youngest_candidate_msg_ts) {
1441 youngest_candidate_msg_ts = stream_iter->current_msg_ts_ns;
1442 youngest_candidate_stream_iter = stream_iter;
1443 } else if (stream_iter->current_msg_ts_ns == youngest_candidate_msg_ts) {
1444 /*
1445 * Order the messages in an arbitrary but
1446 * deterministic way.
1447 */
1448 int ret = common_muxing_compare_messages(
1449 stream_iter->current_msg, youngest_candidate_stream_iter->current_msg);
1450 if (ret < 0) {
1451 /*
1452 * The `youngest_candidate_stream_iter->current_msg`
1453 * should go first. Update the next iterator
1454 * and the current timestamp.
1455 */
1456 youngest_candidate_msg_ts = stream_iter->current_msg_ts_ns;
1457 youngest_candidate_stream_iter = stream_iter;
1458 } else if (ret == 0) {
1459 /* Unable to pick which one should go first. */
1460 BT_COMP_LOGW(
1461 "Cannot deterministically pick next live stream message iterator because they have identical next messages: "
1462 "stream-iter-addr=%p"
1463 "stream-iter-addr=%p",
1464 stream_iter, youngest_candidate_stream_iter);
1465 }
1466 }
1467 trace_idx++;
1468 } else {
1469 /*
1470 * trace_idx is not incremented since
1471 * g_ptr_array_remove_index_fast replaces the
1472 * element at trace_idx with the array's last element.
1473 */
1474 g_ptr_array_remove_index_fast(session->traces, trace_idx);
1475 }
1476 }
1477 if (youngest_candidate_stream_iter) {
1478 *youngest_session_stream_iter = youngest_candidate_stream_iter;
1479 stream_iter_status = LTTNG_LIVE_ITERATOR_STATUS_OK;
1480 } else {
1481 /*
1482 * The only cases where we don't have a candidate for this
1483 * trace is:
1484 * 1. if we reached the end of all the iterators of all the
1485 * traces of this session,
1486 * 2. if we never had live stream iterator in the first place.
1487 *
1488 * In either cases, we return END.
1489 */
1490 BT_ASSERT(session->traces->len == 0);
1491 stream_iter_status = LTTNG_LIVE_ITERATOR_STATUS_END;
1492 }
7cdc2bab 1493end:
4164020e 1494 return stream_iter_status;
14f28187
FD		 1495}
1496
4164020e 1497static inline void put_messages(bt_message_array_const msgs, uint64_t count)
14f28187 1498{
4164020e 1499 uint64_t i;
14f28187 1500
4164020e
SM		 1501 for (i = 0; i < count; i++) {
1502 BT_MESSAGE_PUT_REF_AND_RESET(msgs[i]);
1503 }
7cdc2bab
MD		 1504}
1505
4164020e
SM		 1506bt_message_iterator_class_next_method_status
1507lttng_live_msg_iter_next(bt_self_message_iterator *self_msg_it, bt_message_array_const msgs,
1508 uint64_t capacity, uint64_t *count)
d3eb6e8f 1509{
4164020e
SM		 1510 bt_message_iterator_class_next_method_status status;
1511 enum lttng_live_viewer_status viewer_status;
1512 struct lttng_live_msg_iter *lttng_live_msg_iter =
1513 (struct lttng_live_msg_iter *) bt_self_message_iterator_get_data(self_msg_it);
1514 struct lttng_live_component *lttng_live = lttng_live_msg_iter->lttng_live_comp;
1515 bt_self_component *self_comp = lttng_live_msg_iter->self_comp;
1516 bt_logging_level log_level = lttng_live_msg_iter->log_level;
1517 enum lttng_live_iterator_status stream_iter_status;
1518 uint64_t session_idx;
1519
1520 *count = 0;
1521
1522 BT_ASSERT_DBG(lttng_live_msg_iter);
1523
1524 if (G_UNLIKELY(lttng_live_msg_iter->was_interrupted)) {
1525 /*
1526 * The iterator was interrupted in a previous call to the
1527 * `_next()` method. We currently do not support generating
1528 * messages after such event. The babeltrace2 CLI should never
1529 * be running the graph after being interrupted. So this check
1530 * is to prevent other graph users from using this live
1531 * iterator in an messed up internal state.
1532 */
1533 status = BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_ERROR;
1534 BT_COMP_LOGE_APPEND_CAUSE(
1535 self_comp,
1536 "Message iterator was interrupted during a previous call to the `next()` and currently does not support continuing after such event.");
1537 goto end;
1538 }
1539
1540 /*
1541 * Clear all the invalid message reference that might be left over in
1542 * the output array.
1543 */
1544 memset(msgs, 0, capacity * sizeof(*msgs));
1545
1546 /*
1547 * If no session are exposed on the relay found at the url provided by
1548 * the user, session count will be 0. In this case, we return status
1549 * end to return gracefully.
1550 */
1551 if (lttng_live_msg_iter->sessions->len == 0) {
1552 if (lttng_live->params.sess_not_found_act != SESSION_NOT_FOUND_ACTION_CONTINUE) {
1553 status = BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_END;
1554 goto end;
1555 } else {
1556 /*
1557 * The are no more active session for this session
1558 * name. Retry to create a viewer session for the
1559 * requested session name.
1560 */
1561 viewer_status = lttng_live_create_viewer_session(lttng_live_msg_iter);
1562 if (viewer_status != LTTNG_LIVE_VIEWER_STATUS_OK) {
1563 if (viewer_status == LTTNG_LIVE_VIEWER_STATUS_ERROR) {
1564 status = BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_ERROR;
1565 BT_COMP_LOGE_APPEND_CAUSE(self_comp,
1566 "Error creating LTTng live viewer session");
1567 } else if (viewer_status == LTTNG_LIVE_VIEWER_STATUS_INTERRUPTED) {
1568 status = BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_AGAIN;
1569 } else {
1570 bt_common_abort();
1571 }
1572 goto end;
1573 }
1574 }
1575 }
1576
1577 if (lttng_live_msg_iter->active_stream_iter == 0) {
1578 lttng_live_force_new_streams_and_metadata(lttng_live_msg_iter);
1579 }
1580
1581 /*
1582 * Here the muxing of message is done.
1583 *
1584 * We need to iterate over all the streams of all the traces of all the
1585 * viewer sessions in order to get the message with the smallest
1586 * timestamp. In this case, a session is a viewer session and there is
1587 * one viewer session per consumer daemon. (UST 32bit, UST 64bit and/or
1588 * kernel). Each viewer session can have multiple traces, for example,
1589 * 64bit UST viewer sessions could have multiple per-pid traces.
1590 *
1591 * We iterate over the streams of each traces to update and see what is
1592 * their next message's timestamp. From those timestamps, we select the
1593 * message with the smallest timestamp as the best candidate message
1594 * for that trace and do the same thing across all the sessions.
1595 *
1596 * We then compare the timestamp of best candidate message of all the
1597 * sessions to pick the message with the smallest timestamp and we
1598 * return it.
1599 */
1600 while (*count < capacity) {
1601 struct lttng_live_stream_iterator *youngest_stream_iter = NULL,
1602 *candidate_stream_iter = NULL;
1603 int64_t youngest_msg_ts_ns = INT64_MAX;
1604
1605 BT_ASSERT_DBG(lttng_live_msg_iter->sessions);
1606 session_idx = 0;
1607 while (session_idx < lttng_live_msg_iter->sessions->len) {
1608 struct lttng_live_session *session = (lttng_live_session *) g_ptr_array_index(
1609 lttng_live_msg_iter->sessions, session_idx);
1610
1611 /* Find the best candidate message to send downstream. */
1612 stream_iter_status = next_stream_iterator_for_session(lttng_live_msg_iter, session,
1613 &candidate_stream_iter);
1614
1615 /* If we receive an END status, it means that either:
1616 * - Those traces never had active streams (UST with no
1617 * data produced yet),
1618 * - All live stream iterators have ENDed.*/
1619 if (stream_iter_status == LTTNG_LIVE_ITERATOR_STATUS_END) {
1620 if (session->closed && session->traces->len == 0) {
1621 /*
1622 * Remove the session from the list.
1623 * session_idx is not modified since
1624 * g_ptr_array_remove_index_fast
1625 * replaces the the removed element with
1626 * the array's last element.
1627 */
1628 g_ptr_array_remove_index_fast(lttng_live_msg_iter->sessions, session_idx);
1629 } else {
1630 session_idx++;
1631 }
1632 continue;
1633 }
1634
1635 if (stream_iter_status != LTTNG_LIVE_ITERATOR_STATUS_OK) {
1636 goto return_status;
1637 }
1638
1639 if (G_UNLIKELY(youngest_stream_iter == NULL) ||
1640 candidate_stream_iter->current_msg_ts_ns < youngest_msg_ts_ns) {
1641 youngest_msg_ts_ns = candidate_stream_iter->current_msg_ts_ns;
1642 youngest_stream_iter = candidate_stream_iter;
1643 } else if (candidate_stream_iter->current_msg_ts_ns == youngest_msg_ts_ns) {
1644 /*
1645 * The currently selected message to be sent
1646 * downstream next has the exact same timestamp
1647 * that of the current candidate message. We
1648 * must break the tie in a predictable manner.
1649 */
1650 BT_COMP_LOGD_STR(
1651 "Two of the next message candidates have the same timestamps, pick one deterministically.");
1652 /*
1653 * Order the messages in an arbitrary but
1654 * deterministic way.
1655 */
1656 int ret = common_muxing_compare_messages(candidate_stream_iter->current_msg,
1657 youngest_stream_iter->current_msg);
1658 if (ret < 0) {
1659 /*
1660 * The `candidate_stream_iter->current_msg`
1661 * should go first. Update the next
1662 * iterator and the current timestamp.
1663 */
1664 youngest_msg_ts_ns = candidate_stream_iter->current_msg_ts_ns;
1665 youngest_stream_iter = candidate_stream_iter;
1666 } else if (ret == 0) {
1667 /* Unable to pick which one should go first. */
1668 BT_COMP_LOGW(
1669 "Cannot deterministically pick next live stream message iterator because they have identical next messages: "
1670 "next-stream-iter-addr=%p"
1671 "candidate-stream-iter-addr=%p",
1672 youngest_stream_iter, candidate_stream_iter);
1673 }
1674 }
1675
1676 session_idx++;
1677 }
1678
1679 if (!youngest_stream_iter) {
1680 stream_iter_status = LTTNG_LIVE_ITERATOR_STATUS_AGAIN;
1681 goto return_status;
1682 }
1683
1684 BT_ASSERT_DBG(youngest_stream_iter->current_msg);
1685 /* Ensure monotonicity. */
1686 BT_ASSERT_DBG(lttng_live_msg_iter->last_msg_ts_ns <=
1687 youngest_stream_iter->current_msg_ts_ns);
1688
1689 /*
1690 * Insert the next message to the message batch. This will set
1691 * stream iterator current messsage to NULL so that next time
1692 * we fetch the next message of that stream iterator
1693 */
1694 BT_MESSAGE_MOVE_REF(msgs[*count], youngest_stream_iter->current_msg);
1695 (*count)++;
1696
1697 /* Update the last timestamp in nanoseconds sent downstream. */
1698 lttng_live_msg_iter->last_msg_ts_ns = youngest_msg_ts_ns;
1699 youngest_stream_iter->current_msg_ts_ns = INT64_MAX;
1700
1701 stream_iter_status = LTTNG_LIVE_ITERATOR_STATUS_OK;
1702 }
f79c2d7a
FD		 1703
1704return_status:
4164020e
SM		 1705 switch (stream_iter_status) {
1706 case LTTNG_LIVE_ITERATOR_STATUS_OK:
1707 case LTTNG_LIVE_ITERATOR_STATUS_AGAIN:
1708 /*
1709 * If we gathered messages, return _OK even if the graph was
1710 * interrupted. This allows for the components downstream to at
1711 * least get the thoses messages. If the graph was indeed
1712 * interrupted there should not be another _next() call as the
1713 * application will tear down the graph. This component class
1714 * doesn't support restarting after an interruption.
1715 */
1716 if (*count > 0) {
1717 status = BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_OK;
1718 } else {
1719 status = BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_AGAIN;
1720 }
1721 break;
1722 case LTTNG_LIVE_ITERATOR_STATUS_END:
1723 status = BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_END;
1724 break;
1725 case LTTNG_LIVE_ITERATOR_STATUS_NOMEM:
1726 BT_COMP_LOGE_APPEND_CAUSE(self_comp,
1727 "Memory error preparing the next batch of messages: "
1728 "live-iter-status=%s",
1729 lttng_live_iterator_status_string(stream_iter_status));
1730 status = BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_MEMORY_ERROR;
1731 break;
1732 case LTTNG_LIVE_ITERATOR_STATUS_ERROR:
1733 case LTTNG_LIVE_ITERATOR_STATUS_INVAL:
1734 case LTTNG_LIVE_ITERATOR_STATUS_UNSUPPORTED:
1735 BT_COMP_LOGE_APPEND_CAUSE(self_comp,
1736 "Error preparing the next batch of messages: "
1737 "live-iter-status=%s",
1738 lttng_live_iterator_status_string(stream_iter_status));
1739
1740 status = BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_ERROR;
1741 /* Put all existing messages on error. */
1742 put_messages(msgs, *count);
1743 break;
1744 default:
1745 bt_common_abort();
1746 }
14f28187 1747
f79c2d7a 1748end:
4164020e 1749 return status;
7cdc2bab 1750}
41a2b7ae 1751
4164020e
SM		 1752static struct lttng_live_msg_iter *
1753lttng_live_msg_iter_create(struct lttng_live_component *lttng_live_comp,
1754 bt_self_message_iterator *self_msg_it)
4f74db52 1755{
4164020e
SM		 1756 bt_self_component *self_comp = lttng_live_comp->self_comp;
1757 bt_logging_level log_level = lttng_live_comp->log_level;
4f74db52 1758
4164020e
SM		 1759 struct lttng_live_msg_iter *lttng_live_msg_iter = g_new0(struct lttng_live_msg_iter, 1);
1760 if (!lttng_live_msg_iter) {
1761 BT_COMP_LOGE_APPEND_CAUSE(self_comp, "Failed to allocate lttng_live_msg_iter");
1762 goto end;
1763 }
1764
1765 lttng_live_msg_iter->log_level = lttng_live_comp->log_level;
1766 lttng_live_msg_iter->self_comp = lttng_live_comp->self_comp;
1767 lttng_live_msg_iter->lttng_live_comp = lttng_live_comp;
1768 lttng_live_msg_iter->self_msg_iter = self_msg_it;
1769
1770 lttng_live_msg_iter->active_stream_iter = 0;
1771 lttng_live_msg_iter->last_msg_ts_ns = INT64_MIN;
1772 lttng_live_msg_iter->was_interrupted = false;
4f74db52 1773
4164020e
SM		 1774 lttng_live_msg_iter->sessions =
1775 g_ptr_array_new_with_free_func((GDestroyNotify) lttng_live_destroy_session);
1776 BT_ASSERT(lttng_live_msg_iter->sessions);
1777
1778end:
1779 return lttng_live_msg_iter;
4f74db52
FD		 1780}
1781
4164020e
SM		 1782bt_message_iterator_class_initialize_method_status
1783lttng_live_msg_iter_init(bt_self_message_iterator *self_msg_it,
1784 bt_self_message_iterator_configuration *config,
1785 bt_self_component_port_output *self_port)
7cdc2bab 1786{
4164020e
SM		 1787 bt_message_iterator_class_initialize_method_status status;
1788 struct lttng_live_component *lttng_live;
1789 struct lttng_live_msg_iter *lttng_live_msg_iter;
1790 enum lttng_live_viewer_status viewer_status;
1791 bt_logging_level log_level;
1792 bt_self_component *self_comp = bt_self_message_iterator_borrow_component(self_msg_it);
1793
1794 lttng_live = (lttng_live_component *) bt_self_component_get_data(self_comp);
1795 log_level = lttng_live->log_level;
1796 self_comp = lttng_live->self_comp;
1797
1798 /* There can be only one downstream iterator at the same time. */
1799 BT_ASSERT(!lttng_live->has_msg_iter);
1800 lttng_live->has_msg_iter = true;
1801
1802 lttng_live_msg_iter = lttng_live_msg_iter_create(lttng_live, self_msg_it);
1803 if (!lttng_live_msg_iter) {
4164020e 1804 BT_COMP_LOGE_APPEND_CAUSE(self_comp, "Failed to create lttng_live_msg_iter");
10265ebe 1805 status = BT_MESSAGE_ITERATOR_CLASS_INITIALIZE_METHOD_STATUS_MEMORY_ERROR;
4164020e
SM		 1806 goto error;
1807 }
1808
1809 viewer_status = live_viewer_connection_create(
1810 self_comp, NULL, log_level, lttng_live->params.url->str, false, lttng_live_msg_iter,
1811 &lttng_live_msg_iter->viewer_connection);
1812 if (viewer_status != LTTNG_LIVE_VIEWER_STATUS_OK) {
1813 if (viewer_status == LTTNG_LIVE_VIEWER_STATUS_ERROR) {
1814 BT_COMP_LOGE_APPEND_CAUSE(self_comp, "Failed to create viewer connection");
1815 } else if (viewer_status == LTTNG_LIVE_VIEWER_STATUS_INTERRUPTED) {
1816 /*
1817 * Interruption in the _iter_init() method is not
1818 * supported. Return an error.
1819 */
1820 BT_COMP_LOGE_APPEND_CAUSE(self_comp, "Interrupted while creating viewer connection");
1821 }
10265ebe
SM		 1822
1823 status = BT_MESSAGE_ITERATOR_CLASS_INITIALIZE_METHOD_STATUS_ERROR;
4164020e
SM		 1824 goto error;
1825 }
1826
1827 viewer_status = lttng_live_create_viewer_session(lttng_live_msg_iter);
1828 if (viewer_status != LTTNG_LIVE_VIEWER_STATUS_OK) {
1829 if (viewer_status == LTTNG_LIVE_VIEWER_STATUS_ERROR) {
1830 BT_COMP_LOGE_APPEND_CAUSE(self_comp, "Failed to create viewer session");
1831 } else if (viewer_status == LTTNG_LIVE_VIEWER_STATUS_INTERRUPTED) {
1832 /*
1833 * Interruption in the _iter_init() method is not
1834 * supported. Return an error.
1835 */
1836 BT_COMP_LOGE_APPEND_CAUSE(self_comp, "Interrupted when creating viewer session");
1837 }
10265ebe
SM		 1838
1839 status = BT_MESSAGE_ITERATOR_CLASS_INITIALIZE_METHOD_STATUS_ERROR;
4164020e
SM		 1840 goto error;
1841 }
1842
1843 if (lttng_live_msg_iter->sessions->len == 0) {
1844 switch (lttng_live->params.sess_not_found_act) {
1845 case SESSION_NOT_FOUND_ACTION_CONTINUE:
1846 BT_COMP_LOGI(
1847 "Unable to connect to the requested live viewer session. Keep trying to connect because of "
1848 "%s=\"%s\" component parameter: url=\"%s\"",
1849 SESS_NOT_FOUND_ACTION_PARAM, SESS_NOT_FOUND_ACTION_CONTINUE_STR,
1850 lttng_live->params.url->str);
1851 break;
1852 case SESSION_NOT_FOUND_ACTION_FAIL:
1853 BT_COMP_LOGE_APPEND_CAUSE(
1854 self_comp,
1855 "Unable to connect to the requested live viewer session. Fail the message iterator initialization because of %s=\"%s\" "
1856 "component parameter: url =\"%s\"",
1857 SESS_NOT_FOUND_ACTION_PARAM, SESS_NOT_FOUND_ACTION_FAIL_STR,
1858 lttng_live->params.url->str);
10265ebe 1859 status = BT_MESSAGE_ITERATOR_CLASS_INITIALIZE_METHOD_STATUS_ERROR;
4164020e
SM		 1860 goto error;
1861 case SESSION_NOT_FOUND_ACTION_END:
1862 BT_COMP_LOGI(
1863 "Unable to connect to the requested live viewer session. End gracefully at the first _next() "
1864 "call because of %s=\"%s\" component parameter: "
1865 "url=\"%s\"",
1866 SESS_NOT_FOUND_ACTION_PARAM, SESS_NOT_FOUND_ACTION_END_STR,
1867 lttng_live->params.url->str);
1868 break;
1869 default:
1870 bt_common_abort();
1871 }
1872 }
1873
1874 bt_self_message_iterator_set_data(self_msg_it, lttng_live_msg_iter);
1875 status = BT_MESSAGE_ITERATOR_CLASS_INITIALIZE_METHOD_STATUS_OK;
1876 goto end;
f79c2d7a 1877
14f28187 1878error:
4164020e 1879 lttng_live_msg_iter_destroy(lttng_live_msg_iter);
7cdc2bab 1880end:
4164020e 1881 return status;
7cdc2bab
MD		 1882}
1883
80aff5ef 1884static struct bt_param_validation_map_value_entry_descr list_sessions_params[] = {
88730e42
SM		 1885 {URL_PARAM, BT_PARAM_VALIDATION_MAP_VALUE_ENTRY_MANDATORY,
1886 bt_param_validation_value_descr::makeString()},
4164020e
SM		 1887 BT_PARAM_VALIDATION_MAP_VALUE_ENTRY_END};
1888
1889static bt_component_class_query_method_status
1890lttng_live_query_list_sessions(const bt_value *params, const bt_value **result,
1891 bt_self_component_class *self_comp_class, bt_logging_level log_level)
7cdc2bab 1892{
4164020e
SM		 1893 bt_component_class_query_method_status status;
1894 const bt_value *url_value = NULL;
1895 const char *url;
1896 struct live_viewer_connection *viewer_connection = NULL;
1897 enum lttng_live_viewer_status viewer_status;
1898 enum bt_param_validation_status validation_status;
1899 gchar *validate_error = NULL;
1900
1901 validation_status = bt_param_validation_validate(params, list_sessions_params, &validate_error);
1902 if (validation_status == BT_PARAM_VALIDATION_STATUS_MEMORY_ERROR) {
1903 status = BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_MEMORY_ERROR;
1904 goto error;
1905 } else if (validation_status == BT_PARAM_VALIDATION_STATUS_VALIDATION_ERROR) {
1906 status = BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_ERROR;
1907 BT_COMP_CLASS_LOGE_APPEND_CAUSE(self_comp_class, "%s", validate_error);
1908 goto error;
1909 }
1910
1911 url_value = bt_value_map_borrow_entry_value_const(params, URL_PARAM);
1912 url = bt_value_string_get(url_value);
1913
1914 viewer_status = live_viewer_connection_create(NULL, self_comp_class, log_level, url, true, NULL,
1915 &viewer_connection);
1916 if (viewer_status != LTTNG_LIVE_VIEWER_STATUS_OK) {
1917 if (viewer_status == LTTNG_LIVE_VIEWER_STATUS_ERROR) {
1918 BT_COMP_CLASS_LOGE_APPEND_CAUSE(self_comp_class, "Failed to create viewer connection");
1919 status = BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_ERROR;
1920 } else if (viewer_status == LTTNG_LIVE_VIEWER_STATUS_INTERRUPTED) {
1921 status = BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_AGAIN;
1922 } else {
1923 bt_common_abort();
1924 }
1925 goto error;
1926 }
1927
1928 status = live_viewer_connection_list_sessions(viewer_connection, result);
1929 if (status != BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_OK) {
1930 BT_COMP_CLASS_LOGE_APPEND_CAUSE(self_comp_class, "Failed to list viewer sessions");
1931 goto error;
1932 }
1933
1934 goto end;
c7eee084 1935
7cdc2bab 1936error:
4164020e 1937 BT_VALUE_PUT_REF_AND_RESET(*result);
c7eee084 1938
4164020e
SM		 1939 if (status >= 0) {
1940 status = BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_ERROR;
1941 }
c7eee084 1942
7cdc2bab 1943end:
4164020e
SM		 1944 if (viewer_connection) {
1945 live_viewer_connection_destroy(viewer_connection);
1946 }
80aff5ef 1947
4164020e 1948 g_free(validate_error);
80aff5ef 1949
4164020e 1950 return status;
7cdc2bab
MD		 1951}
1952
4164020e
SM		 1953static bt_component_class_query_method_status
1954lttng_live_query_support_info(const bt_value *params, const bt_value **result,
1955 bt_self_component_class *self_comp_class, bt_logging_level log_level)
312df793 1956{
4164020e
SM		 1957 bt_component_class_query_method_status status = BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_OK;
1958 const bt_value *input_type_value;
1959 const bt_value *input_value;
1960 double weight = 0;
1961 struct bt_common_lttng_live_url_parts parts = {0};
1962
1963 /* Used by the logging macros */
1964 __attribute__((unused)) bt_self_component *self_comp = NULL;
1965
1966 *result = NULL;
1967 input_type_value = bt_value_map_borrow_entry_value_const(params, "type");
1968 if (!input_type_value) {
1969 BT_COMP_CLASS_LOGE_APPEND_CAUSE(self_comp_class, "Missing expected `type` parameter.");
1970 goto error;
1971 }
1972
1973 if (!bt_value_is_string(input_type_value)) {
1974 BT_COMP_CLASS_LOGE_APPEND_CAUSE(self_comp_class, "`type` parameter is not a string value.");
1975 goto error;
1976 }
1977
1978 if (strcmp(bt_value_string_get(input_type_value), "string") != 0) {
1979 /* We don't handle file system paths */
1980 goto create_result;
1981 }
1982
1983 input_value = bt_value_map_borrow_entry_value_const(params, "input");
1984 if (!input_value) {
1985 BT_COMP_CLASS_LOGE_APPEND_CAUSE(self_comp_class, "Missing expected `input` parameter.");
1986 goto error;
1987 }
1988
1989 if (!bt_value_is_string(input_value)) {
1990 BT_COMP_CLASS_LOGE_APPEND_CAUSE(self_comp_class,
1991 "`input` parameter is not a string value.");
1992 goto error;
1993 }
1994
1995 parts = bt_common_parse_lttng_live_url(bt_value_string_get(input_value), NULL, 0);
1996 if (parts.session_name) {
1997 /*
1998 * Looks pretty much like an LTTng live URL: we got the
1999 * session name part, which forms a complete URL.
2000 */
2001 weight = .75;
2002 }
312df793
PP		 2003
2004create_result:
4164020e
SM		 2005 *result = bt_value_real_create_init(weight);
2006 if (!*result) {
2007 status = BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_MEMORY_ERROR;
2008 goto error;
2009 }
312df793 2010
4164020e 2011 goto end;
312df793
PP		 2012
2013error:
4164020e
SM		 2014 if (status >= 0) {
2015 status = BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_ERROR;
2016 }
312df793 2017
4164020e 2018 BT_ASSERT(!*result);
312df793
PP		 2019
2020end:
4164020e
SM		 2021 bt_common_destroy_lttng_live_url_parts(&parts);
2022 return status;
312df793
PP		 2023}
2024
4164020e
SM		 2025bt_component_class_query_method_status lttng_live_query(bt_self_component_class_source *comp_class,
2026 bt_private_query_executor *priv_query_exec,
2027 const char *object, const bt_value *params,
2028 __attribute__((unused)) void *method_data,
2029 const bt_value **result)
7cdc2bab 2030{
4164020e
SM		 2031 bt_component_class_query_method_status status = BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_OK;
2032 bt_self_component *self_comp = NULL;
2033 bt_self_component_class *self_comp_class =
2034 bt_self_component_class_source_as_self_component_class(comp_class);
2035 bt_logging_level log_level = bt_query_executor_get_logging_level(
2036 bt_private_query_executor_as_query_executor_const(priv_query_exec));
2037
2038 if (strcmp(object, "sessions") == 0) {
2039 status = lttng_live_query_list_sessions(params, result, self_comp_class, log_level);
2040 } else if (strcmp(object, "babeltrace.support-info") == 0) {
2041 status = lttng_live_query_support_info(params, result, self_comp_class, log_level);
2042 } else {
2043 BT_COMP_LOGI("Unknown query object `%s`", object);
2044 status = BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_UNKNOWN_OBJECT;
2045 goto end;
2046 }
14f28187
FD		 2047
2048end:
4164020e 2049 return status;
7cdc2bab
MD		 2050}
2051
4164020e 2052static void lttng_live_component_destroy_data(struct lttng_live_component *lttng_live)
7cdc2bab 2053{
4164020e
SM		 2054 if (!lttng_live) {
2055 return;
2056 }
2057 if (lttng_live->params.url) {
2058 g_string_free(lttng_live->params.url, TRUE);
2059 }
2060 g_free(lttng_live);
7cdc2bab
MD		 2061}
2062
14f28187 2063void lttng_live_component_finalize(bt_self_component_source *component)
7cdc2bab 2064{
4164020e
SM		 2065 lttng_live_component *data = (lttng_live_component *) bt_self_component_get_data(
2066 bt_self_component_source_as_self_component(component));
2067
2068 if (!data) {
2069 return;
2070 }
2071 lttng_live_component_destroy_data(data);
7cdc2bab
MD		 2072}
2073
4164020e
SM		 2074static enum session_not_found_action
2075parse_session_not_found_action_param(const bt_value *no_session_param)
14f28187 2076{
4164020e
SM		 2077 enum session_not_found_action action;
2078 const char *no_session_act_str = bt_value_string_get(no_session_param);
2079
2080 if (strcmp(no_session_act_str, SESS_NOT_FOUND_ACTION_CONTINUE_STR) == 0) {
2081 action = SESSION_NOT_FOUND_ACTION_CONTINUE;
2082 } else if (strcmp(no_session_act_str, SESS_NOT_FOUND_ACTION_FAIL_STR) == 0) {
2083 action = SESSION_NOT_FOUND_ACTION_FAIL;
2084 } else {
2085 BT_ASSERT(strcmp(no_session_act_str, SESS_NOT_FOUND_ACTION_END_STR) == 0);
2086 action = SESSION_NOT_FOUND_ACTION_END;
2087 }
2088
2089 return action;
14f28187
FD		 2090}
2091
88730e42
SM		 2092static bt_param_validation_value_descr inputs_elem_descr =
2093 bt_param_validation_value_descr::makeString();
80aff5ef
SM		 2094
2095static const char *sess_not_found_action_choices[] = {
4164020e
SM		 2096 SESS_NOT_FOUND_ACTION_CONTINUE_STR,
2097 SESS_NOT_FOUND_ACTION_FAIL_STR,
2098 SESS_NOT_FOUND_ACTION_END_STR,
80aff5ef
SM		 2099};
2100
2101static struct bt_param_validation_map_value_entry_descr params_descr[] = {
88730e42
SM		 2102 {INPUTS_PARAM, BT_PARAM_VALIDATION_MAP_VALUE_ENTRY_MANDATORY,
2103 bt_param_validation_value_descr::makeArray(1, 1, inputs_elem_descr)},
2104 {SESS_NOT_FOUND_ACTION_PARAM, BT_PARAM_VALIDATION_MAP_VALUE_ENTRY_OPTIONAL,
2105 bt_param_validation_value_descr::makeString(sess_not_found_action_choices)},
4164020e
SM		 2106 BT_PARAM_VALIDATION_MAP_VALUE_ENTRY_END};
2107
2108static bt_component_class_initialize_method_status
2109lttng_live_component_create(const bt_value *params, bt_logging_level log_level,
2110 bt_self_component *self_comp, struct lttng_live_component **component)
7cdc2bab 2111{
4164020e
SM		 2112 struct lttng_live_component *lttng_live = NULL;
2113 const bt_value *inputs_value;
2114 const bt_value *url_value;
2115 const bt_value *value;
2116 const char *url;
2117 enum bt_param_validation_status validation_status;
2118 gchar *validation_error = NULL;
2119 bt_component_class_initialize_method_status status;
2120
2121 validation_status = bt_param_validation_validate(params, params_descr, &validation_error);
2122 if (validation_status == BT_PARAM_VALIDATION_STATUS_MEMORY_ERROR) {
2123 status = BT_COMPONENT_CLASS_INITIALIZE_METHOD_STATUS_MEMORY_ERROR;
2124 goto error;
2125 } else if (validation_status == BT_PARAM_VALIDATION_STATUS_VALIDATION_ERROR) {
2126 BT_COMP_LOGE_APPEND_CAUSE(self_comp, "%s", validation_error);
2127 status = BT_COMPONENT_CLASS_INITIALIZE_METHOD_STATUS_ERROR;
2128 goto error;
2129 }
2130
2131 lttng_live = g_new0(struct lttng_live_component, 1);
2132 if (!lttng_live) {
2133 status = BT_COMPONENT_CLASS_INITIALIZE_METHOD_STATUS_MEMORY_ERROR;
2134 goto end;
2135 }
2136 lttng_live->log_level = log_level;
2137 lttng_live->self_comp = self_comp;
2138 lttng_live->max_query_size = MAX_QUERY_SIZE;
2139 lttng_live->has_msg_iter = false;
2140
2141 inputs_value = bt_value_map_borrow_entry_value_const(params, INPUTS_PARAM);
2142 url_value = bt_value_array_borrow_element_by_index_const(inputs_value, 0);
2143 url = bt_value_string_get(url_value);
2144
2145 lttng_live->params.url = g_string_new(url);
2146 if (!lttng_live->params.url) {
2147 status = BT_COMPONENT_CLASS_INITIALIZE_METHOD_STATUS_MEMORY_ERROR;
2148 goto error;
2149 }
2150
2151 value = bt_value_map_borrow_entry_value_const(params, SESS_NOT_FOUND_ACTION_PARAM);
2152 if (value) {
2153 lttng_live->params.sess_not_found_act = parse_session_not_found_action_param(value);
2154 } else {
2155 BT_COMP_LOGI("Optional `%s` parameter is missing: "
2156 "defaulting to `%s`.",
2157 SESS_NOT_FOUND_ACTION_PARAM, SESS_NOT_FOUND_ACTION_CONTINUE_STR);
2158 lttng_live->params.sess_not_found_act = SESSION_NOT_FOUND_ACTION_CONTINUE;
2159 }
2160
2161 status = BT_COMPONENT_CLASS_INITIALIZE_METHOD_STATUS_OK;
2162 goto end;
7cdc2bab
MD		 2163
2164error:
4164020e
SM		 2165 lttng_live_component_destroy_data(lttng_live);
2166 lttng_live = NULL;
7cdc2bab 2167end:
4164020e 2168 g_free(validation_error);
80aff5ef 2169
4164020e
SM		 2170 *component = lttng_live;
2171 return status;
d3e4dcd8
PP		 2172}
2173
4164020e
SM		 2174bt_component_class_initialize_method_status
2175lttng_live_component_init(bt_self_component_source *self_comp_src,
2176 bt_self_component_source_configuration *config, const bt_value *params,
2177 __attribute__((unused)) void *init_method_data)
f3bc2010 2178{
4164020e
SM		 2179 struct lttng_live_component *lttng_live;
2180 bt_component_class_initialize_method_status ret;
2181 bt_self_component *self_comp = bt_self_component_source_as_self_component(self_comp_src);
2182 bt_logging_level log_level =
2183 bt_component_get_logging_level(bt_self_component_as_component(self_comp));
2184 bt_self_component_add_port_status add_port_status;
2185
2186 ret = lttng_live_component_create(params, log_level, self_comp, &lttng_live);
2187 if (ret != BT_COMPONENT_CLASS_INITIALIZE_METHOD_STATUS_OK) {
2188 goto error;
2189 }
2190
2191 add_port_status = bt_self_component_source_add_output_port(self_comp_src, "out", NULL, NULL);
2192 if (add_port_status != BT_SELF_COMPONENT_ADD_PORT_STATUS_OK) {
2193 ret = (bt_component_class_initialize_method_status) add_port_status;
2194 goto end;
2195 }
2196
2197 bt_self_component_set_data(self_comp, lttng_live);
2198 goto end;
7cdc2bab 2199
19bdff20 2200error:
4164020e
SM		 2201 lttng_live_component_destroy_data(lttng_live);
2202 lttng_live = NULL;
7cdc2bab 2203end:
4164020e 2204 return ret;
d85ef162 2205}
BabelTrace - Trace Format Babel Tower
This page took 0.236761 seconds and 4 git commands to generate.