lib: use BT_LIB_LOG*_APPEND_CAUSE() where appropriate
[babeltrace.git] / src / lib / graph / iterator.c
CommitLineData
47e5a032 1/*
e2f7325d 2 * Copyright 2017-2018 Philippe Proulx <pproulx@efficios.com>
47e5a032 3 * Copyright 2015 Jérémie Galarneau <jeremie.galarneau@efficios.com>
47e5a032
JG
4 *
5 * Permission is hereby granted, free of charge, to any person obtaining a copy
6 * of this software and associated documentation files (the "Software"), to deal
7 * in the Software without restriction, including without limitation the rights
8 * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
9 * copies of the Software, and to permit persons to whom the Software is
10 * furnished to do so, subject to the following conditions:
11 *
12 * The above copyright notice and this permission notice shall be included in
13 * all copies or substantial portions of the Software.
14 *
15 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
16 * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
17 * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
18 * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
19 * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
20 * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
21 * SOFTWARE.
22 */
23
350ad6c1 24#define BT_LOG_TAG "LIB/MSG-ITER"
c2d9d9cf 25#include "lib/logging.h"
5af447e5 26
578e048b 27#include "compat/compiler.h"
e74dbb33 28#include "compat/glib.h"
578e048b
MJ
29#include "lib/trace-ir/clock-class.h"
30#include "lib/trace-ir/clock-snapshot.h"
3fadfbc0
MJ
31#include <babeltrace2/trace-ir/field.h>
32#include <babeltrace2/trace-ir/event-const.h>
578e048b 33#include "lib/trace-ir/event.h"
3fadfbc0 34#include <babeltrace2/trace-ir/packet-const.h>
578e048b
MJ
35#include "lib/trace-ir/packet.h"
36#include "lib/trace-ir/stream.h"
d24d5663
PP
37#include <babeltrace2/trace-ir/clock-class-const.h>
38#include <babeltrace2/trace-ir/stream-class-const.h>
39#include <babeltrace2/trace-ir/stream-const.h>
3fadfbc0 40#include <babeltrace2/graph/connection-const.h>
3fadfbc0 41#include <babeltrace2/graph/component-const.h>
3fadfbc0 42#include <babeltrace2/graph/component-sink-const.h>
3fadfbc0 43#include <babeltrace2/graph/message-const.h>
d24d5663 44#include <babeltrace2/graph/message-iterator.h>
3fadfbc0
MJ
45#include <babeltrace2/graph/self-component-port-input-message-iterator.h>
46#include <babeltrace2/graph/port-output-message-iterator.h>
3fadfbc0 47#include <babeltrace2/graph/message-event-const.h>
5b7b55be
SM
48#include <babeltrace2/graph/message-message-iterator-inactivity-const.h>
49#include <babeltrace2/graph/message-packet-beginning.h>
3fadfbc0
MJ
50#include <babeltrace2/graph/message-packet-beginning-const.h>
51#include <babeltrace2/graph/message-packet-end-const.h>
5b7b55be
SM
52#include <babeltrace2/graph/message-stream-activity-beginning.h>
53#include <babeltrace2/graph/message-stream-activity-beginning-const.h>
54#include <babeltrace2/graph/message-stream-activity-end-const.h>
55#include <babeltrace2/graph/message-stream-beginning.h>
3fadfbc0
MJ
56#include <babeltrace2/graph/message-stream-beginning-const.h>
57#include <babeltrace2/graph/message-stream-end-const.h>
3fadfbc0
MJ
58#include <babeltrace2/graph/port-const.h>
59#include <babeltrace2/graph/graph.h>
60#include <babeltrace2/graph/graph-const.h>
3fadfbc0 61#include <babeltrace2/types.h>
578e048b
MJ
62#include "common/assert.h"
63#include "lib/assert-pre.h"
f6f301d7 64#include "lib/assert-post.h"
fa054faf 65#include <stdint.h>
2ec84d26 66#include <inttypes.h>
0fbb9a9f 67#include <stdlib.h>
3230ee6b 68
578e048b
MJ
69#include "component-class.h"
70#include "component-class-sink-colander.h"
71#include "component.h"
72#include "component-sink.h"
73#include "component-source.h"
74#include "connection.h"
75#include "graph.h"
76#include "message/discarded-items.h"
77#include "message/event.h"
78#include "message/iterator.h"
79#include "message/message.h"
80#include "message/message-iterator-inactivity.h"
81#include "message/stream.h"
82#include "message/packet.h"
83#include "message/stream-activity.h"
d24d5663 84#include "lib/func-status.h"
578e048b 85
d4393e08
PP
86/*
87 * TODO: Use graph's state (number of active iterators, etc.) and
88 * possibly system specifications to make a better guess than this.
89 */
d6e69534 90#define MSG_BATCH_SIZE 15
d4393e08 91
7474e7d3
PP
92#define BT_ASSERT_PRE_ITER_HAS_STATE_TO_SEEK(_iter) \
93 BT_ASSERT_PRE((_iter)->state == BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_ACTIVE || \
94 (_iter)->state == BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_ENDED || \
95 (_iter)->state == BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_LAST_SEEKING_RETURNED_AGAIN || \
96 (_iter)->state == BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_LAST_SEEKING_RETURNED_ERROR, \
97 "Message iterator is in the wrong state: %!+i", _iter)
47e5a032 98
d0fea130 99static inline
003e713f 100void set_self_comp_port_input_msg_iterator_state(
d0fea130
PP
101 struct bt_self_component_port_input_message_iterator *iterator,
102 enum bt_self_component_port_input_message_iterator_state state)
103{
104 BT_ASSERT(iterator);
7474e7d3 105 BT_LIB_LOGD("Updating message iterator's state: new-state=%s",
d0fea130
PP
106 bt_self_component_port_input_message_iterator_state_string(state));
107 iterator->state = state;
108}
109
8ed535b5 110static
d6e69534 111void destroy_base_message_iterator(struct bt_object *obj)
8ed535b5 112{
d6e69534 113 struct bt_message_iterator *iterator = (void *) obj;
d4393e08
PP
114
115 BT_ASSERT(iterator);
116
d6e69534
PP
117 if (iterator->msgs) {
118 g_ptr_array_free(iterator->msgs, TRUE);
119 iterator->msgs = NULL;
d4393e08
PP
120 }
121
122 g_free(iterator);
8ed535b5
PP
123}
124
47e5a032 125static
d6e69534 126void bt_self_component_port_input_message_iterator_destroy(struct bt_object *obj)
47e5a032 127{
d6e69534 128 struct bt_self_component_port_input_message_iterator *iterator;
8738a040 129
f6ccaed9 130 BT_ASSERT(obj);
d3eb6e8f 131
bd14d768 132 /*
d6e69534 133 * The message iterator's reference count is 0 if we're
bd14d768
PP
134 * here. Increment it to avoid a double-destroy (possibly
135 * infinitely recursive). This could happen for example if the
d6e69534 136 * message iterator's finalization function does
d94d92ac
PP
137 * bt_object_get_ref() (or anything that causes
138 * bt_object_get_ref() to be called) on itself (ref. count goes
139 * from 0 to 1), and then bt_object_put_ref(): the reference
140 * count would go from 1 to 0 again and this function would be
141 * called again.
bd14d768 142 */
3fea54f6 143 obj->ref_count++;
07245ac2 144 iterator = (void *) obj;
3f7d4d90 145 BT_LIB_LOGI("Destroying self component input port message iterator object: "
d94d92ac 146 "%!+i", iterator);
d0fea130 147 bt_self_component_port_input_message_iterator_try_finalize(iterator);
d3eb6e8f 148
bd14d768
PP
149 if (iterator->connection) {
150 /*
151 * Remove ourself from the originating connection so
152 * that it does not try to finalize a dangling pointer
153 * later.
154 */
155 bt_connection_remove_iterator(iterator->connection, iterator);
d94d92ac 156 iterator->connection = NULL;
bd14d768
PP
157 }
158
da9c4c52
SM
159 if (iterator->auto_seek.msgs) {
160 while (!g_queue_is_empty(iterator->auto_seek.msgs)) {
5b9e151d 161 bt_object_put_no_null_check(
da9c4c52 162 g_queue_pop_tail(iterator->auto_seek.msgs));
7474e7d3
PP
163 }
164
da9c4c52
SM
165 g_queue_free(iterator->auto_seek.msgs);
166 iterator->auto_seek.msgs = NULL;
7474e7d3
PP
167 }
168
d6e69534 169 destroy_base_message_iterator(obj);
47e5a032
JG
170}
171
bd14d768 172BT_HIDDEN
d0fea130 173void bt_self_component_port_input_message_iterator_try_finalize(
d6e69534 174 struct bt_self_component_port_input_message_iterator *iterator)
bd14d768 175{
d94d92ac
PP
176 typedef void (*method_t)(void *);
177
bd14d768 178 struct bt_component_class *comp_class = NULL;
d94d92ac 179 method_t method = NULL;
bd14d768 180
f6ccaed9 181 BT_ASSERT(iterator);
bd14d768
PP
182
183 switch (iterator->state) {
d6e69534 184 case BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_NON_INITIALIZED:
088d4023 185 /* Skip user finalization if user initialization failed */
d6e69534 186 BT_LIB_LOGD("Not finalizing non-initialized message iterator: "
d94d92ac 187 "%!+i", iterator);
d0fea130 188 goto end;
d6e69534 189 case BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_FINALIZED:
bd14d768 190 /* Already finalized */
d6e69534 191 BT_LIB_LOGD("Not finalizing message iterator: already finalized: "
d94d92ac 192 "%!+i", iterator);
d0fea130
PP
193 goto end;
194 case BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_FINALIZING:
870631a2 195 /* Finalizing */
d0fea130
PP
196 BT_LIB_LOGF("Message iterator is already being finalized: "
197 "%!+i", iterator);
198 abort();
bd14d768
PP
199 default:
200 break;
201 }
202
d6e69534 203 BT_LIB_LOGD("Finalizing message iterator: %!+i", iterator);
d0fea130
PP
204 set_self_comp_port_input_msg_iterator_state(iterator,
205 BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_FINALIZING);
f6ccaed9 206 BT_ASSERT(iterator->upstream_component);
bd14d768
PP
207 comp_class = iterator->upstream_component->class;
208
209 /* Call user-defined destroy method */
210 switch (comp_class->type) {
211 case BT_COMPONENT_CLASS_TYPE_SOURCE:
212 {
d94d92ac
PP
213 struct bt_component_class_source *src_comp_cls =
214 (void *) comp_class;
bd14d768 215
d6e69534 216 method = (method_t) src_comp_cls->methods.msg_iter_finalize;
bd14d768
PP
217 break;
218 }
219 case BT_COMPONENT_CLASS_TYPE_FILTER:
220 {
d94d92ac
PP
221 struct bt_component_class_filter *flt_comp_cls =
222 (void *) comp_class;
bd14d768 223
d6e69534 224 method = (method_t) flt_comp_cls->methods.msg_iter_finalize;
bd14d768
PP
225 break;
226 }
227 default:
228 /* Unreachable */
0fbb9a9f 229 abort();
bd14d768
PP
230 }
231
d94d92ac
PP
232 if (method) {
233 BT_LIB_LOGD("Calling user's finalization method: %!+i",
5af447e5 234 iterator);
d94d92ac 235 method(iterator);
bd14d768
PP
236 }
237
bd14d768
PP
238 iterator->upstream_component = NULL;
239 iterator->upstream_port = NULL;
d0fea130
PP
240 set_self_comp_port_input_msg_iterator_state(iterator,
241 BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_FINALIZED);
d6e69534 242 BT_LIB_LOGD("Finalized message iterator: %!+i", iterator);
d0fea130
PP
243
244end:
245 return;
bd14d768
PP
246}
247
248BT_HIDDEN
d6e69534
PP
249void bt_self_component_port_input_message_iterator_set_connection(
250 struct bt_self_component_port_input_message_iterator *iterator,
bd14d768
PP
251 struct bt_connection *connection)
252{
f6ccaed9 253 BT_ASSERT(iterator);
bd14d768 254 iterator->connection = connection;
3f7d4d90 255 BT_LIB_LOGI("Set message iterator's connection: "
d94d92ac 256 "%![iter-]+i, %![conn-]+x", iterator, connection);
bd14d768
PP
257}
258
90157d89 259static
d6e69534
PP
260int init_message_iterator(struct bt_message_iterator *iterator,
261 enum bt_message_iterator_type type,
90157d89
PP
262 bt_object_release_func destroy)
263{
d4393e08
PP
264 int ret = 0;
265
3fea54f6 266 bt_object_init_shared(&iterator->base, destroy);
90157d89 267 iterator->type = type;
d6e69534
PP
268 iterator->msgs = g_ptr_array_new();
269 if (!iterator->msgs) {
870631a2 270 BT_LIB_LOGE_APPEND_CAUSE("Failed to allocate a GPtrArray.");
d4393e08
PP
271 ret = -1;
272 goto end;
273 }
274
d6e69534 275 g_ptr_array_set_size(iterator->msgs, MSG_BATCH_SIZE);
d4393e08
PP
276
277end:
278 return ret;
90157d89
PP
279}
280
7474e7d3
PP
281static
282bt_bool can_seek_ns_from_origin_true(
283 struct bt_self_component_port_input_message_iterator *iterator,
284 int64_t ns_from_origin)
285{
286 return BT_TRUE;
287}
288
289static
290bt_bool can_seek_beginning_true(
291 struct bt_self_component_port_input_message_iterator *iterator)
292{
293 return BT_TRUE;
294}
295
d94d92ac 296static
d6e69534
PP
297struct bt_self_component_port_input_message_iterator *
298bt_self_component_port_input_message_iterator_create_initial(
3230ee6b 299 struct bt_component *upstream_comp,
d94d92ac 300 struct bt_port *upstream_port)
47e5a032 301{
d4393e08 302 int ret;
d6e69534 303 struct bt_self_component_port_input_message_iterator *iterator = NULL;
47e5a032 304
f6ccaed9
PP
305 BT_ASSERT(upstream_comp);
306 BT_ASSERT(upstream_port);
f6ccaed9 307 BT_ASSERT(bt_port_is_connected(upstream_port));
3f7d4d90 308 BT_LIB_LOGI("Creating initial message iterator on self component input port: "
d94d92ac
PP
309 "%![up-comp-]+c, %![up-port-]+p", upstream_comp, upstream_port);
310 BT_ASSERT(bt_component_get_class_type(upstream_comp) ==
311 BT_COMPONENT_CLASS_TYPE_SOURCE ||
312 bt_component_get_class_type(upstream_comp) ==
313 BT_COMPONENT_CLASS_TYPE_FILTER);
314 iterator = g_new0(
d6e69534 315 struct bt_self_component_port_input_message_iterator, 1);
47e5a032 316 if (!iterator) {
870631a2
PP
317 BT_LIB_LOGE_APPEND_CAUSE(
318 "Failed to allocate one self component input port "
d6e69534 319 "message iterator.");
870631a2 320 goto error;
47e5a032
JG
321 }
322
d6e69534
PP
323 ret = init_message_iterator((void *) iterator,
324 BT_MESSAGE_ITERATOR_TYPE_SELF_COMPONENT_PORT_INPUT,
325 bt_self_component_port_input_message_iterator_destroy);
d4393e08 326 if (ret) {
d6e69534 327 /* init_message_iterator() logs errors */
870631a2 328 goto error;
d4393e08 329 }
3230ee6b 330
54b135a0
SM
331 iterator->last_ns_from_origin = INT64_MIN;
332
da9c4c52
SM
333 iterator->auto_seek.msgs = g_queue_new();
334 if (!iterator->auto_seek.msgs) {
870631a2 335 BT_LIB_LOGE_APPEND_CAUSE("Failed to allocate a GQueue.");
7474e7d3 336 ret = -1;
73d5c1ad 337 goto end;
3230ee6b
PP
338 }
339
bd14d768
PP
340 iterator->upstream_component = upstream_comp;
341 iterator->upstream_port = upstream_port;
d94d92ac 342 iterator->connection = iterator->upstream_port->connection;
5c563278 343 iterator->graph = bt_component_borrow_graph(upstream_comp);
d0fea130
PP
344 set_self_comp_port_input_msg_iterator_state(iterator,
345 BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_NON_INITIALIZED);
7474e7d3
PP
346
347 switch (iterator->upstream_component->class->type) {
348 case BT_COMPONENT_CLASS_TYPE_SOURCE:
349 {
350 struct bt_component_class_source *src_comp_cls =
351 (void *) iterator->upstream_component->class;
352
353 iterator->methods.next =
354 (bt_self_component_port_input_message_iterator_next_method)
355 src_comp_cls->methods.msg_iter_next;
356 iterator->methods.seek_ns_from_origin =
357 (bt_self_component_port_input_message_iterator_seek_ns_from_origin_method)
358 src_comp_cls->methods.msg_iter_seek_ns_from_origin;
359 iterator->methods.seek_beginning =
360 (bt_self_component_port_input_message_iterator_seek_beginning_method)
361 src_comp_cls->methods.msg_iter_seek_beginning;
362 iterator->methods.can_seek_ns_from_origin =
363 (bt_self_component_port_input_message_iterator_can_seek_ns_from_origin_method)
364 src_comp_cls->methods.msg_iter_can_seek_ns_from_origin;
365 iterator->methods.can_seek_beginning =
366 (bt_self_component_port_input_message_iterator_can_seek_beginning_method)
367 src_comp_cls->methods.msg_iter_can_seek_beginning;
368 break;
369 }
370 case BT_COMPONENT_CLASS_TYPE_FILTER:
371 {
372 struct bt_component_class_filter *flt_comp_cls =
373 (void *) iterator->upstream_component->class;
374
375 iterator->methods.next =
376 (bt_self_component_port_input_message_iterator_next_method)
377 flt_comp_cls->methods.msg_iter_next;
378 iterator->methods.seek_ns_from_origin =
379 (bt_self_component_port_input_message_iterator_seek_ns_from_origin_method)
380 flt_comp_cls->methods.msg_iter_seek_ns_from_origin;
381 iterator->methods.seek_beginning =
382 (bt_self_component_port_input_message_iterator_seek_beginning_method)
383 flt_comp_cls->methods.msg_iter_seek_beginning;
384 iterator->methods.can_seek_ns_from_origin =
385 (bt_self_component_port_input_message_iterator_can_seek_ns_from_origin_method)
386 flt_comp_cls->methods.msg_iter_can_seek_ns_from_origin;
387 iterator->methods.can_seek_beginning =
388 (bt_self_component_port_input_message_iterator_can_seek_beginning_method)
389 flt_comp_cls->methods.msg_iter_can_seek_beginning;
390 break;
391 }
392 default:
393 abort();
394 }
395
396 if (iterator->methods.seek_ns_from_origin &&
397 !iterator->methods.can_seek_ns_from_origin) {
398 iterator->methods.can_seek_ns_from_origin =
399 (bt_self_component_port_input_message_iterator_can_seek_ns_from_origin_method)
400 can_seek_ns_from_origin_true;
401 }
402
403 if (iterator->methods.seek_beginning &&
404 !iterator->methods.can_seek_beginning) {
405 iterator->methods.can_seek_beginning =
406 (bt_self_component_port_input_message_iterator_seek_beginning_method)
407 can_seek_beginning_true;
408 }
409
3f7d4d90 410 BT_LIB_LOGI("Created initial message iterator on self component input port: "
d94d92ac
PP
411 "%![up-port-]+p, %![up-comp-]+c, %![iter-]+i",
412 upstream_port, upstream_comp, iterator);
870631a2
PP
413 goto end;
414
415error:
416 BT_OBJECT_PUT_REF_AND_RESET(iterator);
3230ee6b 417
47e5a032 418end:
d94d92ac 419 return iterator;
47e5a032
JG
420}
421
d6e69534
PP
422struct bt_self_component_port_input_message_iterator *
423bt_self_component_port_input_message_iterator_create(
d94d92ac 424 struct bt_self_component_port_input *self_port)
ea8d3e58 425{
d24d5663 426 typedef enum bt_component_class_message_iterator_init_method_status (*init_method_t)(
d94d92ac
PP
427 void *, void *, void *);
428
429 init_method_t init_method = NULL;
d6e69534 430 struct bt_self_component_port_input_message_iterator *iterator =
d94d92ac
PP
431 NULL;
432 struct bt_port *port = (void *) self_port;
433 struct bt_port *upstream_port;
434 struct bt_component *comp;
435 struct bt_component *upstream_comp;
436 struct bt_component_class *upstream_comp_cls;
437
438 BT_ASSERT_PRE_NON_NULL(port, "Port");
0d72b8c3 439 comp = bt_port_borrow_component_inline(port);
d94d92ac
PP
440 BT_ASSERT_PRE(bt_port_is_connected(port),
441 "Port is not connected: %![port-]+p", port);
442 BT_ASSERT_PRE(comp, "Port is not part of a component: %![port-]+p",
443 port);
444 BT_ASSERT_PRE(!bt_component_graph_is_canceled(comp),
445 "Port's component's graph is canceled: "
446 "%![port-]+p, %![comp-]+c", port, comp);
447 BT_ASSERT(port->connection);
448 upstream_port = port->connection->upstream_port;
449 BT_ASSERT(upstream_port);
0d72b8c3 450 upstream_comp = bt_port_borrow_component_inline(upstream_port);
d94d92ac 451 BT_ASSERT(upstream_comp);
5badd463
PP
452 BT_ASSERT_PRE(
453 bt_component_borrow_graph(upstream_comp)->config_state !=
454 BT_GRAPH_CONFIGURATION_STATE_CONFIGURING,
455 "Graph is not configured: %!+g",
456 bt_component_borrow_graph(upstream_comp));
d94d92ac
PP
457 upstream_comp_cls = upstream_comp->class;
458 BT_ASSERT(upstream_comp->class->type ==
459 BT_COMPONENT_CLASS_TYPE_SOURCE ||
460 upstream_comp->class->type ==
461 BT_COMPONENT_CLASS_TYPE_FILTER);
d6e69534 462 iterator = bt_self_component_port_input_message_iterator_create_initial(
d94d92ac
PP
463 upstream_comp, upstream_port);
464 if (!iterator) {
870631a2
PP
465 BT_LIB_LOGE_APPEND_CAUSE(
466 "Cannot create self component input port message iterator.");
467 goto error;
d94d92ac 468 }
890882ef 469
d94d92ac
PP
470 switch (upstream_comp_cls->type) {
471 case BT_COMPONENT_CLASS_TYPE_SOURCE:
472 {
473 struct bt_component_class_source *src_comp_cls =
474 (void *) upstream_comp_cls;
475
476 init_method =
d6e69534 477 (init_method_t) src_comp_cls->methods.msg_iter_init;
d94d92ac
PP
478 break;
479 }
480 case BT_COMPONENT_CLASS_TYPE_FILTER:
481 {
482 struct bt_component_class_filter *flt_comp_cls =
483 (void *) upstream_comp_cls;
484
485 init_method =
d6e69534 486 (init_method_t) flt_comp_cls->methods.msg_iter_init;
d94d92ac
PP
487 break;
488 }
489 default:
490 /* Unreachable */
491 abort();
492 }
493
494 if (init_method) {
d24d5663 495 enum bt_component_class_message_iterator_init_method_status iter_status;
d94d92ac
PP
496
497 BT_LIB_LOGD("Calling user's initialization method: %!+i", iterator);
498 iter_status = init_method(iterator, upstream_comp,
499 upstream_port);
500 BT_LOGD("User method returned: status=%s",
d24d5663
PP
501 bt_common_func_status_string(iter_status));
502 if (iter_status != BT_FUNC_STATUS_OK) {
870631a2
PP
503 BT_LIB_LOGW_APPEND_CAUSE(
504 "Component input port message iterator initialization method failed: "
505 "%![iter-]+i, status=%s",
506 iterator,
507 bt_common_func_status_string(iter_status));
508 goto error;
d94d92ac
PP
509 }
510 }
511
d0fea130
PP
512 set_self_comp_port_input_msg_iterator_state(iterator,
513 BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_ACTIVE);
d94d92ac 514 g_ptr_array_add(port->connection->iterators, iterator);
3f7d4d90 515 BT_LIB_LOGI("Created message iterator on self component input port: "
d94d92ac
PP
516 "%![up-port-]+p, %![up-comp-]+c, %![iter-]+i",
517 upstream_port, upstream_comp, iterator);
870631a2
PP
518 goto end;
519
520error:
521 BT_OBJECT_PUT_REF_AND_RESET(iterator);
d94d92ac
PP
522
523end:
524 return iterator;
ea8d3e58
JG
525}
526
d6e69534
PP
527void *bt_self_message_iterator_get_data(
528 const struct bt_self_message_iterator *self_iterator)
ea8d3e58 529{
d6e69534 530 struct bt_self_component_port_input_message_iterator *iterator =
d94d92ac 531 (void *) self_iterator;
ea8d3e58 532
d6e69534 533 BT_ASSERT_PRE_NON_NULL(iterator, "Message iterator");
d94d92ac 534 return iterator->user_data;
8738a040 535}
413bc2c4 536
d6e69534
PP
537void bt_self_message_iterator_set_data(
538 struct bt_self_message_iterator *self_iterator, void *data)
5c563278 539{
d6e69534 540 struct bt_self_component_port_input_message_iterator *iterator =
d94d92ac 541 (void *) self_iterator;
5c563278 542
d6e69534 543 BT_ASSERT_PRE_NON_NULL(iterator, "Message iterator");
d94d92ac 544 iterator->user_data = data;
3f7d4d90 545 BT_LIB_LOGD("Set message iterator's user data: "
d94d92ac 546 "%!+i, user-data-addr=%p", iterator, data);
5c563278
PP
547}
548
54b135a0
SM
549/*
550 * Validate that the default clock snapshot in `msg` doesn't make us go back in
551 * time.
552 */
553
f6f301d7 554BT_ASSERT_POST_FUNC
54b135a0
SM
555static
556bool clock_snapshots_are_monotonic_one(
557 struct bt_self_component_port_input_message_iterator *iterator,
558 const bt_message *msg)
559{
560 const struct bt_clock_snapshot *clock_snapshot = NULL;
561 bt_message_type message_type = bt_message_get_type(msg);
562 int64_t ns_from_origin;
d24d5663 563 enum bt_clock_snapshot_get_ns_from_origin_status clock_snapshot_status;
54b135a0
SM
564
565 /*
566 * The default is true: if we can't figure out the clock snapshot
567 * (or there is none), assume it is fine.
568 */
569 bool result = true;
570
571 switch (message_type) {
572 case BT_MESSAGE_TYPE_EVENT:
573 {
574 struct bt_message_event *event_msg = (struct bt_message_event *) msg;
575 clock_snapshot = event_msg->default_cs;
576 break;
577 }
578 case BT_MESSAGE_TYPE_MESSAGE_ITERATOR_INACTIVITY:
579 {
580 struct bt_message_message_iterator_inactivity *inactivity_msg =
581 (struct bt_message_message_iterator_inactivity *) msg;
582 clock_snapshot = inactivity_msg->default_cs;
583 break;
584 }
585 case BT_MESSAGE_TYPE_PACKET_BEGINNING:
586 case BT_MESSAGE_TYPE_PACKET_END:
587 {
588 struct bt_message_packet *packet_msg = (struct bt_message_packet *) msg;
589 clock_snapshot = packet_msg->default_cs;
590 break;
591 }
592 case BT_MESSAGE_TYPE_STREAM_ACTIVITY_BEGINNING:
593 case BT_MESSAGE_TYPE_STREAM_ACTIVITY_END:
594 {
595 struct bt_message_stream_activity *str_act_msg =
596 (struct bt_message_stream_activity *) msg;
597
598 if (str_act_msg->default_cs_state == BT_MESSAGE_STREAM_ACTIVITY_CLOCK_SNAPSHOT_STATE_KNOWN) {
599 clock_snapshot = str_act_msg->default_cs;
600 }
601 break;
602 }
603 case BT_MESSAGE_TYPE_STREAM_BEGINNING:
604 case BT_MESSAGE_TYPE_STREAM_END:
605 /* These messages don't have clock snapshots. */
606 goto end;
607 case BT_MESSAGE_TYPE_DISCARDED_EVENTS:
608 case BT_MESSAGE_TYPE_DISCARDED_PACKETS:
609 {
610 struct bt_message_discarded_items *discarded_msg =
611 (struct bt_message_discarded_items *) msg;
612
613 clock_snapshot = discarded_msg->default_begin_cs;
614 break;
615 }
616 }
617
618 if (!clock_snapshot) {
619 goto end;
620 }
621
622 clock_snapshot_status = bt_clock_snapshot_get_ns_from_origin(clock_snapshot, &ns_from_origin);
d24d5663 623 if (clock_snapshot_status != BT_FUNC_STATUS_OK) {
54b135a0
SM
624 goto end;
625 }
626
627 result = ns_from_origin >= iterator->last_ns_from_origin;
628 iterator->last_ns_from_origin = ns_from_origin;
629end:
630 return result;
631}
632
f6f301d7 633BT_ASSERT_POST_FUNC
54b135a0
SM
634static
635bool clock_snapshots_are_monotonic(
636 struct bt_self_component_port_input_message_iterator *iterator,
637 bt_message_array_const msgs, uint64_t msg_count)
638{
639 uint64_t i;
640 bool result;
641
642 for (i = 0; i < msg_count; i++) {
643 if (!clock_snapshots_are_monotonic_one(iterator, msgs[i])) {
644 result = false;
645 goto end;
646 }
647 }
648
649 result = true;
650
651end:
652 return result;
653}
654
655/*
656 * When a new stream begins, verify that the clock class tied to this
657 * stream is compatible with what we've seen before.
658 */
659
f6f301d7 660BT_ASSERT_POST_FUNC
54b135a0
SM
661static
662bool clock_classes_are_compatible_one(struct bt_self_component_port_input_message_iterator *iterator,
663 const struct bt_message *msg)
664{
665 enum bt_message_type message_type = bt_message_get_type(msg);
666 bool result;
667
668 if (message_type == BT_MESSAGE_TYPE_STREAM_BEGINNING) {
669 const struct bt_message_stream *stream_msg = (struct bt_message_stream *) msg;
670 const struct bt_clock_class *clock_class = stream_msg->stream->class->default_clock_class;
671 bt_uuid clock_class_uuid = NULL;
672
673 if (clock_class) {
674 clock_class_uuid = bt_clock_class_get_uuid(clock_class);
675 }
676
677 switch (iterator->clock_expectation.type) {
678 case CLOCK_EXPECTATION_UNSET:
679 /*
680 * This is the first time we see a message with a clock
681 * snapshot: record the properties of that clock, against
682 * which we'll compare the clock properties of the following
683 * messages.
684 */
685
686 if (!clock_class) {
687 iterator->clock_expectation.type = CLOCK_EXPECTATION_NONE;
688 } else if (bt_clock_class_origin_is_unix_epoch(clock_class)) {
689 iterator->clock_expectation.type = CLOCK_EXPECTATION_ORIGIN_UNIX;
690 } else if (clock_class_uuid) {
691 iterator->clock_expectation.type = CLOCK_EXPECTATION_ORIGIN_OTHER_UUID;
692 memcpy(iterator->clock_expectation.uuid, clock_class_uuid, BABELTRACE_UUID_LEN);
693 } else {
694 iterator->clock_expectation.type = CLOCK_EXPECTATION_ORIGIN_OTHER_NO_UUID;
695 }
696 break;
697
698 case CLOCK_EXPECTATION_NONE:
699 if (clock_class) {
f6f301d7 700 BT_ASSERT_POST_MSG("Expecting no clock class, got one: %![cc-]+K",
54b135a0
SM
701 clock_class);
702 result = false;
703 goto end;
704 }
705
706 break;
707
708 case CLOCK_EXPECTATION_ORIGIN_UNIX:
709 if (!clock_class) {
f6f301d7 710 BT_ASSERT_POST_MSG("Expecting a clock class, got none.");
54b135a0
SM
711 result = false;
712 goto end;
713 }
714
715 if (!bt_clock_class_origin_is_unix_epoch(clock_class)) {
f6f301d7 716 BT_ASSERT_POST_MSG("Expecting a clock class with Unix epoch origin: %![cc-]+K",
54b135a0
SM
717 clock_class);
718 result = false;
719 goto end;
720 }
721 break;
722
723 case CLOCK_EXPECTATION_ORIGIN_OTHER_UUID:
724 if (!clock_class) {
f6f301d7 725 BT_ASSERT_POST_MSG("Expecting a clock class, got none.");
54b135a0
SM
726 result = false;
727 goto end;
728 }
729
730 if (bt_clock_class_origin_is_unix_epoch(clock_class)) {
f6f301d7 731 BT_ASSERT_POST_MSG("Expecting a clock class without Unix epoch origin: %![cc-]+K",
54b135a0
SM
732 clock_class);
733 result = false;
734 goto end;
735 }
736
737 if (!clock_class_uuid) {
f6f301d7 738 BT_ASSERT_POST_MSG("Expecting a clock class with UUID: %![cc-]+K",
54b135a0
SM
739 clock_class);
740 result = false;
741 goto end;
742 }
743
744 if (bt_uuid_compare(iterator->clock_expectation.uuid, clock_class_uuid)) {
f6f301d7 745 BT_ASSERT_POST_MSG("Expecting a clock class with UUID, got one "
54b135a0
SM
746 "with a different UUID: %![cc-]+K, expected-uuid=%!u",
747 clock_class, iterator->clock_expectation.uuid);
748 result = false;
749 goto end;
750 }
751 break;
752
753 case CLOCK_EXPECTATION_ORIGIN_OTHER_NO_UUID:
754 if (!clock_class) {
f6f301d7 755 BT_ASSERT_POST_MSG("Expecting a clock class, got none.");
54b135a0
SM
756 result = false;
757 goto end;
758 }
759
760 if (bt_clock_class_origin_is_unix_epoch(clock_class)) {
f6f301d7 761 BT_ASSERT_POST_MSG("Expecting a clock class without Unix epoch origin: %![cc-]+K",
54b135a0
SM
762 clock_class);
763 result = false;
764 goto end;
765 }
766
767 if (clock_class_uuid) {
f6f301d7 768 BT_ASSERT_POST_MSG("Expecting a clock class without UUID: %![cc-]+K",
54b135a0
SM
769 clock_class);
770 result = false;
771 goto end;
772 }
773 break;
774 }
775 }
776
777 result = true;
778
779end:
780 return result;
781}
782
f6f301d7 783BT_ASSERT_POST_FUNC
54b135a0
SM
784static
785bool clock_classes_are_compatible(
786 struct bt_self_component_port_input_message_iterator *iterator,
787 bt_message_array_const msgs, uint64_t msg_count)
788{
789 uint64_t i;
790 bool result;
791
792 for (i = 0; i < msg_count; i++) {
793 if (!clock_classes_are_compatible_one(iterator, msgs[i])) {
794 result = false;
795 goto end;
796 }
797 }
798
799 result = true;
800
801end:
802 return result;
803}
804
805/*
806 * Call the `next` method of the iterator. Do some validation on the returned
807 * messages.
808 */
809
810static
d24d5663
PP
811enum bt_component_class_message_iterator_next_method_status
812call_iterator_next_method(
54b135a0
SM
813 struct bt_self_component_port_input_message_iterator *iterator,
814 bt_message_array_const msgs, uint64_t capacity, uint64_t *user_count)
815{
d24d5663 816 enum bt_component_class_message_iterator_next_method_status status;
54b135a0
SM
817
818 BT_ASSERT(iterator->methods.next);
819 BT_LOGD_STR("Calling user's \"next\" method.");
54b135a0 820 status = iterator->methods.next(iterator, msgs, capacity, user_count);
f6f301d7 821 BT_LOGD("User method returned: status=%s, msg-count=%" PRIu64,
d24d5663 822 bt_common_func_status_string(status), *user_count);
54b135a0 823
d24d5663 824 if (status == BT_FUNC_STATUS_OK) {
f6f301d7 825 BT_ASSERT_POST(clock_classes_are_compatible(iterator, msgs, *user_count),
54b135a0 826 "Clocks are not compatible");
f6f301d7 827 BT_ASSERT_POST(clock_snapshots_are_monotonic(iterator, msgs, *user_count),
54b135a0
SM
828 "Clock snapshots are not monotonic");
829 }
830
831 return status;
832}
833
d24d5663 834enum bt_message_iterator_next_status
d6e69534
PP
835bt_self_component_port_input_message_iterator_next(
836 struct bt_self_component_port_input_message_iterator *iterator,
837 bt_message_array_const *msgs, uint64_t *user_count)
3230ee6b 838{
d24d5663 839 enum bt_message_iterator_next_status status = BT_FUNC_STATUS_OK;
d94d92ac 840
d6e69534
PP
841 BT_ASSERT_PRE_NON_NULL(iterator, "Message iterator");
842 BT_ASSERT_PRE_NON_NULL(msgs, "Message array (output)");
843 BT_ASSERT_PRE_NON_NULL(user_count, "Message count (output)");
f42867e2 844 BT_ASSERT_PRE(iterator->state ==
d6e69534
PP
845 BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_ACTIVE,
846 "Message iterator's \"next\" called, but "
7474e7d3 847 "message iterator is in the wrong state: %!+i", iterator);
f42867e2
PP
848 BT_ASSERT(iterator->upstream_component);
849 BT_ASSERT(iterator->upstream_component->class);
5badd463
PP
850 BT_ASSERT_PRE(
851 bt_component_borrow_graph(iterator->upstream_component)->config_state !=
852 BT_GRAPH_CONFIGURATION_STATE_CONFIGURING,
4725a201
PP
853 "Graph is not configured: %!+g",
854 bt_component_borrow_graph(iterator->upstream_component));
d94d92ac 855 BT_LIB_LOGD("Getting next self component input port "
3f7d4d90
PP
856 "message iterator's messages: %!+i, batch-size=%u",
857 iterator, MSG_BATCH_SIZE);
d3eb6e8f 858
3230ee6b 859 /*
d6e69534 860 * Call the user's "next" method to get the next messages
fa054faf 861 * and status.
3230ee6b 862 */
3f7d4d90 863 *user_count = 0;
d24d5663 864 status = (int) call_iterator_next_method(iterator,
7474e7d3
PP
865 (void *) iterator->base.msgs->pdata, MSG_BATCH_SIZE,
866 user_count);
870631a2
PP
867 BT_LOGD("User method returned: status=%s, msg-count=%" PRIu64,
868 bt_common_func_status_string(status), *user_count);
d4393e08 869 if (status < 0) {
870631a2
PP
870 BT_LIB_LOGW_APPEND_CAUSE(
871 "Component input port message iterator's \"next\" method failed: "
872 "%![iter-]+i, status=%s",
873 iterator, bt_common_func_status_string(status));
f42867e2
PP
874 goto end;
875 }
3230ee6b 876
d0fea130
PP
877 /*
878 * There is no way that this iterator could have been finalized
879 * during its "next" method, as the only way to do this is to
880 * put the last iterator's reference, and this can only be done
881 * by its downstream owner.
7474e7d3
PP
882 *
883 * For the same reason, there is no way that this iterator could
884 * have seeked (cannot seek a self message iterator).
d0fea130
PP
885 */
886 BT_ASSERT(iterator->state ==
887 BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_ACTIVE);
8cf27cc5 888
d4393e08 889 switch (status) {
d24d5663 890 case BT_FUNC_STATUS_OK:
f6f301d7 891 BT_ASSERT_POST(*user_count <= MSG_BATCH_SIZE,
7474e7d3
PP
892 "Invalid returned message count: greater than "
893 "batch size: count=%" PRIu64 ", batch-size=%u",
894 *user_count, MSG_BATCH_SIZE);
d6e69534 895 *msgs = (void *) iterator->base.msgs->pdata;
d4393e08 896 break;
d24d5663 897 case BT_FUNC_STATUS_AGAIN:
d4393e08 898 goto end;
d24d5663 899 case BT_FUNC_STATUS_END:
d0fea130
PP
900 set_self_comp_port_input_msg_iterator_state(iterator,
901 BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_ENDED);
f42867e2 902 goto end;
f42867e2
PP
903 default:
904 /* Unknown non-error status */
905 abort();
41a2b7ae
PP
906 }
907
908end:
3230ee6b
PP
909 return status;
910}
911
d24d5663 912enum bt_message_iterator_next_status bt_port_output_message_iterator_next(
d6e69534
PP
913 struct bt_port_output_message_iterator *iterator,
914 bt_message_array_const *msgs_to_user,
d4393e08 915 uint64_t *count_to_user)
3230ee6b 916{
d24d5663
PP
917 enum bt_message_iterator_next_status status;
918 int graph_status;
3230ee6b 919
d6e69534
PP
920 BT_ASSERT_PRE_NON_NULL(iterator, "Message iterator");
921 BT_ASSERT_PRE_NON_NULL(msgs_to_user, "Message array (output)");
922 BT_ASSERT_PRE_NON_NULL(count_to_user, "Message count (output)");
923 BT_LIB_LOGD("Getting next output port message iterator's messages: "
07245ac2 924 "%!+i", iterator);
d94d92ac
PP
925 graph_status = bt_graph_consume_sink_no_check(iterator->graph,
926 iterator->colander);
07245ac2 927 switch (graph_status) {
d24d5663
PP
928 case BT_FUNC_STATUS_CANCELED:
929 case BT_FUNC_STATUS_AGAIN:
930 case BT_FUNC_STATUS_END:
931 case BT_FUNC_STATUS_MEMORY_ERROR:
d94d92ac 932 status = (int) graph_status;
07245ac2 933 break;
d24d5663
PP
934 case BT_FUNC_STATUS_OK:
935 status = BT_FUNC_STATUS_OK;
d4393e08
PP
936
937 /*
d6e69534 938 * On success, the colander sink moves the messages
d4393e08 939 * to this iterator's array and sets this iterator's
d6e69534 940 * message count: move them to the user.
d4393e08 941 */
d6e69534 942 *msgs_to_user = (void *) iterator->base.msgs->pdata;
d94d92ac 943 *count_to_user = iterator->count;
90157d89 944 break;
90157d89 945 default:
07245ac2 946 /* Other errors */
d24d5663 947 status = BT_FUNC_STATUS_ERROR;
90157d89 948 }
3230ee6b 949
3230ee6b 950 return status;
53d45b87
JG
951}
952
7474e7d3
PP
953struct bt_component *
954bt_self_component_port_input_message_iterator_borrow_component(
d6e69534 955 struct bt_self_component_port_input_message_iterator *iterator)
d94d92ac 956{
d6e69534 957 BT_ASSERT_PRE_NON_NULL(iterator, "Message iterator");
d94d92ac
PP
958 return iterator->upstream_component;
959}
960
7474e7d3
PP
961const struct bt_component *
962bt_self_component_port_input_message_iterator_borrow_component_const(
963 const struct bt_self_component_port_input_message_iterator *iterator)
964{
965 BT_ASSERT_PRE_NON_NULL(iterator, "Message iterator");
966 return iterator->upstream_component;
967}
968
d6e69534
PP
969struct bt_self_component *bt_self_message_iterator_borrow_component(
970 struct bt_self_message_iterator *self_iterator)
413bc2c4 971{
d6e69534 972 struct bt_self_component_port_input_message_iterator *iterator =
d94d92ac 973 (void *) self_iterator;
90157d89 974
d6e69534 975 BT_ASSERT_PRE_NON_NULL(iterator, "Message iterator");
d94d92ac 976 return (void *) iterator->upstream_component;
413bc2c4
JG
977}
978
d6e69534
PP
979struct bt_self_port_output *bt_self_message_iterator_borrow_port(
980 struct bt_self_message_iterator *self_iterator)
91457551 981{
d6e69534 982 struct bt_self_component_port_input_message_iterator *iterator =
d94d92ac
PP
983 (void *) self_iterator;
984
d6e69534 985 BT_ASSERT_PRE_NON_NULL(iterator, "Message iterator");
d94d92ac 986 return (void *) iterator->upstream_port;
91457551 987}
8ed535b5
PP
988
989static
d6e69534 990void bt_port_output_message_iterator_destroy(struct bt_object *obj)
8ed535b5 991{
d6e69534 992 struct bt_port_output_message_iterator *iterator = (void *) obj;
8ed535b5 993
3f7d4d90 994 BT_LIB_LOGI("Destroying output port message iterator object: %!+i",
8ed535b5
PP
995 iterator);
996 BT_LOGD_STR("Putting graph.");
d94d92ac 997 BT_OBJECT_PUT_REF_AND_RESET(iterator->graph);
8ed535b5 998 BT_LOGD_STR("Putting colander sink component.");
d94d92ac 999 BT_OBJECT_PUT_REF_AND_RESET(iterator->colander);
d6e69534 1000 destroy_base_message_iterator(obj);
8ed535b5
PP
1001}
1002
d6e69534 1003struct bt_port_output_message_iterator *
7474e7d3 1004bt_port_output_message_iterator_create(struct bt_graph *graph,
0d72b8c3 1005 const struct bt_port_output *output_port)
8ed535b5 1006{
d6e69534 1007 struct bt_port_output_message_iterator *iterator = NULL;
d94d92ac 1008 struct bt_component_class_sink *colander_comp_cls = NULL;
8ed535b5 1009 struct bt_component *output_port_comp = NULL;
d94d92ac 1010 struct bt_component_sink *colander_comp;
d24d5663 1011 int graph_status;
d94d92ac 1012 struct bt_port_input *colander_in_port = NULL;
8ed535b5 1013 struct bt_component_class_sink_colander_data colander_data;
d4393e08 1014 int ret;
8ed535b5 1015
d94d92ac 1016 BT_ASSERT_PRE_NON_NULL(graph, "Graph");
f42867e2 1017 BT_ASSERT_PRE_NON_NULL(output_port, "Output port");
0d72b8c3
PP
1018 output_port_comp = bt_port_borrow_component_inline(
1019 (const void *) output_port);
f42867e2
PP
1020 BT_ASSERT_PRE(output_port_comp,
1021 "Output port has no component: %!+p", output_port);
d94d92ac
PP
1022 BT_ASSERT_PRE(bt_component_borrow_graph(output_port_comp) ==
1023 (void *) graph,
1024 "Output port is not part of graph: %![graph-]+g, %![port-]+p",
1025 graph, output_port);
2bcba397
PP
1026 BT_ASSERT_PRE(!graph->has_sink,
1027 "Graph already has a sink component: %![graph-]+g");
8ed535b5 1028
d6e69534 1029 /* Create message iterator */
3f7d4d90 1030 BT_LIB_LOGI("Creating message iterator on output port: "
d94d92ac 1031 "%![port-]+p, %![comp-]+c", output_port, output_port_comp);
d6e69534 1032 iterator = g_new0(struct bt_port_output_message_iterator, 1);
8ed535b5 1033 if (!iterator) {
870631a2
PP
1034 BT_LIB_LOGE_APPEND_CAUSE(
1035 "Failed to allocate one output port message iterator.");
8ed535b5
PP
1036 goto error;
1037 }
1038
d6e69534
PP
1039 ret = init_message_iterator((void *) iterator,
1040 BT_MESSAGE_ITERATOR_TYPE_PORT_OUTPUT,
1041 bt_port_output_message_iterator_destroy);
d4393e08 1042 if (ret) {
d6e69534 1043 /* init_message_iterator() logs errors */
65300d60 1044 BT_OBJECT_PUT_REF_AND_RESET(iterator);
d4393e08
PP
1045 goto end;
1046 }
8ed535b5
PP
1047
1048 /* Create colander component */
1049 colander_comp_cls = bt_component_class_sink_colander_get();
1050 if (!colander_comp_cls) {
870631a2
PP
1051 /* bt_component_class_sink_colander_get() logs errors */
1052 BT_LIB_LOGE_APPEND_CAUSE(
1053 "Cannot get colander sink component class.");
8ed535b5
PP
1054 goto error;
1055 }
1056
398454ed
PP
1057 iterator->graph = graph;
1058 bt_object_get_no_null_check(iterator->graph);
d6e69534 1059 colander_data.msgs = (void *) iterator->base.msgs->pdata;
d4393e08 1060 colander_data.count_addr = &iterator->count;
5fd91d88 1061
e874da19
PP
1062 /*
1063 * Hope that nobody uses this very unique name.
1064 *
1065 * We pass `BT_LOGGING_LEVEL_NONE` but the colander component
1066 * class module does not use this level anyway since it belongs
1067 * to the library.
1068 */
d24d5663 1069 graph_status = bt_graph_add_sink_component_with_init_method_data(
5fd91d88
PP
1070 (void *) graph, colander_comp_cls,
1071 "colander-36ac3409-b1a8-4d60-ab1f-4fdf341a8fb1",
e874da19
PP
1072 NULL, &colander_data, BT_LOGGING_LEVEL_NONE,
1073 (void *) &iterator->colander);
d24d5663 1074 if (graph_status != BT_FUNC_STATUS_OK) {
870631a2
PP
1075 BT_LIB_LOGE_APPEND_CAUSE(
1076 "Cannot add colander sink component to graph: "
1077 "%![graph-]+g, status=%s", graph,
d24d5663 1078 bt_common_func_status_string(graph_status));
8ed535b5
PP
1079 goto error;
1080 }
1081
1082 /*
1083 * Connect provided output port to the colander component's
1084 * input port.
1085 */
0d72b8c3
PP
1086 colander_in_port =
1087 (void *) bt_component_sink_borrow_input_port_by_index_const(
1088 (void *) iterator->colander, 0);
f6ccaed9 1089 BT_ASSERT(colander_in_port);
0d72b8c3 1090 graph_status = bt_graph_connect_ports(graph,
8ed535b5 1091 output_port, colander_in_port, NULL);
d24d5663 1092 if (graph_status != BT_FUNC_STATUS_OK) {
870631a2
PP
1093 BT_LIB_LOGW_APPEND_CAUSE(
1094 "Cannot connect colander sink's port: "
d94d92ac
PP
1095 "%![graph-]+g, %![comp-]+c, status=%s", graph,
1096 iterator->colander,
d24d5663 1097 bt_common_func_status_string(graph_status));
8ed535b5
PP
1098 goto error;
1099 }
1100
1101 /*
1102 * At this point everything went fine. Make the graph
d6e69534 1103 * nonconsumable forever so that only this message iterator
8ed535b5 1104 * can consume (thanks to bt_graph_consume_sink_no_check()).
d6e69534
PP
1105 * This avoids leaking the message created by the colander
1106 * sink and moved to the message iterator's message
07245ac2 1107 * member.
8ed535b5 1108 */
d94d92ac 1109 bt_graph_set_can_consume(iterator->graph, false);
5badd463 1110
36d1acad
SM
1111 /* Also set the graph as being configured. */
1112 graph_status = bt_graph_configure(graph);
d24d5663 1113 if (graph_status != BT_FUNC_STATUS_OK) {
870631a2
PP
1114 BT_LIB_LOGW_APPEND_CAUSE(
1115 "Cannot configure graph after having "
1116 "added and connected colander sink: "
1117 "%![graph-]+g, %![comp-]+c, status=%s", graph,
1118 iterator->colander,
d24d5663 1119 bt_common_func_status_string(graph_status));
36d1acad
SM
1120 goto error;
1121 }
8ed535b5
PP
1122 goto end;
1123
1124error:
1125 if (iterator && iterator->graph && iterator->colander) {
1126 int ret;
1127
1128 /* Remove created colander component from graph if any */
1129 colander_comp = iterator->colander;
65300d60 1130 BT_OBJECT_PUT_REF_AND_RESET(iterator->colander);
8ed535b5
PP
1131
1132 /*
1133 * At this point the colander component's reference
1134 * count is 0 because iterator->colander was the only
1135 * owner. We also know that it is not connected because
1136 * this is the last operation before this function
1137 * succeeds.
1138 *
1139 * Since we honor the preconditions here,
1140 * bt_graph_remove_unconnected_component() always
1141 * succeeds.
1142 */
1143 ret = bt_graph_remove_unconnected_component(iterator->graph,
d94d92ac 1144 (void *) colander_comp);
f6ccaed9 1145 BT_ASSERT(ret == 0);
8ed535b5
PP
1146 }
1147
65300d60 1148 BT_OBJECT_PUT_REF_AND_RESET(iterator);
8ed535b5
PP
1149
1150end:
65300d60 1151 bt_object_put_ref(colander_comp_cls);
8ed535b5
PP
1152 return (void *) iterator;
1153}
c5b9b441 1154
7474e7d3
PP
1155bt_bool bt_self_component_port_input_message_iterator_can_seek_ns_from_origin(
1156 struct bt_self_component_port_input_message_iterator *iterator,
1157 int64_t ns_from_origin)
1158{
1159 bt_bool can = BT_FALSE;
1160
1161 BT_ASSERT_PRE_NON_NULL(iterator, "Message iterator");
1162 BT_ASSERT_PRE_ITER_HAS_STATE_TO_SEEK(iterator);
5badd463
PP
1163 BT_ASSERT_PRE(
1164 bt_component_borrow_graph(iterator->upstream_component)->config_state !=
1165 BT_GRAPH_CONFIGURATION_STATE_CONFIGURING,
7474e7d3
PP
1166 "Graph is not configured: %!+g",
1167 bt_component_borrow_graph(iterator->upstream_component));
1168
1169 if (iterator->methods.can_seek_ns_from_origin) {
1170 can = iterator->methods.can_seek_ns_from_origin(iterator,
1171 ns_from_origin);
1172 goto end;
1173 }
1174
1175 /*
1176 * Automatic seeking fall back: if we can seek to the beginning,
1177 * then we can automatically seek to any message.
1178 */
1179 if (iterator->methods.can_seek_beginning) {
1180 can = iterator->methods.can_seek_beginning(iterator);
1181 }
1182
1183end:
1184 return can;
1185}
1186
1187bt_bool bt_self_component_port_input_message_iterator_can_seek_beginning(
1188 struct bt_self_component_port_input_message_iterator *iterator)
1189{
1190 bt_bool can = BT_FALSE;
1191
1192 BT_ASSERT_PRE_NON_NULL(iterator, "Message iterator");
1193 BT_ASSERT_PRE_ITER_HAS_STATE_TO_SEEK(iterator);
5badd463
PP
1194 BT_ASSERT_PRE(
1195 bt_component_borrow_graph(iterator->upstream_component)->config_state !=
1196 BT_GRAPH_CONFIGURATION_STATE_CONFIGURING,
7474e7d3
PP
1197 "Graph is not configured: %!+g",
1198 bt_component_borrow_graph(iterator->upstream_component));
1199
1200 if (iterator->methods.can_seek_beginning) {
1201 can = iterator->methods.can_seek_beginning(iterator);
1202 }
1203
1204 return can;
1205}
1206
1207static inline
003e713f 1208void set_iterator_state_after_seeking(
7474e7d3 1209 struct bt_self_component_port_input_message_iterator *iterator,
d24d5663 1210 int status)
7474e7d3
PP
1211{
1212 enum bt_self_component_port_input_message_iterator_state new_state = 0;
1213
1214 /* Set iterator's state depending on seeking status */
1215 switch (status) {
d24d5663 1216 case BT_FUNC_STATUS_OK:
7474e7d3
PP
1217 new_state = BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_ACTIVE;
1218 break;
d24d5663 1219 case BT_FUNC_STATUS_AGAIN:
7474e7d3
PP
1220 new_state = BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_LAST_SEEKING_RETURNED_AGAIN;
1221 break;
d24d5663
PP
1222 case BT_FUNC_STATUS_ERROR:
1223 case BT_FUNC_STATUS_MEMORY_ERROR:
7474e7d3
PP
1224 new_state = BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_LAST_SEEKING_RETURNED_ERROR;
1225 break;
d24d5663 1226 case BT_FUNC_STATUS_END:
7474e7d3
PP
1227 new_state = BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_ENDED;
1228 break;
1229 default:
1230 abort();
1231 }
1232
1233 set_self_comp_port_input_msg_iterator_state(iterator, new_state);
1234}
1235
54b135a0
SM
1236static
1237void reset_iterator_expectations(
1238 struct bt_self_component_port_input_message_iterator *iterator)
1239{
1240 iterator->last_ns_from_origin = INT64_MIN;
1241 iterator->clock_expectation.type = CLOCK_EXPECTATION_UNSET;
1242}
1243
d24d5663 1244enum bt_message_iterator_seek_beginning_status
7474e7d3
PP
1245bt_self_component_port_input_message_iterator_seek_beginning(
1246 struct bt_self_component_port_input_message_iterator *iterator)
1247{
9275bef4 1248 int status;
7474e7d3
PP
1249
1250 BT_ASSERT_PRE_NON_NULL(iterator, "Message iterator");
1251 BT_ASSERT_PRE_ITER_HAS_STATE_TO_SEEK(iterator);
5badd463
PP
1252 BT_ASSERT_PRE(
1253 bt_component_borrow_graph(iterator->upstream_component)->config_state !=
1254 BT_GRAPH_CONFIGURATION_STATE_CONFIGURING,
7474e7d3
PP
1255 "Graph is not configured: %!+g",
1256 bt_component_borrow_graph(iterator->upstream_component));
1257 BT_ASSERT_PRE(
1258 bt_self_component_port_input_message_iterator_can_seek_beginning(
1259 iterator),
1260 "Message iterator cannot seek beginning: %!+i", iterator);
54b135a0
SM
1261
1262 /*
1263 * We are seeking, reset our expectations about how the following
1264 * messages should look like.
1265 */
1266 reset_iterator_expectations(iterator);
1267
7474e7d3
PP
1268 BT_LIB_LOGD("Calling user's \"seek beginning\" method: %!+i", iterator);
1269 set_self_comp_port_input_msg_iterator_state(iterator,
1270 BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_SEEKING);
1271 status = iterator->methods.seek_beginning(iterator);
1272 BT_LOGD("User method returned: status=%s",
d24d5663
PP
1273 bt_common_func_status_string(status));
1274 BT_ASSERT_POST(status == BT_FUNC_STATUS_OK ||
1275 status == BT_FUNC_STATUS_ERROR ||
1276 status == BT_FUNC_STATUS_MEMORY_ERROR ||
1277 status == BT_FUNC_STATUS_AGAIN,
7474e7d3 1278 "Unexpected status: %![iter-]+i, status=%s",
d24d5663 1279 iterator, bt_common_func_status_string(status));
870631a2
PP
1280 if (status < 0) {
1281 BT_LIB_LOGW_APPEND_CAUSE(
1282 "Component input port message iterator's \"seek beginning\" method failed: "
1283 "%![iter-]+i, status=%s",
1284 iterator, bt_common_func_status_string(status));
1285 }
1286
7474e7d3
PP
1287 set_iterator_state_after_seeking(iterator, status);
1288 return status;
1289}
1290
5b7b55be
SM
1291/*
1292 * Structure used to record the state of a given stream during the fast-forward
1293 * phase of an auto-seek.
1294 */
1295struct auto_seek_stream_state {
1296 /*
1297 * Value representing which step of this timeline we are at.
1298 *
1299 * time --->
1300 * [SB] 1 [SAB] 2 [PB] 3 [PE] 2 [SAE] 1 [SE]
1301 *
1302 * At each point in the timeline, the messages we need to replicate are:
1303 *
1304 * 1: Stream beginning
1305 * 2: Stream beginning, stream activity beginning
1306 * 3: Stream beginning, stream activity beginning, packet beginning
1307 *
1308 * Before "Stream beginning" and after "Stream end", we don't need to
1309 * replicate anything as the stream doesn't exist.
1310 */
1311 enum {
1312 AUTO_SEEK_STREAM_STATE_STREAM_BEGAN,
1313 AUTO_SEEK_STREAM_STATE_STREAM_ACTIVITY_BEGAN,
1314 AUTO_SEEK_STREAM_STATE_PACKET_BEGAN,
1315 } state;
1316
1317 /*
1318 * If `state` is AUTO_SEEK_STREAM_STATE_PACKET_BEGAN, the packet we are
1319 * in. This is a weak reference, since the packet will always be
1320 * alive by the time we use it.
1321 */
1322 struct bt_packet *packet;
1323};
1324
1325static
1326struct auto_seek_stream_state *create_auto_seek_stream_state(void)
1327{
1328 return g_new0(struct auto_seek_stream_state, 1);
1329}
1330
1331static
1332void destroy_auto_seek_stream_state(void *ptr)
1333{
1334 g_free(ptr);
1335}
1336
1337static
1338GHashTable *create_auto_seek_stream_states(void)
1339{
1340 return g_hash_table_new_full(g_direct_hash, g_direct_equal, NULL,
1341 destroy_auto_seek_stream_state);
1342}
1343
1344static
1345void destroy_auto_seek_stream_states(GHashTable *stream_states)
1346{
1347 g_hash_table_destroy(stream_states);
1348}
1349
1350/*
1351 * Handle one message while we are in the fast-forward phase of an auto-seek.
1352 *
1353 * Sets `*got_first` to true if the message's timestamp is greater or equal to
1354 * `ns_from_origin`. In other words, if this is the first message after our
1355 * seek point.
1356 *
1357 * `stream_states` is an hash table of `bt_stream *` (weak reference) to
1358 * `struct auto_seek_stream_state` used to keep the state of each stream
1359 * during the fast-forward.
1360 */
1361
7474e7d3 1362static inline
d24d5663 1363int auto_seek_handle_message(
5b9e151d
PP
1364 struct bt_self_component_port_input_message_iterator *iterator,
1365 int64_t ns_from_origin, const struct bt_message *msg,
5b7b55be 1366 bool *got_first, GHashTable *stream_states)
7474e7d3 1367{
d24d5663 1368 int status = BT_FUNC_STATUS_OK;
5b9e151d 1369 int64_t msg_ns_from_origin;
7474e7d3 1370 const struct bt_clock_snapshot *clk_snapshot = NULL;
5b9e151d
PP
1371 int ret;
1372
1373 BT_ASSERT(msg);
1374 BT_ASSERT(got_first);
7474e7d3
PP
1375
1376 switch (msg->type) {
1377 case BT_MESSAGE_TYPE_EVENT:
1378 {
1379 const struct bt_message_event *event_msg =
1380 (const void *) msg;
1381
2c091c04 1382 clk_snapshot = event_msg->default_cs;
f6f301d7 1383 BT_ASSERT_POST(clk_snapshot,
c7072d5a
PP
1384 "Event message has no default clock snapshot: %!+n",
1385 event_msg);
7474e7d3
PP
1386 break;
1387 }
b9fd9cbb 1388 case BT_MESSAGE_TYPE_MESSAGE_ITERATOR_INACTIVITY:
7474e7d3 1389 {
b9fd9cbb 1390 const struct bt_message_message_iterator_inactivity *inactivity_msg =
7474e7d3
PP
1391 (const void *) msg;
1392
7474e7d3 1393 clk_snapshot = inactivity_msg->default_cs;
16663a5e 1394 BT_ASSERT(clk_snapshot);
7474e7d3
PP
1395 break;
1396 }
16663a5e
PP
1397 case BT_MESSAGE_TYPE_PACKET_BEGINNING:
1398 case BT_MESSAGE_TYPE_PACKET_END:
c7072d5a
PP
1399 {
1400 const struct bt_message_packet *packet_msg =
1401 (const void *) msg;
1402
1403 clk_snapshot = packet_msg->default_cs;
f6f301d7 1404 BT_ASSERT_POST(clk_snapshot,
c7072d5a
PP
1405 "Packet message has no default clock snapshot: %!+n",
1406 packet_msg);
1407 break;
1408 }
16663a5e
PP
1409 case BT_MESSAGE_TYPE_DISCARDED_EVENTS:
1410 case BT_MESSAGE_TYPE_DISCARDED_PACKETS:
7474e7d3 1411 {
5b9e151d
PP
1412 struct bt_message_discarded_items *msg_disc_items =
1413 (void *) msg;
1414
f6f301d7 1415 BT_ASSERT_POST(msg_disc_items->default_begin_cs &&
5b9e151d
PP
1416 msg_disc_items->default_end_cs,
1417 "Discarded events/packets message has no default clock snapshots: %!+n",
1418 msg_disc_items);
1419 ret = bt_clock_snapshot_get_ns_from_origin(
1420 msg_disc_items->default_begin_cs,
1421 &msg_ns_from_origin);
1422 if (ret) {
d24d5663 1423 status = BT_FUNC_STATUS_ERROR;
5b9e151d
PP
1424 goto end;
1425 }
7474e7d3 1426
5b9e151d
PP
1427 if (msg_ns_from_origin >= ns_from_origin) {
1428 *got_first = true;
1429 goto push_msg;
1430 }
1431
1432 ret = bt_clock_snapshot_get_ns_from_origin(
1433 msg_disc_items->default_end_cs,
1434 &msg_ns_from_origin);
1435 if (ret) {
d24d5663 1436 status = BT_FUNC_STATUS_ERROR;
5b9e151d
PP
1437 goto end;
1438 }
1439
1440 if (msg_ns_from_origin >= ns_from_origin) {
1441 /*
1442 * The discarded items message's beginning time
1443 * is before the requested seeking time, but its
1444 * end time is after. Modify the message so as
1445 * to set its beginning time to the requested
1446 * seeking time, and make its item count unknown
1447 * as we don't know if items were really
1448 * discarded within the new time range.
1449 */
1450 uint64_t new_begin_raw_value;
1451
1452 ret = bt_clock_class_clock_value_from_ns_from_origin(
1453 msg_disc_items->default_end_cs->clock_class,
1454 ns_from_origin, &new_begin_raw_value);
1455 if (ret) {
d24d5663 1456 status = BT_FUNC_STATUS_ERROR;
5b9e151d
PP
1457 goto end;
1458 }
1459
1460 bt_clock_snapshot_set_raw_value(
1461 msg_disc_items->default_begin_cs,
1462 new_begin_raw_value);
1463 msg_disc_items->count.base.avail =
1464 BT_PROPERTY_AVAILABILITY_NOT_AVAILABLE;
1465
1466 /*
1467 * It is safe to push it because its beginning
1468 * time is exactly the requested seeking time.
1469 */
1470 goto push_msg;
1471 } else {
1472 goto skip_msg;
1473 }
7474e7d3 1474 }
16663a5e 1475 case BT_MESSAGE_TYPE_STREAM_ACTIVITY_BEGINNING:
7474e7d3 1476 {
16663a5e 1477 const struct bt_message_stream_activity *stream_act_msg =
7474e7d3
PP
1478 (const void *) msg;
1479
16663a5e
PP
1480 switch (stream_act_msg->default_cs_state) {
1481 case BT_MESSAGE_STREAM_ACTIVITY_CLOCK_SNAPSHOT_STATE_UNKNOWN:
1482 case BT_MESSAGE_STREAM_ACTIVITY_CLOCK_SNAPSHOT_STATE_INFINITE:
1483 /*
1484 * -inf is always less than any requested time,
1485 * and we can't assume any specific time for an
1486 * unknown clock snapshot, so skip this.
1487 */
5b9e151d 1488 goto skip_msg;
16663a5e
PP
1489 case BT_MESSAGE_STREAM_ACTIVITY_CLOCK_SNAPSHOT_STATE_KNOWN:
1490 clk_snapshot = stream_act_msg->default_cs;
1491 BT_ASSERT(clk_snapshot);
1492 break;
1493 default:
1494 abort();
1495 }
1496
7474e7d3
PP
1497 break;
1498 }
16663a5e
PP
1499 case BT_MESSAGE_TYPE_STREAM_ACTIVITY_END:
1500 {
1501 const struct bt_message_stream_activity *stream_act_msg =
1502 (const void *) msg;
1503
1504 switch (stream_act_msg->default_cs_state) {
1505 case BT_MESSAGE_STREAM_ACTIVITY_CLOCK_SNAPSHOT_STATE_UNKNOWN:
1506 /*
1507 * We can't assume any specific time for an
1508 * unknown clock snapshot, so skip this.
1509 */
5b9e151d 1510 goto skip_msg;
16663a5e
PP
1511 case BT_MESSAGE_STREAM_ACTIVITY_CLOCK_SNAPSHOT_STATE_INFINITE:
1512 /*
1513 * +inf is always greater than any requested
1514 * time.
1515 */
5b9e151d
PP
1516 *got_first = true;
1517 goto push_msg;
16663a5e
PP
1518 case BT_MESSAGE_STREAM_ACTIVITY_CLOCK_SNAPSHOT_STATE_KNOWN:
1519 clk_snapshot = stream_act_msg->default_cs;
1520 BT_ASSERT(clk_snapshot);
1521 break;
1522 default:
1523 abort();
1524 }
1525
1526 break;
1527 }
1528 case BT_MESSAGE_TYPE_STREAM_BEGINNING:
1529 case BT_MESSAGE_TYPE_STREAM_END:
1530 /* Ignore */
5b9e151d 1531 goto skip_msg;
7474e7d3
PP
1532 default:
1533 abort();
1534 }
1535
5b9e151d
PP
1536 BT_ASSERT(clk_snapshot);
1537 ret = bt_clock_snapshot_get_ns_from_origin(clk_snapshot,
1538 &msg_ns_from_origin);
1539 if (ret) {
d24d5663 1540 status = BT_FUNC_STATUS_ERROR;
7474e7d3
PP
1541 goto end;
1542 }
1543
5b9e151d
PP
1544 if (msg_ns_from_origin >= ns_from_origin) {
1545 *got_first = true;
1546 goto push_msg;
1547 }
1548
1549skip_msg:
5b7b55be
SM
1550 /* This message won't be sent downstream. */
1551 switch (msg->type) {
1552 case BT_MESSAGE_TYPE_STREAM_BEGINNING:
1553 {
1554 const struct bt_message_stream *stream_msg = (const void *) msg;
1555 struct auto_seek_stream_state *stream_state;
5b7b55be
SM
1556
1557 /* Update stream's state: stream began. */
1558 stream_state = create_auto_seek_stream_state();
1559 if (!stream_state) {
d24d5663 1560 status = BT_FUNC_STATUS_MEMORY_ERROR;
5b7b55be
SM
1561 goto end;
1562 }
1563
1564 stream_state->state = AUTO_SEEK_STREAM_STATE_STREAM_BEGAN;
e74dbb33
SM
1565
1566 BT_ASSERT(!bt_g_hash_table_contains(stream_states, stream_msg->stream));
1567 g_hash_table_insert(stream_states, stream_msg->stream, stream_state);
5b7b55be
SM
1568 break;
1569 }
1570 case BT_MESSAGE_TYPE_STREAM_ACTIVITY_BEGINNING:
1571 {
1572 const struct bt_message_stream_activity *stream_act_msg =
1573 (const void *) msg;
1574 struct auto_seek_stream_state *stream_state;
1575
1576 /* Update stream's state: stream activity began. */
1577 stream_state = g_hash_table_lookup(stream_states, stream_act_msg->stream);
1578 BT_ASSERT(stream_state);
1579
1580 BT_ASSERT(stream_state->state == AUTO_SEEK_STREAM_STATE_STREAM_BEGAN);
1581 stream_state->state = AUTO_SEEK_STREAM_STATE_STREAM_ACTIVITY_BEGAN;
1582 BT_ASSERT(!stream_state->packet);
1583 break;
1584 }
1585 case BT_MESSAGE_TYPE_PACKET_BEGINNING:
1586 {
1587 const struct bt_message_packet *packet_msg =
1588 (const void *) msg;
1589 struct auto_seek_stream_state *stream_state;
1590
1591 /* Update stream's state: packet began. */
1592 stream_state = g_hash_table_lookup(stream_states, packet_msg->packet->stream);
1593 BT_ASSERT(stream_state);
1594
1595 BT_ASSERT(stream_state->state == AUTO_SEEK_STREAM_STATE_STREAM_ACTIVITY_BEGAN);
1596 stream_state->state = AUTO_SEEK_STREAM_STATE_PACKET_BEGAN;
1597 BT_ASSERT(!stream_state->packet);
1598 stream_state->packet = packet_msg->packet;
1599 break;
1600 }
1601 case BT_MESSAGE_TYPE_PACKET_END:
1602 {
1603 const struct bt_message_packet *packet_msg =
1604 (const void *) msg;
1605 struct auto_seek_stream_state *stream_state;
1606
1607 /* Update stream's state: packet ended. */
1608 stream_state = g_hash_table_lookup(stream_states, packet_msg->packet->stream);
1609 BT_ASSERT(stream_state);
1610
1611 BT_ASSERT(stream_state->state == AUTO_SEEK_STREAM_STATE_PACKET_BEGAN);
1612 stream_state->state = AUTO_SEEK_STREAM_STATE_STREAM_ACTIVITY_BEGAN;
1613 BT_ASSERT(stream_state->packet);
1614 stream_state->packet = NULL;
1615 break;
1616 }
1617 case BT_MESSAGE_TYPE_STREAM_ACTIVITY_END:
1618 {
1619 const struct bt_message_stream_activity *stream_act_msg =
1620 (const void *) msg;
1621 struct auto_seek_stream_state *stream_state;
1622
1623 /* Update stream's state: stream activity ended. */
1624 stream_state = g_hash_table_lookup(stream_states, stream_act_msg->stream);
1625 BT_ASSERT(stream_state);
1626
1627 BT_ASSERT(stream_state->state == AUTO_SEEK_STREAM_STATE_STREAM_ACTIVITY_BEGAN);
1628 stream_state->state = AUTO_SEEK_STREAM_STATE_STREAM_BEGAN;
1629 BT_ASSERT(!stream_state->packet);
1630 break;
1631 }
1632 case BT_MESSAGE_TYPE_STREAM_END:
1633 {
1634 const struct bt_message_stream *stream_msg = (const void *) msg;
1635 struct auto_seek_stream_state *stream_state;
1636
1637 stream_state = g_hash_table_lookup(stream_states, stream_msg->stream);
1638 BT_ASSERT(stream_state);
1639 BT_ASSERT(stream_state->state == AUTO_SEEK_STREAM_STATE_STREAM_BEGAN);
1640 BT_ASSERT(!stream_state->packet);
1641
1642 /* Update stream's state: this stream doesn't exist anymore. */
1643 g_hash_table_remove(stream_states, stream_msg->stream);
1644 break;
1645 }
1646 default:
1647 break;
1648 }
1649
5b9e151d 1650 bt_object_put_no_null_check(msg);
04c0cec6 1651 msg = NULL;
5b9e151d
PP
1652 goto end;
1653
1654push_msg:
da9c4c52 1655 g_queue_push_tail(iterator->auto_seek.msgs, (void *) msg);
5b9e151d 1656 msg = NULL;
7474e7d3
PP
1657
1658end:
d24d5663 1659 BT_ASSERT(!msg || status != BT_FUNC_STATUS_OK);
5b9e151d 1660 return status;
7474e7d3
PP
1661}
1662
1663static
d24d5663 1664int find_message_ge_ns_from_origin(
7474e7d3 1665 struct bt_self_component_port_input_message_iterator *iterator,
5b7b55be 1666 int64_t ns_from_origin, GHashTable *stream_states)
7474e7d3
PP
1667{
1668 int status;
1669 enum bt_self_component_port_input_message_iterator_state init_state =
1670 iterator->state;
1671 const struct bt_message *messages[MSG_BATCH_SIZE];
1672 uint64_t user_count = 0;
1673 uint64_t i;
5b9e151d 1674 bool got_first = false;
7474e7d3
PP
1675
1676 BT_ASSERT(iterator);
1677 memset(&messages[0], 0, sizeof(messages[0]) * MSG_BATCH_SIZE);
1678
1679 /*
1680 * Make this iterator temporarily active (not seeking) to call
1681 * the "next" method.
1682 */
1683 set_self_comp_port_input_msg_iterator_state(iterator,
1684 BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_ACTIVE);
1685
1686 BT_ASSERT(iterator->methods.next);
1687
e0dade92 1688 while (!got_first) {
7474e7d3
PP
1689 /*
1690 * Call the user's "next" method to get the next
1691 * messages and status.
1692 */
54b135a0 1693 status = call_iterator_next_method(iterator,
7474e7d3 1694 &messages[0], MSG_BATCH_SIZE, &user_count);
870631a2
PP
1695 BT_LOGD("User method returned: status=%s",
1696 bt_common_func_status_string(status));
1697 if (status < 0) {
1698 BT_LIB_LOGW_APPEND_CAUSE(
1699 "Component input port message iterator's \"next\" method failed: "
1700 "%![iter-]+i, status=%s",
1701 iterator, bt_common_func_status_string(status));
1702 }
7474e7d3 1703
7474e7d3
PP
1704 /*
1705 * The user's "next" method must not do any action which
1706 * would change the iterator's state.
1707 */
1708 BT_ASSERT(iterator->state ==
1709 BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_ACTIVE);
7474e7d3
PP
1710
1711 switch (status) {
d24d5663 1712 case BT_FUNC_STATUS_OK:
f6f301d7 1713 BT_ASSERT_POST(user_count <= MSG_BATCH_SIZE,
7474e7d3
PP
1714 "Invalid returned message count: greater than "
1715 "batch size: count=%" PRIu64 ", batch-size=%u",
1716 user_count, MSG_BATCH_SIZE);
1717 break;
d24d5663
PP
1718 case BT_FUNC_STATUS_AGAIN:
1719 case BT_FUNC_STATUS_ERROR:
1720 case BT_FUNC_STATUS_MEMORY_ERROR:
1721 case BT_FUNC_STATUS_END:
7474e7d3
PP
1722 goto end;
1723 default:
1724 abort();
1725 }
1726
7474e7d3 1727 for (i = 0; i < user_count; i++) {
5b9e151d 1728 if (got_first) {
da9c4c52 1729 g_queue_push_tail(iterator->auto_seek.msgs,
5b9e151d
PP
1730 (void *) messages[i]);
1731 messages[i] = NULL;
7474e7d3
PP
1732 continue;
1733 }
1734
5b9e151d 1735 status = auto_seek_handle_message(iterator,
5b7b55be
SM
1736 ns_from_origin, messages[i], &got_first,
1737 stream_states);
d24d5663 1738 if (status == BT_FUNC_STATUS_OK) {
e0dade92 1739 /* Message was either pushed or moved */
5b9e151d
PP
1740 messages[i] = NULL;
1741 } else {
7474e7d3
PP
1742 goto end;
1743 }
7474e7d3
PP
1744 }
1745 }
1746
1747end:
1748 for (i = 0; i < user_count; i++) {
1749 if (messages[i]) {
1750 bt_object_put_no_null_check(messages[i]);
1751 }
1752 }
1753
1754 set_self_comp_port_input_msg_iterator_state(iterator, init_state);
1755 return status;
1756}
1757
5b7b55be
SM
1758/*
1759 * This function is installed as the iterator's next callback after we have
1760 * auto-seeked (seeked to the beginning and fast-forwarded) to send the
1761 * messages saved in iterator->auto_seek.msgs. Once this is done, the original
1762 * next callback is put back.
1763 */
1764
7474e7d3 1765static
d24d5663 1766enum bt_component_class_message_iterator_next_method_status post_auto_seek_next(
7474e7d3
PP
1767 struct bt_self_component_port_input_message_iterator *iterator,
1768 bt_message_array_const msgs, uint64_t capacity,
1769 uint64_t *count)
1770{
da9c4c52 1771 BT_ASSERT(!g_queue_is_empty(iterator->auto_seek.msgs));
5b9e151d 1772 *count = 0;
7474e7d3
PP
1773
1774 /*
1775 * Move auto-seek messages to the output array (which is this
5b9e151d 1776 * iterator's base message array).
7474e7d3 1777 */
da9c4c52
SM
1778 while (capacity > 0 && !g_queue_is_empty(iterator->auto_seek.msgs)) {
1779 msgs[*count] = g_queue_pop_head(iterator->auto_seek.msgs);
5b9e151d
PP
1780 capacity--;
1781 (*count)++;
7474e7d3 1782 }
7474e7d3 1783
5b9e151d
PP
1784 BT_ASSERT(*count > 0);
1785
da9c4c52 1786 if (g_queue_is_empty(iterator->auto_seek.msgs)) {
572075a8
SM
1787 /* No more auto-seek messages, restore user's next callback. */
1788 BT_ASSERT(iterator->auto_seek.original_next_callback);
1789 iterator->methods.next = iterator->auto_seek.original_next_callback;
1790 iterator->auto_seek.original_next_callback = NULL;
7474e7d3
PP
1791 }
1792
d24d5663 1793 return BT_FUNC_STATUS_OK;
7474e7d3
PP
1794}
1795
5b7b55be
SM
1796static inline
1797int clock_raw_value_from_ns_from_origin(const bt_clock_class *clock_class,
1798 int64_t ns_from_origin, uint64_t *raw_value)
1799{
1800
1801 int64_t cc_offset_s = clock_class->offset_seconds;
1802 uint64_t cc_offset_cycles = clock_class->offset_cycles;
1803 uint64_t cc_freq = clock_class->frequency;
1804
1805 return bt_common_clock_value_from_ns_from_origin(cc_offset_s,
1806 cc_offset_cycles, cc_freq, ns_from_origin, raw_value);
1807}
1808
1809
d24d5663 1810enum bt_message_iterator_seek_ns_from_origin_status
7474e7d3
PP
1811bt_self_component_port_input_message_iterator_seek_ns_from_origin(
1812 struct bt_self_component_port_input_message_iterator *iterator,
1813 int64_t ns_from_origin)
1814{
1815 int status;
5b7b55be 1816 GHashTable *stream_states = NULL;
7474e7d3
PP
1817
1818 BT_ASSERT_PRE_NON_NULL(iterator, "Message iterator");
1819 BT_ASSERT_PRE_ITER_HAS_STATE_TO_SEEK(iterator);
5badd463
PP
1820 BT_ASSERT_PRE(
1821 bt_component_borrow_graph(iterator->upstream_component)->config_state !=
1822 BT_GRAPH_CONFIGURATION_STATE_CONFIGURING,
7474e7d3
PP
1823 "Graph is not configured: %!+g",
1824 bt_component_borrow_graph(iterator->upstream_component));
1825 BT_ASSERT_PRE(
1826 bt_self_component_port_input_message_iterator_can_seek_ns_from_origin(
1827 iterator, ns_from_origin),
1828 "Message iterator cannot seek nanoseconds from origin: %!+i, "
1829 "ns-from-origin=%" PRId64, iterator, ns_from_origin);
1830 set_self_comp_port_input_msg_iterator_state(iterator,
1831 BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_SEEKING);
1832
54b135a0
SM
1833 /*
1834 * We are seeking, reset our expectations about how the following
1835 * messages should look like.
1836 */
1837 reset_iterator_expectations(iterator);
1838
7474e7d3 1839 if (iterator->methods.seek_ns_from_origin) {
5b7b55be 1840 /* The iterator knows how to seek to a particular time, let it handle this. */
7474e7d3
PP
1841 BT_LIB_LOGD("Calling user's \"seek nanoseconds from origin\" method: "
1842 "%![iter-]+i, ns=%" PRId64, iterator, ns_from_origin);
1843 status = iterator->methods.seek_ns_from_origin(iterator,
1844 ns_from_origin);
1845 BT_LOGD("User method returned: status=%s",
d24d5663
PP
1846 bt_common_func_status_string(status));
1847 BT_ASSERT_POST(status == BT_FUNC_STATUS_OK ||
1848 status == BT_FUNC_STATUS_ERROR ||
1849 status == BT_FUNC_STATUS_MEMORY_ERROR ||
1850 status == BT_FUNC_STATUS_AGAIN,
7474e7d3 1851 "Unexpected status: %![iter-]+i, status=%s",
d24d5663 1852 iterator, bt_common_func_status_string(status));
870631a2
PP
1853 if (status < 0) {
1854 BT_LIB_LOGW_APPEND_CAUSE(
1855 "Component input port message iterator's \"seek nanoseconds from origin\" method failed: "
1856 "%![iter-]+i, status=%s",
1857 iterator, bt_common_func_status_string(status));
1858 }
7474e7d3 1859 } else {
5b7b55be
SM
1860 /*
1861 * The iterator doesn't know how to seek to a particular time. We will
1862 * seek to the beginning and fast forward to the right place.
1863 */
7474e7d3
PP
1864 BT_ASSERT(iterator->methods.can_seek_beginning(iterator));
1865 BT_ASSERT(iterator->methods.seek_beginning);
1866 BT_LIB_LOGD("Calling user's \"seek beginning\" method: %!+i",
1867 iterator);
1868 status = iterator->methods.seek_beginning(iterator);
1869 BT_LOGD("User method returned: status=%s",
d24d5663
PP
1870 bt_common_func_status_string(status));
1871 BT_ASSERT_POST(status == BT_FUNC_STATUS_OK ||
1872 status == BT_FUNC_STATUS_ERROR ||
1873 status == BT_FUNC_STATUS_MEMORY_ERROR ||
1874 status == BT_FUNC_STATUS_AGAIN,
7474e7d3 1875 "Unexpected status: %![iter-]+i, status=%s",
d24d5663 1876 iterator, bt_common_func_status_string(status));
870631a2
PP
1877 if (status < 0) {
1878 BT_LIB_LOGW_APPEND_CAUSE(
1879 "Component input port message iterator's \"seek beginning\" method failed: "
1880 "%![iter-]+i, status=%s",
1881 iterator, bt_common_func_status_string(status));
1882 }
1883
7474e7d3 1884 switch (status) {
d24d5663 1885 case BT_FUNC_STATUS_OK:
7474e7d3 1886 break;
d24d5663
PP
1887 case BT_FUNC_STATUS_ERROR:
1888 case BT_FUNC_STATUS_MEMORY_ERROR:
1889 case BT_FUNC_STATUS_AGAIN:
7474e7d3
PP
1890 goto end;
1891 default:
1892 abort();
1893 }
1894
1895 /*
1896 * Find the first message which has a default clock
1897 * snapshot greater than or equal to the requested
5b9e151d
PP
1898 * seeking time, and move the received messages from
1899 * this point in the batch to this iterator's auto-seek
1900 * message queue.
7474e7d3 1901 */
da9c4c52 1902 while (!g_queue_is_empty(iterator->auto_seek.msgs)) {
5b9e151d 1903 bt_object_put_no_null_check(
da9c4c52 1904 g_queue_pop_tail(iterator->auto_seek.msgs));
5b9e151d
PP
1905 }
1906
5b7b55be
SM
1907 stream_states = create_auto_seek_stream_states();
1908 if (!stream_states) {
870631a2
PP
1909 BT_LIB_LOGE_APPEND_CAUSE(
1910 "Failed to allocate one GHashTable.");
d24d5663 1911 status = BT_FUNC_STATUS_MEMORY_ERROR;
5b7b55be
SM
1912 goto end;
1913 }
1914
7474e7d3 1915 status = find_message_ge_ns_from_origin(iterator,
5b7b55be 1916 ns_from_origin, stream_states);
7474e7d3 1917 switch (status) {
d24d5663
PP
1918 case BT_FUNC_STATUS_OK:
1919 case BT_FUNC_STATUS_END:
5b7b55be
SM
1920 {
1921 GHashTableIter iter;
1922 gpointer key, value;
1923
1924 /*
1925 * If some streams exist at the seek time, prepend the
1926 * required messages to put those streams in the right
1927 * state.
1928 */
1929 g_hash_table_iter_init(&iter, stream_states);
1930 while (g_hash_table_iter_next (&iter, &key, &value)) {
1931 const bt_stream *stream = key;
1932 struct auto_seek_stream_state *stream_state =
1933 (struct auto_seek_stream_state *) value;
1934 bt_message *msg;
1935 const bt_clock_class *clock_class = bt_stream_class_borrow_default_clock_class_const(
1936 bt_stream_borrow_class_const(stream));
1937 uint64_t raw_value;
1938
1939 if (clock_raw_value_from_ns_from_origin(clock_class, ns_from_origin, &raw_value) != 0) {
870631a2
PP
1940 BT_LIB_LOGW_APPEND_CAUSE(
1941 "Could not convert nanoseconds from origin to clock value: "
1942 "ns-from-origin=%" PRId64 ", %![cc-]+K",
5b7b55be 1943 ns_from_origin, clock_class);
d24d5663 1944 status = BT_FUNC_STATUS_ERROR;
5b7b55be
SM
1945 goto end;
1946 }
1947
1948 switch (stream_state->state) {
1949 case AUTO_SEEK_STREAM_STATE_PACKET_BEGAN:
1950 BT_ASSERT(stream_state->packet);
1951 BT_LIB_LOGD("Creating packet message: %![packet-]+a", stream_state->packet);
1952 msg = bt_message_packet_beginning_create_with_default_clock_snapshot(
1953 (bt_self_message_iterator *) iterator, stream_state->packet, raw_value);
1954 if (!msg) {
d24d5663 1955 status = BT_FUNC_STATUS_MEMORY_ERROR;
5b7b55be
SM
1956 goto end;
1957 }
1958
1959 g_queue_push_head(iterator->auto_seek.msgs, msg);
1960 msg = NULL;
1961 /* fall-thru */
1962 case AUTO_SEEK_STREAM_STATE_STREAM_ACTIVITY_BEGAN:
1963 msg = bt_message_stream_activity_beginning_create(
1964 (bt_self_message_iterator *) iterator, stream);
1965 if (!msg) {
d24d5663 1966 status = BT_FUNC_STATUS_MEMORY_ERROR;
5b7b55be
SM
1967 goto end;
1968 }
1969
1970 bt_message_stream_activity_beginning_set_default_clock_snapshot(msg, raw_value);
1971
1972 g_queue_push_head(iterator->auto_seek.msgs, msg);
1973 msg = NULL;
1974 /* fall-thru */
1975 case AUTO_SEEK_STREAM_STATE_STREAM_BEGAN:
1976 msg = bt_message_stream_beginning_create(
1977 (bt_self_message_iterator *) iterator, stream);
1978 if (!msg) {
d24d5663 1979 status = BT_FUNC_STATUS_MEMORY_ERROR;
5b7b55be
SM
1980 goto end;
1981 }
1982
1983 g_queue_push_head(iterator->auto_seek.msgs, msg);
1984 msg = NULL;
1985 break;
1986 }
1987 }
1988
7474e7d3 1989 /*
5b9e151d
PP
1990 * If there are messages in the auto-seek
1991 * message queue, replace the user's "next"
1992 * method with a custom, temporary "next" method
1993 * which returns them.
7474e7d3 1994 */
da9c4c52 1995 if (!g_queue_is_empty(iterator->auto_seek.msgs)) {
572075a8
SM
1996 BT_ASSERT(!iterator->auto_seek.original_next_callback);
1997 iterator->auto_seek.original_next_callback = iterator->methods.next;
1998
5b9e151d
PP
1999 iterator->methods.next =
2000 (bt_self_component_port_input_message_iterator_next_method)
2001 post_auto_seek_next;
2002 }
2003
2004 /*
d24d5663
PP
2005 * `BT_FUNC_STATUS_END` becomes
2006 * `BT_FUNC_STATUS_OK`: the next
5b9e151d
PP
2007 * time this iterator's "next" method is called,
2008 * it will return
d24d5663 2009 * `BT_FUNC_STATUS_END`.
5b9e151d 2010 */
d24d5663 2011 status = BT_FUNC_STATUS_OK;
7474e7d3 2012 break;
5b7b55be 2013 }
d24d5663
PP
2014 case BT_FUNC_STATUS_ERROR:
2015 case BT_FUNC_STATUS_MEMORY_ERROR:
2016 case BT_FUNC_STATUS_AGAIN:
7474e7d3 2017 goto end;
7474e7d3
PP
2018 default:
2019 abort();
2020 }
2021 }
2022
54b135a0
SM
2023 /*
2024 * The following messages returned by the next method (including
2025 * post_auto_seek_next) must be after (or at) `ns_from_origin`.
2026 */
2027 iterator->last_ns_from_origin = ns_from_origin;
2028
7474e7d3 2029end:
5b7b55be
SM
2030 if (stream_states) {
2031 destroy_auto_seek_stream_states(stream_states);
2032 stream_states = NULL;
2033 }
870631a2 2034
7474e7d3 2035 set_iterator_state_after_seeking(iterator, status);
7474e7d3
PP
2036 return status;
2037}
2038
2039static inline
2040bt_self_component_port_input_message_iterator *
2041borrow_output_port_message_iterator_upstream_iterator(
2042 struct bt_port_output_message_iterator *iterator)
2043{
2044 struct bt_component_class_sink_colander_priv_data *colander_data;
2045
2046 BT_ASSERT(iterator);
2047 colander_data = (void *) iterator->colander->parent.user_data;
2048 BT_ASSERT(colander_data);
2049 BT_ASSERT(colander_data->msg_iter);
2050 return colander_data->msg_iter;
2051}
2052
2053bt_bool bt_port_output_message_iterator_can_seek_ns_from_origin(
2054 struct bt_port_output_message_iterator *iterator,
2055 int64_t ns_from_origin)
2056{
2057 BT_ASSERT_PRE_NON_NULL(iterator, "Message iterator");
2058 return bt_self_component_port_input_message_iterator_can_seek_ns_from_origin(
2059 borrow_output_port_message_iterator_upstream_iterator(
2060 iterator), ns_from_origin);
2061}
2062
2063bt_bool bt_port_output_message_iterator_can_seek_beginning(
2064 struct bt_port_output_message_iterator *iterator)
2065{
2066 BT_ASSERT_PRE_NON_NULL(iterator, "Message iterator");
2067 return bt_self_component_port_input_message_iterator_can_seek_beginning(
2068 borrow_output_port_message_iterator_upstream_iterator(
2069 iterator));
2070}
2071
d24d5663
PP
2072enum bt_message_iterator_seek_ns_from_origin_status
2073bt_port_output_message_iterator_seek_ns_from_origin(
7474e7d3
PP
2074 struct bt_port_output_message_iterator *iterator,
2075 int64_t ns_from_origin)
2076{
2077 BT_ASSERT_PRE_NON_NULL(iterator, "Message iterator");
2078 return bt_self_component_port_input_message_iterator_seek_ns_from_origin(
2079 borrow_output_port_message_iterator_upstream_iterator(iterator),
2080 ns_from_origin);
2081}
2082
d24d5663
PP
2083enum bt_message_iterator_seek_beginning_status
2084bt_port_output_message_iterator_seek_beginning(
7474e7d3
PP
2085 struct bt_port_output_message_iterator *iterator)
2086{
2087 BT_ASSERT_PRE_NON_NULL(iterator, "Message iterator");
2088 return bt_self_component_port_input_message_iterator_seek_beginning(
2089 borrow_output_port_message_iterator_upstream_iterator(
2090 iterator));
2091}
2092
d6e69534
PP
2093void bt_port_output_message_iterator_get_ref(
2094 const struct bt_port_output_message_iterator *iterator)
c5b9b441
PP
2095{
2096 bt_object_get_ref(iterator);
2097}
2098
d6e69534
PP
2099void bt_port_output_message_iterator_put_ref(
2100 const struct bt_port_output_message_iterator *iterator)
c5b9b441
PP
2101{
2102 bt_object_put_ref(iterator);
2103}
2104
d6e69534
PP
2105void bt_self_component_port_input_message_iterator_get_ref(
2106 const struct bt_self_component_port_input_message_iterator *iterator)
c5b9b441
PP
2107{
2108 bt_object_get_ref(iterator);
2109}
2110
d6e69534
PP
2111void bt_self_component_port_input_message_iterator_put_ref(
2112 const struct bt_self_component_port_input_message_iterator *iterator)
c5b9b441
PP
2113{
2114 bt_object_put_ref(iterator);
2115}
This page took 0.172839 seconds and 4 git commands to generate.