projects / babeltrace.git / blob
? search:


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