lib: have default clock snapshot in event message
[babeltrace.git] / 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 "MSG-ITER"
25 #include <babeltrace/lib-logging-internal.h>
26
27 #include <babeltrace/compiler-internal.h>
28 #include <babeltrace/trace-ir/field.h>
29 #include <babeltrace/trace-ir/event-const.h>
30 #include <babeltrace/trace-ir/event-internal.h>
31 #include <babeltrace/trace-ir/packet-const.h>
32 #include <babeltrace/trace-ir/packet-internal.h>
33 #include <babeltrace/trace-ir/stream-internal.h>
34 #include <babeltrace/graph/connection-const.h>
35 #include <babeltrace/graph/connection-internal.h>
36 #include <babeltrace/graph/component-const.h>
37 #include <babeltrace/graph/component-internal.h>
38 #include <babeltrace/graph/component-source-internal.h>
39 #include <babeltrace/graph/component-class-internal.h>
40 #include <babeltrace/graph/component-class-sink-colander-internal.h>
41 #include <babeltrace/graph/component-sink-const.h>
42 #include <babeltrace/graph/component-sink-internal.h>
43 #include <babeltrace/graph/message-const.h>
44 #include <babeltrace/graph/message-iterator-const.h>
45 #include <babeltrace/graph/message-iterator-internal.h>
46 #include <babeltrace/graph/self-component-port-input-message-iterator.h>
47 #include <babeltrace/graph/port-output-message-iterator.h>
48 #include <babeltrace/graph/message-internal.h>
49 #include <babeltrace/graph/message-event-const.h>
50 #include <babeltrace/graph/message-event-internal.h>
51 #include <babeltrace/graph/message-packet-beginning-const.h>
52 #include <babeltrace/graph/message-packet-end-const.h>
53 #include <babeltrace/graph/message-packet-internal.h>
54 #include <babeltrace/graph/message-stream-beginning-const.h>
55 #include <babeltrace/graph/message-stream-end-const.h>
56 #include <babeltrace/graph/message-stream-internal.h>
57 #include <babeltrace/graph/message-inactivity-internal.h>
58 #include <babeltrace/graph/message-discarded-items-internal.h>
59 #include <babeltrace/graph/message-stream-activity-internal.h>
60 #include <babeltrace/graph/port-const.h>
61 #include <babeltrace/graph/graph.h>
62 #include <babeltrace/graph/graph-const.h>
63 #include <babeltrace/graph/graph-internal.h>
64 #include <babeltrace/types.h>
65 #include <babeltrace/assert-internal.h>
66 #include <babeltrace/assert-pre-internal.h>
67 #include <stdint.h>
68 #include <inttypes.h>
69 #include <stdlib.h>
70
71 /*
72 * TODO: Use graph's state (number of active iterators, etc.) and
73 * possibly system specifications to make a better guess than this.
74 */
75 #define MSG_BATCH_SIZE 15
76
77 #define BT_ASSERT_PRE_ITER_HAS_STATE_TO_SEEK(_iter) \
78 BT_ASSERT_PRE((_iter)->state == BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_ACTIVE || \
79 (_iter)->state == BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_ENDED || \
80 (_iter)->state == BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_LAST_SEEKING_RETURNED_AGAIN || \
81 (_iter)->state == BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_LAST_SEEKING_RETURNED_ERROR, \
82 "Message iterator is in the wrong state: %!+i", _iter)
83
84 static inline
85 void _set_self_comp_port_input_msg_iterator_state(
86 struct bt_self_component_port_input_message_iterator *iterator,
87 enum bt_self_component_port_input_message_iterator_state state)
88 {
89 BT_ASSERT(iterator);
90 BT_LIB_LOGD("Updating message iterator's state: new-state=%s",
91 bt_self_component_port_input_message_iterator_state_string(state));
92 iterator->state = state;
93 }
94
95 #ifdef BT_DEV_MODE
96 # define set_self_comp_port_input_msg_iterator_state _set_self_comp_port_input_msg_iterator_state
97 #else
98 # define set_self_comp_port_input_msg_iterator_state(_a, _b)
99 #endif
100
101 static
102 void destroy_base_message_iterator(struct bt_object *obj)
103 {
104 struct bt_message_iterator *iterator = (void *) obj;
105
106 BT_ASSERT(iterator);
107
108 if (iterator->msgs) {
109 g_ptr_array_free(iterator->msgs, TRUE);
110 iterator->msgs = NULL;
111 }
112
113 g_free(iterator);
114 }
115
116 static
117 void bt_self_component_port_input_message_iterator_destroy(struct bt_object *obj)
118 {
119 struct bt_self_component_port_input_message_iterator *iterator;
120
121 BT_ASSERT(obj);
122
123 /*
124 * The message iterator's reference count is 0 if we're
125 * here. Increment it to avoid a double-destroy (possibly
126 * infinitely recursive). This could happen for example if the
127 * message iterator's finalization function does
128 * bt_object_get_ref() (or anything that causes
129 * bt_object_get_ref() to be called) on itself (ref. count goes
130 * from 0 to 1), and then bt_object_put_ref(): the reference
131 * count would go from 1 to 0 again and this function would be
132 * called again.
133 */
134 obj->ref_count++;
135 iterator = (void *) obj;
136 BT_LIB_LOGD("Destroying self component input port message iterator object: "
137 "%!+i", iterator);
138 bt_self_component_port_input_message_iterator_try_finalize(iterator);
139
140 if (iterator->connection) {
141 /*
142 * Remove ourself from the originating connection so
143 * that it does not try to finalize a dangling pointer
144 * later.
145 */
146 bt_connection_remove_iterator(iterator->connection, iterator);
147 iterator->connection = NULL;
148 }
149
150 if (iterator->auto_seek_msgs) {
151 uint64_t i;
152
153 /* Put any remaining message in the auto-seek array */
154 for (i = 0; i < iterator->auto_seek_msgs->len; i++) {
155 if (iterator->auto_seek_msgs->pdata[i]) {
156 bt_object_put_no_null_check(
157 iterator->auto_seek_msgs->pdata[i]);
158 }
159 }
160
161 g_ptr_array_free(iterator->auto_seek_msgs, TRUE);
162 iterator->auto_seek_msgs = NULL;
163 }
164
165 destroy_base_message_iterator(obj);
166 }
167
168 BT_HIDDEN
169 void bt_self_component_port_input_message_iterator_try_finalize(
170 struct bt_self_component_port_input_message_iterator *iterator)
171 {
172 typedef void (*method_t)(void *);
173
174 struct bt_component_class *comp_class = NULL;
175 method_t method = NULL;
176
177 BT_ASSERT(iterator);
178
179 switch (iterator->state) {
180 case BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_NON_INITIALIZED:
181 /* Skip user finalization if user initialization failed */
182 BT_LIB_LOGD("Not finalizing non-initialized message iterator: "
183 "%!+i", iterator);
184 goto end;
185 case BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_FINALIZED:
186 /* Already finalized */
187 BT_LIB_LOGD("Not finalizing message iterator: already finalized: "
188 "%!+i", iterator);
189 goto end;
190 case BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_FINALIZING:
191 /* Already finalized */
192 BT_LIB_LOGF("Message iterator is already being finalized: "
193 "%!+i", iterator);
194 abort();
195 default:
196 break;
197 }
198
199 BT_LIB_LOGD("Finalizing message iterator: %!+i", iterator);
200 set_self_comp_port_input_msg_iterator_state(iterator,
201 BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_FINALIZING);
202 BT_ASSERT(iterator->upstream_component);
203 comp_class = iterator->upstream_component->class;
204
205 /* Call user-defined destroy method */
206 switch (comp_class->type) {
207 case BT_COMPONENT_CLASS_TYPE_SOURCE:
208 {
209 struct bt_component_class_source *src_comp_cls =
210 (void *) comp_class;
211
212 method = (method_t) src_comp_cls->methods.msg_iter_finalize;
213 break;
214 }
215 case BT_COMPONENT_CLASS_TYPE_FILTER:
216 {
217 struct bt_component_class_filter *flt_comp_cls =
218 (void *) comp_class;
219
220 method = (method_t) flt_comp_cls->methods.msg_iter_finalize;
221 break;
222 }
223 default:
224 /* Unreachable */
225 abort();
226 }
227
228 if (method) {
229 BT_LIB_LOGD("Calling user's finalization method: %!+i",
230 iterator);
231 method(iterator);
232 }
233
234 iterator->upstream_component = NULL;
235 iterator->upstream_port = NULL;
236 set_self_comp_port_input_msg_iterator_state(iterator,
237 BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_FINALIZED);
238 BT_LIB_LOGD("Finalized message iterator: %!+i", iterator);
239
240 end:
241 return;
242 }
243
244 BT_HIDDEN
245 void bt_self_component_port_input_message_iterator_set_connection(
246 struct bt_self_component_port_input_message_iterator *iterator,
247 struct bt_connection *connection)
248 {
249 BT_ASSERT(iterator);
250 iterator->connection = connection;
251 BT_LIB_LOGV("Set message iterator's connection: "
252 "%![iter-]+i, %![conn-]+x", iterator, connection);
253 }
254
255 static
256 int init_message_iterator(struct bt_message_iterator *iterator,
257 enum bt_message_iterator_type type,
258 bt_object_release_func destroy)
259 {
260 int ret = 0;
261
262 bt_object_init_shared(&iterator->base, destroy);
263 iterator->type = type;
264 iterator->msgs = g_ptr_array_new();
265 if (!iterator->msgs) {
266 BT_LOGE_STR("Failed to allocate a GPtrArray.");
267 ret = -1;
268 goto end;
269 }
270
271 g_ptr_array_set_size(iterator->msgs, MSG_BATCH_SIZE);
272
273 end:
274 return ret;
275 }
276
277 static
278 bt_bool can_seek_ns_from_origin_true(
279 struct bt_self_component_port_input_message_iterator *iterator,
280 int64_t ns_from_origin)
281 {
282 return BT_TRUE;
283 }
284
285 static
286 bt_bool can_seek_beginning_true(
287 struct bt_self_component_port_input_message_iterator *iterator)
288 {
289 return BT_TRUE;
290 }
291
292 static
293 struct bt_self_component_port_input_message_iterator *
294 bt_self_component_port_input_message_iterator_create_initial(
295 struct bt_component *upstream_comp,
296 struct bt_port *upstream_port)
297 {
298 int ret;
299 struct bt_self_component_port_input_message_iterator *iterator = NULL;
300
301 BT_ASSERT(upstream_comp);
302 BT_ASSERT(upstream_port);
303 BT_ASSERT(bt_port_is_connected(upstream_port));
304 BT_LIB_LOGD("Creating initial message iterator on self component input port: "
305 "%![up-comp-]+c, %![up-port-]+p", upstream_comp, upstream_port);
306 BT_ASSERT(bt_component_get_class_type(upstream_comp) ==
307 BT_COMPONENT_CLASS_TYPE_SOURCE ||
308 bt_component_get_class_type(upstream_comp) ==
309 BT_COMPONENT_CLASS_TYPE_FILTER);
310 iterator = g_new0(
311 struct bt_self_component_port_input_message_iterator, 1);
312 if (!iterator) {
313 BT_LOGE_STR("Failed to allocate one self component input port "
314 "message iterator.");
315 goto end;
316 }
317
318 ret = init_message_iterator((void *) iterator,
319 BT_MESSAGE_ITERATOR_TYPE_SELF_COMPONENT_PORT_INPUT,
320 bt_self_component_port_input_message_iterator_destroy);
321 if (ret) {
322 /* init_message_iterator() logs errors */
323 BT_OBJECT_PUT_REF_AND_RESET(iterator);
324 goto end;
325 }
326
327 iterator->auto_seek_msgs = g_ptr_array_new();
328 if (!iterator->auto_seek_msgs) {
329 BT_LOGE_STR("Failed to allocate a GPtrArray.");
330 ret = -1;
331 goto end;
332 }
333
334 g_ptr_array_set_size(iterator->auto_seek_msgs, MSG_BATCH_SIZE);
335 iterator->upstream_component = upstream_comp;
336 iterator->upstream_port = upstream_port;
337 iterator->connection = iterator->upstream_port->connection;
338 iterator->graph = bt_component_borrow_graph(upstream_comp);
339 set_self_comp_port_input_msg_iterator_state(iterator,
340 BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_NON_INITIALIZED);
341
342 switch (iterator->upstream_component->class->type) {
343 case BT_COMPONENT_CLASS_TYPE_SOURCE:
344 {
345 struct bt_component_class_source *src_comp_cls =
346 (void *) iterator->upstream_component->class;
347
348 iterator->methods.next =
349 (bt_self_component_port_input_message_iterator_next_method)
350 src_comp_cls->methods.msg_iter_next;
351 iterator->methods.seek_ns_from_origin =
352 (bt_self_component_port_input_message_iterator_seek_ns_from_origin_method)
353 src_comp_cls->methods.msg_iter_seek_ns_from_origin;
354 iterator->methods.seek_beginning =
355 (bt_self_component_port_input_message_iterator_seek_beginning_method)
356 src_comp_cls->methods.msg_iter_seek_beginning;
357 iterator->methods.can_seek_ns_from_origin =
358 (bt_self_component_port_input_message_iterator_can_seek_ns_from_origin_method)
359 src_comp_cls->methods.msg_iter_can_seek_ns_from_origin;
360 iterator->methods.can_seek_beginning =
361 (bt_self_component_port_input_message_iterator_can_seek_beginning_method)
362 src_comp_cls->methods.msg_iter_can_seek_beginning;
363 break;
364 }
365 case BT_COMPONENT_CLASS_TYPE_FILTER:
366 {
367 struct bt_component_class_filter *flt_comp_cls =
368 (void *) iterator->upstream_component->class;
369
370 iterator->methods.next =
371 (bt_self_component_port_input_message_iterator_next_method)
372 flt_comp_cls->methods.msg_iter_next;
373 iterator->methods.seek_ns_from_origin =
374 (bt_self_component_port_input_message_iterator_seek_ns_from_origin_method)
375 flt_comp_cls->methods.msg_iter_seek_ns_from_origin;
376 iterator->methods.seek_beginning =
377 (bt_self_component_port_input_message_iterator_seek_beginning_method)
378 flt_comp_cls->methods.msg_iter_seek_beginning;
379 iterator->methods.can_seek_ns_from_origin =
380 (bt_self_component_port_input_message_iterator_can_seek_ns_from_origin_method)
381 flt_comp_cls->methods.msg_iter_can_seek_ns_from_origin;
382 iterator->methods.can_seek_beginning =
383 (bt_self_component_port_input_message_iterator_can_seek_beginning_method)
384 flt_comp_cls->methods.msg_iter_can_seek_beginning;
385 break;
386 }
387 default:
388 abort();
389 }
390
391 if (iterator->methods.seek_ns_from_origin &&
392 !iterator->methods.can_seek_ns_from_origin) {
393 iterator->methods.can_seek_ns_from_origin =
394 (bt_self_component_port_input_message_iterator_can_seek_ns_from_origin_method)
395 can_seek_ns_from_origin_true;
396 }
397
398 if (iterator->methods.seek_beginning &&
399 !iterator->methods.can_seek_beginning) {
400 iterator->methods.can_seek_beginning =
401 (bt_self_component_port_input_message_iterator_seek_beginning_method)
402 can_seek_beginning_true;
403 }
404
405 BT_LIB_LOGD("Created initial message iterator on self component input port: "
406 "%![up-port-]+p, %![up-comp-]+c, %![iter-]+i",
407 upstream_port, upstream_comp, iterator);
408
409 end:
410 return iterator;
411 }
412
413 struct bt_self_component_port_input_message_iterator *
414 bt_self_component_port_input_message_iterator_create(
415 struct bt_self_component_port_input *self_port)
416 {
417 typedef enum bt_self_message_iterator_status (*init_method_t)(
418 void *, void *, void *);
419
420 init_method_t init_method = NULL;
421 struct bt_self_component_port_input_message_iterator *iterator =
422 NULL;
423 struct bt_port *port = (void *) self_port;
424 struct bt_port *upstream_port;
425 struct bt_component *comp;
426 struct bt_component *upstream_comp;
427 struct bt_component_class *upstream_comp_cls;
428
429 BT_ASSERT_PRE_NON_NULL(port, "Port");
430 comp = bt_port_borrow_component_inline(port);
431 BT_ASSERT_PRE(bt_port_is_connected(port),
432 "Port is not connected: %![port-]+p", port);
433 BT_ASSERT_PRE(comp, "Port is not part of a component: %![port-]+p",
434 port);
435 BT_ASSERT_PRE(!bt_component_graph_is_canceled(comp),
436 "Port's component's graph is canceled: "
437 "%![port-]+p, %![comp-]+c", port, comp);
438 BT_ASSERT(port->connection);
439 upstream_port = port->connection->upstream_port;
440 BT_ASSERT(upstream_port);
441 upstream_comp = bt_port_borrow_component_inline(upstream_port);
442 BT_ASSERT(upstream_comp);
443 upstream_comp_cls = upstream_comp->class;
444 BT_ASSERT(upstream_comp->class->type ==
445 BT_COMPONENT_CLASS_TYPE_SOURCE ||
446 upstream_comp->class->type ==
447 BT_COMPONENT_CLASS_TYPE_FILTER);
448 iterator = bt_self_component_port_input_message_iterator_create_initial(
449 upstream_comp, upstream_port);
450 if (!iterator) {
451 BT_LOGW_STR("Cannot create self component input port "
452 "message iterator.");
453 goto end;
454 }
455
456 switch (upstream_comp_cls->type) {
457 case BT_COMPONENT_CLASS_TYPE_SOURCE:
458 {
459 struct bt_component_class_source *src_comp_cls =
460 (void *) upstream_comp_cls;
461
462 init_method =
463 (init_method_t) src_comp_cls->methods.msg_iter_init;
464 break;
465 }
466 case BT_COMPONENT_CLASS_TYPE_FILTER:
467 {
468 struct bt_component_class_filter *flt_comp_cls =
469 (void *) upstream_comp_cls;
470
471 init_method =
472 (init_method_t) flt_comp_cls->methods.msg_iter_init;
473 break;
474 }
475 default:
476 /* Unreachable */
477 abort();
478 }
479
480 if (init_method) {
481 int iter_status;
482
483 BT_LIB_LOGD("Calling user's initialization method: %!+i", iterator);
484 iter_status = init_method(iterator, upstream_comp,
485 upstream_port);
486 BT_LOGD("User method returned: status=%s",
487 bt_message_iterator_status_string(iter_status));
488 if (iter_status != BT_MESSAGE_ITERATOR_STATUS_OK) {
489 BT_LOGW_STR("Initialization method failed.");
490 BT_OBJECT_PUT_REF_AND_RESET(iterator);
491 goto end;
492 }
493 }
494
495 set_self_comp_port_input_msg_iterator_state(iterator,
496 BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_ACTIVE);
497 g_ptr_array_add(port->connection->iterators, iterator);
498 BT_LIB_LOGD("Created message iterator on self component input port: "
499 "%![up-port-]+p, %![up-comp-]+c, %![iter-]+i",
500 upstream_port, upstream_comp, iterator);
501
502 end:
503 return iterator;
504 }
505
506 void *bt_self_message_iterator_get_data(
507 const struct bt_self_message_iterator *self_iterator)
508 {
509 struct bt_self_component_port_input_message_iterator *iterator =
510 (void *) self_iterator;
511
512 BT_ASSERT_PRE_NON_NULL(iterator, "Message iterator");
513 return iterator->user_data;
514 }
515
516 void bt_self_message_iterator_set_data(
517 struct bt_self_message_iterator *self_iterator, void *data)
518 {
519 struct bt_self_component_port_input_message_iterator *iterator =
520 (void *) self_iterator;
521
522 BT_ASSERT_PRE_NON_NULL(iterator, "Message iterator");
523 iterator->user_data = data;
524 BT_LIB_LOGV("Set message iterator's user data: "
525 "%!+i, user-data-addr=%p", iterator, data);
526 }
527
528 BT_ASSERT_PRE_FUNC
529 static inline
530 void bt_message_borrow_packet_stream(const struct bt_message *msg,
531 const struct bt_stream **stream,
532 const struct bt_packet **packet)
533 {
534 BT_ASSERT(msg);
535
536 switch (msg->type) {
537 case BT_MESSAGE_TYPE_EVENT:
538 *packet = bt_event_borrow_packet_const(
539 bt_message_event_borrow_event_const(msg));
540 *stream = bt_packet_borrow_stream_const(*packet);
541 break;
542 case BT_MESSAGE_TYPE_STREAM_BEGINNING:
543 *stream = bt_message_stream_beginning_borrow_stream_const(msg);
544 break;
545 case BT_MESSAGE_TYPE_STREAM_END:
546 *stream = bt_message_stream_end_borrow_stream_const(msg);
547 break;
548 case BT_MESSAGE_TYPE_PACKET_BEGINNING:
549 *packet = bt_message_packet_beginning_borrow_packet_const(msg);
550 *stream = bt_packet_borrow_stream_const(*packet);
551 break;
552 case BT_MESSAGE_TYPE_PACKET_END:
553 *packet = bt_message_packet_end_borrow_packet_const(msg);
554 *stream = bt_packet_borrow_stream_const(*packet);
555 break;
556 default:
557 break;
558 }
559 }
560
561 enum bt_message_iterator_status
562 bt_self_component_port_input_message_iterator_next(
563 struct bt_self_component_port_input_message_iterator *iterator,
564 bt_message_array_const *msgs, uint64_t *user_count)
565 {
566 int status = BT_MESSAGE_ITERATOR_STATUS_OK;
567
568 BT_ASSERT_PRE_NON_NULL(iterator, "Message iterator");
569 BT_ASSERT_PRE_NON_NULL(msgs, "Message array (output)");
570 BT_ASSERT_PRE_NON_NULL(user_count, "Message count (output)");
571 BT_ASSERT_PRE(iterator->state ==
572 BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_ACTIVE,
573 "Message iterator's \"next\" called, but "
574 "message iterator is in the wrong state: %!+i", iterator);
575 BT_ASSERT(iterator->upstream_component);
576 BT_ASSERT(iterator->upstream_component->class);
577 BT_ASSERT_PRE(bt_component_borrow_graph(iterator->upstream_component)->is_configured,
578 "Graph is not configured: %!+g",
579 bt_component_borrow_graph(iterator->upstream_component));
580 BT_LIB_LOGD("Getting next self component input port "
581 "message iterator's messages: %!+i", iterator);
582
583 /*
584 * Call the user's "next" method to get the next messages
585 * and status.
586 */
587 BT_ASSERT(iterator->methods.next);
588 BT_LOGD_STR("Calling user's \"next\" method.");
589 status = iterator->methods.next(iterator,
590 (void *) iterator->base.msgs->pdata, MSG_BATCH_SIZE,
591 user_count);
592 BT_LOGD("User method returned: status=%s",
593 bt_message_iterator_status_string(status));
594 if (status < 0) {
595 BT_LOGW_STR("User method failed.");
596 goto end;
597 }
598
599 #ifdef BT_DEV_MODE
600 /*
601 * There is no way that this iterator could have been finalized
602 * during its "next" method, as the only way to do this is to
603 * put the last iterator's reference, and this can only be done
604 * by its downstream owner.
605 *
606 * For the same reason, there is no way that this iterator could
607 * have seeked (cannot seek a self message iterator).
608 */
609 BT_ASSERT(iterator->state ==
610 BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_ACTIVE);
611 #endif
612
613 switch (status) {
614 case BT_MESSAGE_ITERATOR_STATUS_OK:
615 BT_ASSERT_PRE(*user_count <= MSG_BATCH_SIZE,
616 "Invalid returned message count: greater than "
617 "batch size: count=%" PRIu64 ", batch-size=%u",
618 *user_count, MSG_BATCH_SIZE);
619 *msgs = (void *) iterator->base.msgs->pdata;
620 break;
621 case BT_MESSAGE_ITERATOR_STATUS_AGAIN:
622 goto end;
623 case BT_MESSAGE_ITERATOR_STATUS_END:
624 set_self_comp_port_input_msg_iterator_state(iterator,
625 BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_ENDED);
626 goto end;
627 default:
628 /* Unknown non-error status */
629 abort();
630 }
631
632 end:
633 return status;
634 }
635
636 enum bt_message_iterator_status bt_port_output_message_iterator_next(
637 struct bt_port_output_message_iterator *iterator,
638 bt_message_array_const *msgs_to_user,
639 uint64_t *count_to_user)
640 {
641 enum bt_message_iterator_status status;
642 enum bt_graph_status graph_status;
643
644 BT_ASSERT_PRE_NON_NULL(iterator, "Message iterator");
645 BT_ASSERT_PRE_NON_NULL(msgs_to_user, "Message array (output)");
646 BT_ASSERT_PRE_NON_NULL(count_to_user, "Message count (output)");
647 BT_LIB_LOGD("Getting next output port message iterator's messages: "
648 "%!+i", iterator);
649
650 /*
651 * As soon as the user calls this function, we mark the graph as
652 * being definitely configured.
653 */
654 bt_graph_set_is_configured(iterator->graph, true);
655
656 graph_status = bt_graph_consume_sink_no_check(iterator->graph,
657 iterator->colander);
658 switch (graph_status) {
659 case BT_GRAPH_STATUS_CANCELED:
660 case BT_GRAPH_STATUS_AGAIN:
661 case BT_GRAPH_STATUS_END:
662 case BT_GRAPH_STATUS_NOMEM:
663 status = (int) graph_status;
664 break;
665 case BT_GRAPH_STATUS_OK:
666 status = BT_MESSAGE_ITERATOR_STATUS_OK;
667
668 /*
669 * On success, the colander sink moves the messages
670 * to this iterator's array and sets this iterator's
671 * message count: move them to the user.
672 */
673 *msgs_to_user = (void *) iterator->base.msgs->pdata;
674 *count_to_user = iterator->count;
675 break;
676 default:
677 /* Other errors */
678 status = BT_MESSAGE_ITERATOR_STATUS_ERROR;
679 }
680
681 return status;
682 }
683
684 struct bt_component *
685 bt_self_component_port_input_message_iterator_borrow_component(
686 struct bt_self_component_port_input_message_iterator *iterator)
687 {
688 BT_ASSERT_PRE_NON_NULL(iterator, "Message iterator");
689 return iterator->upstream_component;
690 }
691
692 const struct bt_component *
693 bt_self_component_port_input_message_iterator_borrow_component_const(
694 const struct bt_self_component_port_input_message_iterator *iterator)
695 {
696 BT_ASSERT_PRE_NON_NULL(iterator, "Message iterator");
697 return iterator->upstream_component;
698 }
699
700 struct bt_self_component *bt_self_message_iterator_borrow_component(
701 struct bt_self_message_iterator *self_iterator)
702 {
703 struct bt_self_component_port_input_message_iterator *iterator =
704 (void *) self_iterator;
705
706 BT_ASSERT_PRE_NON_NULL(iterator, "Message iterator");
707 return (void *) iterator->upstream_component;
708 }
709
710 struct bt_self_port_output *bt_self_message_iterator_borrow_port(
711 struct bt_self_message_iterator *self_iterator)
712 {
713 struct bt_self_component_port_input_message_iterator *iterator =
714 (void *) self_iterator;
715
716 BT_ASSERT_PRE_NON_NULL(iterator, "Message iterator");
717 return (void *) iterator->upstream_port;
718 }
719
720 static
721 void bt_port_output_message_iterator_destroy(struct bt_object *obj)
722 {
723 struct bt_port_output_message_iterator *iterator = (void *) obj;
724
725 BT_LIB_LOGD("Destroying output port message iterator object: %!+i",
726 iterator);
727 BT_LOGD_STR("Putting graph.");
728 BT_OBJECT_PUT_REF_AND_RESET(iterator->graph);
729 BT_LOGD_STR("Putting colander sink component.");
730 BT_OBJECT_PUT_REF_AND_RESET(iterator->colander);
731 destroy_base_message_iterator(obj);
732 }
733
734 struct bt_port_output_message_iterator *
735 bt_port_output_message_iterator_create(struct bt_graph *graph,
736 const struct bt_port_output *output_port)
737 {
738 struct bt_port_output_message_iterator *iterator = NULL;
739 struct bt_component_class_sink *colander_comp_cls = NULL;
740 struct bt_component *output_port_comp = NULL;
741 struct bt_component_sink *colander_comp;
742 enum bt_graph_status graph_status;
743 struct bt_port_input *colander_in_port = NULL;
744 struct bt_component_class_sink_colander_data colander_data;
745 int ret;
746
747 BT_ASSERT_PRE_NON_NULL(graph, "Graph");
748 BT_ASSERT_PRE_NON_NULL(output_port, "Output port");
749 output_port_comp = bt_port_borrow_component_inline(
750 (const void *) output_port);
751 BT_ASSERT_PRE(output_port_comp,
752 "Output port has no component: %!+p", output_port);
753 BT_ASSERT_PRE(bt_component_borrow_graph(output_port_comp) ==
754 (void *) graph,
755 "Output port is not part of graph: %![graph-]+g, %![port-]+p",
756 graph, output_port);
757
758 /* Create message iterator */
759 BT_LIB_LOGD("Creating message iterator on output port: "
760 "%![port-]+p, %![comp-]+c", output_port, output_port_comp);
761 iterator = g_new0(struct bt_port_output_message_iterator, 1);
762 if (!iterator) {
763 BT_LOGE_STR("Failed to allocate one output port message iterator.");
764 goto error;
765 }
766
767 ret = init_message_iterator((void *) iterator,
768 BT_MESSAGE_ITERATOR_TYPE_PORT_OUTPUT,
769 bt_port_output_message_iterator_destroy);
770 if (ret) {
771 /* init_message_iterator() logs errors */
772 BT_OBJECT_PUT_REF_AND_RESET(iterator);
773 goto end;
774 }
775
776 /* Create colander component */
777 colander_comp_cls = bt_component_class_sink_colander_get();
778 if (!colander_comp_cls) {
779 BT_LOGW("Cannot get colander sink component class.");
780 goto error;
781 }
782
783 iterator->graph = graph;
784 bt_object_get_no_null_check(iterator->graph);
785 colander_data.msgs = (void *) iterator->base.msgs->pdata;
786 colander_data.count_addr = &iterator->count;
787
788 /* Hope that nobody uses this very unique name */
789 graph_status =
790 bt_graph_add_sink_component_with_init_method_data(
791 (void *) graph, colander_comp_cls,
792 "colander-36ac3409-b1a8-4d60-ab1f-4fdf341a8fb1",
793 NULL, &colander_data, (void *) &iterator->colander);
794 if (graph_status != BT_GRAPH_STATUS_OK) {
795 BT_LIB_LOGW("Cannot add colander sink component to graph: "
796 "%1[graph-]+g, status=%s", graph,
797 bt_graph_status_string(graph_status));
798 goto error;
799 }
800
801 /*
802 * Connect provided output port to the colander component's
803 * input port.
804 */
805 colander_in_port =
806 (void *) bt_component_sink_borrow_input_port_by_index_const(
807 (void *) iterator->colander, 0);
808 BT_ASSERT(colander_in_port);
809 graph_status = bt_graph_connect_ports(graph,
810 output_port, colander_in_port, NULL);
811 if (graph_status != BT_GRAPH_STATUS_OK) {
812 BT_LIB_LOGW("Cannot add colander sink component to graph: "
813 "%![graph-]+g, %![comp-]+c, status=%s", graph,
814 iterator->colander,
815 bt_graph_status_string(graph_status));
816 goto error;
817 }
818
819 /*
820 * At this point everything went fine. Make the graph
821 * nonconsumable forever so that only this message iterator
822 * can consume (thanks to bt_graph_consume_sink_no_check()).
823 * This avoids leaking the message created by the colander
824 * sink and moved to the message iterator's message
825 * member.
826 */
827 bt_graph_set_can_consume(iterator->graph, false);
828 goto end;
829
830 error:
831 if (iterator && iterator->graph && iterator->colander) {
832 int ret;
833
834 /* Remove created colander component from graph if any */
835 colander_comp = iterator->colander;
836 BT_OBJECT_PUT_REF_AND_RESET(iterator->colander);
837
838 /*
839 * At this point the colander component's reference
840 * count is 0 because iterator->colander was the only
841 * owner. We also know that it is not connected because
842 * this is the last operation before this function
843 * succeeds.
844 *
845 * Since we honor the preconditions here,
846 * bt_graph_remove_unconnected_component() always
847 * succeeds.
848 */
849 ret = bt_graph_remove_unconnected_component(iterator->graph,
850 (void *) colander_comp);
851 BT_ASSERT(ret == 0);
852 }
853
854 BT_OBJECT_PUT_REF_AND_RESET(iterator);
855
856 end:
857 bt_object_put_ref(colander_comp_cls);
858 return (void *) iterator;
859 }
860
861 bt_bool bt_self_component_port_input_message_iterator_can_seek_ns_from_origin(
862 struct bt_self_component_port_input_message_iterator *iterator,
863 int64_t ns_from_origin)
864 {
865 bt_bool can = BT_FALSE;
866
867 BT_ASSERT_PRE_NON_NULL(iterator, "Message iterator");
868 BT_ASSERT_PRE_ITER_HAS_STATE_TO_SEEK(iterator);
869 BT_ASSERT_PRE(bt_component_borrow_graph(iterator->upstream_component)->is_configured,
870 "Graph is not configured: %!+g",
871 bt_component_borrow_graph(iterator->upstream_component));
872
873 if (iterator->methods.can_seek_ns_from_origin) {
874 can = iterator->methods.can_seek_ns_from_origin(iterator,
875 ns_from_origin);
876 goto end;
877 }
878
879 /*
880 * Automatic seeking fall back: if we can seek to the beginning,
881 * then we can automatically seek to any message.
882 */
883 if (iterator->methods.can_seek_beginning) {
884 can = iterator->methods.can_seek_beginning(iterator);
885 }
886
887 end:
888 return can;
889 }
890
891 bt_bool bt_self_component_port_input_message_iterator_can_seek_beginning(
892 struct bt_self_component_port_input_message_iterator *iterator)
893 {
894 bt_bool can = BT_FALSE;
895
896 BT_ASSERT_PRE_NON_NULL(iterator, "Message iterator");
897 BT_ASSERT_PRE_ITER_HAS_STATE_TO_SEEK(iterator);
898 BT_ASSERT_PRE(bt_component_borrow_graph(iterator->upstream_component)->is_configured,
899 "Graph is not configured: %!+g",
900 bt_component_borrow_graph(iterator->upstream_component));
901
902 if (iterator->methods.can_seek_beginning) {
903 can = iterator->methods.can_seek_beginning(iterator);
904 }
905
906 return can;
907 }
908
909 static inline
910 void _set_iterator_state_after_seeking(
911 struct bt_self_component_port_input_message_iterator *iterator,
912 enum bt_message_iterator_status status)
913 {
914 enum bt_self_component_port_input_message_iterator_state new_state = 0;
915
916 /* Set iterator's state depending on seeking status */
917 switch (status) {
918 case BT_MESSAGE_ITERATOR_STATUS_OK:
919 new_state = BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_ACTIVE;
920 break;
921 case BT_MESSAGE_ITERATOR_STATUS_AGAIN:
922 new_state = BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_LAST_SEEKING_RETURNED_AGAIN;
923 break;
924 case BT_MESSAGE_ITERATOR_STATUS_ERROR:
925 case BT_MESSAGE_ITERATOR_STATUS_NOMEM:
926 new_state = BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_LAST_SEEKING_RETURNED_ERROR;
927 break;
928 case BT_MESSAGE_ITERATOR_STATUS_END:
929 new_state = BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_ENDED;
930 break;
931 default:
932 abort();
933 }
934
935 set_self_comp_port_input_msg_iterator_state(iterator, new_state);
936 }
937
938 #ifdef BT_DEV_MODE
939 # define set_iterator_state_after_seeking _set_iterator_state_after_seeking
940 #else
941 # define set_iterator_state_after_seeking(_iter, _status)
942 #endif
943
944 enum bt_message_iterator_status
945 bt_self_component_port_input_message_iterator_seek_beginning(
946 struct bt_self_component_port_input_message_iterator *iterator)
947 {
948 int status;
949
950 BT_ASSERT_PRE_NON_NULL(iterator, "Message iterator");
951 BT_ASSERT_PRE_ITER_HAS_STATE_TO_SEEK(iterator);
952 BT_ASSERT_PRE(bt_component_borrow_graph(iterator->upstream_component)->is_configured,
953 "Graph is not configured: %!+g",
954 bt_component_borrow_graph(iterator->upstream_component));
955 BT_ASSERT_PRE(
956 bt_self_component_port_input_message_iterator_can_seek_beginning(
957 iterator),
958 "Message iterator cannot seek beginning: %!+i", iterator);
959 BT_LIB_LOGD("Calling user's \"seek beginning\" method: %!+i", iterator);
960 set_self_comp_port_input_msg_iterator_state(iterator,
961 BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_SEEKING);
962 status = iterator->methods.seek_beginning(iterator);
963 BT_LOGD("User method returned: status=%s",
964 bt_message_iterator_status_string(status));
965 BT_ASSERT_PRE(status == BT_MESSAGE_ITERATOR_STATUS_OK ||
966 status == BT_MESSAGE_ITERATOR_STATUS_ERROR ||
967 status == BT_MESSAGE_ITERATOR_STATUS_NOMEM ||
968 status == BT_MESSAGE_ITERATOR_STATUS_AGAIN,
969 "Unexpected status: %![iter-]+i, status=%s",
970 iterator, bt_self_message_iterator_status_string(status));
971 set_iterator_state_after_seeking(iterator, status);
972 return status;
973 }
974
975 static inline
976 int get_message_ns_from_origin(const struct bt_message *msg,
977 int64_t *ns_from_origin, bool *ignore)
978 {
979 const struct bt_clock_snapshot *clk_snapshot = NULL;
980 int ret = 0;
981
982 switch (msg->type) {
983 case BT_MESSAGE_TYPE_EVENT:
984 {
985 const struct bt_message_event *event_msg =
986 (const void *) msg;
987
988 clk_snapshot = event_msg->default_cs;
989 BT_ASSERT_PRE(clk_snapshot,
990 "Event has no default clock snapshot: %!+e",
991 event_msg->event);
992 break;
993 }
994 case BT_MESSAGE_TYPE_INACTIVITY:
995 {
996 const struct bt_message_inactivity *inactivity_msg =
997 (const void *) msg;
998
999 clk_snapshot = inactivity_msg->default_cs;
1000 BT_ASSERT(clk_snapshot);
1001 break;
1002 }
1003 case BT_MESSAGE_TYPE_PACKET_BEGINNING:
1004 case BT_MESSAGE_TYPE_PACKET_END:
1005 /* Ignore */
1006 goto end;
1007 case BT_MESSAGE_TYPE_DISCARDED_EVENTS:
1008 case BT_MESSAGE_TYPE_DISCARDED_PACKETS:
1009 {
1010 const struct bt_message_discarded_items *disc_items_msg =
1011 (const void *) msg;
1012
1013 clk_snapshot = disc_items_msg->default_begin_cs;
1014 BT_ASSERT_PRE(clk_snapshot,
1015 "Discarded events/packets message has no default clock snapshot: %!+n",
1016 msg);
1017 break;
1018 }
1019 case BT_MESSAGE_TYPE_STREAM_ACTIVITY_BEGINNING:
1020 {
1021 const struct bt_message_stream_activity *stream_act_msg =
1022 (const void *) msg;
1023
1024 switch (stream_act_msg->default_cs_state) {
1025 case BT_MESSAGE_STREAM_ACTIVITY_CLOCK_SNAPSHOT_STATE_UNKNOWN:
1026 case BT_MESSAGE_STREAM_ACTIVITY_CLOCK_SNAPSHOT_STATE_INFINITE:
1027 /*
1028 * -inf is always less than any requested time,
1029 * and we can't assume any specific time for an
1030 * unknown clock snapshot, so skip this.
1031 */
1032 goto set_ignore;
1033 case BT_MESSAGE_STREAM_ACTIVITY_CLOCK_SNAPSHOT_STATE_KNOWN:
1034 clk_snapshot = stream_act_msg->default_cs;
1035 BT_ASSERT(clk_snapshot);
1036 break;
1037 default:
1038 abort();
1039 }
1040
1041 break;
1042 }
1043 case BT_MESSAGE_TYPE_STREAM_ACTIVITY_END:
1044 {
1045 const struct bt_message_stream_activity *stream_act_msg =
1046 (const void *) msg;
1047
1048 switch (stream_act_msg->default_cs_state) {
1049 case BT_MESSAGE_STREAM_ACTIVITY_CLOCK_SNAPSHOT_STATE_UNKNOWN:
1050 /*
1051 * We can't assume any specific time for an
1052 * unknown clock snapshot, so skip this.
1053 */
1054 goto set_ignore;
1055 case BT_MESSAGE_STREAM_ACTIVITY_CLOCK_SNAPSHOT_STATE_INFINITE:
1056 /*
1057 * +inf is always greater than any requested
1058 * time.
1059 */
1060 *ns_from_origin = INT64_MAX;
1061 goto end;
1062 case BT_MESSAGE_STREAM_ACTIVITY_CLOCK_SNAPSHOT_STATE_KNOWN:
1063 clk_snapshot = stream_act_msg->default_cs;
1064 BT_ASSERT(clk_snapshot);
1065 break;
1066 default:
1067 abort();
1068 }
1069
1070 break;
1071 }
1072 case BT_MESSAGE_TYPE_STREAM_BEGINNING:
1073 case BT_MESSAGE_TYPE_STREAM_END:
1074 /* Ignore */
1075 break;
1076 default:
1077 abort();
1078 }
1079
1080 set_ignore:
1081 if (!clk_snapshot) {
1082 *ignore = true;
1083 goto end;
1084 }
1085
1086 ret = bt_clock_snapshot_get_ns_from_origin(clk_snapshot,
1087 ns_from_origin);
1088
1089 end:
1090 return ret;
1091 }
1092
1093 static
1094 enum bt_message_iterator_status find_message_ge_ns_from_origin(
1095 struct bt_self_component_port_input_message_iterator *iterator,
1096 int64_t ns_from_origin)
1097 {
1098 int status;
1099 enum bt_self_component_port_input_message_iterator_state init_state =
1100 iterator->state;
1101 const struct bt_message *messages[MSG_BATCH_SIZE];
1102 uint64_t user_count = 0;
1103 uint64_t i;
1104
1105 BT_ASSERT(iterator);
1106 memset(&messages[0], 0, sizeof(messages[0]) * MSG_BATCH_SIZE);
1107
1108 /*
1109 * Make this iterator temporarily active (not seeking) to call
1110 * the "next" method.
1111 */
1112 set_self_comp_port_input_msg_iterator_state(iterator,
1113 BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_ACTIVE);
1114
1115 BT_ASSERT(iterator->methods.next);
1116
1117 while (true) {
1118 /*
1119 * Call the user's "next" method to get the next
1120 * messages and status.
1121 */
1122 BT_LOGD_STR("Calling user's \"next\" method.");
1123 status = iterator->methods.next(iterator,
1124 &messages[0], MSG_BATCH_SIZE, &user_count);
1125 BT_LOGD("User method returned: status=%s",
1126 bt_message_iterator_status_string(status));
1127
1128 #ifdef BT_DEV_MODE
1129 /*
1130 * The user's "next" method must not do any action which
1131 * would change the iterator's state.
1132 */
1133 BT_ASSERT(iterator->state ==
1134 BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_ACTIVE);
1135 #endif
1136
1137 switch (status) {
1138 case BT_MESSAGE_ITERATOR_STATUS_OK:
1139 BT_ASSERT_PRE(user_count <= MSG_BATCH_SIZE,
1140 "Invalid returned message count: greater than "
1141 "batch size: count=%" PRIu64 ", batch-size=%u",
1142 user_count, MSG_BATCH_SIZE);
1143 break;
1144 case BT_MESSAGE_ITERATOR_STATUS_AGAIN:
1145 case BT_MESSAGE_ITERATOR_STATUS_ERROR:
1146 case BT_MESSAGE_ITERATOR_STATUS_NOMEM:
1147 case BT_MESSAGE_ITERATOR_STATUS_END:
1148 goto end;
1149 default:
1150 abort();
1151 }
1152
1153 /*
1154 * Find first message which has a default clock snapshot
1155 * that is greater than or equal to the requested value.
1156 *
1157 * For event and inactivity messages, compare with the
1158 * default clock snapshot.
1159 *
1160 * For packet beginning messages, compare with the
1161 * default beginning clock snapshot, if any.
1162 *
1163 * For packet end messages, compare with the default end
1164 * clock snapshot, if any.
1165 *
1166 * For stream beginning, stream end, ignore.
1167 */
1168 for (i = 0; i < user_count; i++) {
1169 const struct bt_message *msg = messages[i];
1170 int64_t msg_ns_from_origin;
1171 bool ignore = false;
1172 int ret;
1173
1174 BT_ASSERT(msg);
1175 ret = get_message_ns_from_origin(msg, &msg_ns_from_origin,
1176 &ignore);
1177 if (ret) {
1178 status = BT_MESSAGE_ITERATOR_STATUS_ERROR;
1179 goto end;
1180 }
1181
1182 if (ignore) {
1183 /* Skip message without a clock snapshot */
1184 continue;
1185 }
1186
1187 if (msg_ns_from_origin >= ns_from_origin) {
1188 /*
1189 * We found it: move this message and
1190 * the following ones to the iterator's
1191 * auto-seek message array.
1192 */
1193 uint64_t j;
1194
1195 for (j = i; j < user_count; j++) {
1196 iterator->auto_seek_msgs->pdata[j - i] =
1197 (void *) messages[j];
1198 messages[j] = NULL;
1199 }
1200
1201 iterator->auto_seek_msg_count = user_count - i;
1202 goto end;
1203 }
1204
1205 bt_object_put_no_null_check(msg);
1206 messages[i] = NULL;
1207 }
1208 }
1209
1210 end:
1211 for (i = 0; i < user_count; i++) {
1212 if (messages[i]) {
1213 bt_object_put_no_null_check(messages[i]);
1214 }
1215 }
1216
1217 set_self_comp_port_input_msg_iterator_state(iterator, init_state);
1218 return status;
1219 }
1220
1221 static
1222 enum bt_self_message_iterator_status post_auto_seek_next(
1223 struct bt_self_component_port_input_message_iterator *iterator,
1224 bt_message_array_const msgs, uint64_t capacity,
1225 uint64_t *count)
1226 {
1227 BT_ASSERT(iterator->auto_seek_msg_count <= capacity);
1228 BT_ASSERT(iterator->auto_seek_msg_count > 0);
1229
1230 /*
1231 * Move auto-seek messages to the output array (which is this
1232 * iterator's base message array.
1233 */
1234 memcpy(&msgs[0], &iterator->auto_seek_msgs->pdata[0],
1235 sizeof(msgs[0]) * iterator->auto_seek_msg_count);
1236 memset(&iterator->auto_seek_msgs->pdata[0], 0,
1237 sizeof(iterator->auto_seek_msgs->pdata[0]) *
1238 iterator->auto_seek_msg_count);
1239 *count = iterator->auto_seek_msg_count;
1240
1241 /* Restore real user's "next" method */
1242 switch (iterator->upstream_component->class->type) {
1243 case BT_COMPONENT_CLASS_TYPE_SOURCE:
1244 {
1245 struct bt_component_class_source *src_comp_cls =
1246 (void *) iterator->upstream_component->class;
1247
1248 iterator->methods.next =
1249 (bt_self_component_port_input_message_iterator_next_method)
1250 src_comp_cls->methods.msg_iter_next;
1251 break;
1252 }
1253 case BT_COMPONENT_CLASS_TYPE_FILTER:
1254 {
1255 struct bt_component_class_filter *flt_comp_cls =
1256 (void *) iterator->upstream_component->class;
1257
1258 iterator->methods.next =
1259 (bt_self_component_port_input_message_iterator_next_method)
1260 flt_comp_cls->methods.msg_iter_next;
1261 break;
1262 }
1263 default:
1264 abort();
1265 }
1266
1267 return BT_SELF_MESSAGE_ITERATOR_STATUS_OK;
1268 }
1269
1270 enum bt_message_iterator_status
1271 bt_self_component_port_input_message_iterator_seek_ns_from_origin(
1272 struct bt_self_component_port_input_message_iterator *iterator,
1273 int64_t ns_from_origin)
1274 {
1275 int status;
1276
1277 BT_ASSERT_PRE_NON_NULL(iterator, "Message iterator");
1278 BT_ASSERT_PRE_ITER_HAS_STATE_TO_SEEK(iterator);
1279 BT_ASSERT_PRE(bt_component_borrow_graph(iterator->upstream_component)->is_configured,
1280 "Graph is not configured: %!+g",
1281 bt_component_borrow_graph(iterator->upstream_component));
1282 BT_ASSERT_PRE(
1283 bt_self_component_port_input_message_iterator_can_seek_ns_from_origin(
1284 iterator, ns_from_origin),
1285 "Message iterator cannot seek nanoseconds from origin: %!+i, "
1286 "ns-from-origin=%" PRId64, iterator, ns_from_origin);
1287 set_self_comp_port_input_msg_iterator_state(iterator,
1288 BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_SEEKING);
1289
1290 if (iterator->methods.seek_ns_from_origin) {
1291 BT_LIB_LOGD("Calling user's \"seek nanoseconds from origin\" method: "
1292 "%![iter-]+i, ns=%" PRId64, iterator, ns_from_origin);
1293 status = iterator->methods.seek_ns_from_origin(iterator,
1294 ns_from_origin);
1295 BT_LOGD("User method returned: status=%s",
1296 bt_message_iterator_status_string(status));
1297 BT_ASSERT_PRE(status == BT_MESSAGE_ITERATOR_STATUS_OK ||
1298 status == BT_MESSAGE_ITERATOR_STATUS_ERROR ||
1299 status == BT_MESSAGE_ITERATOR_STATUS_NOMEM ||
1300 status == BT_MESSAGE_ITERATOR_STATUS_AGAIN,
1301 "Unexpected status: %![iter-]+i, status=%s",
1302 iterator,
1303 bt_self_message_iterator_status_string(status));
1304 } else {
1305 /* Start automatic seeking: seek beginning first */
1306 BT_ASSERT(iterator->methods.can_seek_beginning(iterator));
1307 BT_ASSERT(iterator->methods.seek_beginning);
1308 BT_LIB_LOGD("Calling user's \"seek beginning\" method: %!+i",
1309 iterator);
1310 status = iterator->methods.seek_beginning(iterator);
1311 BT_LOGD("User method returned: status=%s",
1312 bt_message_iterator_status_string(status));
1313 BT_ASSERT_PRE(status == BT_MESSAGE_ITERATOR_STATUS_OK ||
1314 status == BT_MESSAGE_ITERATOR_STATUS_ERROR ||
1315 status == BT_MESSAGE_ITERATOR_STATUS_NOMEM ||
1316 status == BT_MESSAGE_ITERATOR_STATUS_AGAIN,
1317 "Unexpected status: %![iter-]+i, status=%s",
1318 iterator,
1319 bt_self_message_iterator_status_string(status));
1320 switch (status) {
1321 case BT_MESSAGE_ITERATOR_STATUS_OK:
1322 break;
1323 case BT_MESSAGE_ITERATOR_STATUS_ERROR:
1324 case BT_MESSAGE_ITERATOR_STATUS_NOMEM:
1325 case BT_MESSAGE_ITERATOR_STATUS_AGAIN:
1326 goto end;
1327 default:
1328 abort();
1329 }
1330
1331 /*
1332 * Find the first message which has a default clock
1333 * snapshot greater than or equal to the requested
1334 * nanoseconds from origin, and move the received
1335 * messages from this point in the batch to this
1336 * iterator's auto-seek message array.
1337 */
1338 status = find_message_ge_ns_from_origin(iterator,
1339 ns_from_origin);
1340 switch (status) {
1341 case BT_MESSAGE_ITERATOR_STATUS_OK:
1342 /*
1343 * Replace the user's "next" method with a
1344 * custom, temporary "next" method which returns
1345 * the messages in the iterator's message array.
1346 */
1347 iterator->methods.next =
1348 (bt_self_component_port_input_message_iterator_next_method)
1349 post_auto_seek_next;
1350 break;
1351 case BT_MESSAGE_ITERATOR_STATUS_ERROR:
1352 case BT_MESSAGE_ITERATOR_STATUS_NOMEM:
1353 case BT_MESSAGE_ITERATOR_STATUS_AGAIN:
1354 goto end;
1355 case BT_MESSAGE_ITERATOR_STATUS_END:
1356 /*
1357 * The iterator reached the end: just return
1358 * `BT_MESSAGE_ITERATOR_STATUS_OK` here, as if
1359 * the seeking operation occured: the next
1360 * "next" method will return
1361 * `BT_MESSAGE_ITERATOR_STATUS_END` itself.
1362 */
1363 break;
1364 default:
1365 abort();
1366 }
1367 }
1368
1369 end:
1370 set_iterator_state_after_seeking(iterator, status);
1371
1372 if (status == BT_MESSAGE_ITERATOR_STATUS_END) {
1373 status = BT_MESSAGE_ITERATOR_STATUS_OK;
1374 }
1375
1376 return status;
1377 }
1378
1379 static inline
1380 bt_self_component_port_input_message_iterator *
1381 borrow_output_port_message_iterator_upstream_iterator(
1382 struct bt_port_output_message_iterator *iterator)
1383 {
1384 struct bt_component_class_sink_colander_priv_data *colander_data;
1385
1386 BT_ASSERT(iterator);
1387 colander_data = (void *) iterator->colander->parent.user_data;
1388 BT_ASSERT(colander_data);
1389 BT_ASSERT(colander_data->msg_iter);
1390 return colander_data->msg_iter;
1391 }
1392
1393 bt_bool bt_port_output_message_iterator_can_seek_ns_from_origin(
1394 struct bt_port_output_message_iterator *iterator,
1395 int64_t ns_from_origin)
1396 {
1397 BT_ASSERT_PRE_NON_NULL(iterator, "Message iterator");
1398 return bt_self_component_port_input_message_iterator_can_seek_ns_from_origin(
1399 borrow_output_port_message_iterator_upstream_iterator(
1400 iterator), ns_from_origin);
1401 }
1402
1403 bt_bool bt_port_output_message_iterator_can_seek_beginning(
1404 struct bt_port_output_message_iterator *iterator)
1405 {
1406 BT_ASSERT_PRE_NON_NULL(iterator, "Message iterator");
1407 return bt_self_component_port_input_message_iterator_can_seek_beginning(
1408 borrow_output_port_message_iterator_upstream_iterator(
1409 iterator));
1410 }
1411
1412 enum bt_message_iterator_status bt_port_output_message_iterator_seek_ns_from_origin(
1413 struct bt_port_output_message_iterator *iterator,
1414 int64_t ns_from_origin)
1415 {
1416 BT_ASSERT_PRE_NON_NULL(iterator, "Message iterator");
1417 return bt_self_component_port_input_message_iterator_seek_ns_from_origin(
1418 borrow_output_port_message_iterator_upstream_iterator(iterator),
1419 ns_from_origin);
1420 }
1421
1422 enum bt_message_iterator_status bt_port_output_message_iterator_seek_beginning(
1423 struct bt_port_output_message_iterator *iterator)
1424 {
1425 BT_ASSERT_PRE_NON_NULL(iterator, "Message iterator");
1426 return bt_self_component_port_input_message_iterator_seek_beginning(
1427 borrow_output_port_message_iterator_upstream_iterator(
1428 iterator));
1429 }
1430
1431 void bt_port_output_message_iterator_get_ref(
1432 const struct bt_port_output_message_iterator *iterator)
1433 {
1434 bt_object_get_ref(iterator);
1435 }
1436
1437 void bt_port_output_message_iterator_put_ref(
1438 const struct bt_port_output_message_iterator *iterator)
1439 {
1440 bt_object_put_ref(iterator);
1441 }
1442
1443 void bt_self_component_port_input_message_iterator_get_ref(
1444 const struct bt_self_component_port_input_message_iterator *iterator)
1445 {
1446 bt_object_get_ref(iterator);
1447 }
1448
1449 void bt_self_component_port_input_message_iterator_put_ref(
1450 const struct bt_self_component_port_input_message_iterator *iterator)
1451 {
1452 bt_object_put_ref(iterator);
1453 }
This page took 0.059371 seconds and 4 git commands to generate.