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