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