2 * Copyright 2017 Philippe Proulx <pproulx@efficios.com>
4 * Permission is hereby granted, free of charge, to any person obtaining a copy
5 * of this software and associated documentation files (the "Software"), to deal
6 * in the Software without restriction, including without limitation the rights
7 * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
8 * copies of the Software, and to permit persons to whom the Software is
9 * furnished to do so, subject to the following conditions:
11 * The above copyright notice and this permission notice shall be included in
12 * all copies or substantial portions of the Software.
14 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
15 * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
16 * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
17 * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
18 * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
19 * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
23 #define BT_COMP_LOG_SELF_COMP (muxer_comp->self_comp)
24 #define BT_LOG_OUTPUT_LEVEL (muxer_comp->log_level)
25 #define BT_LOG_TAG "PLUGIN/FLT.UTILS.MUXER"
26 #include "logging/comp-logging.h"
28 #include "common/macros.h"
29 #include "common/uuid.h"
30 #include <babeltrace2/babeltrace.h>
34 #include "common/assert.h"
35 #include "common/common.h"
39 #include "plugins/common/muxing/muxing.h"
40 #include "plugins/common/param-validation/param-validation.h"
46 bt_self_component_filter
*self_comp_flt
;
47 bt_self_component
*self_comp
;
49 unsigned int next_port_num
;
50 size_t available_input_ports
;
51 bool initializing_muxer_msg_iter
;
52 bt_logging_level log_level
;
55 struct muxer_upstream_msg_iter
{
56 struct muxer_comp
*muxer_comp
;
58 /* Owned by this, NULL if ended */
59 bt_self_component_port_input_message_iterator
*msg_iter
;
61 /* Contains `const bt_message *`, owned by this */
65 enum muxer_msg_iter_clock_class_expectation
{
66 MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_ANY
= 0,
67 MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_NONE
,
68 MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_ABSOLUTE
,
69 MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_NOT_ABS_SPEC_UUID
,
70 MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_NOT_ABS_NO_UUID
,
73 struct muxer_msg_iter
{
74 struct muxer_comp
*muxer_comp
;
77 bt_self_message_iterator
*self_msg_iter
;
80 * Array of struct muxer_upstream_msg_iter * (owned by this).
82 * NOTE: This array is searched in linearly to find the youngest
83 * current message. Keep this until benchmarks confirm that
84 * another data structure is faster than this for our typical
87 GPtrArray
*active_muxer_upstream_msg_iters
;
90 * Array of struct muxer_upstream_msg_iter * (owned by this).
92 * We move ended message iterators from
93 * `active_muxer_upstream_msg_iters` to this array so as to be
94 * able to restore them when seeking.
96 GPtrArray
*ended_muxer_upstream_msg_iters
;
98 /* Last time returned in a message */
99 int64_t last_returned_ts_ns
;
101 /* Clock class expectation state */
102 enum muxer_msg_iter_clock_class_expectation clock_class_expectation
;
105 * Expected clock class UUID, only valid when
106 * clock_class_expectation is
107 * MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_NOT_ABS_SPEC_UUID.
109 bt_uuid_t expected_clock_class_uuid
;
113 void empty_message_queue(struct muxer_upstream_msg_iter
*upstream_msg_iter
)
115 const bt_message
*msg
;
117 while ((msg
= g_queue_pop_head(upstream_msg_iter
->msgs
))) {
118 bt_message_put_ref(msg
);
123 void destroy_muxer_upstream_msg_iter(
124 struct muxer_upstream_msg_iter
*muxer_upstream_msg_iter
)
126 struct muxer_comp
*muxer_comp
;
128 if (!muxer_upstream_msg_iter
) {
132 muxer_comp
= muxer_upstream_msg_iter
->muxer_comp
;
133 BT_COMP_LOGD("Destroying muxer's upstream message iterator wrapper: "
134 "addr=%p, msg-iter-addr=%p, queue-len=%u",
135 muxer_upstream_msg_iter
,
136 muxer_upstream_msg_iter
->msg_iter
,
137 muxer_upstream_msg_iter
->msgs
->length
);
138 bt_self_component_port_input_message_iterator_put_ref(
139 muxer_upstream_msg_iter
->msg_iter
);
141 if (muxer_upstream_msg_iter
->msgs
) {
142 empty_message_queue(muxer_upstream_msg_iter
);
143 g_queue_free(muxer_upstream_msg_iter
->msgs
);
146 g_free(muxer_upstream_msg_iter
);
150 int muxer_msg_iter_add_upstream_msg_iter(struct muxer_msg_iter
*muxer_msg_iter
,
151 bt_self_component_port_input_message_iterator
*self_msg_iter
)
154 struct muxer_upstream_msg_iter
*muxer_upstream_msg_iter
=
155 g_new0(struct muxer_upstream_msg_iter
, 1);
156 struct muxer_comp
*muxer_comp
= muxer_msg_iter
->muxer_comp
;
158 if (!muxer_upstream_msg_iter
) {
159 BT_COMP_LOGE_STR("Failed to allocate one muxer's upstream message iterator wrapper.");
163 muxer_upstream_msg_iter
->muxer_comp
= muxer_comp
;
164 muxer_upstream_msg_iter
->msg_iter
= self_msg_iter
;
165 bt_self_component_port_input_message_iterator_get_ref(muxer_upstream_msg_iter
->msg_iter
);
166 muxer_upstream_msg_iter
->msgs
= g_queue_new();
167 if (!muxer_upstream_msg_iter
->msgs
) {
168 BT_COMP_LOGE_STR("Failed to allocate a GQueue.");
172 g_ptr_array_add(muxer_msg_iter
->active_muxer_upstream_msg_iters
,
173 muxer_upstream_msg_iter
);
174 BT_COMP_LOGD("Added muxer's upstream message iterator wrapper: "
175 "addr=%p, muxer-msg-iter-addr=%p, msg-iter-addr=%p",
176 muxer_upstream_msg_iter
, muxer_msg_iter
,
182 g_free(muxer_upstream_msg_iter
);
190 bt_self_component_add_port_status
add_available_input_port(
191 bt_self_component_filter
*self_comp
)
193 struct muxer_comp
*muxer_comp
= bt_self_component_get_data(
194 bt_self_component_filter_as_self_component(self_comp
));
195 bt_self_component_add_port_status status
=
196 BT_SELF_COMPONENT_ADD_PORT_STATUS_OK
;
197 GString
*port_name
= NULL
;
199 BT_ASSERT(muxer_comp
);
200 port_name
= g_string_new("in");
202 BT_COMP_LOGE_STR("Failed to allocate a GString.");
203 status
= BT_SELF_COMPONENT_ADD_PORT_STATUS_MEMORY_ERROR
;
207 g_string_append_printf(port_name
, "%u", muxer_comp
->next_port_num
);
208 status
= bt_self_component_filter_add_input_port(
209 self_comp
, port_name
->str
, NULL
, NULL
);
210 if (status
!= BT_SELF_COMPONENT_ADD_PORT_STATUS_OK
) {
211 BT_COMP_LOGE("Cannot add input port to muxer component: "
212 "port-name=\"%s\", comp-addr=%p, status=%s",
213 port_name
->str
, self_comp
,
214 bt_common_func_status_string(status
));
218 muxer_comp
->available_input_ports
++;
219 muxer_comp
->next_port_num
++;
220 BT_COMP_LOGI("Added one input port to muxer component: "
221 "port-name=\"%s\", comp-addr=%p",
222 port_name
->str
, self_comp
);
226 g_string_free(port_name
, TRUE
);
233 bt_self_component_add_port_status
create_output_port(
234 bt_self_component_filter
*self_comp
)
236 return bt_self_component_filter_add_output_port(
237 self_comp
, "out", NULL
, NULL
);
241 void destroy_muxer_comp(struct muxer_comp
*muxer_comp
)
250 struct bt_param_validation_map_value_entry_descr muxer_params
[] = {
251 BT_PARAM_VALIDATION_MAP_VALUE_ENTRY_END
255 bt_component_class_initialize_method_status
muxer_init(
256 bt_self_component_filter
*self_comp_flt
,
257 bt_self_component_filter_configuration
*config
,
258 const bt_value
*params
, void *init_data
)
260 bt_component_class_initialize_method_status status
;
261 bt_self_component_add_port_status add_port_status
;
262 bt_self_component
*self_comp
=
263 bt_self_component_filter_as_self_component(self_comp_flt
);
264 struct muxer_comp
*muxer_comp
= g_new0(struct muxer_comp
, 1);
265 bt_logging_level log_level
= bt_component_get_logging_level(
266 bt_self_component_as_component(self_comp
));
267 enum bt_param_validation_status validation_status
;
268 gchar
*validate_error
= NULL
;
270 BT_COMP_LOG_CUR_LVL(BT_LOG_INFO
, log_level
, self_comp
,
271 "Initializing muxer component: "
272 "comp-addr=%p, params-addr=%p", self_comp
, params
);
275 BT_COMP_LOG_CUR_LVL(BT_LOG_ERROR
, log_level
, self_comp
,
276 "Failed to allocate one muxer component.");
277 status
= BT_COMPONENT_CLASS_INITIALIZE_METHOD_STATUS_MEMORY_ERROR
;
281 muxer_comp
->log_level
= log_level
;
282 muxer_comp
->self_comp
= self_comp
;
283 muxer_comp
->self_comp_flt
= self_comp_flt
;
285 validation_status
= bt_param_validation_validate(params
,
286 muxer_params
, &validate_error
);
287 if (validation_status
== BT_PARAM_VALIDATION_STATUS_MEMORY_ERROR
) {
288 status
= BT_COMPONENT_CLASS_INITIALIZE_METHOD_STATUS_MEMORY_ERROR
;
290 } else if (validation_status
== BT_PARAM_VALIDATION_STATUS_VALIDATION_ERROR
) {
291 status
= BT_COMPONENT_CLASS_INITIALIZE_METHOD_STATUS_ERROR
;
292 BT_COMP_LOGE_APPEND_CAUSE(self_comp
, "%s", validate_error
);
296 bt_self_component_set_data(self_comp
, muxer_comp
);
297 add_port_status
= add_available_input_port(self_comp_flt
);
298 if (add_port_status
!= BT_SELF_COMPONENT_ADD_PORT_STATUS_OK
) {
299 BT_COMP_LOGE("Cannot ensure that at least one muxer component's input port is available: "
300 "muxer-comp-addr=%p, status=%s",
302 bt_common_func_status_string(add_port_status
));
303 status
= (int) add_port_status
;
307 add_port_status
= create_output_port(self_comp_flt
);
308 if (add_port_status
!= BT_SELF_COMPONENT_ADD_PORT_STATUS_OK
) {
309 BT_COMP_LOGE("Cannot create muxer component's output port: "
310 "muxer-comp-addr=%p, status=%s",
312 bt_common_func_status_string(add_port_status
));
313 status
= (int) add_port_status
;
317 BT_COMP_LOGI("Initialized muxer component: "
318 "comp-addr=%p, params-addr=%p, muxer-comp-addr=%p",
319 self_comp
, params
, muxer_comp
);
321 status
= BT_COMPONENT_CLASS_INITIALIZE_METHOD_STATUS_OK
;
325 destroy_muxer_comp(muxer_comp
);
326 bt_self_component_set_data(self_comp
, NULL
);
329 g_free(validate_error
);
334 void muxer_finalize(bt_self_component_filter
*self_comp
)
336 struct muxer_comp
*muxer_comp
= bt_self_component_get_data(
337 bt_self_component_filter_as_self_component(self_comp
));
339 BT_COMP_LOGI("Finalizing muxer component: comp-addr=%p",
341 destroy_muxer_comp(muxer_comp
);
345 bt_self_component_port_input_message_iterator_create_from_message_iterator_status
346 create_msg_iter_on_input_port(struct muxer_comp
*muxer_comp
,
347 struct muxer_msg_iter
*muxer_msg_iter
,
348 bt_self_component_port_input
*self_port
,
349 bt_self_component_port_input_message_iterator
**msg_iter
)
351 const bt_port
*port
= bt_self_component_port_as_port(
352 bt_self_component_port_input_as_self_component_port(
354 bt_self_component_port_input_message_iterator_create_from_message_iterator_status
358 BT_ASSERT(bt_port_is_connected(port
));
360 // TODO: Advance the iterator to >= the time of the latest
361 // returned message by the muxer message
362 // iterator which creates it.
363 status
= bt_self_component_port_input_message_iterator_create_from_message_iterator(
364 muxer_msg_iter
->self_msg_iter
, self_port
, msg_iter
);
365 if (status
!= BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_CREATE_FROM_MESSAGE_ITERATOR_STATUS_OK
) {
366 BT_COMP_LOGE("Cannot create upstream message iterator on input port: "
367 "port-addr=%p, port-name=\"%s\"",
368 port
, bt_port_get_name(port
));
372 BT_COMP_LOGI("Created upstream message iterator on input port: "
373 "port-addr=%p, port-name=\"%s\", msg-iter-addr=%p",
374 port
, bt_port_get_name(port
), msg_iter
);
381 bt_component_class_message_iterator_next_method_status
muxer_upstream_msg_iter_next(
382 struct muxer_upstream_msg_iter
*muxer_upstream_msg_iter
,
385 struct muxer_comp
*muxer_comp
=
386 muxer_upstream_msg_iter
->muxer_comp
;
387 bt_component_class_message_iterator_next_method_status status
;
388 bt_message_iterator_next_status input_port_iter_status
;
389 bt_message_array_const msgs
;
393 BT_COMP_LOGD("Calling upstream message iterator's \"next\" method: "
394 "muxer-upstream-msg-iter-wrap-addr=%p, msg-iter-addr=%p",
395 muxer_upstream_msg_iter
,
396 muxer_upstream_msg_iter
->msg_iter
);
397 input_port_iter_status
= bt_self_component_port_input_message_iterator_next(
398 muxer_upstream_msg_iter
->msg_iter
, &msgs
, &count
);
399 BT_COMP_LOGD("Upstream message iterator's \"next\" method returned: "
401 bt_common_func_status_string(input_port_iter_status
));
403 switch (input_port_iter_status
) {
404 case BT_MESSAGE_ITERATOR_NEXT_STATUS_OK
:
406 * Message iterator's current message is
407 * valid: it must be considered for muxing operations.
409 BT_COMP_LOGD_STR("Validated upstream message iterator wrapper.");
410 BT_ASSERT_DBG(count
> 0);
412 /* Move messages to our queue */
413 for (i
= 0; i
< count
; i
++) {
415 * Push to tail in order; other side
416 * (muxer_msg_iter_do_next_one()) consumes
417 * from the head first.
419 g_queue_push_tail(muxer_upstream_msg_iter
->msgs
,
422 status
= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK
;
424 case BT_MESSAGE_ITERATOR_NEXT_STATUS_AGAIN
:
426 * Message iterator's current message is not
427 * valid anymore. Return
428 * BT_MESSAGE_ITERATOR_NEXT_STATUS_AGAIN immediately.
430 status
= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_AGAIN
;
432 case BT_MESSAGE_ITERATOR_NEXT_STATUS_END
: /* Fall-through. */
434 * Message iterator reached the end: release it. It
435 * won't be considered again to find the youngest
439 status
= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK
;
442 /* Error or unsupported status code */
443 BT_COMP_LOGE("Error or unsupported status code: "
444 "status-code=%d", input_port_iter_status
);
445 status
= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_ERROR
;
453 int get_msg_ts_ns(struct muxer_comp
*muxer_comp
,
454 struct muxer_msg_iter
*muxer_msg_iter
,
455 const bt_message
*msg
, int64_t last_returned_ts_ns
,
458 const bt_clock_snapshot
*clock_snapshot
= NULL
;
460 const bt_stream_class
*stream_class
= NULL
;
461 bt_message_type msg_type
;
464 BT_ASSERT_DBG(ts_ns
);
465 BT_COMP_LOGD("Getting message's timestamp: "
466 "muxer-msg-iter-addr=%p, msg-addr=%p, "
467 "last-returned-ts=%" PRId64
,
468 muxer_msg_iter
, msg
, last_returned_ts_ns
);
470 if (G_UNLIKELY(muxer_msg_iter
->clock_class_expectation
==
471 MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_NONE
)) {
472 *ts_ns
= last_returned_ts_ns
;
476 msg_type
= bt_message_get_type(msg
);
478 if (G_UNLIKELY(msg_type
== BT_MESSAGE_TYPE_PACKET_BEGINNING
)) {
479 stream_class
= bt_stream_borrow_class_const(
480 bt_packet_borrow_stream_const(
481 bt_message_packet_beginning_borrow_packet_const(
483 } else if (G_UNLIKELY(msg_type
== BT_MESSAGE_TYPE_PACKET_END
)) {
484 stream_class
= bt_stream_borrow_class_const(
485 bt_packet_borrow_stream_const(
486 bt_message_packet_end_borrow_packet_const(
488 } else if (G_UNLIKELY(msg_type
== BT_MESSAGE_TYPE_DISCARDED_EVENTS
)) {
489 stream_class
= bt_stream_borrow_class_const(
490 bt_message_discarded_events_borrow_stream_const(msg
));
491 } else if (G_UNLIKELY(msg_type
== BT_MESSAGE_TYPE_DISCARDED_PACKETS
)) {
492 stream_class
= bt_stream_borrow_class_const(
493 bt_message_discarded_packets_borrow_stream_const(msg
));
497 case BT_MESSAGE_TYPE_EVENT
:
498 BT_ASSERT_DBG(bt_message_event_borrow_stream_class_default_clock_class_const(
500 clock_snapshot
= bt_message_event_borrow_default_clock_snapshot_const(
503 case BT_MESSAGE_TYPE_PACKET_BEGINNING
:
504 if (bt_stream_class_packets_have_beginning_default_clock_snapshot(
506 clock_snapshot
= bt_message_packet_beginning_borrow_default_clock_snapshot_const(
509 goto no_clock_snapshot
;
513 case BT_MESSAGE_TYPE_PACKET_END
:
514 if (bt_stream_class_packets_have_end_default_clock_snapshot(
516 clock_snapshot
= bt_message_packet_end_borrow_default_clock_snapshot_const(
519 goto no_clock_snapshot
;
523 case BT_MESSAGE_TYPE_STREAM_BEGINNING
:
525 enum bt_message_stream_clock_snapshot_state snapshot_state
=
526 bt_message_stream_beginning_borrow_default_clock_snapshot_const(
527 msg
, &clock_snapshot
);
528 if (snapshot_state
== BT_MESSAGE_STREAM_CLOCK_SNAPSHOT_STATE_UNKNOWN
) {
529 goto no_clock_snapshot
;
534 case BT_MESSAGE_TYPE_STREAM_END
:
536 enum bt_message_stream_clock_snapshot_state snapshot_state
=
537 bt_message_stream_end_borrow_default_clock_snapshot_const(
538 msg
, &clock_snapshot
);
539 if (snapshot_state
== BT_MESSAGE_STREAM_CLOCK_SNAPSHOT_STATE_UNKNOWN
) {
540 goto no_clock_snapshot
;
545 case BT_MESSAGE_TYPE_DISCARDED_EVENTS
:
546 if (bt_stream_class_discarded_events_have_default_clock_snapshots(
548 clock_snapshot
= bt_message_discarded_events_borrow_beginning_default_clock_snapshot_const(
551 goto no_clock_snapshot
;
555 case BT_MESSAGE_TYPE_DISCARDED_PACKETS
:
556 if (bt_stream_class_discarded_packets_have_default_clock_snapshots(
558 clock_snapshot
= bt_message_discarded_packets_borrow_beginning_default_clock_snapshot_const(
561 goto no_clock_snapshot
;
565 case BT_MESSAGE_TYPE_MESSAGE_ITERATOR_INACTIVITY
:
566 clock_snapshot
= bt_message_message_iterator_inactivity_borrow_default_clock_snapshot_const(
570 /* All the other messages have a higher priority */
571 BT_COMP_LOGD_STR("Message has no timestamp: using the last returned timestamp.");
572 *ts_ns
= last_returned_ts_ns
;
576 ret
= bt_clock_snapshot_get_ns_from_origin(clock_snapshot
, ts_ns
);
578 BT_COMP_LOGE("Cannot get nanoseconds from Epoch of clock snapshot: "
579 "clock-snapshot-addr=%p", clock_snapshot
);
586 BT_COMP_LOGD_STR("Message's default clock snapshot is missing: "
587 "using the last returned timestamp.");
588 *ts_ns
= last_returned_ts_ns
;
596 BT_COMP_LOGD("Found message's timestamp: "
597 "muxer-msg-iter-addr=%p, msg-addr=%p, "
598 "last-returned-ts=%" PRId64
", ts=%" PRId64
,
599 muxer_msg_iter
, msg
, last_returned_ts_ns
,
607 int validate_clock_class(struct muxer_msg_iter
*muxer_msg_iter
,
608 struct muxer_comp
*muxer_comp
,
609 const bt_clock_class
*clock_class
)
612 const uint8_t *cc_uuid
;
615 BT_ASSERT_DBG(clock_class
);
616 cc_uuid
= bt_clock_class_get_uuid(clock_class
);
617 cc_name
= bt_clock_class_get_name(clock_class
);
619 if (muxer_msg_iter
->clock_class_expectation
==
620 MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_ANY
) {
622 * This is the first clock class that this muxer message
623 * iterator encounters. Its properties determine what to expect
624 * for the whole lifetime of the iterator.
626 if (bt_clock_class_origin_is_unix_epoch(clock_class
)) {
627 /* Expect absolute clock classes */
628 muxer_msg_iter
->clock_class_expectation
=
629 MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_ABSOLUTE
;
633 * Expect non-absolute clock classes
634 * with a specific UUID.
636 muxer_msg_iter
->clock_class_expectation
=
637 MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_NOT_ABS_SPEC_UUID
;
638 bt_uuid_copy(muxer_msg_iter
->expected_clock_class_uuid
, cc_uuid
);
641 * Expect non-absolute clock classes
644 muxer_msg_iter
->clock_class_expectation
=
645 MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_NOT_ABS_NO_UUID
;
650 switch (muxer_msg_iter
->clock_class_expectation
) {
651 case MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_ABSOLUTE
:
652 if (!bt_clock_class_origin_is_unix_epoch(clock_class
)) {
653 BT_COMP_LOGE("Expecting an absolute clock class, "
654 "but got a non-absolute one: "
655 "clock-class-addr=%p, clock-class-name=\"%s\"",
656 clock_class
, cc_name
);
660 case MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_NOT_ABS_NO_UUID
:
661 if (bt_clock_class_origin_is_unix_epoch(clock_class
)) {
662 BT_COMP_LOGE("Expecting a non-absolute clock class with no UUID, "
663 "but got an absolute one: "
664 "clock-class-addr=%p, clock-class-name=\"%s\"",
665 clock_class
, cc_name
);
670 BT_COMP_LOGE("Expecting a non-absolute clock class with no UUID, "
671 "but got one with a UUID: "
672 "clock-class-addr=%p, clock-class-name=\"%s\", "
673 "uuid=\"" BT_UUID_FMT
"\"",
674 clock_class
, cc_name
, BT_UUID_FMT_VALUES(cc_uuid
));
678 case MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_NOT_ABS_SPEC_UUID
:
679 if (bt_clock_class_origin_is_unix_epoch(clock_class
)) {
680 BT_COMP_LOGE("Expecting a non-absolute clock class with a specific UUID, "
681 "but got an absolute one: "
682 "clock-class-addr=%p, clock-class-name=\"%s\"",
683 clock_class
, cc_name
);
688 BT_COMP_LOGE("Expecting a non-absolute clock class with a specific UUID, "
689 "but got one with no UUID: "
690 "clock-class-addr=%p, clock-class-name=\"%s\"",
691 clock_class
, cc_name
);
695 if (bt_uuid_compare(muxer_msg_iter
->expected_clock_class_uuid
, cc_uuid
) != 0) {
696 BT_COMP_LOGE("Expecting a non-absolute clock class with a specific UUID, "
697 "but got one with different UUID: "
698 "clock-class-addr=%p, clock-class-name=\"%s\", "
699 "expected-uuid=\"" BT_UUID_FMT
"\", "
700 "uuid=\"" BT_UUID_FMT
"\"",
701 clock_class
, cc_name
,
702 BT_UUID_FMT_VALUES(muxer_msg_iter
->expected_clock_class_uuid
),
703 BT_UUID_FMT_VALUES(cc_uuid
));
707 case MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_NONE
:
708 BT_COMP_LOGE("Expecting no clock class, but got one: "
709 "clock-class-addr=%p, clock-class-name=\"%s\"",
710 clock_class
, cc_name
);
714 BT_COMP_LOGF("Unexpected clock class expectation: "
715 "expectation-code=%d",
716 muxer_msg_iter
->clock_class_expectation
);
730 int validate_new_stream_clock_class(struct muxer_msg_iter
*muxer_msg_iter
,
731 struct muxer_comp
*muxer_comp
, const bt_stream
*stream
)
734 const bt_stream_class
*stream_class
=
735 bt_stream_borrow_class_const(stream
);
736 const bt_clock_class
*clock_class
=
737 bt_stream_class_borrow_default_clock_class_const(stream_class
);
740 if (muxer_msg_iter
->clock_class_expectation
==
741 MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_ANY
) {
742 /* Expect no clock class */
743 muxer_msg_iter
->clock_class_expectation
=
744 MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_NONE
;
745 } else if (muxer_msg_iter
->clock_class_expectation
!=
746 MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_NONE
) {
747 BT_COMP_LOGE("Expecting stream class without a default clock class: "
748 "stream-class-addr=%p, stream-class-name=\"%s\", "
749 "stream-class-id=%" PRIu64
,
750 stream_class
, bt_stream_class_get_name(stream_class
),
751 bt_stream_class_get_id(stream_class
));
758 ret
= validate_clock_class(muxer_msg_iter
, muxer_comp
, clock_class
);
765 * This function finds the youngest available message amongst the
766 * non-ended upstream message iterators and returns the upstream
767 * message iterator which has it, or
768 * BT_MESSAGE_ITERATOR_STATUS_END if there's no available
771 * This function does NOT:
773 * * Update any upstream message iterator.
774 * * Check the upstream message iterators to retry.
776 * On sucess, this function sets *muxer_upstream_msg_iter to the
777 * upstream message iterator of which the current message is
778 * the youngest, and sets *ts_ns to its time.
781 bt_component_class_message_iterator_next_method_status
782 muxer_msg_iter_youngest_upstream_msg_iter(
783 struct muxer_comp
*muxer_comp
,
784 struct muxer_msg_iter
*muxer_msg_iter
,
785 struct muxer_upstream_msg_iter
**muxer_upstream_msg_iter
,
790 int64_t youngest_ts_ns
= INT64_MAX
;
791 bt_component_class_message_iterator_next_method_status status
=
792 BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK
;
794 BT_ASSERT_DBG(muxer_comp
);
795 BT_ASSERT_DBG(muxer_msg_iter
);
796 BT_ASSERT_DBG(muxer_upstream_msg_iter
);
797 *muxer_upstream_msg_iter
= NULL
;
799 for (i
= 0; i
< muxer_msg_iter
->active_muxer_upstream_msg_iters
->len
;
801 const bt_message
*msg
;
802 struct muxer_upstream_msg_iter
*cur_muxer_upstream_msg_iter
=
804 muxer_msg_iter
->active_muxer_upstream_msg_iters
,
808 if (!cur_muxer_upstream_msg_iter
->msg_iter
) {
809 /* This upstream message iterator is ended */
810 BT_COMP_LOGT("Skipping ended upstream message iterator: "
811 "muxer-upstream-msg-iter-wrap-addr=%p",
812 cur_muxer_upstream_msg_iter
);
816 BT_ASSERT_DBG(cur_muxer_upstream_msg_iter
->msgs
->length
> 0);
817 msg
= g_queue_peek_head(cur_muxer_upstream_msg_iter
->msgs
);
820 if (G_UNLIKELY(bt_message_get_type(msg
) ==
821 BT_MESSAGE_TYPE_STREAM_BEGINNING
)) {
822 ret
= validate_new_stream_clock_class(
823 muxer_msg_iter
, muxer_comp
,
824 bt_message_stream_beginning_borrow_stream_const(
828 * validate_new_stream_clock_class() logs
831 status
= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_ERROR
;
834 } else if (G_UNLIKELY(bt_message_get_type(msg
) ==
835 BT_MESSAGE_TYPE_MESSAGE_ITERATOR_INACTIVITY
)) {
836 const bt_clock_snapshot
*cs
;
838 cs
= bt_message_message_iterator_inactivity_borrow_default_clock_snapshot_const(
840 ret
= validate_clock_class(muxer_msg_iter
, muxer_comp
,
841 bt_clock_snapshot_borrow_clock_class_const(cs
));
843 /* validate_clock_class() logs errors */
844 status
= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_ERROR
;
849 ret
= get_msg_ts_ns(muxer_comp
, muxer_msg_iter
, msg
,
850 muxer_msg_iter
->last_returned_ts_ns
, &msg_ts_ns
);
852 /* get_msg_ts_ns() logs errors */
853 *muxer_upstream_msg_iter
= NULL
;
854 status
= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_ERROR
;
859 * Update the current message iterator if it has not been set
860 * yet, or if its current message has a timestamp smaller than
861 * the previously selected youngest message.
863 if (G_UNLIKELY(*muxer_upstream_msg_iter
== NULL
) ||
864 msg_ts_ns
< youngest_ts_ns
) {
865 *muxer_upstream_msg_iter
=
866 cur_muxer_upstream_msg_iter
;
867 youngest_ts_ns
= msg_ts_ns
;
868 *ts_ns
= youngest_ts_ns
;
869 } else if (msg_ts_ns
== youngest_ts_ns
) {
871 * The currently selected message to be sent downstream
872 * next has the exact same timestamp that of the
873 * current candidate message. We must break the tie
874 * in a predictable manner.
876 const bt_message
*selected_msg
= g_queue_peek_head(
877 (*muxer_upstream_msg_iter
)->msgs
);
878 BT_COMP_LOGD_STR("Two of the next message candidates have the same timestamps, pick one deterministically.");
881 * Order the messages in an arbitrary but determinitic
884 ret
= common_muxing_compare_messages(msg
, selected_msg
);
887 * The `msg` should go first. Update the next
888 * iterator and the current timestamp.
890 *muxer_upstream_msg_iter
=
891 cur_muxer_upstream_msg_iter
;
892 youngest_ts_ns
= msg_ts_ns
;
893 *ts_ns
= youngest_ts_ns
;
894 } else if (ret
== 0) {
895 /* Unable to pick which one should go first. */
896 BT_COMP_LOGW("Cannot deterministically pick next upstream message iterator because they have identical next messages: "
897 "muxer-upstream-msg-iter-wrap-addr=%p"
898 "cur-muxer-upstream-msg-iter-wrap-addr=%p",
899 *muxer_upstream_msg_iter
,
900 cur_muxer_upstream_msg_iter
);
905 if (!*muxer_upstream_msg_iter
) {
906 status
= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_END
;
915 bt_component_class_message_iterator_next_method_status
916 validate_muxer_upstream_msg_iter(
917 struct muxer_upstream_msg_iter
*muxer_upstream_msg_iter
,
920 struct muxer_comp
*muxer_comp
=
921 muxer_upstream_msg_iter
->muxer_comp
;
922 bt_component_class_message_iterator_next_method_status status
=
923 BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK
;
925 BT_COMP_LOGD("Validating muxer's upstream message iterator wrapper: "
926 "muxer-upstream-msg-iter-wrap-addr=%p",
927 muxer_upstream_msg_iter
);
929 if (muxer_upstream_msg_iter
->msgs
->length
> 0 ||
930 !muxer_upstream_msg_iter
->msg_iter
) {
931 BT_COMP_LOGD("Already valid or not considered: "
932 "queue-len=%u, upstream-msg-iter-addr=%p",
933 muxer_upstream_msg_iter
->msgs
->length
,
934 muxer_upstream_msg_iter
->msg_iter
);
938 /* muxer_upstream_msg_iter_next() logs details/errors */
939 status
= muxer_upstream_msg_iter_next(muxer_upstream_msg_iter
,
947 bt_component_class_message_iterator_next_method_status
948 validate_muxer_upstream_msg_iters(
949 struct muxer_msg_iter
*muxer_msg_iter
)
951 struct muxer_comp
*muxer_comp
= muxer_msg_iter
->muxer_comp
;
952 bt_component_class_message_iterator_next_method_status status
=
953 BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK
;
956 BT_COMP_LOGD("Validating muxer's upstream message iterator wrappers: "
957 "muxer-msg-iter-addr=%p", muxer_msg_iter
);
959 for (i
= 0; i
< muxer_msg_iter
->active_muxer_upstream_msg_iters
->len
;
961 bool is_ended
= false;
962 struct muxer_upstream_msg_iter
*muxer_upstream_msg_iter
=
964 muxer_msg_iter
->active_muxer_upstream_msg_iters
,
967 status
= validate_muxer_upstream_msg_iter(
968 muxer_upstream_msg_iter
, &is_ended
);
969 if (status
!= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK
) {
971 BT_COMP_LOGE("Cannot validate muxer's upstream message iterator wrapper: "
972 "muxer-msg-iter-addr=%p, "
973 "muxer-upstream-msg-iter-wrap-addr=%p",
975 muxer_upstream_msg_iter
);
977 BT_COMP_LOGD("Cannot validate muxer's upstream message iterator wrapper: "
978 "muxer-msg-iter-addr=%p, "
979 "muxer-upstream-msg-iter-wrap-addr=%p",
981 muxer_upstream_msg_iter
);
988 * Move this muxer upstream message iterator to the
989 * array of ended iterators if it's ended.
991 if (G_UNLIKELY(is_ended
)) {
992 BT_COMP_LOGD("Muxer's upstream message iterator wrapper: ended or canceled: "
993 "muxer-msg-iter-addr=%p, "
994 "muxer-upstream-msg-iter-wrap-addr=%p",
995 muxer_msg_iter
, muxer_upstream_msg_iter
);
997 muxer_msg_iter
->ended_muxer_upstream_msg_iters
,
998 muxer_upstream_msg_iter
);
999 muxer_msg_iter
->active_muxer_upstream_msg_iters
->pdata
[i
] = NULL
;
1002 * Use g_ptr_array_remove_fast() because the
1003 * order of those elements is not important.
1005 g_ptr_array_remove_index_fast(
1006 muxer_msg_iter
->active_muxer_upstream_msg_iters
,
1017 bt_component_class_message_iterator_next_method_status
muxer_msg_iter_do_next_one(
1018 struct muxer_comp
*muxer_comp
,
1019 struct muxer_msg_iter
*muxer_msg_iter
,
1020 const bt_message
**msg
)
1022 bt_component_class_message_iterator_next_method_status status
;
1023 struct muxer_upstream_msg_iter
*muxer_upstream_msg_iter
= NULL
;
1024 int64_t next_return_ts
;
1026 status
= validate_muxer_upstream_msg_iters(muxer_msg_iter
);
1027 if (status
!= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK
) {
1028 /* validate_muxer_upstream_msg_iters() logs details */
1033 * At this point we know that all the existing upstream
1034 * message iterators are valid. We can find the one,
1035 * amongst those, of which the current message is the
1038 status
= muxer_msg_iter_youngest_upstream_msg_iter(muxer_comp
,
1039 muxer_msg_iter
, &muxer_upstream_msg_iter
,
1041 if (status
< 0 || status
== BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_END
) {
1043 BT_COMP_LOGE("Cannot find the youngest upstream message iterator wrapper: "
1045 bt_common_func_status_string(status
));
1047 BT_COMP_LOGD("Cannot find the youngest upstream message iterator wrapper: "
1049 bt_common_func_status_string(status
));
1055 if (next_return_ts
< muxer_msg_iter
->last_returned_ts_ns
) {
1056 BT_COMP_LOGE("Youngest upstream message iterator wrapper's timestamp is less than muxer's message iterator's last returned timestamp: "
1057 "muxer-msg-iter-addr=%p, ts=%" PRId64
", "
1058 "last-returned-ts=%" PRId64
,
1059 muxer_msg_iter
, next_return_ts
,
1060 muxer_msg_iter
->last_returned_ts_ns
);
1061 status
= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_ERROR
;
1065 BT_COMP_LOGD("Found youngest upstream message iterator wrapper: "
1066 "muxer-msg-iter-addr=%p, "
1067 "muxer-upstream-msg-iter-wrap-addr=%p, "
1069 muxer_msg_iter
, muxer_upstream_msg_iter
, next_return_ts
);
1070 BT_ASSERT_DBG(status
==
1071 BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK
);
1072 BT_ASSERT_DBG(muxer_upstream_msg_iter
);
1075 * Consume from the queue's head: other side
1076 * (muxer_upstream_msg_iter_next()) writes to the tail.
1078 *msg
= g_queue_pop_head(muxer_upstream_msg_iter
->msgs
);
1079 BT_ASSERT_DBG(*msg
);
1080 muxer_msg_iter
->last_returned_ts_ns
= next_return_ts
;
1087 bt_component_class_message_iterator_next_method_status
muxer_msg_iter_do_next(
1088 struct muxer_comp
*muxer_comp
,
1089 struct muxer_msg_iter
*muxer_msg_iter
,
1090 bt_message_array_const msgs
, uint64_t capacity
,
1093 bt_component_class_message_iterator_next_method_status status
=
1094 BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK
;
1097 while (i
< capacity
&& status
== BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK
) {
1098 status
= muxer_msg_iter_do_next_one(muxer_comp
,
1099 muxer_msg_iter
, &msgs
[i
]);
1100 if (status
== BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK
) {
1107 * Even if muxer_msg_iter_do_next_one() returned
1108 * something else than
1109 * BT_MESSAGE_ITERATOR_STATUS_OK, we accumulated
1110 * message objects in the output message
1111 * array, so we need to return
1112 * BT_MESSAGE_ITERATOR_STATUS_OK so that they are
1113 * transfered to downstream. This other status occurs
1114 * again the next time muxer_msg_iter_do_next() is
1115 * called, possibly without any accumulated
1116 * message, in which case we'll return it.
1119 status
= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK
;
1126 void destroy_muxer_msg_iter(struct muxer_msg_iter
*muxer_msg_iter
)
1128 struct muxer_comp
*muxer_comp
;
1130 if (!muxer_msg_iter
) {
1134 muxer_comp
= muxer_msg_iter
->muxer_comp
;
1135 BT_COMP_LOGD("Destroying muxer component's message iterator: "
1136 "muxer-msg-iter-addr=%p", muxer_msg_iter
);
1138 if (muxer_msg_iter
->active_muxer_upstream_msg_iters
) {
1139 BT_COMP_LOGD_STR("Destroying muxer's active upstream message iterator wrappers.");
1141 muxer_msg_iter
->active_muxer_upstream_msg_iters
, TRUE
);
1144 if (muxer_msg_iter
->ended_muxer_upstream_msg_iters
) {
1145 BT_COMP_LOGD_STR("Destroying muxer's ended upstream message iterator wrappers.");
1147 muxer_msg_iter
->ended_muxer_upstream_msg_iters
, TRUE
);
1150 g_free(muxer_msg_iter
);
1154 bt_component_class_message_iterator_initialize_method_status
1155 muxer_msg_iter_init_upstream_iterators(struct muxer_comp
*muxer_comp
,
1156 struct muxer_msg_iter
*muxer_msg_iter
,
1157 struct bt_self_message_iterator_configuration
*config
)
1161 bt_component_class_message_iterator_initialize_method_status status
;
1162 bool can_seek_forward
= true;
1164 count
= bt_component_filter_get_input_port_count(
1165 bt_self_component_filter_as_component_filter(
1166 muxer_comp
->self_comp_flt
));
1168 BT_COMP_LOGD("No input port to initialize for muxer component's message iterator: "
1169 "muxer-comp-addr=%p, muxer-msg-iter-addr=%p",
1170 muxer_comp
, muxer_msg_iter
);
1171 status
= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_INITIALIZE_METHOD_STATUS_OK
;
1175 for (i
= 0; i
< count
; i
++) {
1176 bt_self_component_port_input_message_iterator
*upstream_msg_iter
;
1177 bt_self_component_port_input
*self_port
=
1178 bt_self_component_filter_borrow_input_port_by_index(
1179 muxer_comp
->self_comp_flt
, i
);
1180 const bt_port
*port
;
1181 bt_self_component_port_input_message_iterator_create_from_message_iterator_status
1185 BT_ASSERT(self_port
);
1186 port
= bt_self_component_port_as_port(
1187 bt_self_component_port_input_as_self_component_port(
1191 if (!bt_port_is_connected(port
)) {
1192 /* Skip non-connected port */
1196 msg_iter_status
= create_msg_iter_on_input_port(muxer_comp
,
1197 muxer_msg_iter
, self_port
, &upstream_msg_iter
);
1198 if (msg_iter_status
!= BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_CREATE_FROM_MESSAGE_ITERATOR_STATUS_OK
) {
1199 /* create_msg_iter_on_input_port() logs errors */
1200 status
= (int) msg_iter_status
;
1204 int_status
= muxer_msg_iter_add_upstream_msg_iter(muxer_msg_iter
,
1206 bt_self_component_port_input_message_iterator_put_ref(
1209 status
= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_INITIALIZE_METHOD_STATUS_ERROR
;
1210 /* muxer_msg_iter_add_upstream_msg_iter() logs errors */
1214 can_seek_forward
= can_seek_forward
&&
1215 bt_self_component_port_input_message_iterator_can_seek_forward(
1220 * This iterator can seek forward if all of its iterators can seek
1223 bt_self_message_iterator_configuration_set_can_seek_forward(
1224 config
, can_seek_forward
);
1226 status
= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_INITIALIZE_METHOD_STATUS_OK
;
1233 bt_component_class_message_iterator_initialize_method_status
muxer_msg_iter_init(
1234 bt_self_message_iterator
*self_msg_iter
,
1235 bt_self_message_iterator_configuration
*config
,
1236 bt_self_component_filter
*self_comp
,
1237 bt_self_component_port_output
*port
)
1239 struct muxer_comp
*muxer_comp
= NULL
;
1240 struct muxer_msg_iter
*muxer_msg_iter
= NULL
;
1241 bt_component_class_message_iterator_initialize_method_status status
;
1243 muxer_comp
= bt_self_component_get_data(
1244 bt_self_component_filter_as_self_component(self_comp
));
1245 BT_ASSERT(muxer_comp
);
1246 BT_COMP_LOGD("Initializing muxer component's message iterator: "
1247 "comp-addr=%p, muxer-comp-addr=%p, msg-iter-addr=%p",
1248 self_comp
, muxer_comp
, self_msg_iter
);
1250 if (muxer_comp
->initializing_muxer_msg_iter
) {
1252 * Weird, unhandled situation detected: downstream
1253 * creates a muxer message iterator while creating
1254 * another muxer message iterator (same component).
1256 BT_COMP_LOGE("Recursive initialization of muxer component's message iterator: "
1257 "comp-addr=%p, muxer-comp-addr=%p, msg-iter-addr=%p",
1258 self_comp
, muxer_comp
, self_msg_iter
);
1259 status
= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_INITIALIZE_METHOD_STATUS_ERROR
;
1263 muxer_comp
->initializing_muxer_msg_iter
= true;
1264 muxer_msg_iter
= g_new0(struct muxer_msg_iter
, 1);
1265 if (!muxer_msg_iter
) {
1266 BT_COMP_LOGE_STR("Failed to allocate one muxer component's message iterator.");
1267 status
= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_INITIALIZE_METHOD_STATUS_MEMORY_ERROR
;
1271 muxer_msg_iter
->muxer_comp
= muxer_comp
;
1272 muxer_msg_iter
->self_msg_iter
= self_msg_iter
;
1273 muxer_msg_iter
->last_returned_ts_ns
= INT64_MIN
;
1274 muxer_msg_iter
->active_muxer_upstream_msg_iters
=
1275 g_ptr_array_new_with_free_func(
1276 (GDestroyNotify
) destroy_muxer_upstream_msg_iter
);
1277 if (!muxer_msg_iter
->active_muxer_upstream_msg_iters
) {
1278 BT_COMP_LOGE_STR("Failed to allocate a GPtrArray.");
1279 status
= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_INITIALIZE_METHOD_STATUS_MEMORY_ERROR
;
1283 muxer_msg_iter
->ended_muxer_upstream_msg_iters
=
1284 g_ptr_array_new_with_free_func(
1285 (GDestroyNotify
) destroy_muxer_upstream_msg_iter
);
1286 if (!muxer_msg_iter
->ended_muxer_upstream_msg_iters
) {
1287 BT_COMP_LOGE_STR("Failed to allocate a GPtrArray.");
1288 status
= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_INITIALIZE_METHOD_STATUS_MEMORY_ERROR
;
1292 status
= muxer_msg_iter_init_upstream_iterators(muxer_comp
,
1293 muxer_msg_iter
, config
);
1295 BT_COMP_LOGE("Cannot initialize connected input ports for muxer component's message iterator: "
1296 "comp-addr=%p, muxer-comp-addr=%p, "
1297 "muxer-msg-iter-addr=%p, msg-iter-addr=%p, ret=%d",
1298 self_comp
, muxer_comp
, muxer_msg_iter
,
1299 self_msg_iter
, status
);
1303 bt_self_message_iterator_set_data(self_msg_iter
, muxer_msg_iter
);
1304 BT_COMP_LOGD("Initialized muxer component's message iterator: "
1305 "comp-addr=%p, muxer-comp-addr=%p, muxer-msg-iter-addr=%p, "
1307 self_comp
, muxer_comp
, muxer_msg_iter
, self_msg_iter
);
1311 destroy_muxer_msg_iter(muxer_msg_iter
);
1312 bt_self_message_iterator_set_data(self_msg_iter
, NULL
);
1315 muxer_comp
->initializing_muxer_msg_iter
= false;
1320 void muxer_msg_iter_finalize(bt_self_message_iterator
*self_msg_iter
)
1322 struct muxer_msg_iter
*muxer_msg_iter
=
1323 bt_self_message_iterator_get_data(self_msg_iter
);
1324 bt_self_component
*self_comp
= NULL
;
1325 struct muxer_comp
*muxer_comp
= NULL
;
1327 self_comp
= bt_self_message_iterator_borrow_component(
1329 BT_ASSERT(self_comp
);
1330 muxer_comp
= bt_self_component_get_data(self_comp
);
1331 BT_COMP_LOGD("Finalizing muxer component's message iterator: "
1332 "comp-addr=%p, muxer-comp-addr=%p, muxer-msg-iter-addr=%p, "
1334 self_comp
, muxer_comp
, muxer_msg_iter
, self_msg_iter
);
1336 if (muxer_msg_iter
) {
1337 destroy_muxer_msg_iter(muxer_msg_iter
);
1342 bt_component_class_message_iterator_next_method_status
muxer_msg_iter_next(
1343 bt_self_message_iterator
*self_msg_iter
,
1344 bt_message_array_const msgs
, uint64_t capacity
,
1347 bt_component_class_message_iterator_next_method_status status
;
1348 struct muxer_msg_iter
*muxer_msg_iter
=
1349 bt_self_message_iterator_get_data(self_msg_iter
);
1350 bt_self_component
*self_comp
= NULL
;
1351 struct muxer_comp
*muxer_comp
= NULL
;
1353 BT_ASSERT_DBG(muxer_msg_iter
);
1354 self_comp
= bt_self_message_iterator_borrow_component(
1356 BT_ASSERT_DBG(self_comp
);
1357 muxer_comp
= bt_self_component_get_data(self_comp
);
1358 BT_ASSERT_DBG(muxer_comp
);
1359 BT_COMP_LOGT("Muxer component's message iterator's \"next\" method called: "
1360 "comp-addr=%p, muxer-comp-addr=%p, muxer-msg-iter-addr=%p, "
1362 self_comp
, muxer_comp
, muxer_msg_iter
, self_msg_iter
);
1364 status
= muxer_msg_iter_do_next(muxer_comp
, muxer_msg_iter
,
1365 msgs
, capacity
, count
);
1367 BT_COMP_LOGE("Cannot get next message: "
1368 "comp-addr=%p, muxer-comp-addr=%p, muxer-msg-iter-addr=%p, "
1369 "msg-iter-addr=%p, status=%s",
1370 self_comp
, muxer_comp
, muxer_msg_iter
, self_msg_iter
,
1371 bt_common_func_status_string(status
));
1373 BT_COMP_LOGT("Returning from muxer component's message iterator's \"next\" method: "
1375 bt_common_func_status_string(status
));
1382 bt_component_class_port_connected_method_status
muxer_input_port_connected(
1383 bt_self_component_filter
*self_comp
,
1384 bt_self_component_port_input
*self_port
,
1385 const bt_port_output
*other_port
)
1387 bt_component_class_port_connected_method_status status
=
1388 BT_COMPONENT_CLASS_PORT_CONNECTED_METHOD_STATUS_OK
;
1389 bt_self_component_add_port_status add_port_status
;
1390 struct muxer_comp
*muxer_comp
= bt_self_component_get_data(
1391 bt_self_component_filter_as_self_component(self_comp
));
1393 add_port_status
= add_available_input_port(self_comp
);
1394 if (add_port_status
) {
1395 BT_COMP_LOGE("Cannot add one muxer component's input port: "
1397 bt_common_func_status_string(status
));
1399 if (add_port_status
==
1400 BT_SELF_COMPONENT_ADD_PORT_STATUS_MEMORY_ERROR
) {
1401 status
= BT_COMPONENT_CLASS_PORT_CONNECTED_METHOD_STATUS_MEMORY_ERROR
;
1403 status
= BT_COMPONENT_CLASS_PORT_CONNECTED_METHOD_STATUS_ERROR
;
1414 bt_component_class_message_iterator_can_seek_beginning_method_status
1415 muxer_upstream_msg_iters_can_all_seek_beginning(
1416 GPtrArray
*muxer_upstream_msg_iters
, bt_bool
*can_seek
)
1418 bt_component_class_message_iterator_can_seek_beginning_method_status status
=
1419 BT_COMPONENT_CLASS_MESSAGE_ITERATOR_CAN_SEEK_BEGINNING_METHOD_STATUS_OK
;
1422 for (i
= 0; i
< muxer_upstream_msg_iters
->len
; i
++) {
1423 struct muxer_upstream_msg_iter
*upstream_msg_iter
=
1424 muxer_upstream_msg_iters
->pdata
[i
];
1425 status
= (int) bt_self_component_port_input_message_iterator_can_seek_beginning(
1426 upstream_msg_iter
->msg_iter
, can_seek
);
1427 if (status
!= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_CAN_SEEK_BEGINNING_METHOD_STATUS_OK
) {
1436 *can_seek
= BT_TRUE
;
1443 bt_component_class_message_iterator_can_seek_beginning_method_status
1444 muxer_msg_iter_can_seek_beginning(
1445 bt_self_message_iterator
*self_msg_iter
, bt_bool
*can_seek
)
1447 struct muxer_msg_iter
*muxer_msg_iter
=
1448 bt_self_message_iterator_get_data(self_msg_iter
);
1449 bt_component_class_message_iterator_can_seek_beginning_method_status status
;
1451 status
= muxer_upstream_msg_iters_can_all_seek_beginning(
1452 muxer_msg_iter
->active_muxer_upstream_msg_iters
, can_seek
);
1453 if (status
!= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_CAN_SEEK_BEGINNING_METHOD_STATUS_OK
) {
1461 status
= muxer_upstream_msg_iters_can_all_seek_beginning(
1462 muxer_msg_iter
->ended_muxer_upstream_msg_iters
, can_seek
);
1469 bt_component_class_message_iterator_seek_beginning_method_status
muxer_msg_iter_seek_beginning(
1470 bt_self_message_iterator
*self_msg_iter
)
1472 struct muxer_msg_iter
*muxer_msg_iter
=
1473 bt_self_message_iterator_get_data(self_msg_iter
);
1474 bt_component_class_message_iterator_seek_beginning_method_status status
=
1475 BT_COMPONENT_CLASS_MESSAGE_ITERATOR_SEEK_BEGINNING_METHOD_STATUS_OK
;
1476 bt_message_iterator_seek_beginning_status seek_beg_status
;
1479 /* Seek all ended upstream iterators first */
1480 for (i
= 0; i
< muxer_msg_iter
->ended_muxer_upstream_msg_iters
->len
;
1482 struct muxer_upstream_msg_iter
*upstream_msg_iter
=
1483 muxer_msg_iter
->ended_muxer_upstream_msg_iters
->pdata
[i
];
1485 seek_beg_status
= bt_self_component_port_input_message_iterator_seek_beginning(
1486 upstream_msg_iter
->msg_iter
);
1487 if (seek_beg_status
!= BT_MESSAGE_ITERATOR_SEEK_BEGINNING_STATUS_OK
) {
1488 status
= (int) seek_beg_status
;
1492 empty_message_queue(upstream_msg_iter
);
1495 /* Seek all previously active upstream iterators */
1496 for (i
= 0; i
< muxer_msg_iter
->active_muxer_upstream_msg_iters
->len
;
1498 struct muxer_upstream_msg_iter
*upstream_msg_iter
=
1499 muxer_msg_iter
->active_muxer_upstream_msg_iters
->pdata
[i
];
1501 seek_beg_status
= bt_self_component_port_input_message_iterator_seek_beginning(
1502 upstream_msg_iter
->msg_iter
);
1503 if (seek_beg_status
!= BT_MESSAGE_ITERATOR_SEEK_BEGINNING_STATUS_OK
) {
1504 status
= (int) seek_beg_status
;
1508 empty_message_queue(upstream_msg_iter
);
1511 /* Make them all active */
1512 for (i
= 0; i
< muxer_msg_iter
->ended_muxer_upstream_msg_iters
->len
;
1514 struct muxer_upstream_msg_iter
*upstream_msg_iter
=
1515 muxer_msg_iter
->ended_muxer_upstream_msg_iters
->pdata
[i
];
1517 g_ptr_array_add(muxer_msg_iter
->active_muxer_upstream_msg_iters
,
1519 muxer_msg_iter
->ended_muxer_upstream_msg_iters
->pdata
[i
] = NULL
;
1523 * GLib < 2.48.0 asserts when g_ptr_array_remove_range() is
1524 * called on an empty array.
1526 if (muxer_msg_iter
->ended_muxer_upstream_msg_iters
->len
> 0) {
1527 g_ptr_array_remove_range(muxer_msg_iter
->ended_muxer_upstream_msg_iters
,
1528 0, muxer_msg_iter
->ended_muxer_upstream_msg_iters
->len
);
1530 muxer_msg_iter
->last_returned_ts_ns
= INT64_MIN
;
1531 muxer_msg_iter
->clock_class_expectation
=
1532 MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_ANY
;