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