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