Replace libuuid with internal implementation
[babeltrace.git] / src / plugins / utils / muxer / muxer.c
1 /*
2 * Copyright 2017 Philippe Proulx <pproulx@efficios.com>
3 *
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:
10 *
11 * The above copyright notice and this permission notice shall be included in
12 * all copies or substantial portions of the Software.
13 *
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
20 * SOFTWARE.
21 */
22
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 "plugins/comp-logging.h"
27
28 #include "common/macros.h"
29 #include "common/uuid.h"
30 #include <babeltrace2/babeltrace.h>
31 #include <glib.h>
32 #include <stdbool.h>
33 #include <inttypes.h>
34 #include "common/assert.h"
35 #include "common/common.h"
36 #include <stdlib.h>
37 #include <string.h>
38
39 #include "muxer.h"
40
41 #define ASSUME_ABSOLUTE_CLOCK_CLASSES_PARAM_NAME "assume-absolute-clock-classes"
42
43 struct muxer_comp {
44 /* Weak refs */
45 bt_self_component_filter *self_comp_flt;
46 bt_self_component *self_comp;
47
48 unsigned int next_port_num;
49 size_t available_input_ports;
50 bool initializing_muxer_msg_iter;
51 bool assume_absolute_clock_classes;
52 bt_logging_level log_level;
53 };
54
55 struct muxer_upstream_msg_iter {
56 struct muxer_comp *muxer_comp;
57
58 /* Owned by this, NULL if ended */
59 bt_self_component_port_input_message_iterator *msg_iter;
60
61 /* Contains `const bt_message *`, owned by this */
62 GQueue *msgs;
63 };
64
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,
71 };
72
73 struct muxer_msg_iter {
74 struct muxer_comp *muxer_comp;
75
76 /*
77 * Array of struct muxer_upstream_msg_iter * (owned by this).
78 *
79 * NOTE: This array is searched in linearly to find the youngest
80 * current message. Keep this until benchmarks confirm that
81 * another data structure is faster than this for our typical
82 * use cases.
83 */
84 GPtrArray *active_muxer_upstream_msg_iters;
85
86 /*
87 * Array of struct muxer_upstream_msg_iter * (owned by this).
88 *
89 * We move ended message iterators from
90 * `active_muxer_upstream_msg_iters` to this array so as to be
91 * able to restore them when seeking.
92 */
93 GPtrArray *ended_muxer_upstream_msg_iters;
94
95 /* Last time returned in a message */
96 int64_t last_returned_ts_ns;
97
98 /* Clock class expectation state */
99 enum muxer_msg_iter_clock_class_expectation clock_class_expectation;
100
101 /*
102 * Expected clock class UUID, only valid when
103 * clock_class_expectation is
104 * MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_NOT_ABS_SPEC_UUID.
105 */
106 bt_uuid_t expected_clock_class_uuid;
107 };
108
109 static
110 void empty_message_queue(struct muxer_upstream_msg_iter *upstream_msg_iter)
111 {
112 const bt_message *msg;
113
114 while ((msg = g_queue_pop_head(upstream_msg_iter->msgs))) {
115 bt_message_put_ref(msg);
116 }
117 }
118
119 static
120 void destroy_muxer_upstream_msg_iter(
121 struct muxer_upstream_msg_iter *muxer_upstream_msg_iter)
122 {
123 struct muxer_comp *muxer_comp;
124
125 if (!muxer_upstream_msg_iter) {
126 return;
127 }
128
129 muxer_comp = muxer_upstream_msg_iter->muxer_comp;
130 BT_COMP_LOGD("Destroying muxer's upstream message iterator wrapper: "
131 "addr=%p, msg-iter-addr=%p, queue-len=%u",
132 muxer_upstream_msg_iter,
133 muxer_upstream_msg_iter->msg_iter,
134 muxer_upstream_msg_iter->msgs->length);
135 bt_self_component_port_input_message_iterator_put_ref(
136 muxer_upstream_msg_iter->msg_iter);
137
138 if (muxer_upstream_msg_iter->msgs) {
139 empty_message_queue(muxer_upstream_msg_iter);
140 g_queue_free(muxer_upstream_msg_iter->msgs);
141 }
142
143 g_free(muxer_upstream_msg_iter);
144 }
145
146 static
147 int muxer_msg_iter_add_upstream_msg_iter(struct muxer_msg_iter *muxer_msg_iter,
148 bt_self_component_port_input_message_iterator *self_msg_iter)
149 {
150 int ret = 0;
151 struct muxer_upstream_msg_iter *muxer_upstream_msg_iter =
152 g_new0(struct muxer_upstream_msg_iter, 1);
153 struct muxer_comp *muxer_comp = muxer_msg_iter->muxer_comp;
154
155 if (!muxer_upstream_msg_iter) {
156 BT_COMP_LOGE_STR("Failed to allocate one muxer's upstream message iterator wrapper.");
157 goto error;
158 }
159
160 muxer_upstream_msg_iter->muxer_comp = muxer_comp;
161 muxer_upstream_msg_iter->msg_iter = self_msg_iter;
162 bt_self_component_port_input_message_iterator_get_ref(muxer_upstream_msg_iter->msg_iter);
163 muxer_upstream_msg_iter->msgs = g_queue_new();
164 if (!muxer_upstream_msg_iter->msgs) {
165 BT_COMP_LOGE_STR("Failed to allocate a GQueue.");
166 goto error;
167 }
168
169 g_ptr_array_add(muxer_msg_iter->active_muxer_upstream_msg_iters,
170 muxer_upstream_msg_iter);
171 BT_COMP_LOGD("Added muxer's upstream message iterator wrapper: "
172 "addr=%p, muxer-msg-iter-addr=%p, msg-iter-addr=%p",
173 muxer_upstream_msg_iter, muxer_msg_iter,
174 self_msg_iter);
175
176 goto end;
177
178 error:
179 g_free(muxer_upstream_msg_iter);
180 ret = -1;
181
182 end:
183 return ret;
184 }
185
186 static
187 bt_self_component_add_port_status add_available_input_port(
188 bt_self_component_filter *self_comp)
189 {
190 struct muxer_comp *muxer_comp = bt_self_component_get_data(
191 bt_self_component_filter_as_self_component(self_comp));
192 bt_self_component_add_port_status status =
193 BT_SELF_COMPONENT_ADD_PORT_STATUS_OK;
194 GString *port_name = NULL;
195
196 BT_ASSERT(muxer_comp);
197 port_name = g_string_new("in");
198 if (!port_name) {
199 BT_COMP_LOGE_STR("Failed to allocate a GString.");
200 status = BT_SELF_COMPONENT_ADD_PORT_STATUS_MEMORY_ERROR;
201 goto end;
202 }
203
204 g_string_append_printf(port_name, "%u", muxer_comp->next_port_num);
205 status = bt_self_component_filter_add_input_port(
206 self_comp, port_name->str, NULL, NULL);
207 if (status != BT_SELF_COMPONENT_ADD_PORT_STATUS_OK) {
208 BT_COMP_LOGE("Cannot add input port to muxer component: "
209 "port-name=\"%s\", comp-addr=%p, status=%s",
210 port_name->str, self_comp,
211 bt_common_func_status_string(status));
212 goto end;
213 }
214
215 muxer_comp->available_input_ports++;
216 muxer_comp->next_port_num++;
217 BT_COMP_LOGI("Added one input port to muxer component: "
218 "port-name=\"%s\", comp-addr=%p",
219 port_name->str, self_comp);
220
221 end:
222 if (port_name) {
223 g_string_free(port_name, TRUE);
224 }
225
226 return status;
227 }
228
229 static
230 bt_self_component_add_port_status create_output_port(
231 bt_self_component_filter *self_comp)
232 {
233 return bt_self_component_filter_add_output_port(
234 self_comp, "out", NULL, NULL);
235 }
236
237 static
238 void destroy_muxer_comp(struct muxer_comp *muxer_comp)
239 {
240 if (!muxer_comp) {
241 return;
242 }
243
244 g_free(muxer_comp);
245 }
246
247 static
248 bt_value *get_default_params(struct muxer_comp *muxer_comp)
249 {
250 bt_value *params;
251 int ret;
252
253 params = bt_value_map_create();
254 if (!params) {
255 BT_COMP_LOGE_STR("Cannot create a map value object.");
256 goto error;
257 }
258
259 ret = bt_value_map_insert_bool_entry(params,
260 ASSUME_ABSOLUTE_CLOCK_CLASSES_PARAM_NAME, false);
261 if (ret) {
262 BT_COMP_LOGE_STR("Cannot add boolean value to map value object.");
263 goto error;
264 }
265
266 goto end;
267
268 error:
269 BT_VALUE_PUT_REF_AND_RESET(params);
270
271 end:
272 return params;
273 }
274
275 static
276 int configure_muxer_comp(struct muxer_comp *muxer_comp,
277 const bt_value *params)
278 {
279 bt_value *default_params = NULL;
280 bt_value *real_params = NULL;
281 const bt_value *assume_absolute_clock_classes = NULL;
282 int ret = 0;
283 bt_bool bool_val;
284
285 default_params = get_default_params(muxer_comp);
286 if (!default_params) {
287 BT_COMP_LOGE("Cannot get default parameters: "
288 "muxer-comp-addr=%p", muxer_comp);
289 goto error;
290 }
291
292 ret = bt_value_map_extend(default_params, params, &real_params);
293 if (ret) {
294 BT_COMP_LOGE("Cannot extend default parameters map value: "
295 "muxer-comp-addr=%p, def-params-addr=%p, "
296 "params-addr=%p", muxer_comp, default_params,
297 params);
298 goto error;
299 }
300
301 assume_absolute_clock_classes = bt_value_map_borrow_entry_value(real_params,
302 ASSUME_ABSOLUTE_CLOCK_CLASSES_PARAM_NAME);
303 if (assume_absolute_clock_classes &&
304 !bt_value_is_bool(assume_absolute_clock_classes)) {
305 BT_COMP_LOGE("Expecting a boolean value for the `%s` parameter: "
306 "muxer-comp-addr=%p, value-type=%s",
307 ASSUME_ABSOLUTE_CLOCK_CLASSES_PARAM_NAME, muxer_comp,
308 bt_common_value_type_string(
309 bt_value_get_type(assume_absolute_clock_classes)));
310 goto error;
311 }
312
313 bool_val = bt_value_bool_get(assume_absolute_clock_classes);
314 muxer_comp->assume_absolute_clock_classes = (bool) bool_val;
315 BT_COMP_LOGI("Configured muxer component: muxer-comp-addr=%p, "
316 "assume-absolute-clock-classes=%d",
317 muxer_comp, muxer_comp->assume_absolute_clock_classes);
318 goto end;
319
320 error:
321 ret = -1;
322
323 end:
324 bt_value_put_ref(default_params);
325 bt_value_put_ref(real_params);
326 return ret;
327 }
328
329 BT_HIDDEN
330 bt_component_class_init_method_status muxer_init(
331 bt_self_component_filter *self_comp_flt,
332 const bt_value *params, void *init_data)
333 {
334 int ret;
335 bt_component_class_init_method_status status =
336 BT_COMPONENT_CLASS_INIT_METHOD_STATUS_OK;
337 bt_self_component_add_port_status add_port_status;
338 bt_self_component *self_comp =
339 bt_self_component_filter_as_self_component(self_comp_flt);
340 struct muxer_comp *muxer_comp = g_new0(struct muxer_comp, 1);
341 bt_logging_level log_level = bt_component_get_logging_level(
342 bt_self_component_as_component(self_comp));
343
344 BT_COMP_LOG_CUR_LVL(BT_LOG_INFO, log_level, self_comp,
345 "Initializing muxer component: "
346 "comp-addr=%p, params-addr=%p", self_comp, params);
347
348 if (!muxer_comp) {
349 BT_COMP_LOG_CUR_LVL(BT_LOG_ERROR, log_level, self_comp,
350 "Failed to allocate one muxer component.");
351 goto error;
352 }
353
354 muxer_comp->log_level = log_level;
355 muxer_comp->self_comp = self_comp;
356 muxer_comp->self_comp_flt = self_comp_flt;
357 ret = configure_muxer_comp(muxer_comp, params);
358 if (ret) {
359 BT_COMP_LOGE("Cannot configure muxer component: "
360 "muxer-comp-addr=%p, params-addr=%p",
361 muxer_comp, params);
362 goto error;
363 }
364
365 bt_self_component_set_data(self_comp, muxer_comp);
366 add_port_status = add_available_input_port(self_comp_flt);
367 if (add_port_status != BT_SELF_COMPONENT_ADD_PORT_STATUS_OK) {
368 BT_COMP_LOGE("Cannot ensure that at least one muxer component's input port is available: "
369 "muxer-comp-addr=%p, status=%s",
370 muxer_comp,
371 bt_common_func_status_string(add_port_status));
372 if (add_port_status ==
373 BT_SELF_COMPONENT_ADD_PORT_STATUS_MEMORY_ERROR) {
374 status = BT_COMPONENT_CLASS_INIT_METHOD_STATUS_MEMORY_ERROR;
375 } else {
376 status = BT_COMPONENT_CLASS_INIT_METHOD_STATUS_ERROR;
377 }
378
379 goto error;
380 }
381
382 add_port_status = create_output_port(self_comp_flt);
383 if (add_port_status != BT_SELF_COMPONENT_ADD_PORT_STATUS_OK) {
384 BT_COMP_LOGE("Cannot create muxer component's output port: "
385 "muxer-comp-addr=%p, status=%s",
386 muxer_comp,
387 bt_common_func_status_string(add_port_status));
388 if (add_port_status ==
389 BT_SELF_COMPONENT_ADD_PORT_STATUS_MEMORY_ERROR) {
390 status = BT_COMPONENT_CLASS_INIT_METHOD_STATUS_MEMORY_ERROR;
391 } else {
392 status = BT_COMPONENT_CLASS_INIT_METHOD_STATUS_ERROR;
393 }
394
395 goto error;
396 }
397
398 BT_COMP_LOGI("Initialized muxer component: "
399 "comp-addr=%p, params-addr=%p, muxer-comp-addr=%p",
400 self_comp, params, muxer_comp);
401
402 goto end;
403
404 error:
405 destroy_muxer_comp(muxer_comp);
406 bt_self_component_set_data(self_comp, NULL);
407
408 if (status == BT_COMPONENT_CLASS_INIT_METHOD_STATUS_OK) {
409 status = BT_COMPONENT_CLASS_INIT_METHOD_STATUS_ERROR;
410 }
411
412 end:
413 return status;
414 }
415
416 BT_HIDDEN
417 void muxer_finalize(bt_self_component_filter *self_comp)
418 {
419 struct muxer_comp *muxer_comp = bt_self_component_get_data(
420 bt_self_component_filter_as_self_component(self_comp));
421
422 BT_COMP_LOGI("Finalizing muxer component: comp-addr=%p",
423 self_comp);
424 destroy_muxer_comp(muxer_comp);
425 }
426
427 static
428 bt_self_component_port_input_message_iterator *
429 create_msg_iter_on_input_port(struct muxer_comp *muxer_comp,
430 bt_self_component_port_input *self_port)
431 {
432 const bt_port *port = bt_self_component_port_as_port(
433 bt_self_component_port_input_as_self_component_port(
434 self_port));
435 bt_self_component_port_input_message_iterator *msg_iter =
436 NULL;
437
438 BT_ASSERT(port);
439 BT_ASSERT(bt_port_is_connected(port));
440
441 // TODO: Advance the iterator to >= the time of the latest
442 // returned message by the muxer message
443 // iterator which creates it.
444 msg_iter = bt_self_component_port_input_message_iterator_create(
445 self_port);
446 if (!msg_iter) {
447 BT_COMP_LOGE("Cannot create upstream message iterator on input port: "
448 "port-addr=%p, port-name=\"%s\"",
449 port, bt_port_get_name(port));
450 goto end;
451 }
452
453 BT_COMP_LOGI("Created upstream message iterator on input port: "
454 "port-addr=%p, port-name=\"%s\", msg-iter-addr=%p",
455 port, bt_port_get_name(port), msg_iter);
456
457 end:
458 return msg_iter;
459 }
460
461 static
462 bt_component_class_message_iterator_next_method_status muxer_upstream_msg_iter_next(
463 struct muxer_upstream_msg_iter *muxer_upstream_msg_iter,
464 bool *is_ended)
465 {
466 struct muxer_comp *muxer_comp =
467 muxer_upstream_msg_iter->muxer_comp;
468 bt_component_class_message_iterator_next_method_status status;
469 bt_message_iterator_next_status input_port_iter_status;
470 bt_message_array_const msgs;
471 uint64_t i;
472 uint64_t count;
473
474 BT_COMP_LOGD("Calling upstream message iterator's \"next\" method: "
475 "muxer-upstream-msg-iter-wrap-addr=%p, msg-iter-addr=%p",
476 muxer_upstream_msg_iter,
477 muxer_upstream_msg_iter->msg_iter);
478 input_port_iter_status = bt_self_component_port_input_message_iterator_next(
479 muxer_upstream_msg_iter->msg_iter, &msgs, &count);
480 BT_COMP_LOGD("Upstream message iterator's \"next\" method returned: "
481 "status=%s",
482 bt_common_func_status_string(input_port_iter_status));
483
484 switch (input_port_iter_status) {
485 case BT_MESSAGE_ITERATOR_NEXT_STATUS_OK:
486 /*
487 * Message iterator's current message is
488 * valid: it must be considered for muxing operations.
489 */
490 BT_COMP_LOGD_STR("Validated upstream message iterator wrapper.");
491 BT_ASSERT(count > 0);
492
493 /* Move messages to our queue */
494 for (i = 0; i < count; i++) {
495 /*
496 * Push to tail in order; other side
497 * (muxer_msg_iter_do_next_one()) consumes
498 * from the head first.
499 */
500 g_queue_push_tail(muxer_upstream_msg_iter->msgs,
501 (void *) msgs[i]);
502 }
503 status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK;
504 break;
505 case BT_MESSAGE_ITERATOR_NEXT_STATUS_AGAIN:
506 /*
507 * Message iterator's current message is not
508 * valid anymore. Return
509 * BT_MESSAGE_ITERATOR_NEXT_STATUS_AGAIN immediately.
510 */
511 status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_AGAIN;
512 break;
513 case BT_MESSAGE_ITERATOR_NEXT_STATUS_END: /* Fall-through. */
514 /*
515 * Message iterator reached the end: release it. It
516 * won't be considered again to find the youngest
517 * message.
518 */
519 *is_ended = true;
520 status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK;
521 break;
522 default:
523 /* Error or unsupported status code */
524 BT_COMP_LOGE("Error or unsupported status code: "
525 "status-code=%d", input_port_iter_status);
526 status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_ERROR;
527 break;
528 }
529
530 return status;
531 }
532
533 static
534 int get_msg_ts_ns(struct muxer_comp *muxer_comp,
535 struct muxer_msg_iter *muxer_msg_iter,
536 const bt_message *msg, int64_t last_returned_ts_ns,
537 int64_t *ts_ns)
538 {
539 const bt_clock_snapshot *clock_snapshot = NULL;
540 int ret = 0;
541 const bt_stream_class *stream_class = NULL;
542 bt_message_type msg_type;
543
544 BT_ASSERT(msg);
545 BT_ASSERT(ts_ns);
546 BT_COMP_LOGD("Getting message's timestamp: "
547 "muxer-msg-iter-addr=%p, msg-addr=%p, "
548 "last-returned-ts=%" PRId64,
549 muxer_msg_iter, msg, last_returned_ts_ns);
550
551 if (G_UNLIKELY(muxer_msg_iter->clock_class_expectation ==
552 MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_NONE)) {
553 *ts_ns = last_returned_ts_ns;
554 goto end;
555 }
556
557 msg_type = bt_message_get_type(msg);
558
559 if (G_UNLIKELY(msg_type == BT_MESSAGE_TYPE_PACKET_BEGINNING)) {
560 stream_class = bt_stream_borrow_class_const(
561 bt_packet_borrow_stream_const(
562 bt_message_packet_beginning_borrow_packet_const(
563 msg)));
564 } else if (G_UNLIKELY(msg_type == BT_MESSAGE_TYPE_PACKET_END)) {
565 stream_class = bt_stream_borrow_class_const(
566 bt_packet_borrow_stream_const(
567 bt_message_packet_end_borrow_packet_const(
568 msg)));
569 } else if (G_UNLIKELY(msg_type == BT_MESSAGE_TYPE_DISCARDED_EVENTS)) {
570 stream_class = bt_stream_borrow_class_const(
571 bt_message_discarded_events_borrow_stream_const(msg));
572 } else if (G_UNLIKELY(msg_type == BT_MESSAGE_TYPE_DISCARDED_PACKETS)) {
573 stream_class = bt_stream_borrow_class_const(
574 bt_message_discarded_packets_borrow_stream_const(msg));
575 }
576
577 switch (msg_type) {
578 case BT_MESSAGE_TYPE_EVENT:
579 BT_ASSERT(bt_message_event_borrow_stream_class_default_clock_class_const(
580 msg));
581 clock_snapshot = bt_message_event_borrow_default_clock_snapshot_const(
582 msg);
583 break;
584 case BT_MESSAGE_TYPE_PACKET_BEGINNING:
585 if (bt_stream_class_packets_have_beginning_default_clock_snapshot(
586 stream_class)) {
587 clock_snapshot = bt_message_packet_beginning_borrow_default_clock_snapshot_const(
588 msg);
589 } else {
590 goto no_clock_snapshot;
591 }
592
593 break;
594 case BT_MESSAGE_TYPE_PACKET_END:
595 if (bt_stream_class_packets_have_end_default_clock_snapshot(
596 stream_class)) {
597 clock_snapshot = bt_message_packet_end_borrow_default_clock_snapshot_const(
598 msg);
599 } else {
600 goto no_clock_snapshot;
601 }
602
603 break;
604 case BT_MESSAGE_TYPE_DISCARDED_EVENTS:
605 if (bt_stream_class_discarded_events_have_default_clock_snapshots(
606 stream_class)) {
607 clock_snapshot = bt_message_discarded_events_borrow_beginning_default_clock_snapshot_const(
608 msg);
609 } else {
610 goto no_clock_snapshot;
611 }
612
613 break;
614 case BT_MESSAGE_TYPE_DISCARDED_PACKETS:
615 if (bt_stream_class_discarded_packets_have_default_clock_snapshots(
616 stream_class)) {
617 clock_snapshot = bt_message_discarded_packets_borrow_beginning_default_clock_snapshot_const(
618 msg);
619 } else {
620 goto no_clock_snapshot;
621 }
622
623 break;
624 case BT_MESSAGE_TYPE_MESSAGE_ITERATOR_INACTIVITY:
625 clock_snapshot = bt_message_message_iterator_inactivity_borrow_default_clock_snapshot_const(
626 msg);
627 break;
628 default:
629 /* All the other messages have a higher priority */
630 BT_COMP_LOGD_STR("Message has no timestamp: using the last returned timestamp.");
631 *ts_ns = last_returned_ts_ns;
632 goto end;
633 }
634
635 ret = bt_clock_snapshot_get_ns_from_origin(clock_snapshot, ts_ns);
636 if (ret) {
637 BT_COMP_LOGE("Cannot get nanoseconds from Epoch of clock snapshot: "
638 "clock-snapshot-addr=%p", clock_snapshot);
639 goto error;
640 }
641
642 goto end;
643
644 no_clock_snapshot:
645 BT_COMP_LOGD_STR("Message's default clock snapshot is missing: "
646 "using the last returned timestamp.");
647 *ts_ns = last_returned_ts_ns;
648 goto end;
649
650 error:
651 ret = -1;
652
653 end:
654 if (ret == 0) {
655 BT_COMP_LOGD("Found message's timestamp: "
656 "muxer-msg-iter-addr=%p, msg-addr=%p, "
657 "last-returned-ts=%" PRId64 ", ts=%" PRId64,
658 muxer_msg_iter, msg, last_returned_ts_ns,
659 *ts_ns);
660 }
661
662 return ret;
663 }
664
665 static inline
666 int validate_clock_class(struct muxer_msg_iter *muxer_msg_iter,
667 struct muxer_comp *muxer_comp,
668 const bt_clock_class *clock_class)
669 {
670 int ret = 0;
671 const uint8_t *cc_uuid;
672 const char *cc_name;
673
674 BT_ASSERT(clock_class);
675 cc_uuid = bt_clock_class_get_uuid(clock_class);
676 cc_name = bt_clock_class_get_name(clock_class);
677
678 if (muxer_msg_iter->clock_class_expectation ==
679 MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_ANY) {
680 /*
681 * This is the first clock class that this muxer
682 * message iterator encounters. Its properties
683 * determine what to expect for the whole lifetime of
684 * the iterator without a true
685 * `assume-absolute-clock-classes` parameter.
686 */
687 if (bt_clock_class_origin_is_unix_epoch(clock_class)) {
688 /* Expect absolute clock classes */
689 muxer_msg_iter->clock_class_expectation =
690 MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_ABSOLUTE;
691 } else {
692 if (cc_uuid) {
693 /*
694 * Expect non-absolute clock classes
695 * with a specific UUID.
696 */
697 muxer_msg_iter->clock_class_expectation =
698 MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_NOT_ABS_SPEC_UUID;
699 bt_uuid_copy(muxer_msg_iter->expected_clock_class_uuid, cc_uuid);
700 } else {
701 /*
702 * Expect non-absolute clock classes
703 * with no UUID.
704 */
705 muxer_msg_iter->clock_class_expectation =
706 MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_NOT_ABS_NO_UUID;
707 }
708 }
709 }
710
711 if (!muxer_comp->assume_absolute_clock_classes) {
712 switch (muxer_msg_iter->clock_class_expectation) {
713 case MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_ABSOLUTE:
714 if (!bt_clock_class_origin_is_unix_epoch(clock_class)) {
715 BT_COMP_LOGE("Expecting an absolute clock class, "
716 "but got a non-absolute one: "
717 "clock-class-addr=%p, clock-class-name=\"%s\"",
718 clock_class, cc_name);
719 goto error;
720 }
721 break;
722 case MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_NOT_ABS_NO_UUID:
723 if (bt_clock_class_origin_is_unix_epoch(clock_class)) {
724 BT_COMP_LOGE("Expecting a non-absolute clock class with no UUID, "
725 "but got an absolute one: "
726 "clock-class-addr=%p, clock-class-name=\"%s\"",
727 clock_class, cc_name);
728 goto error;
729 }
730
731 if (cc_uuid) {
732 BT_COMP_LOGE("Expecting a non-absolute clock class with no UUID, "
733 "but got one with a UUID: "
734 "clock-class-addr=%p, clock-class-name=\"%s\", "
735 "uuid=\"" BT_UUID_FMT "\"",
736 clock_class, cc_name, BT_UUID_FMT_VALUES(cc_uuid));
737 goto error;
738 }
739 break;
740 case MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_NOT_ABS_SPEC_UUID:
741 if (bt_clock_class_origin_is_unix_epoch(clock_class)) {
742 BT_COMP_LOGE("Expecting a non-absolute clock class with a specific UUID, "
743 "but got an absolute one: "
744 "clock-class-addr=%p, clock-class-name=\"%s\"",
745 clock_class, cc_name);
746 goto error;
747 }
748
749 if (!cc_uuid) {
750 BT_COMP_LOGE("Expecting a non-absolute clock class with a specific UUID, "
751 "but got one with no UUID: "
752 "clock-class-addr=%p, clock-class-name=\"%s\"",
753 clock_class, cc_name);
754 goto error;
755 }
756
757 if (bt_uuid_compare(muxer_msg_iter->expected_clock_class_uuid, cc_uuid) != 0) {
758 BT_COMP_LOGE("Expecting a non-absolute clock class with a specific UUID, "
759 "but got one with different UUID: "
760 "clock-class-addr=%p, clock-class-name=\"%s\", "
761 "expected-uuid=\"" BT_UUID_FMT "\", "
762 "uuid=\"" BT_UUID_FMT "\"",
763 clock_class, cc_name,
764 BT_UUID_FMT_VALUES(muxer_msg_iter->expected_clock_class_uuid),
765 BT_UUID_FMT_VALUES(cc_uuid));
766 goto error;
767 }
768 break;
769 case MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_NONE:
770 BT_COMP_LOGE("Expecting no clock class, but got one: "
771 "clock-class-addr=%p, clock-class-name=\"%s\"",
772 clock_class, cc_name);
773 goto error;
774 default:
775 /* Unexpected */
776 BT_COMP_LOGF("Unexpected clock class expectation: "
777 "expectation-code=%d",
778 muxer_msg_iter->clock_class_expectation);
779 abort();
780 }
781 }
782
783 goto end;
784
785 error:
786 ret = -1;
787
788 end:
789 return ret;
790 }
791
792 static inline
793 int validate_new_stream_clock_class(struct muxer_msg_iter *muxer_msg_iter,
794 struct muxer_comp *muxer_comp, const bt_stream *stream)
795 {
796 int ret = 0;
797 const bt_stream_class *stream_class =
798 bt_stream_borrow_class_const(stream);
799 const bt_clock_class *clock_class =
800 bt_stream_class_borrow_default_clock_class_const(stream_class);
801
802 if (!clock_class) {
803 if (muxer_msg_iter->clock_class_expectation ==
804 MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_ANY) {
805 /* Expect no clock class */
806 muxer_msg_iter->clock_class_expectation =
807 MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_NONE;
808 } else {
809 BT_COMP_LOGE("Expecting stream class with a default clock class: "
810 "stream-class-addr=%p, stream-class-name=\"%s\", "
811 "stream-class-id=%" PRIu64,
812 stream_class, bt_stream_class_get_name(stream_class),
813 bt_stream_class_get_id(stream_class));
814 ret = -1;
815 }
816
817 goto end;
818 }
819
820 ret = validate_clock_class(muxer_msg_iter, muxer_comp, clock_class);
821
822 end:
823 return ret;
824 }
825
826 /*
827 * This function finds the youngest available message amongst the
828 * non-ended upstream message iterators and returns the upstream
829 * message iterator which has it, or
830 * BT_MESSAGE_ITERATOR_STATUS_END if there's no available
831 * message.
832 *
833 * This function does NOT:
834 *
835 * * Update any upstream message iterator.
836 * * Check the upstream message iterators to retry.
837 *
838 * On sucess, this function sets *muxer_upstream_msg_iter to the
839 * upstream message iterator of which the current message is
840 * the youngest, and sets *ts_ns to its time.
841 */
842 static
843 bt_component_class_message_iterator_next_method_status
844 muxer_msg_iter_youngest_upstream_msg_iter(
845 struct muxer_comp *muxer_comp,
846 struct muxer_msg_iter *muxer_msg_iter,
847 struct muxer_upstream_msg_iter **muxer_upstream_msg_iter,
848 int64_t *ts_ns)
849 {
850 size_t i;
851 int ret;
852 int64_t youngest_ts_ns = INT64_MAX;
853 bt_component_class_message_iterator_next_method_status status =
854 BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK;
855
856 BT_ASSERT(muxer_comp);
857 BT_ASSERT(muxer_msg_iter);
858 BT_ASSERT(muxer_upstream_msg_iter);
859 *muxer_upstream_msg_iter = NULL;
860
861 for (i = 0; i < muxer_msg_iter->active_muxer_upstream_msg_iters->len;
862 i++) {
863 const bt_message *msg;
864 struct muxer_upstream_msg_iter *cur_muxer_upstream_msg_iter =
865 g_ptr_array_index(
866 muxer_msg_iter->active_muxer_upstream_msg_iters,
867 i);
868 int64_t msg_ts_ns;
869
870 if (!cur_muxer_upstream_msg_iter->msg_iter) {
871 /* This upstream message iterator is ended */
872 BT_COMP_LOGT("Skipping ended upstream message iterator: "
873 "muxer-upstream-msg-iter-wrap-addr=%p",
874 cur_muxer_upstream_msg_iter);
875 continue;
876 }
877
878 BT_ASSERT(cur_muxer_upstream_msg_iter->msgs->length > 0);
879 msg = g_queue_peek_head(cur_muxer_upstream_msg_iter->msgs);
880 BT_ASSERT(msg);
881
882 if (G_UNLIKELY(bt_message_get_type(msg) ==
883 BT_MESSAGE_TYPE_STREAM_BEGINNING)) {
884 ret = validate_new_stream_clock_class(
885 muxer_msg_iter, muxer_comp,
886 bt_message_stream_beginning_borrow_stream_const(
887 msg));
888 if (ret) {
889 /*
890 * validate_new_stream_clock_class() logs
891 * errors.
892 */
893 status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_ERROR;
894 goto end;
895 }
896 } else if (G_UNLIKELY(bt_message_get_type(msg) ==
897 BT_MESSAGE_TYPE_MESSAGE_ITERATOR_INACTIVITY)) {
898 const bt_clock_snapshot *cs;
899
900 cs = bt_message_message_iterator_inactivity_borrow_default_clock_snapshot_const(
901 msg);
902 ret = validate_clock_class(muxer_msg_iter, muxer_comp,
903 bt_clock_snapshot_borrow_clock_class_const(cs));
904 if (ret) {
905 /* validate_clock_class() logs errors */
906 status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_ERROR;
907 goto end;
908 }
909 }
910
911 ret = get_msg_ts_ns(muxer_comp, muxer_msg_iter, msg,
912 muxer_msg_iter->last_returned_ts_ns, &msg_ts_ns);
913 if (ret) {
914 /* get_msg_ts_ns() logs errors */
915 *muxer_upstream_msg_iter = NULL;
916 status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_ERROR;
917 goto end;
918 }
919
920 if (msg_ts_ns <= youngest_ts_ns) {
921 *muxer_upstream_msg_iter =
922 cur_muxer_upstream_msg_iter;
923 youngest_ts_ns = msg_ts_ns;
924 *ts_ns = youngest_ts_ns;
925 }
926 }
927
928 if (!*muxer_upstream_msg_iter) {
929 status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_END;
930 *ts_ns = INT64_MIN;
931 }
932
933 end:
934 return status;
935 }
936
937 static
938 bt_component_class_message_iterator_next_method_status
939 validate_muxer_upstream_msg_iter(
940 struct muxer_upstream_msg_iter *muxer_upstream_msg_iter,
941 bool *is_ended)
942 {
943 struct muxer_comp *muxer_comp =
944 muxer_upstream_msg_iter->muxer_comp;
945 bt_component_class_message_iterator_next_method_status status =
946 BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK;
947
948 BT_COMP_LOGD("Validating muxer's upstream message iterator wrapper: "
949 "muxer-upstream-msg-iter-wrap-addr=%p",
950 muxer_upstream_msg_iter);
951
952 if (muxer_upstream_msg_iter->msgs->length > 0 ||
953 !muxer_upstream_msg_iter->msg_iter) {
954 BT_COMP_LOGD("Already valid or not considered: "
955 "queue-len=%u, upstream-msg-iter-addr=%p",
956 muxer_upstream_msg_iter->msgs->length,
957 muxer_upstream_msg_iter->msg_iter);
958 goto end;
959 }
960
961 /* muxer_upstream_msg_iter_next() logs details/errors */
962 status = muxer_upstream_msg_iter_next(muxer_upstream_msg_iter,
963 is_ended);
964
965 end:
966 return status;
967 }
968
969 static
970 bt_component_class_message_iterator_next_method_status
971 validate_muxer_upstream_msg_iters(
972 struct muxer_msg_iter *muxer_msg_iter)
973 {
974 struct muxer_comp *muxer_comp = muxer_msg_iter->muxer_comp;
975 bt_component_class_message_iterator_next_method_status status =
976 BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK;
977 size_t i;
978
979 BT_COMP_LOGD("Validating muxer's upstream message iterator wrappers: "
980 "muxer-msg-iter-addr=%p", muxer_msg_iter);
981
982 for (i = 0; i < muxer_msg_iter->active_muxer_upstream_msg_iters->len;
983 i++) {
984 bool is_ended = false;
985 struct muxer_upstream_msg_iter *muxer_upstream_msg_iter =
986 g_ptr_array_index(
987 muxer_msg_iter->active_muxer_upstream_msg_iters,
988 i);
989
990 status = validate_muxer_upstream_msg_iter(
991 muxer_upstream_msg_iter, &is_ended);
992 if (status != BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK) {
993 if (status < 0) {
994 BT_COMP_LOGE("Cannot validate muxer's upstream message iterator wrapper: "
995 "muxer-msg-iter-addr=%p, "
996 "muxer-upstream-msg-iter-wrap-addr=%p",
997 muxer_msg_iter,
998 muxer_upstream_msg_iter);
999 } else {
1000 BT_COMP_LOGD("Cannot validate muxer's upstream message iterator wrapper: "
1001 "muxer-msg-iter-addr=%p, "
1002 "muxer-upstream-msg-iter-wrap-addr=%p",
1003 muxer_msg_iter,
1004 muxer_upstream_msg_iter);
1005 }
1006
1007 goto end;
1008 }
1009
1010 /*
1011 * Move this muxer upstream message iterator to the
1012 * array of ended iterators if it's ended.
1013 */
1014 if (G_UNLIKELY(is_ended)) {
1015 BT_COMP_LOGD("Muxer's upstream message iterator wrapper: ended or canceled: "
1016 "muxer-msg-iter-addr=%p, "
1017 "muxer-upstream-msg-iter-wrap-addr=%p",
1018 muxer_msg_iter, muxer_upstream_msg_iter);
1019 g_ptr_array_add(
1020 muxer_msg_iter->ended_muxer_upstream_msg_iters,
1021 muxer_upstream_msg_iter);
1022 muxer_msg_iter->active_muxer_upstream_msg_iters->pdata[i] = NULL;
1023
1024 /*
1025 * Use g_ptr_array_remove_fast() because the
1026 * order of those elements is not important.
1027 */
1028 g_ptr_array_remove_index_fast(
1029 muxer_msg_iter->active_muxer_upstream_msg_iters,
1030 i);
1031 i--;
1032 }
1033 }
1034
1035 end:
1036 return status;
1037 }
1038
1039 static inline
1040 bt_component_class_message_iterator_next_method_status muxer_msg_iter_do_next_one(
1041 struct muxer_comp *muxer_comp,
1042 struct muxer_msg_iter *muxer_msg_iter,
1043 const bt_message **msg)
1044 {
1045 bt_component_class_message_iterator_next_method_status status;
1046 struct muxer_upstream_msg_iter *muxer_upstream_msg_iter = NULL;
1047 int64_t next_return_ts;
1048
1049 status = validate_muxer_upstream_msg_iters(muxer_msg_iter);
1050 if (status != BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK) {
1051 /* validate_muxer_upstream_msg_iters() logs details */
1052 goto end;
1053 }
1054
1055 /*
1056 * At this point we know that all the existing upstream
1057 * message iterators are valid. We can find the one,
1058 * amongst those, of which the current message is the
1059 * youngest.
1060 */
1061 status = muxer_msg_iter_youngest_upstream_msg_iter(muxer_comp,
1062 muxer_msg_iter, &muxer_upstream_msg_iter,
1063 &next_return_ts);
1064 if (status < 0 || status == BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_END) {
1065 if (status < 0) {
1066 BT_COMP_LOGE("Cannot find the youngest upstream message iterator wrapper: "
1067 "status=%s",
1068 bt_common_func_status_string(status));
1069 } else {
1070 BT_COMP_LOGD("Cannot find the youngest upstream message iterator wrapper: "
1071 "status=%s",
1072 bt_common_func_status_string(status));
1073 }
1074
1075 goto end;
1076 }
1077
1078 if (next_return_ts < muxer_msg_iter->last_returned_ts_ns) {
1079 BT_COMP_LOGE("Youngest upstream message iterator wrapper's timestamp is less than muxer's message iterator's last returned timestamp: "
1080 "muxer-msg-iter-addr=%p, ts=%" PRId64 ", "
1081 "last-returned-ts=%" PRId64,
1082 muxer_msg_iter, next_return_ts,
1083 muxer_msg_iter->last_returned_ts_ns);
1084 status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_ERROR;
1085 goto end;
1086 }
1087
1088 BT_COMP_LOGD("Found youngest upstream message iterator wrapper: "
1089 "muxer-msg-iter-addr=%p, "
1090 "muxer-upstream-msg-iter-wrap-addr=%p, "
1091 "ts=%" PRId64,
1092 muxer_msg_iter, muxer_upstream_msg_iter, next_return_ts);
1093 BT_ASSERT(status == BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK);
1094 BT_ASSERT(muxer_upstream_msg_iter);
1095
1096 /*
1097 * Consume from the queue's head: other side
1098 * (muxer_upstream_msg_iter_next()) writes to the tail.
1099 */
1100 *msg = g_queue_pop_head(muxer_upstream_msg_iter->msgs);
1101 BT_ASSERT(*msg);
1102 muxer_msg_iter->last_returned_ts_ns = next_return_ts;
1103
1104 end:
1105 return status;
1106 }
1107
1108 static
1109 bt_component_class_message_iterator_next_method_status muxer_msg_iter_do_next(
1110 struct muxer_comp *muxer_comp,
1111 struct muxer_msg_iter *muxer_msg_iter,
1112 bt_message_array_const msgs, uint64_t capacity,
1113 uint64_t *count)
1114 {
1115 bt_component_class_message_iterator_next_method_status status =
1116 BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK;
1117 uint64_t i = 0;
1118
1119 while (i < capacity && status == BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK) {
1120 status = muxer_msg_iter_do_next_one(muxer_comp,
1121 muxer_msg_iter, &msgs[i]);
1122 if (status == BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK) {
1123 i++;
1124 }
1125 }
1126
1127 if (i > 0) {
1128 /*
1129 * Even if muxer_msg_iter_do_next_one() returned
1130 * something else than
1131 * BT_MESSAGE_ITERATOR_STATUS_OK, we accumulated
1132 * message objects in the output message
1133 * array, so we need to return
1134 * BT_MESSAGE_ITERATOR_STATUS_OK so that they are
1135 * transfered to downstream. This other status occurs
1136 * again the next time muxer_msg_iter_do_next() is
1137 * called, possibly without any accumulated
1138 * message, in which case we'll return it.
1139 */
1140 *count = i;
1141 status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK;
1142 }
1143
1144 return status;
1145 }
1146
1147 static
1148 void destroy_muxer_msg_iter(struct muxer_msg_iter *muxer_msg_iter)
1149 {
1150 struct muxer_comp *muxer_comp;
1151
1152 if (!muxer_msg_iter) {
1153 return;
1154 }
1155
1156 muxer_comp = muxer_msg_iter->muxer_comp;
1157 BT_COMP_LOGD("Destroying muxer component's message iterator: "
1158 "muxer-msg-iter-addr=%p", muxer_msg_iter);
1159
1160 if (muxer_msg_iter->active_muxer_upstream_msg_iters) {
1161 BT_COMP_LOGD_STR("Destroying muxer's active upstream message iterator wrappers.");
1162 g_ptr_array_free(
1163 muxer_msg_iter->active_muxer_upstream_msg_iters, TRUE);
1164 }
1165
1166 if (muxer_msg_iter->ended_muxer_upstream_msg_iters) {
1167 BT_COMP_LOGD_STR("Destroying muxer's ended upstream message iterator wrappers.");
1168 g_ptr_array_free(
1169 muxer_msg_iter->ended_muxer_upstream_msg_iters, TRUE);
1170 }
1171
1172 g_free(muxer_msg_iter);
1173 }
1174
1175 static
1176 int muxer_msg_iter_init_upstream_iterators(struct muxer_comp *muxer_comp,
1177 struct muxer_msg_iter *muxer_msg_iter)
1178 {
1179 int64_t count;
1180 int64_t i;
1181 int ret = 0;
1182
1183 count = bt_component_filter_get_input_port_count(
1184 bt_self_component_filter_as_component_filter(
1185 muxer_comp->self_comp_flt));
1186 if (count < 0) {
1187 BT_COMP_LOGD("No input port to initialize for muxer component's message iterator: "
1188 "muxer-comp-addr=%p, muxer-msg-iter-addr=%p",
1189 muxer_comp, muxer_msg_iter);
1190 goto end;
1191 }
1192
1193 for (i = 0; i < count; i++) {
1194 bt_self_component_port_input_message_iterator *upstream_msg_iter;
1195 bt_self_component_port_input *self_port =
1196 bt_self_component_filter_borrow_input_port_by_index(
1197 muxer_comp->self_comp_flt, i);
1198 const bt_port *port;
1199
1200 BT_ASSERT(self_port);
1201 port = bt_self_component_port_as_port(
1202 bt_self_component_port_input_as_self_component_port(
1203 self_port));
1204 BT_ASSERT(port);
1205
1206 if (!bt_port_is_connected(port)) {
1207 /* Skip non-connected port */
1208 continue;
1209 }
1210
1211 upstream_msg_iter = create_msg_iter_on_input_port(muxer_comp,
1212 self_port);
1213 if (!upstream_msg_iter) {
1214 /* create_msg_iter_on_input_port() logs errors */
1215 BT_ASSERT(!upstream_msg_iter);
1216 ret = -1;
1217 goto end;
1218 }
1219
1220 ret = muxer_msg_iter_add_upstream_msg_iter(muxer_msg_iter,
1221 upstream_msg_iter);
1222 bt_self_component_port_input_message_iterator_put_ref(
1223 upstream_msg_iter);
1224 if (ret) {
1225 /* muxer_msg_iter_add_upstream_msg_iter() logs errors */
1226 goto end;
1227 }
1228 }
1229
1230 end:
1231 return ret;
1232 }
1233
1234 BT_HIDDEN
1235 bt_component_class_message_iterator_init_method_status muxer_msg_iter_init(
1236 bt_self_message_iterator *self_msg_iter,
1237 bt_self_component_filter *self_comp,
1238 bt_self_component_port_output *port)
1239 {
1240 struct muxer_comp *muxer_comp = NULL;
1241 struct muxer_msg_iter *muxer_msg_iter = NULL;
1242 bt_component_class_message_iterator_init_method_status status =
1243 BT_COMPONENT_CLASS_MESSAGE_ITERATOR_INIT_METHOD_STATUS_OK;
1244 int ret;
1245
1246 muxer_comp = bt_self_component_get_data(
1247 bt_self_component_filter_as_self_component(self_comp));
1248 BT_ASSERT(muxer_comp);
1249 BT_COMP_LOGD("Initializing muxer component's message iterator: "
1250 "comp-addr=%p, muxer-comp-addr=%p, msg-iter-addr=%p",
1251 self_comp, muxer_comp, self_msg_iter);
1252
1253 if (muxer_comp->initializing_muxer_msg_iter) {
1254 /*
1255 * Weird, unhandled situation detected: downstream
1256 * creates a muxer message iterator while creating
1257 * another muxer message iterator (same component).
1258 */
1259 BT_COMP_LOGE("Recursive initialization of muxer component's message iterator: "
1260 "comp-addr=%p, muxer-comp-addr=%p, msg-iter-addr=%p",
1261 self_comp, muxer_comp, self_msg_iter);
1262 goto error;
1263 }
1264
1265 muxer_comp->initializing_muxer_msg_iter = true;
1266 muxer_msg_iter = g_new0(struct muxer_msg_iter, 1);
1267 if (!muxer_msg_iter) {
1268 BT_COMP_LOGE_STR("Failed to allocate one muxer component's message iterator.");
1269 goto error;
1270 }
1271
1272 muxer_msg_iter->muxer_comp = muxer_comp;
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 goto error;
1280 }
1281
1282 muxer_msg_iter->ended_muxer_upstream_msg_iters =
1283 g_ptr_array_new_with_free_func(
1284 (GDestroyNotify) destroy_muxer_upstream_msg_iter);
1285 if (!muxer_msg_iter->ended_muxer_upstream_msg_iters) {
1286 BT_COMP_LOGE_STR("Failed to allocate a GPtrArray.");
1287 goto error;
1288 }
1289
1290 ret = muxer_msg_iter_init_upstream_iterators(muxer_comp,
1291 muxer_msg_iter);
1292 if (ret) {
1293 BT_COMP_LOGE("Cannot initialize connected input ports for muxer component's message iterator: "
1294 "comp-addr=%p, muxer-comp-addr=%p, "
1295 "muxer-msg-iter-addr=%p, msg-iter-addr=%p, ret=%d",
1296 self_comp, muxer_comp, muxer_msg_iter,
1297 self_msg_iter, ret);
1298 goto error;
1299 }
1300
1301 bt_self_message_iterator_set_data(self_msg_iter, muxer_msg_iter);
1302 BT_COMP_LOGD("Initialized muxer component's message iterator: "
1303 "comp-addr=%p, muxer-comp-addr=%p, muxer-msg-iter-addr=%p, "
1304 "msg-iter-addr=%p",
1305 self_comp, muxer_comp, muxer_msg_iter, self_msg_iter);
1306 goto end;
1307
1308 error:
1309 destroy_muxer_msg_iter(muxer_msg_iter);
1310 bt_self_message_iterator_set_data(self_msg_iter, NULL);
1311 status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_INIT_METHOD_STATUS_ERROR;
1312
1313 end:
1314 muxer_comp->initializing_muxer_msg_iter = false;
1315 return status;
1316 }
1317
1318 BT_HIDDEN
1319 void muxer_msg_iter_finalize(bt_self_message_iterator *self_msg_iter)
1320 {
1321 struct muxer_msg_iter *muxer_msg_iter =
1322 bt_self_message_iterator_get_data(self_msg_iter);
1323 bt_self_component *self_comp = NULL;
1324 struct muxer_comp *muxer_comp = NULL;
1325
1326 self_comp = bt_self_message_iterator_borrow_component(
1327 self_msg_iter);
1328 BT_ASSERT(self_comp);
1329 muxer_comp = bt_self_component_get_data(self_comp);
1330 BT_COMP_LOGD("Finalizing muxer component's message iterator: "
1331 "comp-addr=%p, muxer-comp-addr=%p, muxer-msg-iter-addr=%p, "
1332 "msg-iter-addr=%p",
1333 self_comp, muxer_comp, muxer_msg_iter, self_msg_iter);
1334
1335 if (muxer_msg_iter) {
1336 destroy_muxer_msg_iter(muxer_msg_iter);
1337 }
1338 }
1339
1340 BT_HIDDEN
1341 bt_component_class_message_iterator_next_method_status muxer_msg_iter_next(
1342 bt_self_message_iterator *self_msg_iter,
1343 bt_message_array_const msgs, uint64_t capacity,
1344 uint64_t *count)
1345 {
1346 bt_component_class_message_iterator_next_method_status status;
1347 struct muxer_msg_iter *muxer_msg_iter =
1348 bt_self_message_iterator_get_data(self_msg_iter);
1349 bt_self_component *self_comp = NULL;
1350 struct muxer_comp *muxer_comp = NULL;
1351
1352 BT_ASSERT(muxer_msg_iter);
1353 self_comp = bt_self_message_iterator_borrow_component(
1354 self_msg_iter);
1355 BT_ASSERT(self_comp);
1356 muxer_comp = bt_self_component_get_data(self_comp);
1357 BT_ASSERT(muxer_comp);
1358 BT_COMP_LOGT("Muxer component's message iterator's \"next\" method called: "
1359 "comp-addr=%p, muxer-comp-addr=%p, muxer-msg-iter-addr=%p, "
1360 "msg-iter-addr=%p",
1361 self_comp, muxer_comp, muxer_msg_iter, self_msg_iter);
1362
1363 status = muxer_msg_iter_do_next(muxer_comp, muxer_msg_iter,
1364 msgs, capacity, count);
1365 if (status < 0) {
1366 BT_COMP_LOGE("Cannot get next message: "
1367 "comp-addr=%p, muxer-comp-addr=%p, muxer-msg-iter-addr=%p, "
1368 "msg-iter-addr=%p, status=%s",
1369 self_comp, muxer_comp, muxer_msg_iter, self_msg_iter,
1370 bt_common_func_status_string(status));
1371 } else {
1372 BT_COMP_LOGT("Returning from muxer component's message iterator's \"next\" method: "
1373 "status=%s",
1374 bt_common_func_status_string(status));
1375 }
1376
1377 return status;
1378 }
1379
1380 BT_HIDDEN
1381 bt_component_class_port_connected_method_status muxer_input_port_connected(
1382 bt_self_component_filter *self_comp,
1383 bt_self_component_port_input *self_port,
1384 const bt_port_output *other_port)
1385 {
1386 bt_component_class_port_connected_method_status status =
1387 BT_COMPONENT_CLASS_PORT_CONNECTED_METHOD_STATUS_OK;
1388 bt_self_component_add_port_status add_port_status;
1389 struct muxer_comp *muxer_comp = bt_self_component_get_data(
1390 bt_self_component_filter_as_self_component(self_comp));
1391
1392 add_port_status = add_available_input_port(self_comp);
1393 if (add_port_status) {
1394 BT_COMP_LOGE("Cannot add one muxer component's input port: "
1395 "status=%s",
1396 bt_common_func_status_string(status));
1397
1398 if (add_port_status ==
1399 BT_SELF_COMPONENT_ADD_PORT_STATUS_MEMORY_ERROR) {
1400 status = BT_COMPONENT_CLASS_PORT_CONNECTED_METHOD_STATUS_MEMORY_ERROR;
1401 } else {
1402 status = BT_COMPONENT_CLASS_PORT_CONNECTED_METHOD_STATUS_ERROR;
1403 }
1404
1405 goto end;
1406 }
1407
1408 end:
1409 return status;
1410 }
1411
1412 static inline
1413 bt_bool muxer_upstream_msg_iters_can_all_seek_beginning(
1414 GPtrArray *muxer_upstream_msg_iters)
1415 {
1416 uint64_t i;
1417 bt_bool ret = BT_TRUE;
1418
1419 for (i = 0; i < muxer_upstream_msg_iters->len; i++) {
1420 struct muxer_upstream_msg_iter *upstream_msg_iter =
1421 muxer_upstream_msg_iters->pdata[i];
1422
1423 if (!bt_self_component_port_input_message_iterator_can_seek_beginning(
1424 upstream_msg_iter->msg_iter)) {
1425 ret = BT_FALSE;
1426 goto end;
1427 }
1428 }
1429
1430 end:
1431 return ret;
1432 }
1433
1434 BT_HIDDEN
1435 bt_bool muxer_msg_iter_can_seek_beginning(
1436 bt_self_message_iterator *self_msg_iter)
1437 {
1438 struct muxer_msg_iter *muxer_msg_iter =
1439 bt_self_message_iterator_get_data(self_msg_iter);
1440 bt_bool ret = BT_TRUE;
1441
1442 if (!muxer_upstream_msg_iters_can_all_seek_beginning(
1443 muxer_msg_iter->active_muxer_upstream_msg_iters)) {
1444 ret = BT_FALSE;
1445 goto end;
1446 }
1447
1448 if (!muxer_upstream_msg_iters_can_all_seek_beginning(
1449 muxer_msg_iter->ended_muxer_upstream_msg_iters)) {
1450 ret = BT_FALSE;
1451 goto end;
1452 }
1453
1454 end:
1455 return ret;
1456 }
1457
1458 BT_HIDDEN
1459 bt_component_class_message_iterator_seek_beginning_method_status muxer_msg_iter_seek_beginning(
1460 bt_self_message_iterator *self_msg_iter)
1461 {
1462 struct muxer_msg_iter *muxer_msg_iter =
1463 bt_self_message_iterator_get_data(self_msg_iter);
1464 bt_component_class_message_iterator_seek_beginning_method_status status =
1465 BT_COMPONENT_CLASS_MESSAGE_ITERATOR_SEEK_BEGINNING_METHOD_STATUS_OK;
1466 bt_message_iterator_seek_beginning_status seek_beg_status;
1467 uint64_t i;
1468
1469 /* Seek all ended upstream iterators first */
1470 for (i = 0; i < muxer_msg_iter->ended_muxer_upstream_msg_iters->len;
1471 i++) {
1472 struct muxer_upstream_msg_iter *upstream_msg_iter =
1473 muxer_msg_iter->ended_muxer_upstream_msg_iters->pdata[i];
1474
1475 seek_beg_status = bt_self_component_port_input_message_iterator_seek_beginning(
1476 upstream_msg_iter->msg_iter);
1477 if (seek_beg_status != BT_MESSAGE_ITERATOR_SEEK_BEGINNING_STATUS_OK) {
1478 status = (int) seek_beg_status;
1479 goto end;
1480 }
1481
1482 empty_message_queue(upstream_msg_iter);
1483 }
1484
1485 /* Seek all previously active upstream iterators */
1486 for (i = 0; i < muxer_msg_iter->active_muxer_upstream_msg_iters->len;
1487 i++) {
1488 struct muxer_upstream_msg_iter *upstream_msg_iter =
1489 muxer_msg_iter->active_muxer_upstream_msg_iters->pdata[i];
1490
1491 seek_beg_status = bt_self_component_port_input_message_iterator_seek_beginning(
1492 upstream_msg_iter->msg_iter);
1493 if (seek_beg_status != BT_MESSAGE_ITERATOR_SEEK_BEGINNING_STATUS_OK) {
1494 status = (int) seek_beg_status;
1495 goto end;
1496 }
1497
1498 empty_message_queue(upstream_msg_iter);
1499 }
1500
1501 /* Make them all active */
1502 for (i = 0; i < muxer_msg_iter->ended_muxer_upstream_msg_iters->len;
1503 i++) {
1504 struct muxer_upstream_msg_iter *upstream_msg_iter =
1505 muxer_msg_iter->ended_muxer_upstream_msg_iters->pdata[i];
1506
1507 g_ptr_array_add(muxer_msg_iter->active_muxer_upstream_msg_iters,
1508 upstream_msg_iter);
1509 muxer_msg_iter->ended_muxer_upstream_msg_iters->pdata[i] = NULL;
1510 }
1511
1512 g_ptr_array_remove_range(muxer_msg_iter->ended_muxer_upstream_msg_iters,
1513 0, muxer_msg_iter->ended_muxer_upstream_msg_iters->len);
1514 muxer_msg_iter->last_returned_ts_ns = INT64_MIN;
1515 muxer_msg_iter->clock_class_expectation =
1516 MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_ANY;
1517
1518 end:
1519 return status;
1520 }
This page took 0.065457 seconds and 4 git commands to generate.