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