projects / babeltrace.git / blob
? search:


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