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