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