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