lib: remove INVALID_PARAMS status
[babeltrace.git] / src / plugins / ctf / lttng-live / lttng-live.c
... / ...
CommitLineData
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 "plugins/comp-logging.h"
35
36#include <glib.h>
37#include <inttypes.h>
38#include <unistd.h>
39
40#include "common/assert.h"
41#include <babeltrace2/babeltrace.h>
42#include "compat/compiler.h"
43#include <babeltrace2/types.h>
44
45#include "data-stream.h"
46#include "metadata.h"
47#include "lttng-live.h"
48
49#define MAX_QUERY_SIZE (256*1024)
50#define URL_PARAM "url"
51#define SESS_NOT_FOUND_ACTION_PARAM "session-not-found-action"
52#define SESS_NOT_FOUND_ACTION_CONTINUE_STR "continue"
53#define SESS_NOT_FOUND_ACTION_FAIL_STR "fail"
54#define SESS_NOT_FOUND_ACTION_END_STR "end"
55
56#define print_dbg(fmt, ...) BT_COMP_LOGD(fmt, ## __VA_ARGS__)
57
58static
59const char *print_live_iterator_status(enum lttng_live_iterator_status status)
60{
61 switch (status) {
62 case LTTNG_LIVE_ITERATOR_STATUS_CONTINUE:
63 return "LTTNG_LIVE_ITERATOR_STATUS_CONTINUE";
64 case LTTNG_LIVE_ITERATOR_STATUS_AGAIN:
65 return "LTTNG_LIVE_ITERATOR_STATUS_AGAIN";
66 case LTTNG_LIVE_ITERATOR_STATUS_END:
67 return "LTTNG_LIVE_ITERATOR_STATUS_END";
68 case LTTNG_LIVE_ITERATOR_STATUS_OK:
69 return "LTTNG_LIVE_ITERATOR_STATUS_OK";
70 case LTTNG_LIVE_ITERATOR_STATUS_INVAL:
71 return "LTTNG_LIVE_ITERATOR_STATUS_INVAL";
72 case LTTNG_LIVE_ITERATOR_STATUS_ERROR:
73 return "LTTNG_LIVE_ITERATOR_STATUS_ERROR";
74 case LTTNG_LIVE_ITERATOR_STATUS_NOMEM:
75 return "LTTNG_LIVE_ITERATOR_STATUS_NOMEM";
76 case LTTNG_LIVE_ITERATOR_STATUS_UNSUPPORTED:
77 return "LTTNG_LIVE_ITERATOR_STATUS_UNSUPPORTED";
78 default:
79 abort();
80 }
81}
82
83static
84const char *print_state(struct lttng_live_stream_iterator *s)
85{
86 switch (s->state) {
87 case LTTNG_LIVE_STREAM_ACTIVE_NO_DATA:
88 return "ACTIVE_NO_DATA";
89 case LTTNG_LIVE_STREAM_QUIESCENT_NO_DATA:
90 return "QUIESCENT_NO_DATA";
91 case LTTNG_LIVE_STREAM_QUIESCENT:
92 return "QUIESCENT";
93 case LTTNG_LIVE_STREAM_ACTIVE_DATA:
94 return "ACTIVE_DATA";
95 case LTTNG_LIVE_STREAM_EOF:
96 return "EOF";
97 default:
98 return "ERROR";
99 }
100}
101
102#define print_stream_state(live_stream_iter) \
103 do { \
104 BT_COMP_LOGD("stream state %s last_inact_ts %" PRId64 \
105 ", curr_inact_ts %" PRId64, \
106 print_state(live_stream_iter), \
107 live_stream_iter->last_inactivity_ts, \
108 live_stream_iter->current_inactivity_ts); \
109 } while (0);
110
111BT_HIDDEN
112bool lttng_live_graph_is_canceled(struct lttng_live_msg_iter *msg_iter)
113{
114 bool ret;
115
116 if (!msg_iter) {
117 ret = false;
118 goto end;
119 }
120
121 ret = bt_self_message_iterator_is_interrupted(
122 msg_iter->self_msg_iter);
123
124end:
125 return ret;
126}
127
128static
129struct lttng_live_trace *lttng_live_find_trace(struct lttng_live_session *session,
130 uint64_t trace_id)
131{
132 uint64_t trace_idx;
133 struct lttng_live_trace *ret_trace = NULL;
134
135 for (trace_idx = 0; trace_idx < session->traces->len; trace_idx++) {
136 struct lttng_live_trace *trace =
137 g_ptr_array_index(session->traces, trace_idx);
138 if (trace->id == trace_id) {
139 ret_trace = trace;
140 goto end;
141 }
142 }
143
144end:
145 return ret_trace;
146}
147
148static
149void lttng_live_destroy_trace(struct lttng_live_trace *trace)
150{
151 bt_logging_level log_level = trace->log_level;
152 bt_self_component *self_comp = trace->self_comp;
153
154 BT_COMP_LOGD("Destroy lttng_live_trace");
155
156 BT_ASSERT(trace->stream_iterators);
157 g_ptr_array_free(trace->stream_iterators, TRUE);
158
159 BT_TRACE_PUT_REF_AND_RESET(trace->trace);
160 BT_TRACE_CLASS_PUT_REF_AND_RESET(trace->trace_class);
161
162 lttng_live_metadata_fini(trace);
163 g_free(trace);
164}
165
166static
167struct lttng_live_trace *lttng_live_create_trace(struct lttng_live_session *session,
168 uint64_t trace_id)
169{
170 struct lttng_live_trace *trace = NULL;
171 bt_logging_level log_level = session->log_level;
172 bt_self_component *self_comp = session->self_comp;
173
174 trace = g_new0(struct lttng_live_trace, 1);
175 if (!trace) {
176 goto error;
177 }
178 trace->log_level = session->log_level;
179 trace->self_comp = session->self_comp;
180 trace->session = session;
181 trace->id = trace_id;
182 trace->trace_class = NULL;
183 trace->trace = NULL;
184 trace->stream_iterators = g_ptr_array_new_with_free_func(
185 (GDestroyNotify) lttng_live_stream_iterator_destroy);
186 BT_ASSERT(trace->stream_iterators);
187 trace->new_metadata_needed = true;
188 g_ptr_array_add(session->traces, trace);
189
190 BT_COMP_LOGI("Create trace");
191 goto end;
192error:
193 g_free(trace);
194 trace = NULL;
195end:
196 return trace;
197}
198
199BT_HIDDEN
200struct lttng_live_trace *lttng_live_borrow_trace(
201 struct lttng_live_session *session, uint64_t trace_id)
202{
203 struct lttng_live_trace *trace;
204
205 trace = lttng_live_find_trace(session, trace_id);
206 if (trace) {
207 goto end;
208 }
209
210 /* The session is the owner of the newly created trace. */
211 trace = lttng_live_create_trace(session, trace_id);
212
213end:
214 return trace;
215}
216
217BT_HIDDEN
218int lttng_live_add_session(struct lttng_live_msg_iter *lttng_live_msg_iter,
219 uint64_t session_id, const char *hostname,
220 const char *session_name)
221{
222 int ret = 0;
223 struct lttng_live_session *session;
224 bt_logging_level log_level = lttng_live_msg_iter->log_level;
225 bt_self_component *self_comp = lttng_live_msg_iter->self_comp;
226
227 session = g_new0(struct lttng_live_session, 1);
228 if (!session) {
229 goto error;
230 }
231
232 session->log_level = lttng_live_msg_iter->log_level;
233 session->self_comp = lttng_live_msg_iter->self_comp;
234 session->id = session_id;
235 session->traces = g_ptr_array_new_with_free_func(
236 (GDestroyNotify) lttng_live_destroy_trace);
237 BT_ASSERT(session->traces);
238 session->lttng_live_msg_iter = lttng_live_msg_iter;
239 session->new_streams_needed = true;
240 session->hostname = g_string_new(hostname);
241 BT_ASSERT(session->hostname);
242
243 session->session_name = g_string_new(session_name);
244 BT_ASSERT(session->session_name);
245
246 BT_COMP_LOGI("Reading from session: %" PRIu64 " hostname: %s session_name: %s",
247 session->id, hostname, session_name);
248 g_ptr_array_add(lttng_live_msg_iter->sessions, session);
249 goto end;
250error:
251 BT_COMP_LOGE("Error adding session");
252 g_free(session);
253 ret = -1;
254end:
255 return ret;
256}
257
258static
259void lttng_live_destroy_session(struct lttng_live_session *session)
260{
261 bt_logging_level log_level;
262 bt_self_component *self_comp;
263
264 if (!session) {
265 goto end;
266 }
267
268 log_level = session->log_level;
269 self_comp = session->self_comp;
270 BT_COMP_LOGD("Destroy lttng live session");
271 if (session->id != -1ULL) {
272 if (lttng_live_detach_session(session)) {
273 if (session->lttng_live_msg_iter &&
274 !lttng_live_graph_is_canceled(
275 session->lttng_live_msg_iter)) {
276 /* Old relayd cannot detach sessions. */
277 BT_COMP_LOGD("Unable to detach lttng live session %" PRIu64,
278 session->id);
279 }
280 }
281 session->id = -1ULL;
282 }
283
284 if (session->traces) {
285 g_ptr_array_free(session->traces, TRUE);
286 }
287
288 if (session->hostname) {
289 g_string_free(session->hostname, TRUE);
290 }
291 if (session->session_name) {
292 g_string_free(session->session_name, TRUE);
293 }
294 g_free(session);
295
296end:
297 return;
298}
299
300static
301void lttng_live_msg_iter_destroy(struct lttng_live_msg_iter *lttng_live_msg_iter)
302{
303 if (!lttng_live_msg_iter) {
304 goto end;
305 }
306
307 if (lttng_live_msg_iter->sessions) {
308 g_ptr_array_free(lttng_live_msg_iter->sessions, TRUE);
309 }
310
311 BT_OBJECT_PUT_REF_AND_RESET(lttng_live_msg_iter->viewer_connection);
312 BT_ASSERT(lttng_live_msg_iter->lttng_live_comp);
313 BT_ASSERT(lttng_live_msg_iter->lttng_live_comp->has_msg_iter);
314
315 /* All stream iterators must be destroyed at this point. */
316 BT_ASSERT(lttng_live_msg_iter->active_stream_iter == 0);
317 lttng_live_msg_iter->lttng_live_comp->has_msg_iter = false;
318
319 g_free(lttng_live_msg_iter);
320
321end:
322 return;
323}
324
325BT_HIDDEN
326void lttng_live_msg_iter_finalize(bt_self_message_iterator *self_msg_iter)
327{
328 struct lttng_live_msg_iter *lttng_live_msg_iter;
329
330 BT_ASSERT(self_msg_iter);
331
332 lttng_live_msg_iter = bt_self_message_iterator_get_data(self_msg_iter);
333 BT_ASSERT(lttng_live_msg_iter);
334 lttng_live_msg_iter_destroy(lttng_live_msg_iter);
335}
336
337static
338enum lttng_live_iterator_status lttng_live_iterator_next_check_stream_state(
339 struct lttng_live_stream_iterator *lttng_live_stream)
340{
341 bt_logging_level log_level = lttng_live_stream->log_level;
342 bt_self_component *self_comp = lttng_live_stream->self_comp;
343
344 switch (lttng_live_stream->state) {
345 case LTTNG_LIVE_STREAM_QUIESCENT:
346 case LTTNG_LIVE_STREAM_ACTIVE_DATA:
347 break;
348 case LTTNG_LIVE_STREAM_ACTIVE_NO_DATA:
349 /* Invalid state. */
350 BT_COMP_LOGF("Unexpected stream state \"ACTIVE_NO_DATA\"");
351 abort();
352 case LTTNG_LIVE_STREAM_QUIESCENT_NO_DATA:
353 /* Invalid state. */
354 BT_COMP_LOGF("Unexpected stream state \"QUIESCENT_NO_DATA\"");
355 abort();
356 case LTTNG_LIVE_STREAM_EOF:
357 break;
358 }
359 return LTTNG_LIVE_ITERATOR_STATUS_OK;
360}
361
362/*
363 * For active no data stream, fetch next data. It can be either:
364 * - quiescent: need to put it in the prio heap at quiescent end
365 * timestamp,
366 * - have data: need to wire up first event into the prio heap,
367 * - have no data on this stream at this point: need to retry (AGAIN) or
368 * return EOF.
369 */
370static
371enum lttng_live_iterator_status lttng_live_iterator_next_handle_one_no_data_stream(
372 struct lttng_live_msg_iter *lttng_live_msg_iter,
373 struct lttng_live_stream_iterator *lttng_live_stream)
374{
375 bt_logging_level log_level = lttng_live_msg_iter->log_level;
376 bt_self_component *self_comp = lttng_live_msg_iter->self_comp;
377 enum lttng_live_iterator_status ret =
378 LTTNG_LIVE_ITERATOR_STATUS_OK;
379 struct packet_index index;
380 enum lttng_live_stream_state orig_state = lttng_live_stream->state;
381
382 if (lttng_live_stream->trace->new_metadata_needed) {
383 ret = LTTNG_LIVE_ITERATOR_STATUS_CONTINUE;
384 goto end;
385 }
386 if (lttng_live_stream->trace->session->new_streams_needed) {
387 ret = LTTNG_LIVE_ITERATOR_STATUS_CONTINUE;
388 goto end;
389 }
390 if (lttng_live_stream->state != LTTNG_LIVE_STREAM_ACTIVE_NO_DATA &&
391 lttng_live_stream->state != LTTNG_LIVE_STREAM_QUIESCENT_NO_DATA) {
392 goto end;
393 }
394 ret = lttng_live_get_next_index(lttng_live_msg_iter, lttng_live_stream,
395 &index);
396 if (ret != LTTNG_LIVE_ITERATOR_STATUS_OK) {
397 goto end;
398 }
399 BT_ASSERT(lttng_live_stream->state != LTTNG_LIVE_STREAM_EOF);
400 if (lttng_live_stream->state == LTTNG_LIVE_STREAM_QUIESCENT) {
401 uint64_t last_inact_ts = lttng_live_stream->last_inactivity_ts,
402 curr_inact_ts = lttng_live_stream->current_inactivity_ts;
403
404 if (orig_state == LTTNG_LIVE_STREAM_QUIESCENT_NO_DATA &&
405 last_inact_ts == curr_inact_ts) {
406 ret = LTTNG_LIVE_ITERATOR_STATUS_AGAIN;
407 print_stream_state(lttng_live_stream);
408 } else {
409 ret = LTTNG_LIVE_ITERATOR_STATUS_CONTINUE;
410 }
411 goto end;
412 }
413 lttng_live_stream->base_offset = index.offset;
414 lttng_live_stream->offset = index.offset;
415 lttng_live_stream->len = index.packet_size / CHAR_BIT;
416end:
417 if (ret == LTTNG_LIVE_ITERATOR_STATUS_OK) {
418 ret = lttng_live_iterator_next_check_stream_state(lttng_live_stream);
419 }
420 return ret;
421}
422
423/*
424 * Creation of the message requires the ctf trace class to be created
425 * beforehand, but the live protocol gives us all streams (including metadata)
426 * at once. So we split it in three steps: getting streams, getting metadata
427 * (which creates the ctf trace class), and then creating the per-stream
428 * messages.
429 */
430static
431enum lttng_live_iterator_status lttng_live_get_session(
432 struct lttng_live_msg_iter *lttng_live_msg_iter,
433 struct lttng_live_session *session)
434{
435 enum lttng_live_iterator_status status;
436 uint64_t trace_idx;
437 int ret = 0;
438
439 if (!session->attached) {
440 ret = lttng_live_attach_session(session);
441 if (ret) {
442 if (lttng_live_msg_iter && lttng_live_graph_is_canceled(
443 lttng_live_msg_iter)) {
444 status = LTTNG_LIVE_ITERATOR_STATUS_AGAIN;
445 } else {
446 status = LTTNG_LIVE_ITERATOR_STATUS_ERROR;
447 }
448 goto end;
449 }
450 }
451
452 status = lttng_live_get_new_streams(session);
453 if (status != LTTNG_LIVE_ITERATOR_STATUS_OK &&
454 status != LTTNG_LIVE_ITERATOR_STATUS_END) {
455 goto end;
456 }
457 for (trace_idx = 0; trace_idx < session->traces->len; trace_idx++) {
458 struct lttng_live_trace *trace =
459 g_ptr_array_index(session->traces, trace_idx);
460
461 status = lttng_live_metadata_update(trace);
462 if (status != LTTNG_LIVE_ITERATOR_STATUS_OK &&
463 status != LTTNG_LIVE_ITERATOR_STATUS_END) {
464 goto end;
465 }
466 }
467 status = lttng_live_lazy_msg_init(session);
468
469end:
470 return status;
471}
472
473BT_HIDDEN
474void lttng_live_need_new_streams(struct lttng_live_msg_iter *lttng_live_msg_iter)
475{
476 uint64_t session_idx;
477
478 for (session_idx = 0; session_idx < lttng_live_msg_iter->sessions->len;
479 session_idx++) {
480 struct lttng_live_session *session =
481 g_ptr_array_index(lttng_live_msg_iter->sessions, session_idx);
482 session->new_streams_needed = true;
483 }
484}
485
486static
487void lttng_live_force_new_streams_and_metadata(struct lttng_live_msg_iter *lttng_live_msg_iter)
488{
489 uint64_t session_idx, trace_idx;
490
491 for (session_idx = 0; session_idx < lttng_live_msg_iter->sessions->len;
492 session_idx++) {
493 struct lttng_live_session *session =
494 g_ptr_array_index(lttng_live_msg_iter->sessions, session_idx);
495 session->new_streams_needed = true;
496 for (trace_idx = 0; trace_idx < session->traces->len;
497 trace_idx++) {
498 struct lttng_live_trace *trace =
499 g_ptr_array_index(session->traces, trace_idx);
500 trace->new_metadata_needed = true;
501 }
502 }
503}
504
505static
506enum lttng_live_iterator_status
507lttng_live_iterator_handle_new_streams_and_metadata(
508 struct lttng_live_msg_iter *lttng_live_msg_iter)
509{
510 enum lttng_live_iterator_status ret =
511 LTTNG_LIVE_ITERATOR_STATUS_OK;
512 uint64_t session_idx = 0, nr_sessions_opened = 0;
513 struct lttng_live_session *session;
514 enum session_not_found_action sess_not_found_act =
515 lttng_live_msg_iter->lttng_live_comp->params.sess_not_found_act;
516
517 /*
518 * In a remotely distant future, we could add a "new
519 * session" flag to the protocol, which would tell us that we
520 * need to query for new sessions even though we have sessions
521 * currently ongoing.
522 */
523 if (lttng_live_msg_iter->sessions->len == 0) {
524 if (sess_not_found_act != SESSION_NOT_FOUND_ACTION_CONTINUE) {
525 ret = LTTNG_LIVE_ITERATOR_STATUS_END;
526 goto end;
527 } else {
528 /*
529 * Retry to create a viewer session for the requested
530 * session name.
531 */
532 if (lttng_live_create_viewer_session(lttng_live_msg_iter)) {
533 ret = LTTNG_LIVE_ITERATOR_STATUS_ERROR;
534 goto end;
535 }
536 }
537 }
538
539 for (session_idx = 0; session_idx < lttng_live_msg_iter->sessions->len;
540 session_idx++) {
541 session = g_ptr_array_index(lttng_live_msg_iter->sessions,
542 session_idx);
543 ret = lttng_live_get_session(lttng_live_msg_iter, session);
544 switch (ret) {
545 case LTTNG_LIVE_ITERATOR_STATUS_OK:
546 break;
547 case LTTNG_LIVE_ITERATOR_STATUS_END:
548 ret = LTTNG_LIVE_ITERATOR_STATUS_OK;
549 break;
550 default:
551 goto end;
552 }
553 if (!session->closed) {
554 nr_sessions_opened++;
555 }
556 }
557end:
558 if (ret == LTTNG_LIVE_ITERATOR_STATUS_OK &&
559 sess_not_found_act != SESSION_NOT_FOUND_ACTION_CONTINUE &&
560 nr_sessions_opened == 0) {
561 ret = LTTNG_LIVE_ITERATOR_STATUS_END;
562 }
563 return ret;
564}
565
566static
567enum lttng_live_iterator_status emit_inactivity_message(
568 struct lttng_live_msg_iter *lttng_live_msg_iter,
569 struct lttng_live_stream_iterator *stream_iter,
570 bt_message **message, uint64_t timestamp)
571{
572 enum lttng_live_iterator_status ret =
573 LTTNG_LIVE_ITERATOR_STATUS_OK;
574 bt_message *msg = NULL;
575
576 BT_ASSERT(stream_iter->trace->clock_class);
577
578 msg = bt_message_message_iterator_inactivity_create(
579 lttng_live_msg_iter->self_msg_iter,
580 stream_iter->trace->clock_class,
581 timestamp);
582 if (!msg) {
583 goto error;
584 }
585
586 *message = msg;
587end:
588 return ret;
589
590error:
591 ret = LTTNG_LIVE_ITERATOR_STATUS_ERROR;
592 bt_message_put_ref(msg);
593 goto end;
594}
595
596static
597enum lttng_live_iterator_status lttng_live_iterator_next_handle_one_quiescent_stream(
598 struct lttng_live_msg_iter *lttng_live_msg_iter,
599 struct lttng_live_stream_iterator *lttng_live_stream,
600 bt_message **message)
601{
602 enum lttng_live_iterator_status ret =
603 LTTNG_LIVE_ITERATOR_STATUS_OK;
604
605 if (lttng_live_stream->state != LTTNG_LIVE_STREAM_QUIESCENT) {
606 return LTTNG_LIVE_ITERATOR_STATUS_OK;
607 }
608
609 if (lttng_live_stream->current_inactivity_ts ==
610 lttng_live_stream->last_inactivity_ts) {
611 lttng_live_stream->state = LTTNG_LIVE_STREAM_QUIESCENT_NO_DATA;
612 ret = LTTNG_LIVE_ITERATOR_STATUS_CONTINUE;
613 goto end;
614 }
615
616 ret = emit_inactivity_message(lttng_live_msg_iter, lttng_live_stream,
617 message, lttng_live_stream->current_inactivity_ts);
618
619 lttng_live_stream->last_inactivity_ts =
620 lttng_live_stream->current_inactivity_ts;
621end:
622 return ret;
623}
624
625static
626int live_get_msg_ts_ns(struct lttng_live_stream_iterator *stream_iter,
627 struct lttng_live_msg_iter *lttng_live_msg_iter,
628 const bt_message *msg, int64_t last_msg_ts_ns,
629 int64_t *ts_ns)
630{
631 const bt_clock_class *clock_class = NULL;
632 const bt_clock_snapshot *clock_snapshot = NULL;
633 int ret = 0;
634 bt_logging_level log_level = lttng_live_msg_iter->log_level;
635 bt_self_component *self_comp = lttng_live_msg_iter->self_comp;
636
637 BT_ASSERT(msg);
638 BT_ASSERT(ts_ns);
639
640 BT_COMP_LOGD("Getting message's timestamp: iter-data-addr=%p, msg-addr=%p, "
641 "last-msg-ts=%" PRId64, lttng_live_msg_iter, msg,
642 last_msg_ts_ns);
643
644 switch (bt_message_get_type(msg)) {
645 case BT_MESSAGE_TYPE_EVENT:
646 clock_class =
647 bt_message_event_borrow_stream_class_default_clock_class_const(
648 msg);
649 BT_ASSERT(clock_class);
650
651 clock_snapshot = bt_message_event_borrow_default_clock_snapshot_const(
652 msg);
653 break;
654 case BT_MESSAGE_TYPE_PACKET_BEGINNING:
655 clock_class =
656 bt_message_packet_beginning_borrow_stream_class_default_clock_class_const(
657 msg);
658 BT_ASSERT(clock_class);
659
660 clock_snapshot = bt_message_packet_beginning_borrow_default_clock_snapshot_const(
661 msg);
662 break;
663 case BT_MESSAGE_TYPE_PACKET_END:
664 clock_class =
665 bt_message_packet_end_borrow_stream_class_default_clock_class_const(
666 msg);
667 BT_ASSERT(clock_class);
668
669 clock_snapshot = bt_message_packet_end_borrow_default_clock_snapshot_const(
670 msg);
671 break;
672 case BT_MESSAGE_TYPE_DISCARDED_EVENTS:
673 clock_class =
674 bt_message_discarded_events_borrow_stream_class_default_clock_class_const(
675 msg);
676 BT_ASSERT(clock_class);
677
678 clock_snapshot = bt_message_discarded_events_borrow_beginning_default_clock_snapshot_const(
679 msg);
680 break;
681 case BT_MESSAGE_TYPE_DISCARDED_PACKETS:
682 clock_class =
683 bt_message_discarded_packets_borrow_stream_class_default_clock_class_const(
684 msg);
685 BT_ASSERT(clock_class);
686
687 clock_snapshot = bt_message_discarded_packets_borrow_beginning_default_clock_snapshot_const(
688 msg);
689 break;
690 case BT_MESSAGE_TYPE_MESSAGE_ITERATOR_INACTIVITY:
691 clock_snapshot =
692 bt_message_message_iterator_inactivity_borrow_default_clock_snapshot_const(
693 msg);
694 break;
695 default:
696 /* All the other messages have a higher priority */
697 BT_COMP_LOGD_STR("Message has no timestamp: using the last message timestamp.");
698 *ts_ns = last_msg_ts_ns;
699 goto end;
700 }
701
702 clock_class = bt_clock_snapshot_borrow_clock_class_const(clock_snapshot);
703 BT_ASSERT(clock_class);
704
705 ret = bt_clock_snapshot_get_ns_from_origin(clock_snapshot, ts_ns);
706 if (ret) {
707 BT_COMP_LOGE("Cannot get nanoseconds from Epoch of clock snapshot: "
708 "clock-snapshot-addr=%p", clock_snapshot);
709 goto error;
710 }
711
712 goto end;
713
714error:
715 ret = -1;
716
717end:
718 if (ret == 0) {
719 BT_COMP_LOGD("Found message's timestamp: "
720 "iter-data-addr=%p, msg-addr=%p, "
721 "last-msg-ts=%" PRId64 ", ts=%" PRId64,
722 lttng_live_msg_iter, msg, last_msg_ts_ns, *ts_ns);
723 }
724
725 return ret;
726}
727
728static
729enum lttng_live_iterator_status lttng_live_iterator_next_handle_one_active_data_stream(
730 struct lttng_live_msg_iter *lttng_live_msg_iter,
731 struct lttng_live_stream_iterator *lttng_live_stream,
732 bt_message **message)
733{
734 enum lttng_live_iterator_status ret = LTTNG_LIVE_ITERATOR_STATUS_OK;
735 enum bt_msg_iter_status status;
736 uint64_t session_idx, trace_idx;
737 bt_logging_level log_level = lttng_live_msg_iter->log_level;
738 bt_self_component *self_comp = lttng_live_msg_iter->self_comp;
739
740 for (session_idx = 0; session_idx < lttng_live_msg_iter->sessions->len;
741 session_idx++) {
742 struct lttng_live_session *session =
743 g_ptr_array_index(lttng_live_msg_iter->sessions, session_idx);
744
745 if (session->new_streams_needed) {
746 ret = LTTNG_LIVE_ITERATOR_STATUS_CONTINUE;
747 goto end;
748 }
749 for (trace_idx = 0; trace_idx < session->traces->len;
750 trace_idx++) {
751 struct lttng_live_trace *trace =
752 g_ptr_array_index(session->traces, trace_idx);
753 if (trace->new_metadata_needed) {
754 ret = LTTNG_LIVE_ITERATOR_STATUS_CONTINUE;
755 goto end;
756 }
757 }
758 }
759
760 if (lttng_live_stream->state != LTTNG_LIVE_STREAM_ACTIVE_DATA) {
761 ret = LTTNG_LIVE_ITERATOR_STATUS_ERROR;
762 goto end;
763 }
764
765 status = bt_msg_iter_get_next_message(lttng_live_stream->msg_iter,
766 lttng_live_msg_iter->self_msg_iter, message);
767 switch (status) {
768 case BT_MSG_ITER_STATUS_EOF:
769 ret = LTTNG_LIVE_ITERATOR_STATUS_END;
770 break;
771 case BT_MSG_ITER_STATUS_OK:
772 ret = LTTNG_LIVE_ITERATOR_STATUS_OK;
773 break;
774 case BT_MSG_ITER_STATUS_AGAIN:
775 /*
776 * Continue immediately (end of packet). The next
777 * get_index may return AGAIN to delay the following
778 * attempt.
779 */
780 ret = LTTNG_LIVE_ITERATOR_STATUS_CONTINUE;
781 break;
782 case BT_MSG_ITER_STATUS_INVAL:
783 /* No argument provided by the user, so don't return INVAL. */
784 case BT_MSG_ITER_STATUS_ERROR:
785 default:
786 ret = LTTNG_LIVE_ITERATOR_STATUS_ERROR;
787 BT_COMP_LOGW("CTF msg iterator return an error or failed msg_iter=%p",
788 lttng_live_stream->msg_iter);
789 break;
790 }
791
792end:
793 return ret;
794}
795
796/*
797 * helper function:
798 * handle_no_data_streams()
799 * retry:
800 * - for each ACTIVE_NO_DATA stream:
801 * - query relayd for stream data, or quiescence info.
802 * - if need metadata, get metadata, goto retry.
803 * - if new stream, get new stream as ACTIVE_NO_DATA, goto retry
804 * - if quiescent, move to QUIESCENT streams
805 * - if fetched data, move to ACTIVE_DATA streams
806 * (at this point each stream either has data, or is quiescent)
807 *
808 *
809 * iterator_next:
810 * handle_new_streams_and_metadata()
811 * - query relayd for known streams, add them as ACTIVE_NO_DATA
812 * - query relayd for metadata
813 *
814 * call handle_active_no_data_streams()
815 *
816 * handle_quiescent_streams()
817 * - if at least one stream is ACTIVE_DATA:
818 * - peek stream event with lowest timestamp -> next_ts
819 * - for each quiescent stream
820 * - if next_ts >= quiescent end
821 * - set state to ACTIVE_NO_DATA
822 * - else
823 * - for each quiescent stream
824 * - set state to ACTIVE_NO_DATA
825 *
826 * call handle_active_no_data_streams()
827 *
828 * handle_active_data_streams()
829 * - if at least one stream is ACTIVE_DATA:
830 * - get stream event with lowest timestamp from heap
831 * - make that stream event the current message.
832 * - move this stream heap position to its next event
833 * - if we need to fetch data from relayd, move
834 * stream to ACTIVE_NO_DATA.
835 * - return OK
836 * - return AGAIN
837 *
838 * end criterion: ctrl-c on client. If relayd exits or the session
839 * closes on the relay daemon side, we keep on waiting for streams.
840 * Eventually handle --end timestamp (also an end criterion).
841 *
842 * When disconnected from relayd: try to re-connect endlessly.
843 */
844static
845enum lttng_live_iterator_status lttng_live_iterator_next_on_stream(
846 struct lttng_live_msg_iter *lttng_live_msg_iter,
847 struct lttng_live_stream_iterator *stream_iter,
848 bt_message **curr_msg)
849{
850 bt_logging_level log_level = lttng_live_msg_iter->log_level;
851 bt_self_component *self_comp = lttng_live_msg_iter->self_comp;
852 enum lttng_live_iterator_status live_status;
853
854retry:
855 print_stream_state(stream_iter);
856 live_status = lttng_live_iterator_handle_new_streams_and_metadata(
857 lttng_live_msg_iter);
858 if (live_status != LTTNG_LIVE_ITERATOR_STATUS_OK) {
859 goto end;
860 }
861 live_status = lttng_live_iterator_next_handle_one_no_data_stream(
862 lttng_live_msg_iter, stream_iter);
863 if (live_status != LTTNG_LIVE_ITERATOR_STATUS_OK) {
864 goto end;
865 }
866 live_status = lttng_live_iterator_next_handle_one_quiescent_stream(
867 lttng_live_msg_iter, stream_iter, curr_msg);
868 if (live_status != LTTNG_LIVE_ITERATOR_STATUS_OK) {
869 BT_ASSERT(!*curr_msg);
870 goto end;
871 }
872 if (*curr_msg) {
873 goto end;
874 }
875 live_status = lttng_live_iterator_next_handle_one_active_data_stream(
876 lttng_live_msg_iter, stream_iter, curr_msg);
877 if (live_status != LTTNG_LIVE_ITERATOR_STATUS_OK) {
878 BT_ASSERT(!*curr_msg);
879 }
880
881end:
882 if (live_status == LTTNG_LIVE_ITERATOR_STATUS_CONTINUE) {
883 goto retry;
884 }
885
886 return live_status;
887}
888
889static
890enum lttng_live_iterator_status next_stream_iterator_for_trace(
891 struct lttng_live_msg_iter *lttng_live_msg_iter,
892 struct lttng_live_trace *live_trace,
893 struct lttng_live_stream_iterator **candidate_stream_iter)
894{
895 struct lttng_live_stream_iterator *curr_candidate_stream_iter = NULL;
896 enum lttng_live_iterator_status stream_iter_status;;
897 int64_t curr_candidate_msg_ts = INT64_MAX;
898 uint64_t stream_iter_idx;
899 bt_logging_level log_level = lttng_live_msg_iter->log_level;
900 bt_self_component *self_comp = lttng_live_msg_iter->self_comp;
901
902 BT_ASSERT(live_trace);
903 BT_ASSERT(live_trace->stream_iterators);
904 /*
905 * Update the current message of every stream iterators of this trace.
906 * The current msg of every stream must have a timestamp equal or
907 * larger than the last message returned by this iterator. We must
908 * ensure monotonicity.
909 */
910 stream_iter_idx = 0;
911 while (stream_iter_idx < live_trace->stream_iterators->len) {
912 bool stream_iter_is_ended = false;
913 struct lttng_live_stream_iterator *stream_iter =
914 g_ptr_array_index(live_trace->stream_iterators,
915 stream_iter_idx);
916
917 /*
918 * Find if there is are now current message for this stream
919 * iterator get it.
920 */
921 while (!stream_iter->current_msg) {
922 bt_message *msg = NULL;
923 int64_t curr_msg_ts_ns = INT64_MAX;
924 stream_iter_status = lttng_live_iterator_next_on_stream(
925 lttng_live_msg_iter, stream_iter, &msg);
926
927 BT_COMP_LOGD("live stream iterator returned status :%s",
928 print_live_iterator_status(stream_iter_status));
929 if (stream_iter_status == LTTNG_LIVE_ITERATOR_STATUS_END) {
930 stream_iter_is_ended = true;
931 break;
932 }
933
934 if (stream_iter_status != LTTNG_LIVE_ITERATOR_STATUS_OK) {
935 goto end;
936 }
937
938 BT_ASSERT(msg);
939
940 /*
941 * Get the timestamp in nanoseconds from origin of this
942 * messsage.
943 */
944 live_get_msg_ts_ns(stream_iter, lttng_live_msg_iter,
945 msg, lttng_live_msg_iter->last_msg_ts_ns,
946 &curr_msg_ts_ns);
947
948 /*
949 * Check if the message of the current live stream
950 * iterator occured at the exact same time or after the
951 * last message returned by this component's message
952 * iterator. If not, we return an error.
953 */
954 if (curr_msg_ts_ns >= lttng_live_msg_iter->last_msg_ts_ns) {
955 stream_iter->current_msg = msg;
956 stream_iter->current_msg_ts_ns = curr_msg_ts_ns;
957 } else {
958 /*
959 * We received a message in the past. To ensure
960 * monotonicity, we can't send it forward.
961 */
962 BT_COMP_LOGE("Message's timestamp is less than "
963 "lttng-live's message iterator's last "
964 "returned timestamp: "
965 "lttng-live-msg-iter-addr=%p, ts=%" PRId64 ", "
966 "last-msg-ts=%" PRId64,
967 lttng_live_msg_iter, curr_msg_ts_ns,
968 lttng_live_msg_iter->last_msg_ts_ns);
969 stream_iter_status = LTTNG_LIVE_ITERATOR_STATUS_ERROR;
970 goto end;
971 }
972 }
973
974 if (!stream_iter_is_ended &&
975 stream_iter->current_msg_ts_ns <= curr_candidate_msg_ts) {
976 /*
977 * Update the current best candidate message for the
978 * stream iterator of thise live trace to be forwarded
979 * downstream.
980 */
981 curr_candidate_msg_ts = stream_iter->current_msg_ts_ns;
982 curr_candidate_stream_iter = stream_iter;
983 }
984
985 if (stream_iter_is_ended) {
986 /*
987 * The live stream iterator is ENDed. We remove that
988 * iterator from the list and we restart the iteration
989 * at the beginning of the live stream iterator array
990 * to because the removal will shuffle the array.
991 */
992 g_ptr_array_remove_index_fast(live_trace->stream_iterators,
993 stream_iter_idx);
994 stream_iter_idx = 0;
995 } else {
996 stream_iter_idx++;
997 }
998 }
999
1000 if (curr_candidate_stream_iter) {
1001 *candidate_stream_iter = curr_candidate_stream_iter;
1002 stream_iter_status = LTTNG_LIVE_ITERATOR_STATUS_OK;
1003 } else {
1004 /*
1005 * The only case where we don't have a candidate for this trace
1006 * is if we reached the end of all the iterators.
1007 */
1008 BT_ASSERT(live_trace->stream_iterators->len == 0);
1009 stream_iter_status = LTTNG_LIVE_ITERATOR_STATUS_END;
1010 }
1011
1012end:
1013 return stream_iter_status;
1014}
1015
1016static
1017enum lttng_live_iterator_status next_stream_iterator_for_session(
1018 struct lttng_live_msg_iter *lttng_live_msg_iter,
1019 struct lttng_live_session *session,
1020 struct lttng_live_stream_iterator **candidate_session_stream_iter)
1021{
1022 enum lttng_live_iterator_status stream_iter_status;
1023 uint64_t trace_idx = 0;
1024 int64_t curr_candidate_msg_ts = INT64_MAX;
1025 struct lttng_live_stream_iterator *curr_candidate_stream_iter = NULL;
1026
1027 /*
1028 * Make sure we are attached to the session and look for new streams
1029 * and metadata.
1030 */
1031 stream_iter_status = lttng_live_get_session(lttng_live_msg_iter, session);
1032 if (stream_iter_status != LTTNG_LIVE_ITERATOR_STATUS_OK &&
1033 stream_iter_status != LTTNG_LIVE_ITERATOR_STATUS_CONTINUE &&
1034 stream_iter_status != LTTNG_LIVE_ITERATOR_STATUS_END) {
1035 goto end;
1036 }
1037
1038 BT_ASSERT(session->traces);
1039
1040 /*
1041 * Use while loops here rather then for loops so we can restart the
1042 * iteration if an element is removed from the array during the
1043 * looping.
1044 */
1045 while (trace_idx < session->traces->len) {
1046 bool trace_is_ended = false;
1047 struct lttng_live_stream_iterator *stream_iter;
1048 struct lttng_live_trace *trace =
1049 g_ptr_array_index(session->traces, trace_idx);
1050
1051 stream_iter_status = next_stream_iterator_for_trace(
1052 lttng_live_msg_iter, trace, &stream_iter);
1053 if (stream_iter_status == LTTNG_LIVE_ITERATOR_STATUS_END) {
1054 /*
1055 * All the live stream iterators for this trace are
1056 * ENDed. Remove the trace from this session.
1057 */
1058 trace_is_ended = true;
1059 } else if (stream_iter_status != LTTNG_LIVE_ITERATOR_STATUS_OK) {
1060 goto end;
1061 }
1062
1063 if (!trace_is_ended) {
1064 BT_ASSERT(stream_iter);
1065
1066 if (stream_iter->current_msg_ts_ns <= curr_candidate_msg_ts) {
1067 curr_candidate_msg_ts = stream_iter->current_msg_ts_ns;
1068 curr_candidate_stream_iter = stream_iter;
1069 }
1070 trace_idx++;
1071 } else {
1072 g_ptr_array_remove_index_fast(session->traces, trace_idx);
1073 trace_idx = 0;
1074 }
1075 }
1076 if (curr_candidate_stream_iter) {
1077 *candidate_session_stream_iter = curr_candidate_stream_iter;
1078 stream_iter_status = LTTNG_LIVE_ITERATOR_STATUS_OK;
1079 } else {
1080 /*
1081 * The only cases where we don't have a candidate for this
1082 * trace is:
1083 * 1. if we reached the end of all the iterators of all the
1084 * traces of this session,
1085 * 2. if we never had live stream iterator in the first place.
1086 *
1087 * In either cases, we return END.
1088 */
1089 BT_ASSERT(session->traces->len == 0);
1090 stream_iter_status = LTTNG_LIVE_ITERATOR_STATUS_END;
1091 }
1092end:
1093 return stream_iter_status;
1094}
1095
1096static inline
1097void put_messages(bt_message_array_const msgs, uint64_t count)
1098{
1099 uint64_t i;
1100
1101 for (i = 0; i < count; i++) {
1102 BT_MESSAGE_PUT_REF_AND_RESET(msgs[i]);
1103 }
1104}
1105
1106BT_HIDDEN
1107bt_component_class_message_iterator_next_method_status lttng_live_msg_iter_next(
1108 bt_self_message_iterator *self_msg_it,
1109 bt_message_array_const msgs, uint64_t capacity,
1110 uint64_t *count)
1111{
1112 bt_component_class_message_iterator_next_method_status status;
1113 struct lttng_live_msg_iter *lttng_live_msg_iter =
1114 bt_self_message_iterator_get_data(self_msg_it);
1115 struct lttng_live_component *lttng_live =
1116 lttng_live_msg_iter->lttng_live_comp;
1117 enum lttng_live_iterator_status stream_iter_status;
1118 uint64_t session_idx;
1119
1120 *count = 0;
1121
1122 BT_ASSERT(lttng_live_msg_iter);
1123
1124 /*
1125 * Clear all the invalid message reference that might be left over in
1126 * the output array.
1127 */
1128 memset(msgs, 0, capacity * sizeof(*msgs));
1129
1130 /*
1131 * If no session are exposed on the relay found at the url provided by
1132 * the user, session count will be 0. In this case, we return status
1133 * end to return gracefully.
1134 */
1135 if (lttng_live_msg_iter->sessions->len == 0) {
1136 if (lttng_live->params.sess_not_found_act !=
1137 SESSION_NOT_FOUND_ACTION_CONTINUE) {
1138 status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_END;
1139 goto no_session;
1140 } else {
1141 /*
1142 * The are no more active session for this session
1143 * name. Retry to create a viewer session for the
1144 * requested session name.
1145 */
1146 if (lttng_live_create_viewer_session(lttng_live_msg_iter)) {
1147 status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_ERROR;
1148 goto no_session;
1149 }
1150 }
1151 }
1152
1153 if (lttng_live_msg_iter->active_stream_iter == 0) {
1154 lttng_live_force_new_streams_and_metadata(lttng_live_msg_iter);
1155 }
1156
1157 /*
1158 * Here the muxing of message is done.
1159 *
1160 * We need to iterate over all the streams of all the traces of all the
1161 * viewer sessions in order to get the message with the smallest
1162 * timestamp. In this case, a session is a viewer session and there is
1163 * one viewer session per consumer daemon. (UST 32bit, UST 64bit and/or
1164 * kernel). Each viewer session can have multiple traces, for example,
1165 * 64bit UST viewer sessions could have multiple per-pid traces.
1166 *
1167 * We iterate over the streams of each traces to update and see what is
1168 * their next message's timestamp. From those timestamps, we select the
1169 * message with the smallest timestamp as the best candidate message
1170 * for that trace and do the same thing across all the sessions.
1171 *
1172 * We then compare the timestamp of best candidate message of all the
1173 * sessions to pick the message with the smallest timestamp and we
1174 * return it.
1175 */
1176 while (*count < capacity) {
1177 struct lttng_live_stream_iterator *next_stream_iter = NULL,
1178 *candidate_stream_iter = NULL;
1179 int64_t next_msg_ts_ns = INT64_MAX;
1180
1181 BT_ASSERT(lttng_live_msg_iter->sessions);
1182 session_idx = 0;
1183 /*
1184 * Use a while loop instead of a for loop so we can restart the
1185 * iteration if we remove an element. We can safely call
1186 * next_stream_iterator_for_session() multiple times on the
1187 * same session as we only fetch a new message if there is no
1188 * current next message for each live stream iterator.
1189 * If all live stream iterator of that session already have a
1190 * current next message, the function will simply exit return
1191 * the same candidate live stream iterator every time.
1192 */
1193 while (session_idx < lttng_live_msg_iter->sessions->len) {
1194 struct lttng_live_session *session =
1195 g_ptr_array_index(lttng_live_msg_iter->sessions,
1196 session_idx);
1197
1198 /* Find the best candidate message to send downstream. */
1199 stream_iter_status = next_stream_iterator_for_session(
1200 lttng_live_msg_iter, session,
1201 &candidate_stream_iter);
1202
1203 /* If we receive an END status, it means that either:
1204 * - Those traces never had active streams (UST with no
1205 * data produced yet),
1206 * - All live stream iterators have ENDed.*/
1207 if (stream_iter_status == LTTNG_LIVE_ITERATOR_STATUS_END) {
1208 if (session->closed && session->traces->len == 0) {
1209 /*
1210 * Remove the session from the list and restart the
1211 * iteration at the beginning of the array since the
1212 * removal shuffle the elements of the array.
1213 */
1214 g_ptr_array_remove_index_fast(
1215 lttng_live_msg_iter->sessions,
1216 session_idx);
1217 session_idx = 0;
1218 } else {
1219 session_idx++;
1220 }
1221 continue;
1222 }
1223
1224 if (stream_iter_status != LTTNG_LIVE_ITERATOR_STATUS_OK) {
1225 goto end;
1226 }
1227
1228 if (candidate_stream_iter->current_msg_ts_ns <= next_msg_ts_ns) {
1229 next_msg_ts_ns = candidate_stream_iter->current_msg_ts_ns;
1230 next_stream_iter = candidate_stream_iter;
1231 }
1232
1233 session_idx++;
1234 }
1235
1236 if (!next_stream_iter) {
1237 stream_iter_status = LTTNG_LIVE_ITERATOR_STATUS_AGAIN;
1238 goto end;
1239 }
1240
1241 BT_ASSERT(next_stream_iter->current_msg);
1242 /* Ensure monotonicity. */
1243 BT_ASSERT(lttng_live_msg_iter->last_msg_ts_ns <=
1244 next_stream_iter->current_msg_ts_ns);
1245
1246 /*
1247 * Insert the next message to the message batch. This will set
1248 * stream iterator current messsage to NULL so that next time
1249 * we fetch the next message of that stream iterator
1250 */
1251 BT_MESSAGE_MOVE_REF(msgs[*count], next_stream_iter->current_msg);
1252 (*count)++;
1253
1254 /* Update the last timestamp in nanoseconds sent downstream. */
1255 lttng_live_msg_iter->last_msg_ts_ns = next_msg_ts_ns;
1256 next_stream_iter->current_msg_ts_ns = INT64_MAX;
1257
1258 stream_iter_status = LTTNG_LIVE_ITERATOR_STATUS_OK;
1259 }
1260end:
1261 switch (stream_iter_status) {
1262 case LTTNG_LIVE_ITERATOR_STATUS_OK:
1263 case LTTNG_LIVE_ITERATOR_STATUS_AGAIN:
1264 if (*count > 0) {
1265 /*
1266 * We received a again status but we have some messages
1267 * to send downstream. We send them and return OK for
1268 * now. On the next call we return again if there are
1269 * still no new message to send.
1270 */
1271 status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK;
1272 } else {
1273 status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_AGAIN;
1274 }
1275 break;
1276 case LTTNG_LIVE_ITERATOR_STATUS_END:
1277 status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_END;
1278 break;
1279 case LTTNG_LIVE_ITERATOR_STATUS_NOMEM:
1280 status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_MEMORY_ERROR;
1281 break;
1282 case LTTNG_LIVE_ITERATOR_STATUS_ERROR:
1283 case LTTNG_LIVE_ITERATOR_STATUS_INVAL:
1284 case LTTNG_LIVE_ITERATOR_STATUS_UNSUPPORTED:
1285 status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_ERROR;
1286 /* Put all existing messages on error. */
1287 put_messages(msgs, *count);
1288 break;
1289 default:
1290 abort();
1291 }
1292
1293no_session:
1294 return status;
1295}
1296
1297BT_HIDDEN
1298bt_component_class_message_iterator_init_method_status lttng_live_msg_iter_init(
1299 bt_self_message_iterator *self_msg_it,
1300 bt_self_component_source *self_comp_src,
1301 bt_self_component_port_output *self_port)
1302{
1303 bt_component_class_message_iterator_init_method_status ret =
1304 BT_COMPONENT_CLASS_MESSAGE_ITERATOR_INIT_METHOD_STATUS_OK;
1305 bt_self_component *self_comp =
1306 bt_self_component_source_as_self_component(self_comp_src);
1307 struct lttng_live_component *lttng_live;
1308 struct lttng_live_msg_iter *lttng_live_msg_iter;
1309 bt_logging_level log_level;
1310
1311 BT_ASSERT(self_msg_it);
1312
1313 lttng_live = bt_self_component_get_data(self_comp);
1314 log_level = lttng_live->log_level;
1315 self_comp = lttng_live->self_comp;
1316
1317 /* There can be only one downstream iterator at the same time. */
1318 BT_ASSERT(!lttng_live->has_msg_iter);
1319 lttng_live->has_msg_iter = true;
1320
1321 lttng_live_msg_iter = g_new0(struct lttng_live_msg_iter, 1);
1322 if (!lttng_live_msg_iter) {
1323 ret = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_INIT_METHOD_STATUS_ERROR;
1324 goto end;
1325 }
1326
1327 lttng_live_msg_iter->log_level = lttng_live->log_level;
1328 lttng_live_msg_iter->self_comp = lttng_live->self_comp;
1329 lttng_live_msg_iter->lttng_live_comp = lttng_live;
1330 lttng_live_msg_iter->self_msg_iter = self_msg_it;
1331
1332 lttng_live_msg_iter->active_stream_iter = 0;
1333 lttng_live_msg_iter->last_msg_ts_ns = INT64_MIN;
1334 lttng_live_msg_iter->sessions = g_ptr_array_new_with_free_func(
1335 (GDestroyNotify) lttng_live_destroy_session);
1336 BT_ASSERT(lttng_live_msg_iter->sessions);
1337
1338 lttng_live_msg_iter->viewer_connection =
1339 live_viewer_connection_create(lttng_live->params.url->str, false,
1340 lttng_live_msg_iter);
1341 if (!lttng_live_msg_iter->viewer_connection) {
1342 goto error;
1343 }
1344
1345 if (lttng_live_create_viewer_session(lttng_live_msg_iter)) {
1346 goto error;
1347 }
1348 if (lttng_live_msg_iter->sessions->len == 0) {
1349 switch (lttng_live->params.sess_not_found_act) {
1350 case SESSION_NOT_FOUND_ACTION_CONTINUE:
1351 BT_COMP_LOGI("Unable to connect to the requested live viewer "
1352 "session. Keep trying to connect because of "
1353 "%s=\"%s\" component parameter: url=\"%s\"",
1354 SESS_NOT_FOUND_ACTION_PARAM,
1355 SESS_NOT_FOUND_ACTION_CONTINUE_STR,
1356 lttng_live->params.url->str);
1357 break;
1358 case SESSION_NOT_FOUND_ACTION_FAIL:
1359 BT_COMP_LOGE("Unable to connect to the requested live viewer "
1360 "session. Fail the message iterator"
1361 "initialization because of %s=\"%s\" "
1362 "component parameter: url =\"%s\"",
1363 SESS_NOT_FOUND_ACTION_PARAM,
1364 SESS_NOT_FOUND_ACTION_FAIL_STR,
1365 lttng_live->params.url->str);
1366 goto error;
1367 case SESSION_NOT_FOUND_ACTION_END:
1368 BT_COMP_LOGI("Unable to connect to the requested live viewer "
1369 "session. End gracefully at the first _next() "
1370 "call because of %s=\"%s\" component parameter: "
1371 "url=\"%s\"", SESS_NOT_FOUND_ACTION_PARAM,
1372 SESS_NOT_FOUND_ACTION_END_STR,
1373 lttng_live->params.url->str);
1374 break;
1375 }
1376 }
1377
1378 bt_self_message_iterator_set_data(self_msg_it, lttng_live_msg_iter);
1379
1380 goto end;
1381error:
1382 ret = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_INIT_METHOD_STATUS_ERROR;
1383 lttng_live_msg_iter_destroy(lttng_live_msg_iter);
1384end:
1385 return ret;
1386}
1387
1388static
1389bt_component_class_query_method_status lttng_live_query_list_sessions(
1390 const bt_value *params, const bt_value **result,
1391 bt_logging_level log_level)
1392{
1393 bt_component_class_query_method_status status =
1394 BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_OK;
1395 const bt_value *url_value = NULL;
1396 const char *url;
1397 struct live_viewer_connection *viewer_connection = NULL;
1398 bt_self_component *self_comp = NULL;
1399
1400 url_value = bt_value_map_borrow_entry_value_const(params, URL_PARAM);
1401 if (!url_value) {
1402 BT_COMP_LOGE("Mandatory `%s` parameter missing", URL_PARAM);
1403 status = BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_ERROR;
1404 goto error;
1405 }
1406
1407 if (!bt_value_is_string(url_value)) {
1408 BT_COMP_LOGE("`%s` parameter is required to be a string value",
1409 URL_PARAM);
1410 status = BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_ERROR;
1411 goto error;
1412 }
1413
1414 url = bt_value_string_get(url_value);
1415
1416 viewer_connection = live_viewer_connection_create(url, true, NULL);
1417 if (!viewer_connection) {
1418 goto error;
1419 }
1420
1421 status = live_viewer_connection_list_sessions(viewer_connection,
1422 result);
1423 if (status != BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_OK) {
1424 goto error;
1425 }
1426
1427 goto end;
1428
1429error:
1430 BT_VALUE_PUT_REF_AND_RESET(*result);
1431
1432 if (status >= 0) {
1433 status = BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_ERROR;
1434 }
1435
1436end:
1437 if (viewer_connection) {
1438 live_viewer_connection_destroy(viewer_connection);
1439 }
1440 return status;
1441}
1442
1443BT_HIDDEN
1444bt_component_class_query_method_status lttng_live_query(
1445 bt_self_component_class_source *comp_class,
1446 const bt_query_executor *query_exec,
1447 const char *object, const bt_value *params,
1448 bt_logging_level log_level, const bt_value **result)
1449{
1450 bt_component_class_query_method_status status =
1451 BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_OK;
1452 bt_self_component *self_comp = NULL;
1453
1454 if (strcmp(object, "sessions") == 0) {
1455 status = lttng_live_query_list_sessions(params, result,
1456 log_level);
1457 } else {
1458 BT_COMP_LOGI("Unknown query object `%s`", object);
1459 status = BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_INVALID_OBJECT;
1460 goto end;
1461 }
1462
1463end:
1464 return status;
1465}
1466
1467static
1468void lttng_live_component_destroy_data(struct lttng_live_component *lttng_live)
1469{
1470 if (!lttng_live) {
1471 return;
1472 }
1473 if (lttng_live->params.url) {
1474 g_string_free(lttng_live->params.url, TRUE);
1475 }
1476 g_free(lttng_live);
1477}
1478
1479BT_HIDDEN
1480void lttng_live_component_finalize(bt_self_component_source *component)
1481{
1482 void *data = bt_self_component_get_data(
1483 bt_self_component_source_as_self_component(component));
1484
1485 if (!data) {
1486 return;
1487 }
1488 lttng_live_component_destroy_data(data);
1489}
1490
1491static
1492enum session_not_found_action parse_session_not_found_action_param(
1493 const bt_value *no_session_param)
1494{
1495 enum session_not_found_action action;
1496 const char *no_session_act_str;
1497 no_session_act_str = bt_value_string_get(no_session_param);
1498 if (strcmp(no_session_act_str, SESS_NOT_FOUND_ACTION_CONTINUE_STR) == 0) {
1499 action = SESSION_NOT_FOUND_ACTION_CONTINUE;
1500 } else if (strcmp(no_session_act_str, SESS_NOT_FOUND_ACTION_FAIL_STR) == 0) {
1501 action = SESSION_NOT_FOUND_ACTION_FAIL;
1502 } else if (strcmp(no_session_act_str, SESS_NOT_FOUND_ACTION_END_STR) == 0) {
1503 action = SESSION_NOT_FOUND_ACTION_END;
1504 } else {
1505 action = -1;
1506 }
1507
1508 return action;
1509}
1510
1511struct lttng_live_component *lttng_live_component_create(const bt_value *params,
1512 bt_logging_level log_level, bt_self_component *self_comp)
1513{
1514 struct lttng_live_component *lttng_live;
1515 const bt_value *value = NULL;
1516 const char *url;
1517
1518 lttng_live = g_new0(struct lttng_live_component, 1);
1519 if (!lttng_live) {
1520 goto end;
1521 }
1522 lttng_live->log_level = log_level;
1523 lttng_live->self_comp = self_comp;
1524 lttng_live->max_query_size = MAX_QUERY_SIZE;
1525 lttng_live->has_msg_iter = false;
1526
1527 value = bt_value_map_borrow_entry_value_const(params, URL_PARAM);
1528 if (!value || !bt_value_is_string(value)) {
1529 BT_COMP_LOGW("Mandatory `%s` parameter missing or not a string",
1530 URL_PARAM);
1531 goto error;
1532 }
1533 url = bt_value_string_get(value);
1534 lttng_live->params.url = g_string_new(url);
1535 if (!lttng_live->params.url) {
1536 goto error;
1537 }
1538
1539 value = bt_value_map_borrow_entry_value_const(params,
1540 SESS_NOT_FOUND_ACTION_PARAM);
1541
1542 if (value && bt_value_is_string(value)) {
1543 lttng_live->params.sess_not_found_act =
1544 parse_session_not_found_action_param(value);
1545 if (lttng_live->params.sess_not_found_act == -1) {
1546 BT_COMP_LOGE("Unexpected value for `%s` parameter: "
1547 "value=\"%s\"", SESS_NOT_FOUND_ACTION_PARAM,
1548 bt_value_string_get(value));
1549 goto error;
1550 }
1551 } else {
1552 BT_COMP_LOGW("Optional `%s` parameter is missing or "
1553 "not a string value. Defaulting to %s=\"%s\".",
1554 SESS_NOT_FOUND_ACTION_PARAM,
1555 SESS_NOT_FOUND_ACTION_PARAM,
1556 SESS_NOT_FOUND_ACTION_CONTINUE_STR);
1557 lttng_live->params.sess_not_found_act =
1558 SESSION_NOT_FOUND_ACTION_CONTINUE;
1559 }
1560
1561 goto end;
1562
1563error:
1564 lttng_live_component_destroy_data(lttng_live);
1565 lttng_live = NULL;
1566end:
1567 return lttng_live;
1568}
1569
1570BT_HIDDEN
1571bt_component_class_init_method_status lttng_live_component_init(
1572 bt_self_component_source *self_comp_src,
1573 const bt_value *params, __attribute__((unused)) void *init_method_data)
1574{
1575 struct lttng_live_component *lttng_live;
1576 bt_component_class_init_method_status ret =
1577 BT_COMPONENT_CLASS_INIT_METHOD_STATUS_OK;
1578 bt_self_component *self_comp =
1579 bt_self_component_source_as_self_component(self_comp_src);
1580 bt_logging_level log_level = bt_component_get_logging_level(
1581 bt_self_component_as_component(self_comp));
1582 bt_self_component_add_port_status add_port_status;
1583
1584 lttng_live = lttng_live_component_create(params, log_level, self_comp);
1585 if (!lttng_live) {
1586 ret = BT_COMPONENT_CLASS_INIT_METHOD_STATUS_MEMORY_ERROR;
1587 goto error;
1588 }
1589
1590 add_port_status = bt_self_component_source_add_output_port(
1591 self_comp_src, "out", NULL, NULL);
1592 switch (add_port_status) {
1593 case BT_SELF_COMPONENT_ADD_PORT_STATUS_ERROR:
1594 ret = BT_COMPONENT_CLASS_INIT_METHOD_STATUS_ERROR;
1595 goto error;
1596 case BT_SELF_COMPONENT_ADD_PORT_STATUS_MEMORY_ERROR:
1597 ret = BT_COMPONENT_CLASS_INIT_METHOD_STATUS_MEMORY_ERROR;
1598 goto error;
1599 default:
1600 break;
1601 }
1602
1603 bt_self_component_set_data(self_comp, lttng_live);
1604 goto end;
1605
1606error:
1607 lttng_live_component_destroy_data(lttng_live);
1608 lttng_live = NULL;
1609end:
1610 return ret;
1611}
This page took 0.036973 seconds and 4 git commands to generate.