lib: rename bt_plugin_create_all_*() -> bt_plugin_find_all_*()
[babeltrace.git] / lib / graph / iterator.c
... / ...
CommitLineData
1/*
2 * Copyright 2017-2018 Philippe Proulx <pproulx@efficios.com>
3 * Copyright 2015 Jérémie Galarneau <jeremie.galarneau@efficios.com>
4 *
5 * Permission is hereby granted, free of charge, to any person obtaining a copy
6 * of this software and associated documentation files (the "Software"), to deal
7 * in the Software without restriction, including without limitation the rights
8 * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
9 * copies of the Software, and to permit persons to whom the Software is
10 * furnished to do so, subject to the following conditions:
11 *
12 * The above copyright notice and this permission notice shall be included in
13 * all copies or substantial portions of the Software.
14 *
15 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
16 * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
17 * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
18 * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
19 * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
20 * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
21 * SOFTWARE.
22 */
23
24#define BT_LOG_TAG "NOTIF-ITER"
25#include <babeltrace/lib-logging-internal.h>
26
27#include <babeltrace/compiler-internal.h>
28#include <babeltrace/trace-ir/field.h>
29#include <babeltrace/trace-ir/event-const.h>
30#include <babeltrace/trace-ir/event-internal.h>
31#include <babeltrace/trace-ir/packet-const.h>
32#include <babeltrace/trace-ir/packet-internal.h>
33#include <babeltrace/trace-ir/stream-internal.h>
34#include <babeltrace/graph/connection-const.h>
35#include <babeltrace/graph/connection-internal.h>
36#include <babeltrace/graph/component-const.h>
37#include <babeltrace/graph/component-internal.h>
38#include <babeltrace/graph/component-source-internal.h>
39#include <babeltrace/graph/component-class-internal.h>
40#include <babeltrace/graph/component-class-sink-colander-internal.h>
41#include <babeltrace/graph/component-sink-const.h>
42#include <babeltrace/graph/notification-const.h>
43#include <babeltrace/graph/notification-iterator.h>
44#include <babeltrace/graph/notification-iterator-internal.h>
45#include <babeltrace/graph/self-component-port-input-notification-iterator.h>
46#include <babeltrace/graph/port-output-notification-iterator.h>
47#include <babeltrace/graph/notification-internal.h>
48#include <babeltrace/graph/notification-event-const.h>
49#include <babeltrace/graph/notification-event-internal.h>
50#include <babeltrace/graph/notification-packet-const.h>
51#include <babeltrace/graph/notification-packet-internal.h>
52#include <babeltrace/graph/notification-stream-const.h>
53#include <babeltrace/graph/notification-stream-internal.h>
54#include <babeltrace/graph/port-const.h>
55#include <babeltrace/graph/graph.h>
56#include <babeltrace/graph/graph-const.h>
57#include <babeltrace/graph/graph-internal.h>
58#include <babeltrace/types.h>
59#include <babeltrace/assert-internal.h>
60#include <babeltrace/assert-pre-internal.h>
61#include <stdint.h>
62#include <inttypes.h>
63#include <stdlib.h>
64
65/*
66 * TODO: Use graph's state (number of active iterators, etc.) and
67 * possibly system specifications to make a better guess than this.
68 */
69#define NOTIF_BATCH_SIZE 15
70
71struct stream_state {
72 const struct bt_stream *stream; /* owned by this */
73 const struct bt_packet *cur_packet; /* owned by this */
74 uint64_t expected_notif_seq_num;
75 bt_bool is_ended;
76};
77
78BT_ASSERT_PRE_FUNC
79static
80void destroy_stream_state(struct stream_state *stream_state)
81{
82 if (!stream_state) {
83 return;
84 }
85
86 BT_LOGV("Destroying stream state: stream-state-addr=%p", stream_state);
87 BT_LOGV_STR("Putting stream state's current packet.");
88 BT_OBJECT_PUT_REF_AND_RESET(stream_state->cur_packet);
89 BT_LOGV_STR("Putting stream state's stream.");
90 BT_OBJECT_PUT_REF_AND_RESET(stream_state->stream);
91 g_free(stream_state);
92}
93
94BT_ASSERT_PRE_FUNC
95static
96struct stream_state *create_stream_state(const struct bt_stream *stream)
97{
98 struct stream_state *stream_state = g_new0(struct stream_state, 1);
99
100 if (!stream_state) {
101 BT_LOGE_STR("Failed to allocate one stream state.");
102 goto end;
103 }
104
105 /*
106 * We keep a reference to the stream until we know it's ended.
107 */
108 stream_state->stream = stream;
109 bt_object_get_no_null_check(stream_state->stream);
110 BT_LIB_LOGV("Created stream state: %![stream-]+s, "
111 "stream-state-addr=%p",
112 stream, stream_state);
113
114end:
115 return stream_state;
116}
117
118static
119void destroy_base_notification_iterator(struct bt_object *obj)
120{
121 struct bt_notification_iterator *iterator = (void *) obj;
122
123 BT_ASSERT(iterator);
124
125 if (iterator->notifs) {
126 g_ptr_array_free(iterator->notifs, TRUE);
127 iterator->notifs = NULL;
128 }
129
130 g_free(iterator);
131}
132
133static
134void bt_self_component_port_input_notification_iterator_destroy(struct bt_object *obj)
135{
136 struct bt_self_component_port_input_notification_iterator *iterator;
137
138 BT_ASSERT(obj);
139
140 /*
141 * The notification iterator's reference count is 0 if we're
142 * here. Increment it to avoid a double-destroy (possibly
143 * infinitely recursive). This could happen for example if the
144 * notification iterator's finalization function does
145 * bt_object_get_ref() (or anything that causes
146 * bt_object_get_ref() to be called) on itself (ref. count goes
147 * from 0 to 1), and then bt_object_put_ref(): the reference
148 * count would go from 1 to 0 again and this function would be
149 * called again.
150 */
151 obj->ref_count++;
152 iterator = (void *) obj;
153 BT_LIB_LOGD("Destroying self component input port notification iterator object: "
154 "%!+i", iterator);
155 bt_self_component_port_input_notification_iterator_finalize(iterator);
156
157 if (iterator->stream_states) {
158 /*
159 * Remove our destroy listener from each stream which
160 * has a state in this iterator. Otherwise the destroy
161 * listener would be called with an invalid/other
162 * notification iterator object.
163 */
164 g_hash_table_destroy(iterator->stream_states);
165 iterator->stream_states = NULL;
166 }
167
168 if (iterator->connection) {
169 /*
170 * Remove ourself from the originating connection so
171 * that it does not try to finalize a dangling pointer
172 * later.
173 */
174 bt_connection_remove_iterator(iterator->connection, iterator);
175 iterator->connection = NULL;
176 }
177
178 destroy_base_notification_iterator(obj);
179}
180
181BT_HIDDEN
182void bt_self_component_port_input_notification_iterator_finalize(
183 struct bt_self_component_port_input_notification_iterator *iterator)
184{
185 typedef void (*method_t)(void *);
186
187 struct bt_component_class *comp_class = NULL;
188 method_t method = NULL;
189
190 BT_ASSERT(iterator);
191
192 switch (iterator->state) {
193 case BT_SELF_COMPONENT_PORT_INPUT_NOTIFICATION_ITERATOR_STATE_NON_INITIALIZED:
194 /* Skip user finalization if user initialization failed */
195 BT_LIB_LOGD("Not finalizing non-initialized notification iterator: "
196 "%!+i", iterator);
197 return;
198 case BT_SELF_COMPONENT_PORT_INPUT_NOTIFICATION_ITERATOR_STATE_FINALIZED:
199 case BT_SELF_COMPONENT_PORT_INPUT_NOTIFICATION_ITERATOR_STATE_FINALIZED_AND_ENDED:
200 /* Already finalized */
201 BT_LIB_LOGD("Not finalizing notification iterator: already finalized: "
202 "%!+i", iterator);
203 return;
204 default:
205 break;
206 }
207
208 BT_LIB_LOGD("Finalizing notification iterator: %!+i", iterator);
209
210 if (iterator->state == BT_SELF_COMPONENT_PORT_INPUT_NOTIFICATION_ITERATOR_STATE_ENDED) {
211 BT_LIB_LOGD("Updating notification iterator's state: "
212 "new-state=BT_SELF_COMPONENT_PORT_INPUT_NOTIFICATION_ITERATOR_STATE_FINALIZED_AND_ENDED");
213 iterator->state = BT_SELF_COMPONENT_PORT_INPUT_NOTIFICATION_ITERATOR_STATE_FINALIZED_AND_ENDED;
214 } else {
215 BT_LIB_LOGD("Updating notification iterator's state: "
216 "new-state=BT_SELF_COMPONENT_PORT_INPUT_NOTIFICATION_ITERATOR_STATE_FINALIZED");
217 iterator->state = BT_SELF_COMPONENT_PORT_INPUT_NOTIFICATION_ITERATOR_STATE_FINALIZED;
218 }
219
220 BT_ASSERT(iterator->upstream_component);
221 comp_class = iterator->upstream_component->class;
222
223 /* Call user-defined destroy method */
224 switch (comp_class->type) {
225 case BT_COMPONENT_CLASS_TYPE_SOURCE:
226 {
227 struct bt_component_class_source *src_comp_cls =
228 (void *) comp_class;
229
230 method = (method_t) src_comp_cls->methods.notif_iter_finalize;
231 break;
232 }
233 case BT_COMPONENT_CLASS_TYPE_FILTER:
234 {
235 struct bt_component_class_filter *flt_comp_cls =
236 (void *) comp_class;
237
238 method = (method_t) flt_comp_cls->methods.notif_iter_finalize;
239 break;
240 }
241 default:
242 /* Unreachable */
243 abort();
244 }
245
246 if (method) {
247 BT_LIB_LOGD("Calling user's finalization method: %!+i",
248 iterator);
249 method(iterator);
250 }
251
252 iterator->upstream_component = NULL;
253 iterator->upstream_port = NULL;
254 BT_LIB_LOGD("Finalized notification iterator: %!+i", iterator);
255}
256
257BT_HIDDEN
258void bt_self_component_port_input_notification_iterator_set_connection(
259 struct bt_self_component_port_input_notification_iterator *iterator,
260 struct bt_connection *connection)
261{
262 BT_ASSERT(iterator);
263 iterator->connection = connection;
264 BT_LIB_LOGV("Set notification iterator's connection: "
265 "%![iter-]+i, %![conn-]+x", iterator, connection);
266}
267
268static
269int init_notification_iterator(struct bt_notification_iterator *iterator,
270 enum bt_notification_iterator_type type,
271 bt_object_release_func destroy)
272{
273 int ret = 0;
274
275 bt_object_init_shared(&iterator->base, destroy);
276 iterator->type = type;
277 iterator->notifs = g_ptr_array_new();
278 if (!iterator->notifs) {
279 BT_LOGE_STR("Failed to allocate a GPtrArray.");
280 ret = -1;
281 goto end;
282 }
283
284 g_ptr_array_set_size(iterator->notifs, NOTIF_BATCH_SIZE);
285
286end:
287 return ret;
288}
289
290static
291struct bt_self_component_port_input_notification_iterator *
292bt_self_component_port_input_notification_iterator_create_initial(
293 struct bt_component *upstream_comp,
294 struct bt_port *upstream_port)
295{
296 int ret;
297 struct bt_self_component_port_input_notification_iterator *iterator = NULL;
298
299 BT_ASSERT(upstream_comp);
300 BT_ASSERT(upstream_port);
301 BT_ASSERT(bt_port_is_connected(upstream_port));
302 BT_LIB_LOGD("Creating initial notification iterator on self component input port: "
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(
309 struct bt_self_component_port_input_notification_iterator, 1);
310 if (!iterator) {
311 BT_LOGE_STR("Failed to allocate one self component input port "
312 "notification iterator.");
313 goto end;
314 }
315
316 ret = init_notification_iterator((void *) iterator,
317 BT_NOTIFICATION_ITERATOR_TYPE_SELF_COMPONENT_PORT_INPUT,
318 bt_self_component_port_input_notification_iterator_destroy);
319 if (ret) {
320 /* init_notification_iterator() logs errors */
321 BT_OBJECT_PUT_REF_AND_RESET(iterator);
322 goto end;
323 }
324
325 iterator->stream_states = g_hash_table_new_full(g_direct_hash,
326 g_direct_equal, NULL, (GDestroyNotify) destroy_stream_state);
327 if (!iterator->stream_states) {
328 BT_LOGE_STR("Failed to allocate a GHashTable.");
329 BT_OBJECT_PUT_REF_AND_RESET(iterator);
330 goto end;
331 }
332
333 iterator->upstream_component = upstream_comp;
334 iterator->upstream_port = upstream_port;
335 iterator->connection = iterator->upstream_port->connection;
336 iterator->graph = bt_component_borrow_graph(upstream_comp);
337 iterator->state = BT_SELF_COMPONENT_PORT_INPUT_NOTIFICATION_ITERATOR_STATE_NON_INITIALIZED;
338 BT_LIB_LOGD("Created initial notification iterator on self component input port: "
339 "%![up-port-]+p, %![up-comp-]+c, %![iter-]+i",
340 upstream_port, upstream_comp, iterator);
341
342end:
343 return iterator;
344}
345
346struct bt_self_component_port_input_notification_iterator *
347bt_self_component_port_input_notification_iterator_create(
348 struct bt_self_component_port_input *self_port)
349{
350 typedef enum bt_self_notification_iterator_status (*init_method_t)(
351 void *, void *, void *);
352
353 init_method_t init_method = NULL;
354 struct bt_self_component_port_input_notification_iterator *iterator =
355 NULL;
356 struct bt_port *port = (void *) self_port;
357 struct bt_port *upstream_port;
358 struct bt_component *comp;
359 struct bt_component *upstream_comp;
360 struct bt_component_class *upstream_comp_cls;
361
362 BT_ASSERT_PRE_NON_NULL(port, "Port");
363 comp = bt_port_borrow_component_inline(port);
364 BT_ASSERT_PRE(bt_port_is_connected(port),
365 "Port is not connected: %![port-]+p", port);
366 BT_ASSERT_PRE(comp, "Port is not part of a component: %![port-]+p",
367 port);
368 BT_ASSERT_PRE(!bt_component_graph_is_canceled(comp),
369 "Port's component's graph is canceled: "
370 "%![port-]+p, %![comp-]+c", port, comp);
371 BT_ASSERT(port->connection);
372 upstream_port = port->connection->upstream_port;
373 BT_ASSERT(upstream_port);
374 upstream_comp = bt_port_borrow_component_inline(upstream_port);
375 BT_ASSERT(upstream_comp);
376 upstream_comp_cls = upstream_comp->class;
377 BT_ASSERT(upstream_comp->class->type ==
378 BT_COMPONENT_CLASS_TYPE_SOURCE ||
379 upstream_comp->class->type ==
380 BT_COMPONENT_CLASS_TYPE_FILTER);
381 iterator = bt_self_component_port_input_notification_iterator_create_initial(
382 upstream_comp, upstream_port);
383 if (!iterator) {
384 BT_LOGW_STR("Cannot create self component input port "
385 "notification iterator.");
386 goto end;
387 }
388
389 switch (upstream_comp_cls->type) {
390 case BT_COMPONENT_CLASS_TYPE_SOURCE:
391 {
392 struct bt_component_class_source *src_comp_cls =
393 (void *) upstream_comp_cls;
394
395 init_method =
396 (init_method_t) src_comp_cls->methods.notif_iter_init;
397 break;
398 }
399 case BT_COMPONENT_CLASS_TYPE_FILTER:
400 {
401 struct bt_component_class_filter *flt_comp_cls =
402 (void *) upstream_comp_cls;
403
404 init_method =
405 (init_method_t) flt_comp_cls->methods.notif_iter_init;
406 break;
407 }
408 default:
409 /* Unreachable */
410 abort();
411 }
412
413 if (init_method) {
414 int iter_status;
415
416 BT_LIB_LOGD("Calling user's initialization method: %!+i", iterator);
417 iter_status = init_method(iterator, upstream_comp,
418 upstream_port);
419 BT_LOGD("User method returned: status=%s",
420 bt_notification_iterator_status_string(iter_status));
421 if (iter_status != BT_NOTIFICATION_ITERATOR_STATUS_OK) {
422 BT_LOGW_STR("Initialization method failed.");
423 goto end;
424 }
425 }
426
427 iterator->state = BT_SELF_COMPONENT_PORT_INPUT_NOTIFICATION_ITERATOR_STATE_ACTIVE;
428 g_ptr_array_add(port->connection->iterators, iterator);
429 BT_LIB_LOGD("Created notification iterator on self component input port: "
430 "%![up-port-]+p, %![up-comp-]+c, %![iter-]+i",
431 upstream_port, upstream_comp, iterator);
432
433end:
434 return iterator;
435}
436
437void *bt_self_notification_iterator_get_data(
438 const struct bt_self_notification_iterator *self_iterator)
439{
440 struct bt_self_component_port_input_notification_iterator *iterator =
441 (void *) self_iterator;
442
443 BT_ASSERT_PRE_NON_NULL(iterator, "Notification iterator");
444 return iterator->user_data;
445}
446
447void bt_self_notification_iterator_set_data(
448 struct bt_self_notification_iterator *self_iterator, void *data)
449{
450 struct bt_self_component_port_input_notification_iterator *iterator =
451 (void *) self_iterator;
452
453 BT_ASSERT_PRE_NON_NULL(iterator, "Notification iterator");
454 iterator->user_data = data;
455 BT_LIB_LOGV("Set notification iterator's user data: "
456 "%!+i, user-data-addr=%p", iterator, data);
457}
458
459BT_ASSERT_PRE_FUNC
460static inline
461void bt_notification_borrow_packet_stream(const struct bt_notification *notif,
462 const struct bt_stream **stream,
463 const struct bt_packet **packet)
464{
465 BT_ASSERT(notif);
466
467 switch (notif->type) {
468 case BT_NOTIFICATION_TYPE_EVENT:
469 *packet = bt_event_borrow_packet_const(
470 bt_notification_event_borrow_event_const(notif));
471 *stream = bt_packet_borrow_stream_const(*packet);
472 break;
473 case BT_NOTIFICATION_TYPE_STREAM_BEGINNING:
474 *stream = bt_notification_stream_beginning_borrow_stream_const(notif);
475 break;
476 case BT_NOTIFICATION_TYPE_STREAM_END:
477 *stream = bt_notification_stream_end_borrow_stream_const(notif);
478 break;
479 case BT_NOTIFICATION_TYPE_PACKET_BEGINNING:
480 *packet = bt_notification_packet_beginning_borrow_packet_const(notif);
481 *stream = bt_packet_borrow_stream_const(*packet);
482 break;
483 case BT_NOTIFICATION_TYPE_PACKET_END:
484 *packet = bt_notification_packet_end_borrow_packet_const(notif);
485 *stream = bt_packet_borrow_stream_const(*packet);
486 break;
487 default:
488 break;
489 }
490}
491
492BT_ASSERT_PRE_FUNC
493static inline
494bool validate_notification(
495 struct bt_self_component_port_input_notification_iterator *iterator,
496 const struct bt_notification *c_notif)
497{
498 bool is_valid = true;
499 struct stream_state *stream_state;
500 const struct bt_stream *stream = NULL;
501 const struct bt_packet *packet = NULL;
502 struct bt_notification *notif = (void *) c_notif;
503
504 BT_ASSERT(notif);
505 bt_notification_borrow_packet_stream(c_notif, &stream, &packet);
506
507 if (!stream) {
508 /* we don't care about notifications not attached to streams */
509 goto end;
510 }
511
512 stream_state = g_hash_table_lookup(iterator->stream_states, stream);
513 if (!stream_state) {
514 /*
515 * No stream state for this stream: this notification
516 * MUST be a BT_NOTIFICATION_TYPE_STREAM_BEGINNING notification
517 * and its sequence number must be 0.
518 */
519 if (c_notif->type != BT_NOTIFICATION_TYPE_STREAM_BEGINNING) {
520 BT_ASSERT_PRE_MSG("Unexpected notification: missing a "
521 "BT_NOTIFICATION_TYPE_STREAM_BEGINNING "
522 "notification prior to this notification: "
523 "%![stream-]+s", stream);
524 is_valid = false;
525 goto end;
526 }
527
528 if (c_notif->seq_num == -1ULL) {
529 notif->seq_num = 0;
530 }
531
532 if (c_notif->seq_num != 0) {
533 BT_ASSERT_PRE_MSG("Unexpected notification sequence "
534 "number for this notification iterator: "
535 "this is the first notification for this "
536 "stream, expecting sequence number 0: "
537 "seq-num=%" PRIu64 ", %![stream-]+s",
538 c_notif->seq_num, stream);
539 is_valid = false;
540 goto end;
541 }
542
543 stream_state = create_stream_state(stream);
544 if (!stream_state) {
545 abort();
546 }
547
548 g_hash_table_insert(iterator->stream_states,
549 (void *) stream, stream_state);
550 stream_state->expected_notif_seq_num++;
551 goto end;
552 }
553
554 if (stream_state->is_ended) {
555 /*
556 * There's a new notification which has a reference to a
557 * stream which, from this iterator's point of view, is
558 * ended ("end of stream" notification was returned).
559 * This is bad: the API guarantees that it can never
560 * happen.
561 */
562 BT_ASSERT_PRE_MSG("Stream is already ended: %![stream-]+s",
563 stream);
564 is_valid = false;
565 goto end;
566 }
567
568 if (c_notif->seq_num == -1ULL) {
569 notif->seq_num = stream_state->expected_notif_seq_num;
570 }
571
572 if (c_notif->seq_num != -1ULL &&
573 c_notif->seq_num != stream_state->expected_notif_seq_num) {
574 BT_ASSERT_PRE_MSG("Unexpected notification sequence number: "
575 "seq-num=%" PRIu64 ", "
576 "expected-seq-num=%" PRIu64 ", %![stream-]+s",
577 c_notif->seq_num, stream_state->expected_notif_seq_num,
578 stream);
579 is_valid = false;
580 goto end;
581 }
582
583 switch (c_notif->type) {
584 case BT_NOTIFICATION_TYPE_STREAM_BEGINNING:
585 BT_ASSERT_PRE_MSG("Unexpected BT_NOTIFICATION_TYPE_STREAM_BEGINNING "
586 "notification at this point: notif-seq-num=%" PRIu64 ", "
587 "%![stream-]+s", c_notif->seq_num, stream);
588 is_valid = false;
589 goto end;
590 case BT_NOTIFICATION_TYPE_STREAM_END:
591 if (stream_state->cur_packet) {
592 BT_ASSERT_PRE_MSG("Unexpected BT_NOTIFICATION_TYPE_STREAM_END "
593 "notification: missing a "
594 "BT_NOTIFICATION_TYPE_PACKET_END notification "
595 "prior to this notification: "
596 "notif-seq-num=%" PRIu64 ", "
597 "%![stream-]+s", c_notif->seq_num, stream);
598 is_valid = false;
599 goto end;
600 }
601 stream_state->expected_notif_seq_num++;
602 stream_state->is_ended = true;
603 goto end;
604 case BT_NOTIFICATION_TYPE_PACKET_BEGINNING:
605 if (stream_state->cur_packet) {
606 BT_ASSERT_PRE_MSG("Unexpected BT_NOTIFICATION_TYPE_PACKET_BEGINNING "
607 "notification at this point: missing a "
608 "BT_NOTIFICATION_TYPE_PACKET_END notification "
609 "prior to this notification: "
610 "notif-seq-num=%" PRIu64 ", %![stream-]+s, "
611 "%![packet-]+a", c_notif->seq_num, stream,
612 packet);
613 is_valid = false;
614 goto end;
615 }
616 stream_state->expected_notif_seq_num++;
617 stream_state->cur_packet = packet;
618 bt_object_get_no_null_check(stream_state->cur_packet);
619 goto end;
620 case BT_NOTIFICATION_TYPE_PACKET_END:
621 if (!stream_state->cur_packet) {
622 BT_ASSERT_PRE_MSG("Unexpected BT_NOTIFICATION_TYPE_PACKET_END "
623 "notification at this point: missing a "
624 "BT_NOTIFICATION_TYPE_PACKET_BEGINNING notification "
625 "prior to this notification: "
626 "notif-seq-num=%" PRIu64 ", %![stream-]+s, "
627 "%![packet-]+a", c_notif->seq_num, stream,
628 packet);
629 is_valid = false;
630 goto end;
631 }
632 stream_state->expected_notif_seq_num++;
633 BT_OBJECT_PUT_REF_AND_RESET(stream_state->cur_packet);
634 goto end;
635 case BT_NOTIFICATION_TYPE_EVENT:
636 if (packet != stream_state->cur_packet) {
637 BT_ASSERT_PRE_MSG("Unexpected packet for "
638 "BT_NOTIFICATION_TYPE_EVENT notification: "
639 "notif-seq-num=%" PRIu64 ", %![stream-]+s, "
640 "%![notif-packet-]+a, %![expected-packet-]+a",
641 c_notif->seq_num, stream,
642 stream_state->cur_packet, packet);
643 is_valid = false;
644 goto end;
645 }
646 stream_state->expected_notif_seq_num++;
647 goto end;
648 default:
649 break;
650 }
651
652end:
653 return is_valid;
654}
655
656BT_ASSERT_PRE_FUNC
657static inline
658bool validate_notifications(
659 struct bt_self_component_port_input_notification_iterator *iterator,
660 uint64_t count)
661{
662 bool ret = true;
663 bt_notification_array_const notifs =
664 (void *) iterator->base.notifs->pdata;
665 uint64_t i;
666
667 for (i = 0; i < count; i++) {
668 ret = validate_notification(iterator, notifs[i]);
669 if (!ret) {
670 break;
671 }
672 }
673
674 return ret;
675}
676
677BT_ASSERT_PRE_FUNC
678static inline bool self_comp_port_input_notif_iter_can_end(
679 struct bt_self_component_port_input_notification_iterator *iterator)
680{
681 GHashTableIter iter;
682 gpointer stream_key, state_value;
683 bool ret = true;
684
685 /*
686 * Verify that this iterator received a
687 * BT_NOTIFICATION_TYPE_STREAM_END notification for each stream
688 * which has a state.
689 */
690
691 g_hash_table_iter_init(&iter, iterator->stream_states);
692
693 while (g_hash_table_iter_next(&iter, &stream_key, &state_value)) {
694 struct stream_state *stream_state = (void *) state_value;
695
696 BT_ASSERT(stream_state);
697 BT_ASSERT(stream_key);
698
699 if (!stream_state->is_ended) {
700 BT_ASSERT_PRE_MSG("Ending notification iterator, "
701 "but stream is not ended: "
702 "%![stream-]s", stream_key);
703 ret = false;
704 goto end;
705 }
706 }
707
708end:
709 return ret;
710}
711
712enum bt_notification_iterator_status
713bt_self_component_port_input_notification_iterator_next(
714 struct bt_self_component_port_input_notification_iterator *iterator,
715 bt_notification_array_const *notifs, uint64_t *user_count)
716{
717 typedef enum bt_self_notification_iterator_status (*method_t)(
718 void *, bt_notification_array_const, uint64_t, uint64_t *);
719
720 method_t method = NULL;
721 struct bt_component_class *comp_cls;
722 int status = BT_NOTIFICATION_ITERATOR_STATUS_OK;
723
724 BT_ASSERT_PRE_NON_NULL(iterator, "Notification iterator");
725 BT_ASSERT_PRE_NON_NULL(notifs, "Notification array (output)");
726 BT_ASSERT_PRE_NON_NULL(user_count, "Notification count (output)");
727 BT_ASSERT_PRE(iterator->state ==
728 BT_SELF_COMPONENT_PORT_INPUT_NOTIFICATION_ITERATOR_STATE_ACTIVE,
729 "Notification iterator's \"next\" called, but "
730 "iterator is in the wrong state: %!+i", iterator);
731 BT_ASSERT(iterator->upstream_component);
732 BT_ASSERT(iterator->upstream_component->class);
733 BT_LIB_LOGD("Getting next self component input port "
734 "notification iterator's notifications: %!+i", iterator);
735 comp_cls = iterator->upstream_component->class;
736
737 /* Pick the appropriate "next" method */
738 switch (comp_cls->type) {
739 case BT_COMPONENT_CLASS_TYPE_SOURCE:
740 {
741 struct bt_component_class_source *src_comp_cls =
742 (void *) comp_cls;
743
744 method = (method_t) src_comp_cls->methods.notif_iter_next;
745 break;
746 }
747 case BT_COMPONENT_CLASS_TYPE_FILTER:
748 {
749 struct bt_component_class_filter *flt_comp_cls =
750 (void *) comp_cls;
751
752 method = (method_t) flt_comp_cls->methods.notif_iter_next;
753 break;
754 }
755 default:
756 abort();
757 }
758
759 /*
760 * Call the user's "next" method to get the next notifications
761 * and status.
762 */
763 BT_ASSERT(method);
764 BT_LOGD_STR("Calling user's \"next\" method.");
765 status = method(iterator,
766 (void *) iterator->base.notifs->pdata,
767 NOTIF_BATCH_SIZE, user_count);
768 BT_LOGD("User method returned: status=%s",
769 bt_notification_iterator_status_string(status));
770 if (status < 0) {
771 BT_LOGW_STR("User method failed.");
772 goto end;
773 }
774
775 if (iterator->state == BT_SELF_COMPONENT_PORT_INPUT_NOTIFICATION_ITERATOR_STATE_FINALIZED ||
776 iterator->state == BT_SELF_COMPONENT_PORT_INPUT_NOTIFICATION_ITERATOR_STATE_FINALIZED_AND_ENDED) {
777 /*
778 * The user's "next" method, somehow, cancelled its own
779 * notification iterator. This can happen, for example,
780 * when the user's method removes the port on which
781 * there's the connection from which the iterator was
782 * created. In this case, said connection is ended, and
783 * all its notification iterators are finalized.
784 *
785 * Only bt_object_put_ref() the returned notification if
786 * the status is BT_NOTIFICATION_ITERATOR_STATUS_OK
787 * because otherwise this field could be garbage.
788 */
789 if (status == BT_NOTIFICATION_ITERATOR_STATUS_OK) {
790 uint64_t i;
791 bt_notification_array_const notifs =
792 (void *) iterator->base.notifs->pdata;
793
794 for (i = 0; i < *user_count; i++) {
795 bt_object_put_ref(notifs[i]);
796 }
797 }
798
799 status = BT_NOTIFICATION_ITERATOR_STATUS_CANCELED;
800 goto end;
801 }
802
803 switch (status) {
804 case BT_NOTIFICATION_ITERATOR_STATUS_OK:
805 BT_ASSERT_PRE(validate_notifications(iterator, *user_count),
806 "Notifications are invalid at this point: "
807 "%![notif-iter-]+i, count=%" PRIu64,
808 iterator, *user_count);
809 *notifs = (void *) iterator->base.notifs->pdata;
810 break;
811 case BT_NOTIFICATION_ITERATOR_STATUS_AGAIN:
812 goto end;
813 case BT_NOTIFICATION_ITERATOR_STATUS_END:
814 BT_ASSERT_PRE(self_comp_port_input_notif_iter_can_end(iterator),
815 "Notification iterator cannot end at this point: "
816 "%!+i", iterator);
817 BT_ASSERT(iterator->state ==
818 BT_SELF_COMPONENT_PORT_INPUT_NOTIFICATION_ITERATOR_STATE_ACTIVE);
819 iterator->state = BT_SELF_COMPONENT_PORT_INPUT_NOTIFICATION_ITERATOR_STATE_ENDED;
820 BT_LOGD("Set new status: status=%s",
821 bt_notification_iterator_status_string(status));
822 goto end;
823 default:
824 /* Unknown non-error status */
825 abort();
826 }
827
828end:
829 return status;
830}
831
832enum bt_notification_iterator_status
833bt_port_output_notification_iterator_next(
834 struct bt_port_output_notification_iterator *iterator,
835 bt_notification_array_const *notifs_to_user,
836 uint64_t *count_to_user)
837{
838 enum bt_notification_iterator_status status;
839 enum bt_graph_status graph_status;
840
841 BT_ASSERT_PRE_NON_NULL(iterator, "Notification iterator");
842 BT_ASSERT_PRE_NON_NULL(notifs_to_user, "Notification array (output)");
843 BT_ASSERT_PRE_NON_NULL(count_to_user, "Notification count (output)");
844 BT_LIB_LOGD("Getting next output port notification iterator's notifications: "
845 "%!+i", iterator);
846
847 graph_status = bt_graph_consume_sink_no_check(iterator->graph,
848 iterator->colander);
849 switch (graph_status) {
850 case BT_GRAPH_STATUS_CANCELED:
851 case BT_GRAPH_STATUS_AGAIN:
852 case BT_GRAPH_STATUS_END:
853 case BT_GRAPH_STATUS_NOMEM:
854 status = (int) graph_status;
855 break;
856 case BT_GRAPH_STATUS_OK:
857 status = BT_NOTIFICATION_ITERATOR_STATUS_OK;
858
859 /*
860 * On success, the colander sink moves the notifications
861 * to this iterator's array and sets this iterator's
862 * notification count: move them to the user.
863 */
864 *notifs_to_user = (void *) iterator->base.notifs->pdata;
865 *count_to_user = iterator->count;
866 break;
867 default:
868 /* Other errors */
869 status = BT_NOTIFICATION_ITERATOR_STATUS_ERROR;
870 }
871
872 return status;
873}
874
875struct bt_component *bt_self_component_port_input_notification_iterator_borrow_component(
876 struct bt_self_component_port_input_notification_iterator *iterator)
877{
878 BT_ASSERT_PRE_NON_NULL(iterator, "Notification iterator");
879 return iterator->upstream_component;
880}
881
882struct bt_self_component *bt_self_notification_iterator_borrow_component(
883 struct bt_self_notification_iterator *self_iterator)
884{
885 struct bt_self_component_port_input_notification_iterator *iterator =
886 (void *) self_iterator;
887
888 BT_ASSERT_PRE_NON_NULL(iterator, "Notification iterator");
889 return (void *) iterator->upstream_component;
890}
891
892struct bt_self_port_output *bt_self_notification_iterator_borrow_port(
893 struct bt_self_notification_iterator *self_iterator)
894{
895 struct bt_self_component_port_input_notification_iterator *iterator =
896 (void *) self_iterator;
897
898 BT_ASSERT_PRE_NON_NULL(iterator, "Notification iterator");
899 return (void *) iterator->upstream_port;
900}
901
902static
903void bt_port_output_notification_iterator_destroy(struct bt_object *obj)
904{
905 struct bt_port_output_notification_iterator *iterator = (void *) obj;
906
907 BT_LIB_LOGD("Destroying output port notification iterator object: %!+i",
908 iterator);
909 BT_LOGD_STR("Putting graph.");
910 BT_OBJECT_PUT_REF_AND_RESET(iterator->graph);
911 BT_LOGD_STR("Putting colander sink component.");
912 BT_OBJECT_PUT_REF_AND_RESET(iterator->colander);
913 destroy_base_notification_iterator(obj);
914}
915
916struct bt_port_output_notification_iterator *
917bt_port_output_notification_iterator_create(
918 struct bt_graph *graph,
919 const struct bt_port_output *output_port)
920{
921 struct bt_port_output_notification_iterator *iterator = NULL;
922 struct bt_component_class_sink *colander_comp_cls = NULL;
923 struct bt_component *output_port_comp = NULL;
924 struct bt_component_sink *colander_comp;
925 enum bt_graph_status graph_status;
926 struct bt_port_input *colander_in_port = NULL;
927 struct bt_component_class_sink_colander_data colander_data;
928 int ret;
929
930 BT_ASSERT_PRE_NON_NULL(graph, "Graph");
931 BT_ASSERT_PRE_NON_NULL(output_port, "Output port");
932 output_port_comp = bt_port_borrow_component_inline(
933 (const void *) output_port);
934 BT_ASSERT_PRE(output_port_comp,
935 "Output port has no component: %!+p", output_port);
936 BT_ASSERT_PRE(bt_component_borrow_graph(output_port_comp) ==
937 (void *) graph,
938 "Output port is not part of graph: %![graph-]+g, %![port-]+p",
939 graph, output_port);
940
941 /* Create notification iterator */
942 BT_LIB_LOGD("Creating notification iterator on output port: "
943 "%![port-]+p, %![comp-]+c", output_port, output_port_comp);
944 iterator = g_new0(struct bt_port_output_notification_iterator, 1);
945 if (!iterator) {
946 BT_LOGE_STR("Failed to allocate one output port notification iterator.");
947 goto error;
948 }
949
950 ret = init_notification_iterator((void *) iterator,
951 BT_NOTIFICATION_ITERATOR_TYPE_PORT_OUTPUT,
952 bt_port_output_notification_iterator_destroy);
953 if (ret) {
954 /* init_notification_iterator() logs errors */
955 BT_OBJECT_PUT_REF_AND_RESET(iterator);
956 goto end;
957 }
958
959 /* Create colander component */
960 colander_comp_cls = bt_component_class_sink_colander_get();
961 if (!colander_comp_cls) {
962 BT_LOGW("Cannot get colander sink component class.");
963 goto error;
964 }
965
966 iterator->graph = graph;
967 bt_object_get_no_null_check(iterator->graph);
968 colander_data.notifs = (void *) iterator->base.notifs->pdata;
969 colander_data.count_addr = &iterator->count;
970
971 /* Hope that nobody uses this very unique name */
972 graph_status =
973 bt_graph_add_sink_component_with_init_method_data(
974 (void *) graph, colander_comp_cls,
975 "colander-36ac3409-b1a8-4d60-ab1f-4fdf341a8fb1",
976 NULL, &colander_data, (void *) &iterator->colander);
977 if (graph_status != BT_GRAPH_STATUS_OK) {
978 BT_LIB_LOGW("Cannot add colander sink component to graph: "
979 "%1[graph-]+g, status=%s", graph,
980 bt_graph_status_string(graph_status));
981 goto error;
982 }
983
984 /*
985 * Connect provided output port to the colander component's
986 * input port.
987 */
988 colander_in_port =
989 (void *) bt_component_sink_borrow_input_port_by_index_const(
990 (void *) iterator->colander, 0);
991 BT_ASSERT(colander_in_port);
992 graph_status = bt_graph_connect_ports(graph,
993 output_port, colander_in_port, NULL);
994 if (graph_status != BT_GRAPH_STATUS_OK) {
995 BT_LIB_LOGW("Cannot add colander sink component to graph: "
996 "%![graph-]+g, %![comp-]+c, status=%s", graph,
997 iterator->colander,
998 bt_graph_status_string(graph_status));
999 goto error;
1000 }
1001
1002 /*
1003 * At this point everything went fine. Make the graph
1004 * nonconsumable forever so that only this notification iterator
1005 * can consume (thanks to bt_graph_consume_sink_no_check()).
1006 * This avoids leaking the notification created by the colander
1007 * sink and moved to the notification iterator's notification
1008 * member.
1009 */
1010 bt_graph_set_can_consume(iterator->graph, false);
1011 goto end;
1012
1013error:
1014 if (iterator && iterator->graph && iterator->colander) {
1015 int ret;
1016
1017 /* Remove created colander component from graph if any */
1018 colander_comp = iterator->colander;
1019 BT_OBJECT_PUT_REF_AND_RESET(iterator->colander);
1020
1021 /*
1022 * At this point the colander component's reference
1023 * count is 0 because iterator->colander was the only
1024 * owner. We also know that it is not connected because
1025 * this is the last operation before this function
1026 * succeeds.
1027 *
1028 * Since we honor the preconditions here,
1029 * bt_graph_remove_unconnected_component() always
1030 * succeeds.
1031 */
1032 ret = bt_graph_remove_unconnected_component(iterator->graph,
1033 (void *) colander_comp);
1034 BT_ASSERT(ret == 0);
1035 }
1036
1037 BT_OBJECT_PUT_REF_AND_RESET(iterator);
1038
1039end:
1040 bt_object_put_ref(colander_comp_cls);
1041 return (void *) iterator;
1042}
1043
1044void bt_port_output_notification_iterator_get_ref(
1045 const struct bt_port_output_notification_iterator *iterator)
1046{
1047 bt_object_get_ref(iterator);
1048}
1049
1050void bt_port_output_notification_iterator_put_ref(
1051 const struct bt_port_output_notification_iterator *iterator)
1052{
1053 bt_object_put_ref(iterator);
1054}
1055
1056void bt_self_component_port_input_notification_iterator_get_ref(
1057 const struct bt_self_component_port_input_notification_iterator *iterator)
1058{
1059 bt_object_get_ref(iterator);
1060}
1061
1062void bt_self_component_port_input_notification_iterator_put_ref(
1063 const struct bt_self_component_port_input_notification_iterator *iterator)
1064{
1065 bt_object_put_ref(iterator);
1066}
This page took 0.029043 seconds and 4 git commands to generate.