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
252 bt_component_class_initialize_method_status
muxer_init(
253 bt_self_component_filter
*self_comp_flt
,
254 bt_self_component_filter_configuration
*config
,
255 const bt_value
*params
, void *init_data
)
257 bt_component_class_initialize_method_status status
;
258 bt_self_component_add_port_status add_port_status
;
259 bt_self_component
*self_comp
=
260 bt_self_component_filter_as_self_component(self_comp_flt
);
261 struct muxer_comp
*muxer_comp
= g_new0(struct muxer_comp
, 1);
262 bt_logging_level log_level
= bt_component_get_logging_level(
263 bt_self_component_as_component(self_comp
));
264 enum bt_param_validation_status validation_status
;
265 gchar
*validate_error
= NULL
;
267 BT_COMP_LOG_CUR_LVL(BT_LOG_INFO
, log_level
, self_comp
,
268 "Initializing muxer component: "
269 "comp-addr=%p, params-addr=%p", self_comp
, params
);
273 * Don't use BT_COMP_LOGE_APPEND_CAUSE, as `muxer_comp` is not
276 BT_COMP_LOG_CUR_LVL(BT_LOG_ERROR
, log_level
, self_comp
,
277 "Failed to allocate one muxer component.");
278 BT_CURRENT_THREAD_ERROR_APPEND_CAUSE_FROM_COMPONENT(self_comp
,
279 "Failed to allocate one muxer component.");
280 status
= BT_COMPONENT_CLASS_INITIALIZE_METHOD_STATUS_MEMORY_ERROR
;
284 muxer_comp
->log_level
= log_level
;
285 muxer_comp
->self_comp
= self_comp
;
286 muxer_comp
->self_comp_flt
= self_comp_flt
;
288 validation_status
= bt_param_validation_validate(params
,
289 muxer_params
, &validate_error
);
290 if (validation_status
== BT_PARAM_VALIDATION_STATUS_MEMORY_ERROR
) {
291 status
= BT_COMPONENT_CLASS_INITIALIZE_METHOD_STATUS_MEMORY_ERROR
;
293 } else if (validation_status
== BT_PARAM_VALIDATION_STATUS_VALIDATION_ERROR
) {
294 status
= BT_COMPONENT_CLASS_INITIALIZE_METHOD_STATUS_ERROR
;
295 BT_COMP_LOGE_APPEND_CAUSE(self_comp
, "%s", validate_error
);
299 bt_self_component_set_data(self_comp
, muxer_comp
);
300 add_port_status
= add_available_input_port(self_comp_flt
);
301 if (add_port_status
!= BT_SELF_COMPONENT_ADD_PORT_STATUS_OK
) {
302 BT_COMP_LOGE_APPEND_CAUSE(self_comp
,
303 "Cannot ensure that at least one muxer component's input port is available: "
304 "muxer-comp-addr=%p, status=%s",
305 muxer_comp
, bt_common_func_status_string(add_port_status
));
306 status
= (int) add_port_status
;
310 add_port_status
= create_output_port(self_comp_flt
);
311 if (add_port_status
!= BT_SELF_COMPONENT_ADD_PORT_STATUS_OK
) {
312 BT_COMP_LOGE_APPEND_CAUSE(self_comp
,
313 "Cannot create muxer component's output port: "
314 "muxer-comp-addr=%p, status=%s",
315 muxer_comp
, bt_common_func_status_string(add_port_status
));
316 status
= (int) add_port_status
;
320 BT_COMP_LOGI("Initialized muxer component: "
321 "comp-addr=%p, params-addr=%p, muxer-comp-addr=%p",
322 self_comp
, params
, muxer_comp
);
324 status
= BT_COMPONENT_CLASS_INITIALIZE_METHOD_STATUS_OK
;
328 destroy_muxer_comp(muxer_comp
);
329 bt_self_component_set_data(self_comp
, NULL
);
332 g_free(validate_error
);
337 void muxer_finalize(bt_self_component_filter
*self_comp
)
339 struct muxer_comp
*muxer_comp
= bt_self_component_get_data(
340 bt_self_component_filter_as_self_component(self_comp
));
342 BT_COMP_LOGI("Finalizing muxer component: comp-addr=%p",
344 destroy_muxer_comp(muxer_comp
);
348 bt_message_iterator_create_from_message_iterator_status
349 create_msg_iter_on_input_port(struct muxer_comp
*muxer_comp
,
350 struct muxer_msg_iter
*muxer_msg_iter
,
351 bt_self_component_port_input
*self_port
,
352 bt_message_iterator
**msg_iter
)
354 const bt_port
*port
= bt_self_component_port_as_port(
355 bt_self_component_port_input_as_self_component_port(
357 bt_message_iterator_create_from_message_iterator_status
361 BT_ASSERT(bt_port_is_connected(port
));
363 // TODO: Advance the iterator to >= the time of the latest
364 // returned message by the muxer message
365 // iterator which creates it.
366 status
= bt_message_iterator_create_from_message_iterator(
367 muxer_msg_iter
->self_msg_iter
, self_port
, msg_iter
);
368 if (status
!= BT_MESSAGE_ITERATOR_CREATE_FROM_MESSAGE_ITERATOR_STATUS_OK
) {
369 BT_COMP_LOGE_APPEND_CAUSE(muxer_comp
->self_comp
,
370 "Cannot create upstream message iterator on input port: "
371 "port-addr=%p, port-name=\"%s\"",
372 port
, bt_port_get_name(port
));
376 BT_COMP_LOGI("Created upstream message iterator on input port: "
377 "port-addr=%p, port-name=\"%s\", msg-iter-addr=%p",
378 port
, bt_port_get_name(port
), msg_iter
);
385 bt_message_iterator_class_next_method_status
muxer_upstream_msg_iter_next(
386 struct muxer_upstream_msg_iter
*muxer_upstream_msg_iter
,
389 struct muxer_comp
*muxer_comp
= muxer_upstream_msg_iter
->muxer_comp
;
390 bt_message_iterator_class_next_method_status status
;
391 bt_message_iterator_next_status input_port_iter_status
;
392 bt_message_array_const msgs
;
396 BT_COMP_LOGD("Calling upstream message iterator's \"next\" method: "
397 "muxer-upstream-msg-iter-wrap-addr=%p, msg-iter-addr=%p",
398 muxer_upstream_msg_iter
,
399 muxer_upstream_msg_iter
->msg_iter
);
400 input_port_iter_status
= bt_message_iterator_next(
401 muxer_upstream_msg_iter
->msg_iter
, &msgs
, &count
);
402 BT_COMP_LOGD("Upstream message iterator's \"next\" method returned: "
404 bt_common_func_status_string(input_port_iter_status
));
406 switch (input_port_iter_status
) {
407 case BT_MESSAGE_ITERATOR_NEXT_STATUS_OK
:
409 * Message iterator's current message is
410 * valid: it must be considered for muxing operations.
412 BT_COMP_LOGD_STR("Validated upstream message iterator wrapper.");
413 BT_ASSERT_DBG(count
> 0);
415 g_ptr_array_set_size(muxer_upstream_msg_iter
->msgs
, count
);
416 muxer_upstream_msg_iter
->next_msg
= 0;
418 /* Move messages to our queue */
419 for (i
= 0; i
< count
; i
++) {
421 * Push to tail in order; other side
422 * (muxer_msg_iter_do_next_one()) consumes
423 * from the head first.
425 g_ptr_array_index(muxer_upstream_msg_iter
->msgs
, i
)
426 = (gpointer
*) msgs
[i
];
428 status
= BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_OK
;
430 case BT_MESSAGE_ITERATOR_NEXT_STATUS_AGAIN
:
432 * Message iterator's current message is not
433 * valid anymore. Return
434 * BT_MESSAGE_ITERATOR_NEXT_STATUS_AGAIN immediately.
436 status
= BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_AGAIN
;
438 case BT_MESSAGE_ITERATOR_NEXT_STATUS_END
: /* Fall-through. */
440 * Message iterator reached the end: release it. It
441 * won't be considered again to find the youngest
445 status
= BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_OK
;
447 case BT_MESSAGE_ITERATOR_NEXT_STATUS_ERROR
:
448 case BT_MESSAGE_ITERATOR_NEXT_STATUS_MEMORY_ERROR
:
449 /* Error status code */
450 BT_COMP_LOGE_APPEND_CAUSE(muxer_comp
->self_comp
,
451 "Upstream iterator's next method returned an error: status=%s",
452 bt_common_func_status_string(input_port_iter_status
));
453 status
= (int) input_port_iter_status
;
456 /* Unsupported status code */
457 BT_COMP_LOGE_APPEND_CAUSE(muxer_comp
->self_comp
,
458 "Unsupported status code: status=%s",
459 bt_common_func_status_string(input_port_iter_status
));
460 status
= BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_ERROR
;
468 int get_msg_ts_ns(struct muxer_comp
*muxer_comp
,
469 struct muxer_msg_iter
*muxer_msg_iter
,
470 const bt_message
*msg
, int64_t last_returned_ts_ns
,
473 const bt_clock_snapshot
*clock_snapshot
= NULL
;
475 const bt_stream_class
*stream_class
= NULL
;
476 bt_message_type msg_type
;
479 BT_ASSERT_DBG(ts_ns
);
480 BT_COMP_LOGD("Getting message's timestamp: "
481 "muxer-msg-iter-addr=%p, msg-addr=%p, "
482 "last-returned-ts=%" PRId64
,
483 muxer_msg_iter
, msg
, last_returned_ts_ns
);
485 if (G_UNLIKELY(muxer_msg_iter
->clock_class_expectation
==
486 MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_NONE
)) {
487 *ts_ns
= last_returned_ts_ns
;
491 msg_type
= bt_message_get_type(msg
);
493 if (G_UNLIKELY(msg_type
== BT_MESSAGE_TYPE_PACKET_BEGINNING
)) {
494 stream_class
= bt_stream_borrow_class_const(
495 bt_packet_borrow_stream_const(
496 bt_message_packet_beginning_borrow_packet_const(
498 } else if (G_UNLIKELY(msg_type
== BT_MESSAGE_TYPE_PACKET_END
)) {
499 stream_class
= bt_stream_borrow_class_const(
500 bt_packet_borrow_stream_const(
501 bt_message_packet_end_borrow_packet_const(
503 } else if (G_UNLIKELY(msg_type
== BT_MESSAGE_TYPE_DISCARDED_EVENTS
)) {
504 stream_class
= bt_stream_borrow_class_const(
505 bt_message_discarded_events_borrow_stream_const(msg
));
506 } else if (G_UNLIKELY(msg_type
== BT_MESSAGE_TYPE_DISCARDED_PACKETS
)) {
507 stream_class
= bt_stream_borrow_class_const(
508 bt_message_discarded_packets_borrow_stream_const(msg
));
512 case BT_MESSAGE_TYPE_EVENT
:
513 BT_ASSERT_DBG(bt_message_event_borrow_stream_class_default_clock_class_const(
515 clock_snapshot
= bt_message_event_borrow_default_clock_snapshot_const(
518 case BT_MESSAGE_TYPE_PACKET_BEGINNING
:
519 if (bt_stream_class_packets_have_beginning_default_clock_snapshot(
521 clock_snapshot
= bt_message_packet_beginning_borrow_default_clock_snapshot_const(
524 goto no_clock_snapshot
;
528 case BT_MESSAGE_TYPE_PACKET_END
:
529 if (bt_stream_class_packets_have_end_default_clock_snapshot(
531 clock_snapshot
= bt_message_packet_end_borrow_default_clock_snapshot_const(
534 goto no_clock_snapshot
;
538 case BT_MESSAGE_TYPE_STREAM_BEGINNING
:
540 enum bt_message_stream_clock_snapshot_state snapshot_state
=
541 bt_message_stream_beginning_borrow_default_clock_snapshot_const(
542 msg
, &clock_snapshot
);
543 if (snapshot_state
== BT_MESSAGE_STREAM_CLOCK_SNAPSHOT_STATE_UNKNOWN
) {
544 goto no_clock_snapshot
;
549 case BT_MESSAGE_TYPE_STREAM_END
:
551 enum bt_message_stream_clock_snapshot_state snapshot_state
=
552 bt_message_stream_end_borrow_default_clock_snapshot_const(
553 msg
, &clock_snapshot
);
554 if (snapshot_state
== BT_MESSAGE_STREAM_CLOCK_SNAPSHOT_STATE_UNKNOWN
) {
555 goto no_clock_snapshot
;
560 case BT_MESSAGE_TYPE_DISCARDED_EVENTS
:
561 if (bt_stream_class_discarded_events_have_default_clock_snapshots(
563 clock_snapshot
= bt_message_discarded_events_borrow_beginning_default_clock_snapshot_const(
566 goto no_clock_snapshot
;
570 case BT_MESSAGE_TYPE_DISCARDED_PACKETS
:
571 if (bt_stream_class_discarded_packets_have_default_clock_snapshots(
573 clock_snapshot
= bt_message_discarded_packets_borrow_beginning_default_clock_snapshot_const(
576 goto no_clock_snapshot
;
580 case BT_MESSAGE_TYPE_MESSAGE_ITERATOR_INACTIVITY
:
581 clock_snapshot
= bt_message_message_iterator_inactivity_borrow_clock_snapshot_const(
585 /* All the other messages have a higher priority */
586 BT_COMP_LOGD_STR("Message has no timestamp: using the last returned timestamp.");
587 *ts_ns
= last_returned_ts_ns
;
591 ret
= bt_clock_snapshot_get_ns_from_origin(clock_snapshot
, ts_ns
);
593 BT_COMP_LOGE_APPEND_CAUSE(muxer_comp
->self_comp
,
594 "Cannot get nanoseconds from Epoch of clock snapshot: "
595 "clock-snapshot-addr=%p", clock_snapshot
);
602 BT_COMP_LOGD_STR("Message's default clock snapshot is missing: "
603 "using the last returned timestamp.");
604 *ts_ns
= last_returned_ts_ns
;
612 BT_COMP_LOGD("Found message's timestamp: "
613 "muxer-msg-iter-addr=%p, msg-addr=%p, "
614 "last-returned-ts=%" PRId64
", ts=%" PRId64
,
615 muxer_msg_iter
, msg
, last_returned_ts_ns
,
623 int validate_clock_class(struct muxer_msg_iter
*muxer_msg_iter
,
624 struct muxer_comp
*muxer_comp
,
625 const bt_clock_class
*clock_class
)
628 const uint8_t *cc_uuid
;
631 BT_ASSERT_DBG(clock_class
);
632 cc_uuid
= bt_clock_class_get_uuid(clock_class
);
633 cc_name
= bt_clock_class_get_name(clock_class
);
635 if (muxer_msg_iter
->clock_class_expectation
==
636 MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_ANY
) {
638 * This is the first clock class that this muxer message
639 * iterator encounters. Its properties determine what to expect
640 * for the whole lifetime of the iterator.
642 if (bt_clock_class_origin_is_unix_epoch(clock_class
)) {
643 /* Expect absolute clock classes */
644 muxer_msg_iter
->clock_class_expectation
=
645 MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_ABSOLUTE
;
649 * Expect non-absolute clock classes
650 * with a specific UUID.
652 muxer_msg_iter
->clock_class_expectation
=
653 MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_NOT_ABS_SPEC_UUID
;
654 bt_uuid_copy(muxer_msg_iter
->expected_clock_class_uuid
, cc_uuid
);
657 * Expect non-absolute clock classes
660 muxer_msg_iter
->clock_class_expectation
=
661 MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_NOT_ABS_NO_UUID
;
666 switch (muxer_msg_iter
->clock_class_expectation
) {
667 case MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_ABSOLUTE
:
668 if (!bt_clock_class_origin_is_unix_epoch(clock_class
)) {
669 BT_COMP_LOGE_APPEND_CAUSE(muxer_comp
->self_comp
,
670 "Expecting an absolute clock class, "
671 "but got a non-absolute one: "
672 "clock-class-addr=%p, clock-class-name=\"%s\"",
673 clock_class
, cc_name
);
677 case MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_NOT_ABS_NO_UUID
:
678 if (bt_clock_class_origin_is_unix_epoch(clock_class
)) {
679 BT_COMP_LOGE_APPEND_CAUSE(muxer_comp
->self_comp
,
680 "Expecting a non-absolute clock class with no 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_APPEND_CAUSE(muxer_comp
->self_comp
,
689 "Expecting a non-absolute clock class with no UUID, "
690 "but got one with a UUID: "
691 "clock-class-addr=%p, clock-class-name=\"%s\", "
692 "uuid=\"" BT_UUID_FMT
"\"",
693 clock_class
, cc_name
, BT_UUID_FMT_VALUES(cc_uuid
));
697 case MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_NOT_ABS_SPEC_UUID
:
698 if (bt_clock_class_origin_is_unix_epoch(clock_class
)) {
699 BT_COMP_LOGE_APPEND_CAUSE(muxer_comp
->self_comp
,
700 "Expecting a non-absolute clock class with a specific UUID, "
701 "but got an absolute one: "
702 "clock-class-addr=%p, clock-class-name=\"%s\"",
703 clock_class
, cc_name
);
708 BT_COMP_LOGE_APPEND_CAUSE(muxer_comp
->self_comp
,
709 "Expecting a non-absolute clock class with a specific UUID, "
710 "but got one with no UUID: "
711 "clock-class-addr=%p, clock-class-name=\"%s\"",
712 clock_class
, cc_name
);
716 if (bt_uuid_compare(muxer_msg_iter
->expected_clock_class_uuid
, cc_uuid
) != 0) {
717 BT_COMP_LOGE_APPEND_CAUSE(muxer_comp
->self_comp
,
718 "Expecting a non-absolute clock class with a specific UUID, "
719 "but got one with different UUID: "
720 "clock-class-addr=%p, clock-class-name=\"%s\", "
721 "expected-uuid=\"" BT_UUID_FMT
"\", "
722 "uuid=\"" BT_UUID_FMT
"\"",
723 clock_class
, cc_name
,
724 BT_UUID_FMT_VALUES(muxer_msg_iter
->expected_clock_class_uuid
),
725 BT_UUID_FMT_VALUES(cc_uuid
));
729 case MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_NONE
:
730 BT_COMP_LOGE_APPEND_CAUSE(muxer_comp
->self_comp
,
731 "Expecting no clock class, but got one: "
732 "clock-class-addr=%p, clock-class-name=\"%s\"",
733 clock_class
, cc_name
);
737 BT_COMP_LOGF("Unexpected clock class expectation: "
738 "expectation-code=%d",
739 muxer_msg_iter
->clock_class_expectation
);
753 int validate_new_stream_clock_class(struct muxer_msg_iter
*muxer_msg_iter
,
754 struct muxer_comp
*muxer_comp
, const bt_stream
*stream
)
757 const bt_stream_class
*stream_class
=
758 bt_stream_borrow_class_const(stream
);
759 const bt_clock_class
*clock_class
=
760 bt_stream_class_borrow_default_clock_class_const(stream_class
);
763 if (muxer_msg_iter
->clock_class_expectation
==
764 MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_ANY
) {
765 /* Expect no clock class */
766 muxer_msg_iter
->clock_class_expectation
=
767 MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_NONE
;
768 } else if (muxer_msg_iter
->clock_class_expectation
!=
769 MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_NONE
) {
770 BT_COMP_LOGE_APPEND_CAUSE(muxer_comp
->self_comp
,
771 "Expecting stream class without a default clock class: "
772 "stream-class-addr=%p, stream-class-name=\"%s\", "
773 "stream-class-id=%" PRIu64
,
774 stream_class
, bt_stream_class_get_name(stream_class
),
775 bt_stream_class_get_id(stream_class
));
782 ret
= validate_clock_class(muxer_msg_iter
, muxer_comp
, clock_class
);
789 * This function finds the youngest available message amongst the
790 * non-ended upstream message iterators and returns the upstream
791 * message iterator which has it, or
792 * BT_MESSAGE_ITERATOR_STATUS_END if there's no available
795 * This function does NOT:
797 * * Update any upstream message iterator.
798 * * Check the upstream message iterators to retry.
800 * On sucess, this function sets *muxer_upstream_msg_iter to the
801 * upstream message iterator of which the current message is
802 * the youngest, and sets *ts_ns to its time.
805 bt_message_iterator_class_next_method_status
806 muxer_msg_iter_youngest_upstream_msg_iter(
807 struct muxer_comp
*muxer_comp
,
808 struct muxer_msg_iter
*muxer_msg_iter
,
809 struct muxer_upstream_msg_iter
**muxer_upstream_msg_iter
,
814 int64_t youngest_ts_ns
= INT64_MAX
;
815 bt_message_iterator_class_next_method_status status
=
816 BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_OK
;
818 BT_ASSERT_DBG(muxer_comp
);
819 BT_ASSERT_DBG(muxer_msg_iter
);
820 BT_ASSERT_DBG(muxer_upstream_msg_iter
);
821 *muxer_upstream_msg_iter
= NULL
;
823 for (i
= 0; i
< muxer_msg_iter
->active_muxer_upstream_msg_iters
->len
;
825 const bt_message
*msg
;
826 struct muxer_upstream_msg_iter
*cur_muxer_upstream_msg_iter
=
828 muxer_msg_iter
->active_muxer_upstream_msg_iters
,
832 if (!cur_muxer_upstream_msg_iter
->msg_iter
) {
833 /* This upstream message iterator is ended */
834 BT_COMP_LOGT("Skipping ended upstream message iterator: "
835 "muxer-upstream-msg-iter-wrap-addr=%p",
836 cur_muxer_upstream_msg_iter
);
840 BT_ASSERT_DBG(cur_muxer_upstream_msg_iter
->next_msg
<
841 cur_muxer_upstream_msg_iter
->msgs
->len
);
842 msg
= g_ptr_array_index(cur_muxer_upstream_msg_iter
->msgs
,
843 cur_muxer_upstream_msg_iter
->next_msg
);
846 if (G_UNLIKELY(bt_message_get_type(msg
) ==
847 BT_MESSAGE_TYPE_STREAM_BEGINNING
)) {
848 ret
= validate_new_stream_clock_class(
849 muxer_msg_iter
, muxer_comp
,
850 bt_message_stream_beginning_borrow_stream_const(
854 * validate_new_stream_clock_class() logs
857 status
= BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_ERROR
;
860 } else if (G_UNLIKELY(bt_message_get_type(msg
) ==
861 BT_MESSAGE_TYPE_MESSAGE_ITERATOR_INACTIVITY
)) {
862 const bt_clock_snapshot
*cs
;
864 cs
= bt_message_message_iterator_inactivity_borrow_clock_snapshot_const(
866 ret
= validate_clock_class(muxer_msg_iter
, muxer_comp
,
867 bt_clock_snapshot_borrow_clock_class_const(cs
));
869 /* validate_clock_class() logs errors */
870 status
= BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_ERROR
;
875 ret
= get_msg_ts_ns(muxer_comp
, muxer_msg_iter
, msg
,
876 muxer_msg_iter
->last_returned_ts_ns
, &msg_ts_ns
);
878 /* get_msg_ts_ns() logs errors */
879 *muxer_upstream_msg_iter
= NULL
;
880 status
= BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_ERROR
;
885 * Update the current message iterator if it has not been set
886 * yet, or if its current message has a timestamp smaller than
887 * the previously selected youngest message.
889 if (G_UNLIKELY(*muxer_upstream_msg_iter
== NULL
) ||
890 msg_ts_ns
< youngest_ts_ns
) {
891 *muxer_upstream_msg_iter
=
892 cur_muxer_upstream_msg_iter
;
893 youngest_ts_ns
= msg_ts_ns
;
894 *ts_ns
= youngest_ts_ns
;
895 } else if (msg_ts_ns
== youngest_ts_ns
) {
897 * The currently selected message to be sent downstream
898 * next has the exact same timestamp that of the
899 * current candidate message. We must break the tie
900 * in a predictable manner.
902 BT_ASSERT_DBG((*muxer_upstream_msg_iter
)->next_msg
<
903 (*muxer_upstream_msg_iter
)->msgs
->len
);
904 const bt_message
*selected_msg
=
905 g_ptr_array_index((*muxer_upstream_msg_iter
)->msgs
,
906 (*muxer_upstream_msg_iter
)->next_msg
);
907 BT_COMP_LOGD_STR("Two of the next message candidates have the same timestamps, pick one deterministically.");
910 * Order the messages in an arbitrary but determinitic
913 ret
= common_muxing_compare_messages(msg
, selected_msg
);
916 * The `msg` should go first. Update the next
917 * iterator and the current timestamp.
919 *muxer_upstream_msg_iter
=
920 cur_muxer_upstream_msg_iter
;
921 youngest_ts_ns
= msg_ts_ns
;
922 *ts_ns
= youngest_ts_ns
;
923 } else if (ret
== 0) {
924 /* Unable to pick which one should go first. */
925 BT_COMP_LOGW("Cannot deterministically pick next upstream message iterator because they have identical next messages: "
926 "muxer-upstream-msg-iter-wrap-addr=%p"
927 "cur-muxer-upstream-msg-iter-wrap-addr=%p",
928 *muxer_upstream_msg_iter
,
929 cur_muxer_upstream_msg_iter
);
934 if (!*muxer_upstream_msg_iter
) {
935 status
= BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_END
;
944 bt_message_iterator_class_next_method_status
945 validate_muxer_upstream_msg_iter(
946 struct muxer_upstream_msg_iter
*muxer_upstream_msg_iter
,
949 struct muxer_comp
*muxer_comp
= muxer_upstream_msg_iter
->muxer_comp
;
950 bt_message_iterator_class_next_method_status status
;
952 BT_COMP_LOGD("Validating muxer's upstream message iterator wrapper: "
953 "muxer-upstream-msg-iter-wrap-addr=%p",
954 muxer_upstream_msg_iter
);
956 if (muxer_upstream_msg_iter
->next_msg
< muxer_upstream_msg_iter
->msgs
->len
||
957 !muxer_upstream_msg_iter
->msg_iter
) {
958 BT_COMP_LOGD("Already valid or not considered: "
959 "queue-len=%u, next-msg=%u, upstream-msg-iter-addr=%p",
960 muxer_upstream_msg_iter
->msgs
->len
,
961 muxer_upstream_msg_iter
->next_msg
,
962 muxer_upstream_msg_iter
->msg_iter
);
963 status
= BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_OK
;
967 /* muxer_upstream_msg_iter_next() logs details/errors */
968 status
= muxer_upstream_msg_iter_next(muxer_upstream_msg_iter
,
976 bt_message_iterator_class_next_method_status
977 validate_muxer_upstream_msg_iters(
978 struct muxer_msg_iter
*muxer_msg_iter
)
980 struct muxer_comp
*muxer_comp
= muxer_msg_iter
->muxer_comp
;
981 bt_message_iterator_class_next_method_status status
;
984 BT_COMP_LOGD("Validating muxer's upstream message iterator wrappers: "
985 "muxer-msg-iter-addr=%p", muxer_msg_iter
);
987 for (i
= 0; i
< muxer_msg_iter
->active_muxer_upstream_msg_iters
->len
;
989 bool is_ended
= false;
990 struct muxer_upstream_msg_iter
*muxer_upstream_msg_iter
=
992 muxer_msg_iter
->active_muxer_upstream_msg_iters
,
995 status
= validate_muxer_upstream_msg_iter(
996 muxer_upstream_msg_iter
, &is_ended
);
997 if (status
!= BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_OK
) {
999 BT_COMP_LOGE_APPEND_CAUSE(muxer_comp
->self_comp
,
1000 "Cannot validate muxer's upstream message iterator wrapper: "
1001 "muxer-msg-iter-addr=%p, "
1002 "muxer-upstream-msg-iter-wrap-addr=%p",
1004 muxer_upstream_msg_iter
);
1006 BT_COMP_LOGD("Cannot validate muxer's upstream message iterator wrapper: "
1007 "muxer-msg-iter-addr=%p, "
1008 "muxer-upstream-msg-iter-wrap-addr=%p",
1010 muxer_upstream_msg_iter
);
1017 * Move this muxer upstream message iterator to the
1018 * array of ended iterators if it's ended.
1020 if (G_UNLIKELY(is_ended
)) {
1021 BT_COMP_LOGD("Muxer's upstream message iterator wrapper: ended or canceled: "
1022 "muxer-msg-iter-addr=%p, "
1023 "muxer-upstream-msg-iter-wrap-addr=%p",
1024 muxer_msg_iter
, muxer_upstream_msg_iter
);
1026 muxer_msg_iter
->ended_muxer_upstream_msg_iters
,
1027 muxer_upstream_msg_iter
);
1028 muxer_msg_iter
->active_muxer_upstream_msg_iters
->pdata
[i
] = NULL
;
1031 * Use g_ptr_array_remove_fast() because the
1032 * order of those elements is not important.
1034 g_ptr_array_remove_index_fast(
1035 muxer_msg_iter
->active_muxer_upstream_msg_iters
,
1041 status
= BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_OK
;
1048 bt_message_iterator_class_next_method_status
muxer_msg_iter_do_next_one(
1049 struct muxer_comp
*muxer_comp
,
1050 struct muxer_msg_iter
*muxer_msg_iter
,
1051 const bt_message
**msg
)
1053 bt_message_iterator_class_next_method_status status
;
1054 struct muxer_upstream_msg_iter
*muxer_upstream_msg_iter
= NULL
;
1055 /* Initialize to avoid -Wmaybe-uninitialized warning with gcc 4.8. */
1056 int64_t next_return_ts
= 0;
1058 status
= validate_muxer_upstream_msg_iters(muxer_msg_iter
);
1059 if (status
!= BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_OK
) {
1060 /* validate_muxer_upstream_msg_iters() logs details */
1065 * At this point we know that all the existing upstream
1066 * message iterators are valid. We can find the one,
1067 * amongst those, of which the current message is the
1070 status
= muxer_msg_iter_youngest_upstream_msg_iter(muxer_comp
,
1071 muxer_msg_iter
, &muxer_upstream_msg_iter
,
1073 if (status
< 0 || status
== BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_END
) {
1075 BT_COMP_LOGE_APPEND_CAUSE(muxer_comp
->self_comp
,
1076 "Cannot find the youngest upstream message iterator wrapper: "
1078 bt_common_func_status_string(status
));
1080 BT_COMP_LOGD("Cannot find the youngest upstream message iterator wrapper: "
1082 bt_common_func_status_string(status
));
1088 if (next_return_ts
< muxer_msg_iter
->last_returned_ts_ns
) {
1089 BT_COMP_LOGE_APPEND_CAUSE(muxer_comp
->self_comp
,
1090 "Youngest upstream message iterator wrapper's timestamp is less than muxer's message iterator's last returned timestamp: "
1091 "muxer-msg-iter-addr=%p, ts=%" PRId64
", "
1092 "last-returned-ts=%" PRId64
,
1093 muxer_msg_iter
, next_return_ts
,
1094 muxer_msg_iter
->last_returned_ts_ns
);
1095 status
= BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_ERROR
;
1099 BT_COMP_LOGD("Found youngest upstream message iterator wrapper: "
1100 "muxer-msg-iter-addr=%p, "
1101 "muxer-upstream-msg-iter-wrap-addr=%p, "
1103 muxer_msg_iter
, muxer_upstream_msg_iter
, next_return_ts
);
1104 BT_ASSERT_DBG(status
==
1105 BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_OK
);
1106 BT_ASSERT_DBG(muxer_upstream_msg_iter
);
1109 * Consume from the queue's head: other side
1110 * (muxer_upstream_msg_iter_next()) writes to the tail.
1112 *msg
= g_ptr_array_index(muxer_upstream_msg_iter
->msgs
,
1113 muxer_upstream_msg_iter
->next_msg
);
1114 g_ptr_array_index(muxer_upstream_msg_iter
->msgs
,
1115 muxer_upstream_msg_iter
->next_msg
) = NULL
;
1116 ++muxer_upstream_msg_iter
->next_msg
;
1117 BT_ASSERT_DBG(*msg
);
1118 muxer_msg_iter
->last_returned_ts_ns
= next_return_ts
;
1125 bt_message_iterator_class_next_method_status
muxer_msg_iter_do_next(
1126 struct muxer_comp
*muxer_comp
,
1127 struct muxer_msg_iter
*muxer_msg_iter
,
1128 bt_message_array_const msgs
, uint64_t capacity
,
1131 bt_message_iterator_class_next_method_status status
;
1134 if (G_UNLIKELY(muxer_msg_iter
->next_saved_error
)) {
1136 * Last time we were called, we hit an error but had some
1137 * messages to deliver, so we stashed the error here. Return
1140 BT_CURRENT_THREAD_MOVE_ERROR_AND_RESET(muxer_msg_iter
->next_saved_error
);
1141 status
= muxer_msg_iter
->next_saved_status
;
1146 status
= muxer_msg_iter_do_next_one(muxer_comp
,
1147 muxer_msg_iter
, &msgs
[i
]);
1148 if (status
== BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_OK
) {
1151 } while (i
< capacity
&& status
== BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_OK
);
1155 * Even if muxer_msg_iter_do_next_one() returned
1156 * something else than
1157 * BT_MESSAGE_ITERATOR_STATUS_OK, we accumulated
1158 * message objects in the output message
1159 * array, so we need to return
1160 * BT_MESSAGE_ITERATOR_STATUS_OK so that they are
1161 * transfered to downstream. This other status occurs
1162 * again the next time muxer_msg_iter_do_next() is
1163 * called, possibly without any accumulated
1164 * message, in which case we'll return it.
1168 * Save this error for the next _next call. Assume that
1169 * this component always appends error causes when
1170 * returning an error status code, which will cause the
1171 * current thread error to be non-NULL.
1173 muxer_msg_iter
->next_saved_error
= bt_current_thread_take_error();
1174 BT_ASSERT(muxer_msg_iter
->next_saved_error
);
1175 muxer_msg_iter
->next_saved_status
= status
;
1179 status
= BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_OK
;
1187 void destroy_muxer_msg_iter(struct muxer_msg_iter
*muxer_msg_iter
)
1189 struct muxer_comp
*muxer_comp
;
1191 if (!muxer_msg_iter
) {
1195 muxer_comp
= muxer_msg_iter
->muxer_comp
;
1196 BT_COMP_LOGD("Destroying muxer component's message iterator: "
1197 "muxer-msg-iter-addr=%p", muxer_msg_iter
);
1199 if (muxer_msg_iter
->active_muxer_upstream_msg_iters
) {
1200 BT_COMP_LOGD_STR("Destroying muxer's active upstream message iterator wrappers.");
1202 muxer_msg_iter
->active_muxer_upstream_msg_iters
, TRUE
);
1205 if (muxer_msg_iter
->ended_muxer_upstream_msg_iters
) {
1206 BT_COMP_LOGD_STR("Destroying muxer's ended upstream message iterator wrappers.");
1208 muxer_msg_iter
->ended_muxer_upstream_msg_iters
, TRUE
);
1211 g_free(muxer_msg_iter
);
1215 bt_message_iterator_class_initialize_method_status
1216 muxer_msg_iter_init_upstream_iterators(struct muxer_comp
*muxer_comp
,
1217 struct muxer_msg_iter
*muxer_msg_iter
,
1218 struct bt_self_message_iterator_configuration
*config
)
1222 bt_message_iterator_class_initialize_method_status status
;
1223 bool can_seek_forward
= true;
1225 count
= bt_component_filter_get_input_port_count(
1226 bt_self_component_filter_as_component_filter(
1227 muxer_comp
->self_comp_flt
));
1229 BT_COMP_LOGD("No input port to initialize for muxer component's message iterator: "
1230 "muxer-comp-addr=%p, muxer-msg-iter-addr=%p",
1231 muxer_comp
, muxer_msg_iter
);
1232 status
= BT_MESSAGE_ITERATOR_CLASS_INITIALIZE_METHOD_STATUS_OK
;
1236 for (i
= 0; i
< count
; i
++) {
1237 bt_message_iterator
*upstream_msg_iter
;
1238 bt_self_component_port_input
*self_port
=
1239 bt_self_component_filter_borrow_input_port_by_index(
1240 muxer_comp
->self_comp_flt
, i
);
1241 const bt_port
*port
;
1242 bt_message_iterator_create_from_message_iterator_status
1246 BT_ASSERT(self_port
);
1247 port
= bt_self_component_port_as_port(
1248 bt_self_component_port_input_as_self_component_port(
1252 if (!bt_port_is_connected(port
)) {
1253 /* Skip non-connected port */
1257 msg_iter_status
= create_msg_iter_on_input_port(muxer_comp
,
1258 muxer_msg_iter
, self_port
, &upstream_msg_iter
);
1259 if (msg_iter_status
!= BT_MESSAGE_ITERATOR_CREATE_FROM_MESSAGE_ITERATOR_STATUS_OK
) {
1260 /* create_msg_iter_on_input_port() logs errors */
1261 status
= (int) msg_iter_status
;
1265 int_status
= muxer_msg_iter_add_upstream_msg_iter(muxer_msg_iter
,
1267 bt_message_iterator_put_ref(
1270 status
= BT_MESSAGE_ITERATOR_CLASS_INITIALIZE_METHOD_STATUS_ERROR
;
1271 /* muxer_msg_iter_add_upstream_msg_iter() logs errors */
1275 can_seek_forward
= can_seek_forward
&&
1276 bt_message_iterator_can_seek_forward(
1281 * This iterator can seek forward if all of its iterators can seek
1284 bt_self_message_iterator_configuration_set_can_seek_forward(
1285 config
, can_seek_forward
);
1287 status
= BT_MESSAGE_ITERATOR_CLASS_INITIALIZE_METHOD_STATUS_OK
;
1294 bt_message_iterator_class_initialize_method_status
muxer_msg_iter_init(
1295 bt_self_message_iterator
*self_msg_iter
,
1296 bt_self_message_iterator_configuration
*config
,
1297 bt_self_component_port_output
*port
)
1299 struct muxer_comp
*muxer_comp
= NULL
;
1300 struct muxer_msg_iter
*muxer_msg_iter
= NULL
;
1301 bt_message_iterator_class_initialize_method_status status
;
1302 bt_self_component
*self_comp
=
1303 bt_self_message_iterator_borrow_component(self_msg_iter
);
1305 muxer_comp
= bt_self_component_get_data(self_comp
);
1306 BT_ASSERT(muxer_comp
);
1307 BT_COMP_LOGD("Initializing muxer component's message iterator: "
1308 "comp-addr=%p, muxer-comp-addr=%p, msg-iter-addr=%p",
1309 self_comp
, muxer_comp
, self_msg_iter
);
1311 if (muxer_comp
->initializing_muxer_msg_iter
) {
1313 * Weird, unhandled situation detected: downstream
1314 * creates a muxer message iterator while creating
1315 * another muxer message iterator (same component).
1317 BT_COMP_LOGE_APPEND_CAUSE(muxer_comp
->self_comp
,
1318 "Recursive initialization of muxer component's message iterator: "
1319 "comp-addr=%p, muxer-comp-addr=%p, msg-iter-addr=%p",
1320 self_comp
, muxer_comp
, self_msg_iter
);
1321 status
= BT_MESSAGE_ITERATOR_CLASS_INITIALIZE_METHOD_STATUS_ERROR
;
1325 muxer_comp
->initializing_muxer_msg_iter
= true;
1326 muxer_msg_iter
= g_new0(struct muxer_msg_iter
, 1);
1327 if (!muxer_msg_iter
) {
1328 BT_COMP_LOGE_APPEND_CAUSE(muxer_comp
->self_comp
,
1329 "Failed to allocate one muxer component's message iterator.");
1330 status
= BT_MESSAGE_ITERATOR_CLASS_INITIALIZE_METHOD_STATUS_MEMORY_ERROR
;
1334 muxer_msg_iter
->muxer_comp
= muxer_comp
;
1335 muxer_msg_iter
->self_msg_iter
= self_msg_iter
;
1336 muxer_msg_iter
->last_returned_ts_ns
= INT64_MIN
;
1337 muxer_msg_iter
->active_muxer_upstream_msg_iters
=
1338 g_ptr_array_new_with_free_func(
1339 (GDestroyNotify
) destroy_muxer_upstream_msg_iter
);
1340 if (!muxer_msg_iter
->active_muxer_upstream_msg_iters
) {
1341 BT_COMP_LOGE_APPEND_CAUSE(muxer_comp
->self_comp
, "Failed to allocate a GPtrArray.");
1342 status
= BT_MESSAGE_ITERATOR_CLASS_INITIALIZE_METHOD_STATUS_MEMORY_ERROR
;
1346 muxer_msg_iter
->ended_muxer_upstream_msg_iters
=
1347 g_ptr_array_new_with_free_func(
1348 (GDestroyNotify
) destroy_muxer_upstream_msg_iter
);
1349 if (!muxer_msg_iter
->ended_muxer_upstream_msg_iters
) {
1350 BT_COMP_LOGE_APPEND_CAUSE(muxer_comp
->self_comp
, "Failed to allocate a GPtrArray.");
1351 status
= BT_MESSAGE_ITERATOR_CLASS_INITIALIZE_METHOD_STATUS_MEMORY_ERROR
;
1355 status
= muxer_msg_iter_init_upstream_iterators(muxer_comp
,
1356 muxer_msg_iter
, config
);
1358 BT_COMP_LOGE_APPEND_CAUSE(muxer_comp
->self_comp
,
1359 "Cannot initialize connected input ports for muxer component's message iterator: "
1360 "comp-addr=%p, muxer-comp-addr=%p, "
1361 "muxer-msg-iter-addr=%p, msg-iter-addr=%p, ret=%d",
1362 self_comp
, muxer_comp
, muxer_msg_iter
,
1363 self_msg_iter
, status
);
1367 bt_self_message_iterator_set_data(self_msg_iter
, muxer_msg_iter
);
1368 BT_COMP_LOGD("Initialized muxer component's message iterator: "
1369 "comp-addr=%p, muxer-comp-addr=%p, muxer-msg-iter-addr=%p, "
1371 self_comp
, muxer_comp
, muxer_msg_iter
, self_msg_iter
);
1375 destroy_muxer_msg_iter(muxer_msg_iter
);
1376 bt_self_message_iterator_set_data(self_msg_iter
, NULL
);
1379 muxer_comp
->initializing_muxer_msg_iter
= false;
1384 void muxer_msg_iter_finalize(bt_self_message_iterator
*self_msg_iter
)
1386 struct muxer_msg_iter
*muxer_msg_iter
=
1387 bt_self_message_iterator_get_data(self_msg_iter
);
1388 bt_self_component
*self_comp
= NULL
;
1389 struct muxer_comp
*muxer_comp
= NULL
;
1391 self_comp
= bt_self_message_iterator_borrow_component(
1393 BT_ASSERT(self_comp
);
1394 muxer_comp
= bt_self_component_get_data(self_comp
);
1395 BT_COMP_LOGD("Finalizing muxer component's message iterator: "
1396 "comp-addr=%p, muxer-comp-addr=%p, muxer-msg-iter-addr=%p, "
1398 self_comp
, muxer_comp
, muxer_msg_iter
, self_msg_iter
);
1400 if (muxer_msg_iter
) {
1401 destroy_muxer_msg_iter(muxer_msg_iter
);
1406 bt_message_iterator_class_next_method_status
muxer_msg_iter_next(
1407 bt_self_message_iterator
*self_msg_iter
,
1408 bt_message_array_const msgs
, uint64_t capacity
,
1411 bt_message_iterator_class_next_method_status status
;
1412 struct muxer_msg_iter
*muxer_msg_iter
=
1413 bt_self_message_iterator_get_data(self_msg_iter
);
1414 bt_self_component
*self_comp
= NULL
;
1415 struct muxer_comp
*muxer_comp
= NULL
;
1417 BT_ASSERT_DBG(muxer_msg_iter
);
1418 self_comp
= bt_self_message_iterator_borrow_component(
1420 BT_ASSERT_DBG(self_comp
);
1421 muxer_comp
= bt_self_component_get_data(self_comp
);
1422 BT_ASSERT_DBG(muxer_comp
);
1423 BT_COMP_LOGT("Muxer component's message iterator's \"next\" method called: "
1424 "comp-addr=%p, muxer-comp-addr=%p, muxer-msg-iter-addr=%p, "
1426 self_comp
, muxer_comp
, muxer_msg_iter
, self_msg_iter
);
1428 status
= muxer_msg_iter_do_next(muxer_comp
, muxer_msg_iter
,
1429 msgs
, capacity
, count
);
1431 BT_COMP_LOGE_APPEND_CAUSE(self_comp
,
1432 "Cannot get next message: "
1433 "comp-addr=%p, muxer-comp-addr=%p, muxer-msg-iter-addr=%p, "
1434 "msg-iter-addr=%p, status=%s",
1435 self_comp
, muxer_comp
, muxer_msg_iter
, self_msg_iter
,
1436 bt_common_func_status_string(status
));
1438 BT_COMP_LOGT("Returning from muxer component's message iterator's \"next\" method: "
1440 bt_common_func_status_string(status
));
1447 bt_component_class_port_connected_method_status
muxer_input_port_connected(
1448 bt_self_component_filter
*self_comp
,
1449 bt_self_component_port_input
*self_port
,
1450 const bt_port_output
*other_port
)
1452 bt_component_class_port_connected_method_status status
=
1453 BT_COMPONENT_CLASS_PORT_CONNECTED_METHOD_STATUS_OK
;
1454 bt_self_component_add_port_status add_port_status
;
1455 struct muxer_comp
*muxer_comp
= bt_self_component_get_data(
1456 bt_self_component_filter_as_self_component(self_comp
));
1458 add_port_status
= add_available_input_port(self_comp
);
1459 if (add_port_status
) {
1460 BT_COMP_LOGE_APPEND_CAUSE(muxer_comp
->self_comp
,
1461 "Cannot add one muxer component's input port: status=%s",
1462 bt_common_func_status_string(add_port_status
));
1464 if (add_port_status
==
1465 BT_SELF_COMPONENT_ADD_PORT_STATUS_MEMORY_ERROR
) {
1466 status
= BT_COMPONENT_CLASS_PORT_CONNECTED_METHOD_STATUS_MEMORY_ERROR
;
1468 status
= BT_COMPONENT_CLASS_PORT_CONNECTED_METHOD_STATUS_ERROR
;
1479 bt_message_iterator_class_can_seek_beginning_method_status
1480 muxer_upstream_msg_iters_can_all_seek_beginning(
1481 struct muxer_comp
*muxer_comp
,
1482 GPtrArray
*muxer_upstream_msg_iters
, bt_bool
*can_seek
)
1484 bt_message_iterator_class_can_seek_beginning_method_status status
=
1485 BT_MESSAGE_ITERATOR_CLASS_CAN_SEEK_BEGINNING_METHOD_STATUS_OK
;
1488 for (i
= 0; i
< muxer_upstream_msg_iters
->len
; i
++) {
1489 struct muxer_upstream_msg_iter
*upstream_msg_iter
=
1490 muxer_upstream_msg_iters
->pdata
[i
];
1491 status
= (int) bt_message_iterator_can_seek_beginning(
1492 upstream_msg_iter
->msg_iter
, can_seek
);
1493 if (status
!= BT_MESSAGE_ITERATOR_CLASS_CAN_SEEK_BEGINNING_METHOD_STATUS_OK
) {
1494 BT_COMP_LOGE_APPEND_CAUSE(muxer_comp
->self_comp
,
1495 "Failed to determine whether upstream message iterator can seek beginning: "
1496 "msg-iter-addr=%p", upstream_msg_iter
->msg_iter
);
1505 *can_seek
= BT_TRUE
;
1512 bt_message_iterator_class_can_seek_beginning_method_status
1513 muxer_msg_iter_can_seek_beginning(
1514 bt_self_message_iterator
*self_msg_iter
, bt_bool
*can_seek
)
1516 struct muxer_msg_iter
*muxer_msg_iter
=
1517 bt_self_message_iterator_get_data(self_msg_iter
);
1518 bt_message_iterator_class_can_seek_beginning_method_status status
;
1520 status
= muxer_upstream_msg_iters_can_all_seek_beginning(
1521 muxer_msg_iter
->muxer_comp
,
1522 muxer_msg_iter
->active_muxer_upstream_msg_iters
, can_seek
);
1523 if (status
!= BT_MESSAGE_ITERATOR_CLASS_CAN_SEEK_BEGINNING_METHOD_STATUS_OK
) {
1531 status
= muxer_upstream_msg_iters_can_all_seek_beginning(
1532 muxer_msg_iter
->muxer_comp
,
1533 muxer_msg_iter
->ended_muxer_upstream_msg_iters
, can_seek
);
1540 bt_message_iterator_class_seek_beginning_method_status
muxer_msg_iter_seek_beginning(
1541 bt_self_message_iterator
*self_msg_iter
)
1543 struct muxer_msg_iter
*muxer_msg_iter
=
1544 bt_self_message_iterator_get_data(self_msg_iter
);
1545 bt_message_iterator_class_seek_beginning_method_status status
=
1546 BT_MESSAGE_ITERATOR_CLASS_SEEK_BEGINNING_METHOD_STATUS_OK
;
1547 bt_message_iterator_seek_beginning_status seek_beg_status
;
1550 /* Seek all ended upstream iterators first */
1551 for (i
= 0; i
< muxer_msg_iter
->ended_muxer_upstream_msg_iters
->len
;
1553 struct muxer_upstream_msg_iter
*upstream_msg_iter
=
1554 muxer_msg_iter
->ended_muxer_upstream_msg_iters
->pdata
[i
];
1556 seek_beg_status
= bt_message_iterator_seek_beginning(
1557 upstream_msg_iter
->msg_iter
);
1558 if (seek_beg_status
!= BT_MESSAGE_ITERATOR_SEEK_BEGINNING_STATUS_OK
) {
1559 status
= (int) seek_beg_status
;
1563 empty_message_queue(upstream_msg_iter
);
1566 /* Seek all previously active upstream iterators */
1567 for (i
= 0; i
< muxer_msg_iter
->active_muxer_upstream_msg_iters
->len
;
1569 struct muxer_upstream_msg_iter
*upstream_msg_iter
=
1570 muxer_msg_iter
->active_muxer_upstream_msg_iters
->pdata
[i
];
1572 seek_beg_status
= bt_message_iterator_seek_beginning(
1573 upstream_msg_iter
->msg_iter
);
1574 if (seek_beg_status
!= BT_MESSAGE_ITERATOR_SEEK_BEGINNING_STATUS_OK
) {
1575 status
= (int) seek_beg_status
;
1579 empty_message_queue(upstream_msg_iter
);
1582 /* Make them all active */
1583 for (i
= 0; i
< muxer_msg_iter
->ended_muxer_upstream_msg_iters
->len
;
1585 struct muxer_upstream_msg_iter
*upstream_msg_iter
=
1586 muxer_msg_iter
->ended_muxer_upstream_msg_iters
->pdata
[i
];
1588 g_ptr_array_add(muxer_msg_iter
->active_muxer_upstream_msg_iters
,
1590 muxer_msg_iter
->ended_muxer_upstream_msg_iters
->pdata
[i
] = NULL
;
1594 * GLib < 2.48.0 asserts when g_ptr_array_remove_range() is
1595 * called on an empty array.
1597 if (muxer_msg_iter
->ended_muxer_upstream_msg_iters
->len
> 0) {
1598 g_ptr_array_remove_range(muxer_msg_iter
->ended_muxer_upstream_msg_iters
,
1599 0, muxer_msg_iter
->ended_muxer_upstream_msg_iters
->len
);
1601 muxer_msg_iter
->last_returned_ts_ns
= INT64_MIN
;
1602 muxer_msg_iter
->clock_class_expectation
=
1603 MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_ANY
;