projects / babeltrace.git / blob
? search:


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