projects / babeltrace.git / blob
? search:


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