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