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