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