projects / babeltrace.git / blob
? search:


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