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