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