2 * SPDX-License-Identifier: MIT
4 * Copyright 2017 Philippe Proulx <pproulx@efficios.com>
7 #define BT_COMP_LOG_SELF_COMP (muxer_comp->self_comp)
8 #define BT_LOG_OUTPUT_LEVEL (muxer_comp->log_level)
9 #define BT_LOG_TAG "PLUGIN/FLT.UTILS.MUXER"
10 #include "logging/comp-logging.h"
12 #include "common/macros.h"
13 #include "common/uuid.h"
14 #include <babeltrace2/babeltrace.h>
18 #include "common/assert.h"
19 #include "common/common.h"
23 #include "plugins/common/muxing/muxing.h"
24 #include "plugins/common/param-validation/param-validation.h"
30 bt_self_component_filter
*self_comp_flt
;
31 bt_self_component
*self_comp
;
33 unsigned int next_port_num
;
34 size_t available_input_ports
;
35 bool initializing_muxer_msg_iter
;
36 bt_logging_level log_level
;
39 struct muxer_upstream_msg_iter
{
40 struct muxer_comp
*muxer_comp
;
42 /* Owned by this, NULL if ended */
43 bt_message_iterator
*msg_iter
;
45 /* Contains `const bt_message *`, owned by this */
48 /* Index of the next message in `msgs` to return */
52 enum muxer_msg_iter_clock_class_expectation
{
53 MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_ANY
= 0,
54 MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_NONE
,
55 MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_ABSOLUTE
,
56 MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_NOT_ABS_SPEC_UUID
,
57 MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_NOT_ABS_NO_UUID
,
60 struct muxer_msg_iter
{
61 struct muxer_comp
*muxer_comp
;
64 bt_self_message_iterator
*self_msg_iter
;
67 * Array of struct muxer_upstream_msg_iter * (owned by this).
69 * NOTE: This array is searched in linearly to find the youngest
70 * current message. Keep this until benchmarks confirm that
71 * another data structure is faster than this for our typical
74 GPtrArray
*active_muxer_upstream_msg_iters
;
77 * Array of struct muxer_upstream_msg_iter * (owned by this).
79 * We move ended message iterators from
80 * `active_muxer_upstream_msg_iters` to this array so as to be
81 * able to restore them when seeking.
83 GPtrArray
*ended_muxer_upstream_msg_iters
;
85 /* Last time returned in a message */
86 int64_t last_returned_ts_ns
;
88 /* Clock class expectation state */
89 enum muxer_msg_iter_clock_class_expectation clock_class_expectation
;
92 * Expected clock class UUID, only valid when
93 * clock_class_expectation is
94 * MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_NOT_ABS_SPEC_UUID.
96 bt_uuid_t expected_clock_class_uuid
;
99 * Saved error. If we hit an error in the _next method, but have some
100 * messages ready to return, we save the error here and return it on
101 * the next _next call.
103 bt_message_iterator_class_next_method_status next_saved_status
;
104 const struct bt_error
*next_saved_error
;
108 void empty_message_queue(struct muxer_upstream_msg_iter
*upstream_msg_iter
)
110 g_ptr_array_set_size(upstream_msg_iter
->msgs
, 0);
114 void destroy_muxer_upstream_msg_iter(
115 struct muxer_upstream_msg_iter
*muxer_upstream_msg_iter
)
117 struct muxer_comp
*muxer_comp
;
119 if (!muxer_upstream_msg_iter
) {
123 muxer_comp
= muxer_upstream_msg_iter
->muxer_comp
;
124 BT_COMP_LOGD("Destroying muxer's upstream message iterator wrapper: "
125 "addr=%p, msg-iter-addr=%p, queue-len=%u, next-msg=%u",
126 muxer_upstream_msg_iter
,
127 muxer_upstream_msg_iter
->msg_iter
,
128 muxer_upstream_msg_iter
->msgs
->len
,
129 muxer_upstream_msg_iter
->next_msg
);
131 bt_message_iterator_put_ref(
132 muxer_upstream_msg_iter
->msg_iter
);
134 if (muxer_upstream_msg_iter
->msgs
) {
135 g_ptr_array_free(muxer_upstream_msg_iter
->msgs
, TRUE
);
138 g_free(muxer_upstream_msg_iter
);
142 int muxer_msg_iter_add_upstream_msg_iter(struct muxer_msg_iter
*muxer_msg_iter
,
143 bt_message_iterator
*self_msg_iter
)
146 struct muxer_upstream_msg_iter
*muxer_upstream_msg_iter
=
147 g_new0(struct muxer_upstream_msg_iter
, 1);
148 struct muxer_comp
*muxer_comp
= muxer_msg_iter
->muxer_comp
;
150 if (!muxer_upstream_msg_iter
) {
151 BT_COMP_LOGE_APPEND_CAUSE(muxer_comp
->self_comp
,
152 "Failed to allocate one muxer's upstream message iterator wrapper.");
156 muxer_upstream_msg_iter
->muxer_comp
= muxer_comp
;
157 muxer_upstream_msg_iter
->msg_iter
= self_msg_iter
;
158 bt_message_iterator_get_ref(muxer_upstream_msg_iter
->msg_iter
);
159 muxer_upstream_msg_iter
->msgs
=
160 g_ptr_array_new_with_free_func((GDestroyNotify
) bt_message_put_ref
);
161 if (!muxer_upstream_msg_iter
->msgs
) {
162 BT_COMP_LOGE_APPEND_CAUSE(muxer_comp
->self_comp
,
163 "Failed to allocate a GPtrArray.");
167 g_ptr_array_add(muxer_msg_iter
->active_muxer_upstream_msg_iters
,
168 muxer_upstream_msg_iter
);
169 BT_COMP_LOGD("Added muxer's upstream message iterator wrapper: "
170 "addr=%p, muxer-msg-iter-addr=%p, msg-iter-addr=%p",
171 muxer_upstream_msg_iter
, muxer_msg_iter
,
177 destroy_muxer_upstream_msg_iter(muxer_upstream_msg_iter
);
185 bt_self_component_add_port_status
add_available_input_port(
186 bt_self_component_filter
*self_comp
)
188 struct muxer_comp
*muxer_comp
= bt_self_component_get_data(
189 bt_self_component_filter_as_self_component(self_comp
));
190 bt_self_component_add_port_status status
=
191 BT_SELF_COMPONENT_ADD_PORT_STATUS_OK
;
192 GString
*port_name
= NULL
;
194 BT_ASSERT(muxer_comp
);
195 port_name
= g_string_new("in");
197 BT_COMP_LOGE_APPEND_CAUSE(muxer_comp
->self_comp
, "Failed to allocate a GString.");
198 status
= BT_SELF_COMPONENT_ADD_PORT_STATUS_MEMORY_ERROR
;
202 g_string_append_printf(port_name
, "%u", muxer_comp
->next_port_num
);
203 status
= bt_self_component_filter_add_input_port(
204 self_comp
, port_name
->str
, NULL
, NULL
);
205 if (status
!= BT_SELF_COMPONENT_ADD_PORT_STATUS_OK
) {
206 BT_COMP_LOGE_APPEND_CAUSE(muxer_comp
->self_comp
,
207 "Cannot add input port to muxer component: "
208 "port-name=\"%s\", comp-addr=%p, status=%s",
209 port_name
->str
, self_comp
,
210 bt_common_func_status_string(status
));
214 muxer_comp
->available_input_ports
++;
215 muxer_comp
->next_port_num
++;
216 BT_COMP_LOGI("Added one input port to muxer component: "
217 "port-name=\"%s\", comp-addr=%p",
218 port_name
->str
, self_comp
);
222 g_string_free(port_name
, TRUE
);
229 bt_self_component_add_port_status
create_output_port(
230 bt_self_component_filter
*self_comp
)
232 return bt_self_component_filter_add_output_port(
233 self_comp
, "out", NULL
, NULL
);
237 void destroy_muxer_comp(struct muxer_comp
*muxer_comp
)
247 struct bt_param_validation_map_value_entry_descr muxer_params
[] = {
248 BT_PARAM_VALIDATION_MAP_VALUE_ENTRY_END
251 bt_component_class_initialize_method_status
muxer_init(
252 bt_self_component_filter
*self_comp_flt
,
253 bt_self_component_filter_configuration
*config
,
254 const bt_value
*params
, void *init_data
)
256 bt_component_class_initialize_method_status status
;
257 bt_self_component_add_port_status add_port_status
;
258 bt_self_component
*self_comp
=
259 bt_self_component_filter_as_self_component(self_comp_flt
);
260 struct muxer_comp
*muxer_comp
= g_new0(struct muxer_comp
, 1);
261 bt_logging_level log_level
= bt_component_get_logging_level(
262 bt_self_component_as_component(self_comp
));
263 enum bt_param_validation_status validation_status
;
264 gchar
*validate_error
= NULL
;
266 BT_COMP_LOG_CUR_LVL(BT_LOG_INFO
, log_level
, self_comp
,
267 "Initializing muxer component: "
268 "comp-addr=%p, params-addr=%p", self_comp
, params
);
272 * Don't use BT_COMP_LOGE_APPEND_CAUSE, as `muxer_comp` is not
275 BT_COMP_LOG_CUR_LVL(BT_LOG_ERROR
, log_level
, self_comp
,
276 "Failed to allocate one muxer component.");
277 BT_CURRENT_THREAD_ERROR_APPEND_CAUSE_FROM_COMPONENT(self_comp
,
278 "Failed to allocate one muxer component.");
279 status
= BT_COMPONENT_CLASS_INITIALIZE_METHOD_STATUS_MEMORY_ERROR
;
283 muxer_comp
->log_level
= log_level
;
284 muxer_comp
->self_comp
= self_comp
;
285 muxer_comp
->self_comp_flt
= self_comp_flt
;
287 validation_status
= bt_param_validation_validate(params
,
288 muxer_params
, &validate_error
);
289 if (validation_status
== BT_PARAM_VALIDATION_STATUS_MEMORY_ERROR
) {
290 status
= BT_COMPONENT_CLASS_INITIALIZE_METHOD_STATUS_MEMORY_ERROR
;
292 } else if (validation_status
== BT_PARAM_VALIDATION_STATUS_VALIDATION_ERROR
) {
293 status
= BT_COMPONENT_CLASS_INITIALIZE_METHOD_STATUS_ERROR
;
294 BT_COMP_LOGE_APPEND_CAUSE(self_comp
, "%s", validate_error
);
298 bt_self_component_set_data(self_comp
, muxer_comp
);
299 add_port_status
= add_available_input_port(self_comp_flt
);
300 if (add_port_status
!= BT_SELF_COMPONENT_ADD_PORT_STATUS_OK
) {
301 BT_COMP_LOGE_APPEND_CAUSE(self_comp
,
302 "Cannot ensure that at least one muxer component's input port is available: "
303 "muxer-comp-addr=%p, status=%s",
304 muxer_comp
, bt_common_func_status_string(add_port_status
));
305 status
= (int) add_port_status
;
309 add_port_status
= create_output_port(self_comp_flt
);
310 if (add_port_status
!= BT_SELF_COMPONENT_ADD_PORT_STATUS_OK
) {
311 BT_COMP_LOGE_APPEND_CAUSE(self_comp
,
312 "Cannot create muxer component's output port: "
313 "muxer-comp-addr=%p, status=%s",
314 muxer_comp
, bt_common_func_status_string(add_port_status
));
315 status
= (int) add_port_status
;
319 BT_COMP_LOGI("Initialized muxer component: "
320 "comp-addr=%p, params-addr=%p, muxer-comp-addr=%p",
321 self_comp
, params
, muxer_comp
);
323 status
= BT_COMPONENT_CLASS_INITIALIZE_METHOD_STATUS_OK
;
327 destroy_muxer_comp(muxer_comp
);
328 bt_self_component_set_data(self_comp
, NULL
);
331 g_free(validate_error
);
335 void muxer_finalize(bt_self_component_filter
*self_comp
)
337 struct muxer_comp
*muxer_comp
= bt_self_component_get_data(
338 bt_self_component_filter_as_self_component(self_comp
));
340 BT_COMP_LOGI("Finalizing muxer component: comp-addr=%p",
342 destroy_muxer_comp(muxer_comp
);
346 bt_message_iterator_create_from_message_iterator_status
347 create_msg_iter_on_input_port(struct muxer_comp
*muxer_comp
,
348 struct muxer_msg_iter
*muxer_msg_iter
,
349 bt_self_component_port_input
*self_port
,
350 bt_message_iterator
**msg_iter
)
352 const bt_port
*port
= bt_self_component_port_as_port(
353 bt_self_component_port_input_as_self_component_port(
355 bt_message_iterator_create_from_message_iterator_status
359 BT_ASSERT(bt_port_is_connected(port
));
361 // TODO: Advance the iterator to >= the time of the latest
362 // returned message by the muxer message
363 // iterator which creates it.
364 status
= bt_message_iterator_create_from_message_iterator(
365 muxer_msg_iter
->self_msg_iter
, self_port
, msg_iter
);
366 if (status
!= BT_MESSAGE_ITERATOR_CREATE_FROM_MESSAGE_ITERATOR_STATUS_OK
) {
367 BT_COMP_LOGE_APPEND_CAUSE(muxer_comp
->self_comp
,
368 "Cannot create upstream message iterator on input port: "
369 "port-addr=%p, port-name=\"%s\"",
370 port
, bt_port_get_name(port
));
374 BT_COMP_LOGI("Created upstream message iterator on input port: "
375 "port-addr=%p, port-name=\"%s\", msg-iter-addr=%p",
376 port
, bt_port_get_name(port
), msg_iter
);
383 bt_message_iterator_class_next_method_status
muxer_upstream_msg_iter_next(
384 struct muxer_upstream_msg_iter
*muxer_upstream_msg_iter
,
387 struct muxer_comp
*muxer_comp
= muxer_upstream_msg_iter
->muxer_comp
;
388 bt_message_iterator_class_next_method_status status
;
389 bt_message_iterator_next_status input_port_iter_status
;
390 bt_message_array_const msgs
;
394 BT_COMP_LOGD("Calling upstream message iterator's \"next\" method: "
395 "muxer-upstream-msg-iter-wrap-addr=%p, msg-iter-addr=%p",
396 muxer_upstream_msg_iter
,
397 muxer_upstream_msg_iter
->msg_iter
);
398 input_port_iter_status
= bt_message_iterator_next(
399 muxer_upstream_msg_iter
->msg_iter
, &msgs
, &count
);
400 BT_COMP_LOGD("Upstream message iterator's \"next\" method returned: "
402 bt_common_func_status_string(input_port_iter_status
));
404 switch (input_port_iter_status
) {
405 case BT_MESSAGE_ITERATOR_NEXT_STATUS_OK
:
407 * Message iterator's current message is
408 * valid: it must be considered for muxing operations.
410 BT_COMP_LOGD_STR("Validated upstream message iterator wrapper.");
411 BT_ASSERT_DBG(count
> 0);
413 g_ptr_array_set_size(muxer_upstream_msg_iter
->msgs
, count
);
414 muxer_upstream_msg_iter
->next_msg
= 0;
416 /* Move messages to our queue */
417 for (i
= 0; i
< count
; i
++) {
419 * Push to tail in order; other side
420 * (muxer_msg_iter_do_next_one()) consumes
421 * from the head first.
423 g_ptr_array_index(muxer_upstream_msg_iter
->msgs
, i
)
424 = (gpointer
*) msgs
[i
];
426 status
= BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_OK
;
428 case BT_MESSAGE_ITERATOR_NEXT_STATUS_AGAIN
:
430 * Message iterator's current message is not
431 * valid anymore. Return
432 * BT_MESSAGE_ITERATOR_NEXT_STATUS_AGAIN immediately.
434 status
= BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_AGAIN
;
436 case BT_MESSAGE_ITERATOR_NEXT_STATUS_END
: /* Fall-through. */
438 * Message iterator reached the end: release it. It
439 * won't be considered again to find the youngest
443 status
= BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_OK
;
445 case BT_MESSAGE_ITERATOR_NEXT_STATUS_ERROR
:
446 case BT_MESSAGE_ITERATOR_NEXT_STATUS_MEMORY_ERROR
:
447 /* Error status code */
448 BT_COMP_LOGE_APPEND_CAUSE(muxer_comp
->self_comp
,
449 "Upstream iterator's next method returned an error: status=%s",
450 bt_common_func_status_string(input_port_iter_status
));
451 status
= (int) input_port_iter_status
;
454 /* Unsupported status code */
455 BT_COMP_LOGE_APPEND_CAUSE(muxer_comp
->self_comp
,
456 "Unsupported status code: status=%s",
457 bt_common_func_status_string(input_port_iter_status
));
458 status
= BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_ERROR
;
466 int get_msg_ts_ns(struct muxer_comp
*muxer_comp
,
467 struct muxer_msg_iter
*muxer_msg_iter
,
468 const bt_message
*msg
, int64_t last_returned_ts_ns
,
471 const bt_clock_snapshot
*clock_snapshot
= NULL
;
473 const bt_stream_class
*stream_class
= NULL
;
474 bt_message_type msg_type
;
477 BT_ASSERT_DBG(ts_ns
);
478 BT_COMP_LOGD("Getting message's timestamp: "
479 "muxer-msg-iter-addr=%p, msg-addr=%p, "
480 "last-returned-ts=%" PRId64
,
481 muxer_msg_iter
, msg
, last_returned_ts_ns
);
483 if (G_UNLIKELY(muxer_msg_iter
->clock_class_expectation
==
484 MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_NONE
)) {
485 *ts_ns
= last_returned_ts_ns
;
489 msg_type
= bt_message_get_type(msg
);
491 if (G_UNLIKELY(msg_type
== BT_MESSAGE_TYPE_PACKET_BEGINNING
)) {
492 stream_class
= bt_stream_borrow_class_const(
493 bt_packet_borrow_stream_const(
494 bt_message_packet_beginning_borrow_packet_const(
496 } else if (G_UNLIKELY(msg_type
== BT_MESSAGE_TYPE_PACKET_END
)) {
497 stream_class
= bt_stream_borrow_class_const(
498 bt_packet_borrow_stream_const(
499 bt_message_packet_end_borrow_packet_const(
501 } else if (G_UNLIKELY(msg_type
== BT_MESSAGE_TYPE_DISCARDED_EVENTS
)) {
502 stream_class
= bt_stream_borrow_class_const(
503 bt_message_discarded_events_borrow_stream_const(msg
));
504 } else if (G_UNLIKELY(msg_type
== BT_MESSAGE_TYPE_DISCARDED_PACKETS
)) {
505 stream_class
= bt_stream_borrow_class_const(
506 bt_message_discarded_packets_borrow_stream_const(msg
));
510 case BT_MESSAGE_TYPE_EVENT
:
511 BT_ASSERT_DBG(bt_message_event_borrow_stream_class_default_clock_class_const(
513 clock_snapshot
= bt_message_event_borrow_default_clock_snapshot_const(
516 case BT_MESSAGE_TYPE_PACKET_BEGINNING
:
517 if (bt_stream_class_packets_have_beginning_default_clock_snapshot(
519 clock_snapshot
= bt_message_packet_beginning_borrow_default_clock_snapshot_const(
522 goto no_clock_snapshot
;
526 case BT_MESSAGE_TYPE_PACKET_END
:
527 if (bt_stream_class_packets_have_end_default_clock_snapshot(
529 clock_snapshot
= bt_message_packet_end_borrow_default_clock_snapshot_const(
532 goto no_clock_snapshot
;
536 case BT_MESSAGE_TYPE_STREAM_BEGINNING
:
538 enum bt_message_stream_clock_snapshot_state snapshot_state
=
539 bt_message_stream_beginning_borrow_default_clock_snapshot_const(
540 msg
, &clock_snapshot
);
541 if (snapshot_state
== BT_MESSAGE_STREAM_CLOCK_SNAPSHOT_STATE_UNKNOWN
) {
542 goto no_clock_snapshot
;
547 case BT_MESSAGE_TYPE_STREAM_END
:
549 enum bt_message_stream_clock_snapshot_state snapshot_state
=
550 bt_message_stream_end_borrow_default_clock_snapshot_const(
551 msg
, &clock_snapshot
);
552 if (snapshot_state
== BT_MESSAGE_STREAM_CLOCK_SNAPSHOT_STATE_UNKNOWN
) {
553 goto no_clock_snapshot
;
558 case BT_MESSAGE_TYPE_DISCARDED_EVENTS
:
559 if (bt_stream_class_discarded_events_have_default_clock_snapshots(
561 clock_snapshot
= bt_message_discarded_events_borrow_beginning_default_clock_snapshot_const(
564 goto no_clock_snapshot
;
568 case BT_MESSAGE_TYPE_DISCARDED_PACKETS
:
569 if (bt_stream_class_discarded_packets_have_default_clock_snapshots(
571 clock_snapshot
= bt_message_discarded_packets_borrow_beginning_default_clock_snapshot_const(
574 goto no_clock_snapshot
;
578 case BT_MESSAGE_TYPE_MESSAGE_ITERATOR_INACTIVITY
:
579 clock_snapshot
= bt_message_message_iterator_inactivity_borrow_clock_snapshot_const(
583 /* All the other messages have a higher priority */
584 BT_COMP_LOGD_STR("Message has no timestamp: using the last returned timestamp.");
585 *ts_ns
= last_returned_ts_ns
;
589 ret
= bt_clock_snapshot_get_ns_from_origin(clock_snapshot
, ts_ns
);
591 BT_COMP_LOGE_APPEND_CAUSE(muxer_comp
->self_comp
,
592 "Cannot get nanoseconds from Epoch of clock snapshot: "
593 "clock-snapshot-addr=%p", clock_snapshot
);
600 BT_COMP_LOGD_STR("Message's default clock snapshot is missing: "
601 "using the last returned timestamp.");
602 *ts_ns
= last_returned_ts_ns
;
610 BT_COMP_LOGD("Found message's timestamp: "
611 "muxer-msg-iter-addr=%p, msg-addr=%p, "
612 "last-returned-ts=%" PRId64
", ts=%" PRId64
,
613 muxer_msg_iter
, msg
, last_returned_ts_ns
,
621 int validate_clock_class(struct muxer_msg_iter
*muxer_msg_iter
,
622 struct muxer_comp
*muxer_comp
,
623 const bt_clock_class
*clock_class
)
626 const uint8_t *cc_uuid
;
629 BT_ASSERT_DBG(clock_class
);
630 cc_uuid
= bt_clock_class_get_uuid(clock_class
);
631 cc_name
= bt_clock_class_get_name(clock_class
);
633 if (muxer_msg_iter
->clock_class_expectation
==
634 MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_ANY
) {
636 * This is the first clock class that this muxer message
637 * iterator encounters. Its properties determine what to expect
638 * for the whole lifetime of the iterator.
640 if (bt_clock_class_origin_is_unix_epoch(clock_class
)) {
641 /* Expect absolute clock classes */
642 muxer_msg_iter
->clock_class_expectation
=
643 MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_ABSOLUTE
;
647 * Expect non-absolute clock classes
648 * with a specific UUID.
650 muxer_msg_iter
->clock_class_expectation
=
651 MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_NOT_ABS_SPEC_UUID
;
652 bt_uuid_copy(muxer_msg_iter
->expected_clock_class_uuid
, cc_uuid
);
655 * Expect non-absolute clock classes
658 muxer_msg_iter
->clock_class_expectation
=
659 MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_NOT_ABS_NO_UUID
;
664 switch (muxer_msg_iter
->clock_class_expectation
) {
665 case MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_ABSOLUTE
:
666 if (!bt_clock_class_origin_is_unix_epoch(clock_class
)) {
667 BT_COMP_LOGE_APPEND_CAUSE(muxer_comp
->self_comp
,
668 "Expecting an absolute clock class, "
669 "but got a non-absolute one: "
670 "clock-class-addr=%p, clock-class-name=\"%s\"",
671 clock_class
, cc_name
);
675 case MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_NOT_ABS_NO_UUID
:
676 if (bt_clock_class_origin_is_unix_epoch(clock_class
)) {
677 BT_COMP_LOGE_APPEND_CAUSE(muxer_comp
->self_comp
,
678 "Expecting a non-absolute clock class with no UUID, "
679 "but got an absolute one: "
680 "clock-class-addr=%p, clock-class-name=\"%s\"",
681 clock_class
, cc_name
);
686 BT_COMP_LOGE_APPEND_CAUSE(muxer_comp
->self_comp
,
687 "Expecting a non-absolute clock class with no UUID, "
688 "but got one with a UUID: "
689 "clock-class-addr=%p, clock-class-name=\"%s\", "
690 "uuid=\"" BT_UUID_FMT
"\"",
691 clock_class
, cc_name
, BT_UUID_FMT_VALUES(cc_uuid
));
695 case MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_NOT_ABS_SPEC_UUID
:
696 if (bt_clock_class_origin_is_unix_epoch(clock_class
)) {
697 BT_COMP_LOGE_APPEND_CAUSE(muxer_comp
->self_comp
,
698 "Expecting a non-absolute clock class with a specific UUID, "
699 "but got an absolute one: "
700 "clock-class-addr=%p, clock-class-name=\"%s\"",
701 clock_class
, cc_name
);
706 BT_COMP_LOGE_APPEND_CAUSE(muxer_comp
->self_comp
,
707 "Expecting a non-absolute clock class with a specific UUID, "
708 "but got one with no UUID: "
709 "clock-class-addr=%p, clock-class-name=\"%s\"",
710 clock_class
, cc_name
);
714 if (bt_uuid_compare(muxer_msg_iter
->expected_clock_class_uuid
, cc_uuid
) != 0) {
715 BT_COMP_LOGE_APPEND_CAUSE(muxer_comp
->self_comp
,
716 "Expecting a non-absolute clock class with a specific UUID, "
717 "but got one with different UUID: "
718 "clock-class-addr=%p, clock-class-name=\"%s\", "
719 "expected-uuid=\"" BT_UUID_FMT
"\", "
720 "uuid=\"" BT_UUID_FMT
"\"",
721 clock_class
, cc_name
,
722 BT_UUID_FMT_VALUES(muxer_msg_iter
->expected_clock_class_uuid
),
723 BT_UUID_FMT_VALUES(cc_uuid
));
727 case MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_NONE
:
728 BT_COMP_LOGE_APPEND_CAUSE(muxer_comp
->self_comp
,
729 "Expecting no clock class, but got one: "
730 "clock-class-addr=%p, clock-class-name=\"%s\"",
731 clock_class
, cc_name
);
735 BT_COMP_LOGF("Unexpected clock class expectation: "
736 "expectation-code=%d",
737 muxer_msg_iter
->clock_class_expectation
);
751 int validate_new_stream_clock_class(struct muxer_msg_iter
*muxer_msg_iter
,
752 struct muxer_comp
*muxer_comp
, const bt_stream
*stream
)
755 const bt_stream_class
*stream_class
=
756 bt_stream_borrow_class_const(stream
);
757 const bt_clock_class
*clock_class
=
758 bt_stream_class_borrow_default_clock_class_const(stream_class
);
761 if (muxer_msg_iter
->clock_class_expectation
==
762 MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_ANY
) {
763 /* Expect no clock class */
764 muxer_msg_iter
->clock_class_expectation
=
765 MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_NONE
;
766 } else if (muxer_msg_iter
->clock_class_expectation
!=
767 MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_NONE
) {
768 BT_COMP_LOGE_APPEND_CAUSE(muxer_comp
->self_comp
,
769 "Expecting stream class without a default clock class: "
770 "stream-class-addr=%p, stream-class-name=\"%s\", "
771 "stream-class-id=%" PRIu64
,
772 stream_class
, bt_stream_class_get_name(stream_class
),
773 bt_stream_class_get_id(stream_class
));
780 ret
= validate_clock_class(muxer_msg_iter
, muxer_comp
, clock_class
);
787 * This function finds the youngest available message amongst the
788 * non-ended upstream message iterators and returns the upstream
789 * message iterator which has it, or
790 * BT_MESSAGE_ITERATOR_STATUS_END if there's no available
793 * This function does NOT:
795 * * Update any upstream message iterator.
796 * * Check the upstream message iterators to retry.
798 * On sucess, this function sets *muxer_upstream_msg_iter to the
799 * upstream message iterator of which the current message is
800 * the youngest, and sets *ts_ns to its time.
803 bt_message_iterator_class_next_method_status
804 muxer_msg_iter_youngest_upstream_msg_iter(
805 struct muxer_comp
*muxer_comp
,
806 struct muxer_msg_iter
*muxer_msg_iter
,
807 struct muxer_upstream_msg_iter
**muxer_upstream_msg_iter
,
812 int64_t youngest_ts_ns
= INT64_MAX
;
813 bt_message_iterator_class_next_method_status status
=
814 BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_OK
;
816 BT_ASSERT_DBG(muxer_comp
);
817 BT_ASSERT_DBG(muxer_msg_iter
);
818 BT_ASSERT_DBG(muxer_upstream_msg_iter
);
819 *muxer_upstream_msg_iter
= NULL
;
821 for (i
= 0; i
< muxer_msg_iter
->active_muxer_upstream_msg_iters
->len
;
823 const bt_message
*msg
;
824 struct muxer_upstream_msg_iter
*cur_muxer_upstream_msg_iter
=
826 muxer_msg_iter
->active_muxer_upstream_msg_iters
,
830 if (!cur_muxer_upstream_msg_iter
->msg_iter
) {
831 /* This upstream message iterator is ended */
832 BT_COMP_LOGT("Skipping ended upstream message iterator: "
833 "muxer-upstream-msg-iter-wrap-addr=%p",
834 cur_muxer_upstream_msg_iter
);
838 BT_ASSERT_DBG(cur_muxer_upstream_msg_iter
->next_msg
<
839 cur_muxer_upstream_msg_iter
->msgs
->len
);
840 msg
= g_ptr_array_index(cur_muxer_upstream_msg_iter
->msgs
,
841 cur_muxer_upstream_msg_iter
->next_msg
);
844 if (G_UNLIKELY(bt_message_get_type(msg
) ==
845 BT_MESSAGE_TYPE_STREAM_BEGINNING
)) {
846 ret
= validate_new_stream_clock_class(
847 muxer_msg_iter
, muxer_comp
,
848 bt_message_stream_beginning_borrow_stream_const(
852 * validate_new_stream_clock_class() logs
855 status
= BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_ERROR
;
858 } else if (G_UNLIKELY(bt_message_get_type(msg
) ==
859 BT_MESSAGE_TYPE_MESSAGE_ITERATOR_INACTIVITY
)) {
860 const bt_clock_snapshot
*cs
;
862 cs
= bt_message_message_iterator_inactivity_borrow_clock_snapshot_const(
864 ret
= validate_clock_class(muxer_msg_iter
, muxer_comp
,
865 bt_clock_snapshot_borrow_clock_class_const(cs
));
867 /* validate_clock_class() logs errors */
868 status
= BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_ERROR
;
873 ret
= get_msg_ts_ns(muxer_comp
, muxer_msg_iter
, msg
,
874 muxer_msg_iter
->last_returned_ts_ns
, &msg_ts_ns
);
876 /* get_msg_ts_ns() logs errors */
877 *muxer_upstream_msg_iter
= NULL
;
878 status
= BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_ERROR
;
883 * Update the current message iterator if it has not been set
884 * yet, or if its current message has a timestamp smaller than
885 * the previously selected youngest message.
887 if (G_UNLIKELY(*muxer_upstream_msg_iter
== NULL
) ||
888 msg_ts_ns
< youngest_ts_ns
) {
889 *muxer_upstream_msg_iter
=
890 cur_muxer_upstream_msg_iter
;
891 youngest_ts_ns
= msg_ts_ns
;
892 *ts_ns
= youngest_ts_ns
;
893 } else if (msg_ts_ns
== youngest_ts_ns
) {
895 * The currently selected message to be sent downstream
896 * next has the exact same timestamp that of the
897 * current candidate message. We must break the tie
898 * in a predictable manner.
900 BT_ASSERT_DBG((*muxer_upstream_msg_iter
)->next_msg
<
901 (*muxer_upstream_msg_iter
)->msgs
->len
);
902 const bt_message
*selected_msg
=
903 g_ptr_array_index((*muxer_upstream_msg_iter
)->msgs
,
904 (*muxer_upstream_msg_iter
)->next_msg
);
905 BT_COMP_LOGD_STR("Two of the next message candidates have the same timestamps, pick one deterministically.");
908 * Order the messages in an arbitrary but determinitic
911 ret
= common_muxing_compare_messages(msg
, selected_msg
);
914 * The `msg` should go first. Update the next
915 * iterator and the current timestamp.
917 *muxer_upstream_msg_iter
=
918 cur_muxer_upstream_msg_iter
;
919 youngest_ts_ns
= msg_ts_ns
;
920 *ts_ns
= youngest_ts_ns
;
921 } else if (ret
== 0) {
922 /* Unable to pick which one should go first. */
923 BT_COMP_LOGW("Cannot deterministically pick next upstream message iterator because they have identical next messages: "
924 "muxer-upstream-msg-iter-wrap-addr=%p"
925 "cur-muxer-upstream-msg-iter-wrap-addr=%p",
926 *muxer_upstream_msg_iter
,
927 cur_muxer_upstream_msg_iter
);
932 if (!*muxer_upstream_msg_iter
) {
933 status
= BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_END
;
942 bt_message_iterator_class_next_method_status
943 validate_muxer_upstream_msg_iter(
944 struct muxer_upstream_msg_iter
*muxer_upstream_msg_iter
,
947 struct muxer_comp
*muxer_comp
= muxer_upstream_msg_iter
->muxer_comp
;
948 bt_message_iterator_class_next_method_status status
;
950 BT_COMP_LOGD("Validating muxer's upstream message iterator wrapper: "
951 "muxer-upstream-msg-iter-wrap-addr=%p",
952 muxer_upstream_msg_iter
);
954 if (muxer_upstream_msg_iter
->next_msg
< muxer_upstream_msg_iter
->msgs
->len
||
955 !muxer_upstream_msg_iter
->msg_iter
) {
956 BT_COMP_LOGD("Already valid or not considered: "
957 "queue-len=%u, next-msg=%u, upstream-msg-iter-addr=%p",
958 muxer_upstream_msg_iter
->msgs
->len
,
959 muxer_upstream_msg_iter
->next_msg
,
960 muxer_upstream_msg_iter
->msg_iter
);
961 status
= BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_OK
;
965 /* muxer_upstream_msg_iter_next() logs details/errors */
966 status
= muxer_upstream_msg_iter_next(muxer_upstream_msg_iter
,
974 bt_message_iterator_class_next_method_status
975 validate_muxer_upstream_msg_iters(
976 struct muxer_msg_iter
*muxer_msg_iter
)
978 struct muxer_comp
*muxer_comp
= muxer_msg_iter
->muxer_comp
;
979 bt_message_iterator_class_next_method_status status
;
982 BT_COMP_LOGD("Validating muxer's upstream message iterator wrappers: "
983 "muxer-msg-iter-addr=%p", muxer_msg_iter
);
985 for (i
= 0; i
< muxer_msg_iter
->active_muxer_upstream_msg_iters
->len
;
987 bool is_ended
= false;
988 struct muxer_upstream_msg_iter
*muxer_upstream_msg_iter
=
990 muxer_msg_iter
->active_muxer_upstream_msg_iters
,
993 status
= validate_muxer_upstream_msg_iter(
994 muxer_upstream_msg_iter
, &is_ended
);
995 if (status
!= BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_OK
) {
997 BT_COMP_LOGE_APPEND_CAUSE(muxer_comp
->self_comp
,
998 "Cannot validate muxer's upstream message iterator wrapper: "
999 "muxer-msg-iter-addr=%p, "
1000 "muxer-upstream-msg-iter-wrap-addr=%p",
1002 muxer_upstream_msg_iter
);
1004 BT_COMP_LOGD("Cannot validate muxer's upstream message iterator wrapper: "
1005 "muxer-msg-iter-addr=%p, "
1006 "muxer-upstream-msg-iter-wrap-addr=%p",
1008 muxer_upstream_msg_iter
);
1015 * Move this muxer upstream message iterator to the
1016 * array of ended iterators if it's ended.
1018 if (G_UNLIKELY(is_ended
)) {
1019 BT_COMP_LOGD("Muxer's upstream message iterator wrapper: ended or canceled: "
1020 "muxer-msg-iter-addr=%p, "
1021 "muxer-upstream-msg-iter-wrap-addr=%p",
1022 muxer_msg_iter
, muxer_upstream_msg_iter
);
1024 muxer_msg_iter
->ended_muxer_upstream_msg_iters
,
1025 muxer_upstream_msg_iter
);
1026 muxer_msg_iter
->active_muxer_upstream_msg_iters
->pdata
[i
] = NULL
;
1029 * Use g_ptr_array_remove_fast() because the
1030 * order of those elements is not important.
1032 g_ptr_array_remove_index_fast(
1033 muxer_msg_iter
->active_muxer_upstream_msg_iters
,
1039 status
= BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_OK
;
1046 bt_message_iterator_class_next_method_status
muxer_msg_iter_do_next_one(
1047 struct muxer_comp
*muxer_comp
,
1048 struct muxer_msg_iter
*muxer_msg_iter
,
1049 const bt_message
**msg
)
1051 bt_message_iterator_class_next_method_status status
;
1052 struct muxer_upstream_msg_iter
*muxer_upstream_msg_iter
= NULL
;
1053 /* Initialize to avoid -Wmaybe-uninitialized warning with gcc 4.8. */
1054 int64_t next_return_ts
= 0;
1056 status
= validate_muxer_upstream_msg_iters(muxer_msg_iter
);
1057 if (status
!= BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_OK
) {
1058 /* validate_muxer_upstream_msg_iters() logs details */
1063 * At this point we know that all the existing upstream
1064 * message iterators are valid. We can find the one,
1065 * amongst those, of which the current message is the
1068 status
= muxer_msg_iter_youngest_upstream_msg_iter(muxer_comp
,
1069 muxer_msg_iter
, &muxer_upstream_msg_iter
,
1071 if (status
< 0 || status
== BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_END
) {
1073 BT_COMP_LOGE_APPEND_CAUSE(muxer_comp
->self_comp
,
1074 "Cannot find the youngest upstream message iterator wrapper: "
1076 bt_common_func_status_string(status
));
1078 BT_COMP_LOGD("Cannot find the youngest upstream message iterator wrapper: "
1080 bt_common_func_status_string(status
));
1086 if (next_return_ts
< muxer_msg_iter
->last_returned_ts_ns
) {
1087 BT_COMP_LOGE_APPEND_CAUSE(muxer_comp
->self_comp
,
1088 "Youngest upstream message iterator wrapper's timestamp is less than muxer's message iterator's last returned timestamp: "
1089 "muxer-msg-iter-addr=%p, ts=%" PRId64
", "
1090 "last-returned-ts=%" PRId64
,
1091 muxer_msg_iter
, next_return_ts
,
1092 muxer_msg_iter
->last_returned_ts_ns
);
1093 status
= BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_ERROR
;
1097 BT_COMP_LOGD("Found youngest upstream message iterator wrapper: "
1098 "muxer-msg-iter-addr=%p, "
1099 "muxer-upstream-msg-iter-wrap-addr=%p, "
1101 muxer_msg_iter
, muxer_upstream_msg_iter
, next_return_ts
);
1102 BT_ASSERT_DBG(status
==
1103 BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_OK
);
1104 BT_ASSERT_DBG(muxer_upstream_msg_iter
);
1107 * Consume from the queue's head: other side
1108 * (muxer_upstream_msg_iter_next()) writes to the tail.
1110 *msg
= g_ptr_array_index(muxer_upstream_msg_iter
->msgs
,
1111 muxer_upstream_msg_iter
->next_msg
);
1112 g_ptr_array_index(muxer_upstream_msg_iter
->msgs
,
1113 muxer_upstream_msg_iter
->next_msg
) = NULL
;
1114 ++muxer_upstream_msg_iter
->next_msg
;
1115 BT_ASSERT_DBG(*msg
);
1116 muxer_msg_iter
->last_returned_ts_ns
= next_return_ts
;
1123 bt_message_iterator_class_next_method_status
muxer_msg_iter_do_next(
1124 struct muxer_comp
*muxer_comp
,
1125 struct muxer_msg_iter
*muxer_msg_iter
,
1126 bt_message_array_const msgs
, uint64_t capacity
,
1129 bt_message_iterator_class_next_method_status status
;
1132 if (G_UNLIKELY(muxer_msg_iter
->next_saved_error
)) {
1134 * Last time we were called, we hit an error but had some
1135 * messages to deliver, so we stashed the error here. Return
1138 BT_CURRENT_THREAD_MOVE_ERROR_AND_RESET(muxer_msg_iter
->next_saved_error
);
1139 status
= muxer_msg_iter
->next_saved_status
;
1144 status
= muxer_msg_iter_do_next_one(muxer_comp
,
1145 muxer_msg_iter
, &msgs
[i
]);
1146 if (status
== BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_OK
) {
1149 } while (i
< capacity
&& status
== BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_OK
);
1153 * Even if muxer_msg_iter_do_next_one() returned
1154 * something else than
1155 * BT_MESSAGE_ITERATOR_STATUS_OK, we accumulated
1156 * message objects in the output message
1157 * array, so we need to return
1158 * BT_MESSAGE_ITERATOR_STATUS_OK so that they are
1159 * transfered to downstream. This other status occurs
1160 * again the next time muxer_msg_iter_do_next() is
1161 * called, possibly without any accumulated
1162 * message, in which case we'll return it.
1166 * Save this error for the next _next call. Assume that
1167 * this component always appends error causes when
1168 * returning an error status code, which will cause the
1169 * current thread error to be non-NULL.
1171 muxer_msg_iter
->next_saved_error
= bt_current_thread_take_error();
1172 BT_ASSERT(muxer_msg_iter
->next_saved_error
);
1173 muxer_msg_iter
->next_saved_status
= status
;
1177 status
= BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_OK
;
1185 void destroy_muxer_msg_iter(struct muxer_msg_iter
*muxer_msg_iter
)
1187 struct muxer_comp
*muxer_comp
;
1189 if (!muxer_msg_iter
) {
1193 muxer_comp
= muxer_msg_iter
->muxer_comp
;
1194 BT_COMP_LOGD("Destroying muxer component's message iterator: "
1195 "muxer-msg-iter-addr=%p", muxer_msg_iter
);
1197 if (muxer_msg_iter
->active_muxer_upstream_msg_iters
) {
1198 BT_COMP_LOGD_STR("Destroying muxer's active upstream message iterator wrappers.");
1200 muxer_msg_iter
->active_muxer_upstream_msg_iters
, TRUE
);
1203 if (muxer_msg_iter
->ended_muxer_upstream_msg_iters
) {
1204 BT_COMP_LOGD_STR("Destroying muxer's ended upstream message iterator wrappers.");
1206 muxer_msg_iter
->ended_muxer_upstream_msg_iters
, TRUE
);
1209 g_free(muxer_msg_iter
);
1213 bt_message_iterator_class_initialize_method_status
1214 muxer_msg_iter_init_upstream_iterators(struct muxer_comp
*muxer_comp
,
1215 struct muxer_msg_iter
*muxer_msg_iter
,
1216 struct bt_self_message_iterator_configuration
*config
)
1220 bt_message_iterator_class_initialize_method_status status
;
1221 bool can_seek_forward
= true;
1223 count
= bt_component_filter_get_input_port_count(
1224 bt_self_component_filter_as_component_filter(
1225 muxer_comp
->self_comp_flt
));
1227 BT_COMP_LOGD("No input port to initialize for muxer component's message iterator: "
1228 "muxer-comp-addr=%p, muxer-msg-iter-addr=%p",
1229 muxer_comp
, muxer_msg_iter
);
1230 status
= BT_MESSAGE_ITERATOR_CLASS_INITIALIZE_METHOD_STATUS_OK
;
1234 for (i
= 0; i
< count
; i
++) {
1235 bt_message_iterator
*upstream_msg_iter
;
1236 bt_self_component_port_input
*self_port
=
1237 bt_self_component_filter_borrow_input_port_by_index(
1238 muxer_comp
->self_comp_flt
, i
);
1239 const bt_port
*port
;
1240 bt_message_iterator_create_from_message_iterator_status
1244 BT_ASSERT(self_port
);
1245 port
= bt_self_component_port_as_port(
1246 bt_self_component_port_input_as_self_component_port(
1250 if (!bt_port_is_connected(port
)) {
1251 /* Skip non-connected port */
1255 msg_iter_status
= create_msg_iter_on_input_port(muxer_comp
,
1256 muxer_msg_iter
, self_port
, &upstream_msg_iter
);
1257 if (msg_iter_status
!= BT_MESSAGE_ITERATOR_CREATE_FROM_MESSAGE_ITERATOR_STATUS_OK
) {
1258 /* create_msg_iter_on_input_port() logs errors */
1259 status
= (int) msg_iter_status
;
1263 int_status
= muxer_msg_iter_add_upstream_msg_iter(muxer_msg_iter
,
1265 bt_message_iterator_put_ref(
1268 status
= BT_MESSAGE_ITERATOR_CLASS_INITIALIZE_METHOD_STATUS_ERROR
;
1269 /* muxer_msg_iter_add_upstream_msg_iter() logs errors */
1273 can_seek_forward
= can_seek_forward
&&
1274 bt_message_iterator_can_seek_forward(
1279 * This iterator can seek forward if all of its iterators can seek
1282 bt_self_message_iterator_configuration_set_can_seek_forward(
1283 config
, can_seek_forward
);
1285 status
= BT_MESSAGE_ITERATOR_CLASS_INITIALIZE_METHOD_STATUS_OK
;
1291 bt_message_iterator_class_initialize_method_status
muxer_msg_iter_init(
1292 bt_self_message_iterator
*self_msg_iter
,
1293 bt_self_message_iterator_configuration
*config
,
1294 bt_self_component_port_output
*port
)
1296 struct muxer_comp
*muxer_comp
= NULL
;
1297 struct muxer_msg_iter
*muxer_msg_iter
= NULL
;
1298 bt_message_iterator_class_initialize_method_status status
;
1299 bt_self_component
*self_comp
=
1300 bt_self_message_iterator_borrow_component(self_msg_iter
);
1302 muxer_comp
= bt_self_component_get_data(self_comp
);
1303 BT_ASSERT(muxer_comp
);
1304 BT_COMP_LOGD("Initializing muxer component's message iterator: "
1305 "comp-addr=%p, muxer-comp-addr=%p, msg-iter-addr=%p",
1306 self_comp
, muxer_comp
, self_msg_iter
);
1308 if (muxer_comp
->initializing_muxer_msg_iter
) {
1310 * Weird, unhandled situation detected: downstream
1311 * creates a muxer message iterator while creating
1312 * another muxer message iterator (same component).
1314 BT_COMP_LOGE_APPEND_CAUSE(muxer_comp
->self_comp
,
1315 "Recursive initialization of muxer component's message iterator: "
1316 "comp-addr=%p, muxer-comp-addr=%p, msg-iter-addr=%p",
1317 self_comp
, muxer_comp
, self_msg_iter
);
1318 status
= BT_MESSAGE_ITERATOR_CLASS_INITIALIZE_METHOD_STATUS_ERROR
;
1322 muxer_comp
->initializing_muxer_msg_iter
= true;
1323 muxer_msg_iter
= g_new0(struct muxer_msg_iter
, 1);
1324 if (!muxer_msg_iter
) {
1325 BT_COMP_LOGE_APPEND_CAUSE(muxer_comp
->self_comp
,
1326 "Failed to allocate one muxer component's message iterator.");
1327 status
= BT_MESSAGE_ITERATOR_CLASS_INITIALIZE_METHOD_STATUS_MEMORY_ERROR
;
1331 muxer_msg_iter
->muxer_comp
= muxer_comp
;
1332 muxer_msg_iter
->self_msg_iter
= self_msg_iter
;
1333 muxer_msg_iter
->last_returned_ts_ns
= INT64_MIN
;
1334 muxer_msg_iter
->active_muxer_upstream_msg_iters
=
1335 g_ptr_array_new_with_free_func(
1336 (GDestroyNotify
) destroy_muxer_upstream_msg_iter
);
1337 if (!muxer_msg_iter
->active_muxer_upstream_msg_iters
) {
1338 BT_COMP_LOGE_APPEND_CAUSE(muxer_comp
->self_comp
, "Failed to allocate a GPtrArray.");
1339 status
= BT_MESSAGE_ITERATOR_CLASS_INITIALIZE_METHOD_STATUS_MEMORY_ERROR
;
1343 muxer_msg_iter
->ended_muxer_upstream_msg_iters
=
1344 g_ptr_array_new_with_free_func(
1345 (GDestroyNotify
) destroy_muxer_upstream_msg_iter
);
1346 if (!muxer_msg_iter
->ended_muxer_upstream_msg_iters
) {
1347 BT_COMP_LOGE_APPEND_CAUSE(muxer_comp
->self_comp
, "Failed to allocate a GPtrArray.");
1348 status
= BT_MESSAGE_ITERATOR_CLASS_INITIALIZE_METHOD_STATUS_MEMORY_ERROR
;
1352 status
= muxer_msg_iter_init_upstream_iterators(muxer_comp
,
1353 muxer_msg_iter
, config
);
1355 BT_COMP_LOGE_APPEND_CAUSE(muxer_comp
->self_comp
,
1356 "Cannot initialize connected input ports for muxer component's message iterator: "
1357 "comp-addr=%p, muxer-comp-addr=%p, "
1358 "muxer-msg-iter-addr=%p, msg-iter-addr=%p, ret=%d",
1359 self_comp
, muxer_comp
, muxer_msg_iter
,
1360 self_msg_iter
, status
);
1364 bt_self_message_iterator_set_data(self_msg_iter
, muxer_msg_iter
);
1365 BT_COMP_LOGD("Initialized muxer component's message iterator: "
1366 "comp-addr=%p, muxer-comp-addr=%p, muxer-msg-iter-addr=%p, "
1368 self_comp
, muxer_comp
, muxer_msg_iter
, self_msg_iter
);
1372 destroy_muxer_msg_iter(muxer_msg_iter
);
1373 bt_self_message_iterator_set_data(self_msg_iter
, NULL
);
1376 muxer_comp
->initializing_muxer_msg_iter
= false;
1380 void muxer_msg_iter_finalize(bt_self_message_iterator
*self_msg_iter
)
1382 struct muxer_msg_iter
*muxer_msg_iter
=
1383 bt_self_message_iterator_get_data(self_msg_iter
);
1384 bt_self_component
*self_comp
= NULL
;
1385 struct muxer_comp
*muxer_comp
= NULL
;
1387 self_comp
= bt_self_message_iterator_borrow_component(
1389 BT_ASSERT(self_comp
);
1390 muxer_comp
= bt_self_component_get_data(self_comp
);
1391 BT_COMP_LOGD("Finalizing muxer component's message iterator: "
1392 "comp-addr=%p, muxer-comp-addr=%p, muxer-msg-iter-addr=%p, "
1394 self_comp
, muxer_comp
, muxer_msg_iter
, self_msg_iter
);
1396 if (muxer_msg_iter
) {
1397 destroy_muxer_msg_iter(muxer_msg_iter
);
1401 bt_message_iterator_class_next_method_status
muxer_msg_iter_next(
1402 bt_self_message_iterator
*self_msg_iter
,
1403 bt_message_array_const msgs
, uint64_t capacity
,
1406 bt_message_iterator_class_next_method_status status
;
1407 struct muxer_msg_iter
*muxer_msg_iter
=
1408 bt_self_message_iterator_get_data(self_msg_iter
);
1409 bt_self_component
*self_comp
= NULL
;
1410 struct muxer_comp
*muxer_comp
= NULL
;
1412 BT_ASSERT_DBG(muxer_msg_iter
);
1413 self_comp
= bt_self_message_iterator_borrow_component(
1415 BT_ASSERT_DBG(self_comp
);
1416 muxer_comp
= bt_self_component_get_data(self_comp
);
1417 BT_ASSERT_DBG(muxer_comp
);
1418 BT_COMP_LOGT("Muxer component's message iterator's \"next\" method called: "
1419 "comp-addr=%p, muxer-comp-addr=%p, muxer-msg-iter-addr=%p, "
1421 self_comp
, muxer_comp
, muxer_msg_iter
, self_msg_iter
);
1423 status
= muxer_msg_iter_do_next(muxer_comp
, muxer_msg_iter
,
1424 msgs
, capacity
, count
);
1426 BT_COMP_LOGE_APPEND_CAUSE(self_comp
,
1427 "Cannot get next message: "
1428 "comp-addr=%p, muxer-comp-addr=%p, muxer-msg-iter-addr=%p, "
1429 "msg-iter-addr=%p, status=%s",
1430 self_comp
, muxer_comp
, muxer_msg_iter
, self_msg_iter
,
1431 bt_common_func_status_string(status
));
1433 BT_COMP_LOGT("Returning from muxer component's message iterator's \"next\" method: "
1435 bt_common_func_status_string(status
));
1441 bt_component_class_port_connected_method_status
muxer_input_port_connected(
1442 bt_self_component_filter
*self_comp
,
1443 bt_self_component_port_input
*self_port
,
1444 const bt_port_output
*other_port
)
1446 bt_component_class_port_connected_method_status status
=
1447 BT_COMPONENT_CLASS_PORT_CONNECTED_METHOD_STATUS_OK
;
1448 bt_self_component_add_port_status add_port_status
;
1449 struct muxer_comp
*muxer_comp
= bt_self_component_get_data(
1450 bt_self_component_filter_as_self_component(self_comp
));
1452 add_port_status
= add_available_input_port(self_comp
);
1453 if (add_port_status
) {
1454 BT_COMP_LOGE_APPEND_CAUSE(muxer_comp
->self_comp
,
1455 "Cannot add one muxer component's input port: status=%s",
1456 bt_common_func_status_string(add_port_status
));
1458 if (add_port_status
==
1459 BT_SELF_COMPONENT_ADD_PORT_STATUS_MEMORY_ERROR
) {
1460 status
= BT_COMPONENT_CLASS_PORT_CONNECTED_METHOD_STATUS_MEMORY_ERROR
;
1462 status
= BT_COMPONENT_CLASS_PORT_CONNECTED_METHOD_STATUS_ERROR
;
1473 bt_message_iterator_class_can_seek_beginning_method_status
1474 muxer_upstream_msg_iters_can_all_seek_beginning(
1475 struct muxer_comp
*muxer_comp
,
1476 GPtrArray
*muxer_upstream_msg_iters
, bt_bool
*can_seek
)
1478 bt_message_iterator_class_can_seek_beginning_method_status status
=
1479 BT_MESSAGE_ITERATOR_CLASS_CAN_SEEK_BEGINNING_METHOD_STATUS_OK
;
1482 for (i
= 0; i
< muxer_upstream_msg_iters
->len
; i
++) {
1483 struct muxer_upstream_msg_iter
*upstream_msg_iter
=
1484 muxer_upstream_msg_iters
->pdata
[i
];
1485 status
= (int) bt_message_iterator_can_seek_beginning(
1486 upstream_msg_iter
->msg_iter
, can_seek
);
1487 if (status
!= BT_MESSAGE_ITERATOR_CLASS_CAN_SEEK_BEGINNING_METHOD_STATUS_OK
) {
1488 BT_COMP_LOGE_APPEND_CAUSE(muxer_comp
->self_comp
,
1489 "Failed to determine whether upstream message iterator can seek beginning: "
1490 "msg-iter-addr=%p", upstream_msg_iter
->msg_iter
);
1499 *can_seek
= BT_TRUE
;
1505 bt_message_iterator_class_can_seek_beginning_method_status
1506 muxer_msg_iter_can_seek_beginning(
1507 bt_self_message_iterator
*self_msg_iter
, bt_bool
*can_seek
)
1509 struct muxer_msg_iter
*muxer_msg_iter
=
1510 bt_self_message_iterator_get_data(self_msg_iter
);
1511 bt_message_iterator_class_can_seek_beginning_method_status status
;
1513 status
= muxer_upstream_msg_iters_can_all_seek_beginning(
1514 muxer_msg_iter
->muxer_comp
,
1515 muxer_msg_iter
->active_muxer_upstream_msg_iters
, can_seek
);
1516 if (status
!= BT_MESSAGE_ITERATOR_CLASS_CAN_SEEK_BEGINNING_METHOD_STATUS_OK
) {
1524 status
= muxer_upstream_msg_iters_can_all_seek_beginning(
1525 muxer_msg_iter
->muxer_comp
,
1526 muxer_msg_iter
->ended_muxer_upstream_msg_iters
, can_seek
);
1532 bt_message_iterator_class_seek_beginning_method_status
muxer_msg_iter_seek_beginning(
1533 bt_self_message_iterator
*self_msg_iter
)
1535 struct muxer_msg_iter
*muxer_msg_iter
=
1536 bt_self_message_iterator_get_data(self_msg_iter
);
1537 bt_message_iterator_class_seek_beginning_method_status status
=
1538 BT_MESSAGE_ITERATOR_CLASS_SEEK_BEGINNING_METHOD_STATUS_OK
;
1539 bt_message_iterator_seek_beginning_status seek_beg_status
;
1542 /* Seek all ended upstream iterators first */
1543 for (i
= 0; i
< muxer_msg_iter
->ended_muxer_upstream_msg_iters
->len
;
1545 struct muxer_upstream_msg_iter
*upstream_msg_iter
=
1546 muxer_msg_iter
->ended_muxer_upstream_msg_iters
->pdata
[i
];
1548 seek_beg_status
= bt_message_iterator_seek_beginning(
1549 upstream_msg_iter
->msg_iter
);
1550 if (seek_beg_status
!= BT_MESSAGE_ITERATOR_SEEK_BEGINNING_STATUS_OK
) {
1551 status
= (int) seek_beg_status
;
1555 empty_message_queue(upstream_msg_iter
);
1558 /* Seek all previously active upstream iterators */
1559 for (i
= 0; i
< muxer_msg_iter
->active_muxer_upstream_msg_iters
->len
;
1561 struct muxer_upstream_msg_iter
*upstream_msg_iter
=
1562 muxer_msg_iter
->active_muxer_upstream_msg_iters
->pdata
[i
];
1564 seek_beg_status
= bt_message_iterator_seek_beginning(
1565 upstream_msg_iter
->msg_iter
);
1566 if (seek_beg_status
!= BT_MESSAGE_ITERATOR_SEEK_BEGINNING_STATUS_OK
) {
1567 status
= (int) seek_beg_status
;
1571 empty_message_queue(upstream_msg_iter
);
1574 /* Make them all active */
1575 for (i
= 0; i
< muxer_msg_iter
->ended_muxer_upstream_msg_iters
->len
;
1577 struct muxer_upstream_msg_iter
*upstream_msg_iter
=
1578 muxer_msg_iter
->ended_muxer_upstream_msg_iters
->pdata
[i
];
1580 g_ptr_array_add(muxer_msg_iter
->active_muxer_upstream_msg_iters
,
1582 muxer_msg_iter
->ended_muxer_upstream_msg_iters
->pdata
[i
] = NULL
;
1586 * GLib < 2.48.0 asserts when g_ptr_array_remove_range() is
1587 * called on an empty array.
1589 if (muxer_msg_iter
->ended_muxer_upstream_msg_iters
->len
> 0) {
1590 g_ptr_array_remove_range(muxer_msg_iter
->ended_muxer_upstream_msg_iters
,
1591 0, muxer_msg_iter
->ended_muxer_upstream_msg_iters
->len
);
1593 muxer_msg_iter
->last_returned_ts_ns
= INT64_MIN
;
1594 muxer_msg_iter
->clock_class_expectation
=
1595 MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_ANY
;