bt_port_output_notification_iterator_create(): remove colander comp. name
[babeltrace.git] / lib / graph / iterator.c
CommitLineData
47e5a032 1/*
47e5a032 2 * Copyright 2015 Jérémie Galarneau <jeremie.galarneau@efficios.com>
3230ee6b 3 * Copyright 2017 Philippe Proulx <pproulx@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
5af447e5
PP
24#define BT_LOG_TAG "NOTIF-ITER"
25#include <babeltrace/lib-logging-internal.h>
26
3d9990ac 27#include <babeltrace/compiler-internal.h>
65300d60 28#include <babeltrace/object.h>
56e18c4c 29#include <babeltrace/trace-ir/fields.h>
56e18c4c
PP
30#include <babeltrace/trace-ir/event-internal.h>
31#include <babeltrace/trace-ir/packet-internal.h>
32#include <babeltrace/trace-ir/stream-internal.h>
73d5c1ad 33#include <babeltrace/graph/connection.h>
bd14d768 34#include <babeltrace/graph/connection-internal.h>
b2e0c907 35#include <babeltrace/graph/component.h>
e5be10ef 36#include <babeltrace/graph/component-internal.h>
b2e0c907
PP
37#include <babeltrace/graph/component-source-internal.h>
38#include <babeltrace/graph/component-class-internal.h>
8ed535b5
PP
39#include <babeltrace/graph/component-class-sink-colander-internal.h>
40#include <babeltrace/graph/component-sink.h>
fa054faf 41#include <babeltrace/graph/notification.h>
b2e0c907
PP
42#include <babeltrace/graph/notification-iterator.h>
43#include <babeltrace/graph/notification-iterator-internal.h>
d94d92ac
PP
44#include <babeltrace/graph/self-component-port-input-notification-iterator.h>
45#include <babeltrace/graph/port-output-notification-iterator.h>
e7fa96c3 46#include <babeltrace/graph/notification-internal.h>
3230ee6b
PP
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>
a2d06fd5 54#include <babeltrace/graph/private-graph.h>
8ed535b5 55#include <babeltrace/graph/graph-internal.h>
c55a9f58 56#include <babeltrace/types.h>
f6ccaed9 57#include <babeltrace/assert-internal.h>
f42867e2 58#include <babeltrace/assert-pre-internal.h>
fa054faf 59#include <stdint.h>
2ec84d26 60#include <inttypes.h>
0fbb9a9f 61#include <stdlib.h>
3230ee6b 62
d4393e08
PP
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
3230ee6b 69struct stream_state {
50842bdc
PP
70 struct bt_stream *stream; /* owned by this */
71 struct bt_packet *cur_packet; /* owned by this */
f42867e2 72 uint64_t expected_notif_seq_num;
c55a9f58 73 bt_bool is_ended;
3230ee6b
PP
74};
75
26e21a82 76BT_ASSERT_PRE_FUNC
3230ee6b
PP
77static
78void destroy_stream_state(struct stream_state *stream_state)
79{
80 if (!stream_state) {
81 return;
82 }
83
5af447e5
PP
84 BT_LOGV("Destroying stream state: stream-state-addr=%p", stream_state);
85 BT_LOGV_STR("Putting stream state's current packet.");
d94d92ac 86 BT_OBJECT_PUT_REF_AND_RESET(stream_state->cur_packet);
5af447e5 87 BT_LOGV_STR("Putting stream state's stream.");
d94d92ac 88 BT_OBJECT_PUT_REF_AND_RESET(stream_state->stream);
3230ee6b
PP
89 g_free(stream_state);
90}
91
26e21a82 92BT_ASSERT_PRE_FUNC
3230ee6b 93static
50842bdc 94struct stream_state *create_stream_state(struct bt_stream *stream)
3230ee6b
PP
95{
96 struct stream_state *stream_state = g_new0(struct stream_state, 1);
97
98 if (!stream_state) {
5af447e5 99 BT_LOGE_STR("Failed to allocate one stream state.");
3230ee6b
PP
100 goto end;
101 }
102
103 /*
f42867e2 104 * We keep a reference to the stream until we know it's ended.
3230ee6b 105 */
65300d60 106 stream_state->stream = bt_object_get_ref(stream);
d94d92ac 107 BT_LIB_LOGV("Created stream state: %![stream-]+s, "
5af447e5 108 "stream-state-addr=%p",
d94d92ac 109 stream, stream_state);
3230ee6b
PP
110
111end:
112 return stream_state;
113}
47e5a032 114
8ed535b5
PP
115static
116void destroy_base_notification_iterator(struct bt_object *obj)
117{
d4393e08
PP
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);
d94d92ac 124 iterator->notifs = NULL;
d4393e08
PP
125 }
126
127 g_free(iterator);
8ed535b5
PP
128}
129
47e5a032 130static
d94d92ac 131void bt_self_component_port_input_notification_iterator_destroy(struct bt_object *obj)
47e5a032 132{
d94d92ac 133 struct bt_self_component_port_input_notification_iterator *iterator;
8738a040 134
f6ccaed9 135 BT_ASSERT(obj);
d3eb6e8f 136
bd14d768
PP
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
d94d92ac
PP
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.
bd14d768 147 */
3fea54f6 148 obj->ref_count++;
07245ac2 149 iterator = (void *) obj;
d94d92ac
PP
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);
d3eb6e8f 153
3230ee6b
PP
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 */
3230ee6b 161 g_hash_table_destroy(iterator->stream_states);
d94d92ac 162 iterator->stream_states = NULL;
3230ee6b
PP
163 }
164
bd14d768
PP
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);
d94d92ac 172 iterator->connection = NULL;
bd14d768
PP
173 }
174
8ed535b5 175 destroy_base_notification_iterator(obj);
47e5a032
JG
176}
177
bd14d768 178BT_HIDDEN
d94d92ac
PP
179void bt_self_component_port_input_notification_iterator_finalize(
180 struct bt_self_component_port_input_notification_iterator *iterator)
bd14d768 181{
d94d92ac
PP
182 typedef void (*method_t)(void *);
183
bd14d768 184 struct bt_component_class *comp_class = NULL;
d94d92ac 185 method_t method = NULL;
bd14d768 186
f6ccaed9 187 BT_ASSERT(iterator);
bd14d768
PP
188
189 switch (iterator->state) {
d94d92ac 190 case BT_SELF_COMPONENT_PORT_INPUT_NOTIFICATION_ITERATOR_STATE_NON_INITIALIZED:
088d4023 191 /* Skip user finalization if user initialization failed */
d94d92ac
PP
192 BT_LIB_LOGD("Not finalizing non-initialized notification iterator: "
193 "%!+i", iterator);
088d4023 194 return;
d94d92ac
PP
195 case BT_SELF_COMPONENT_PORT_INPUT_NOTIFICATION_ITERATOR_STATE_FINALIZED:
196 case BT_SELF_COMPONENT_PORT_INPUT_NOTIFICATION_ITERATOR_STATE_FINALIZED_AND_ENDED:
bd14d768 197 /* Already finalized */
d94d92ac
PP
198 BT_LIB_LOGD("Not finalizing notification iterator: already finalized: "
199 "%!+i", iterator);
bd14d768
PP
200 return;
201 default:
202 break;
203 }
204
d94d92ac 205 BT_LIB_LOGD("Finalizing notification iterator: %!+i", iterator);
5af447e5 206
d94d92ac
PP
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;
df14f8af 211 } else {
d94d92ac
PP
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;
df14f8af
MD
215 }
216
f6ccaed9 217 BT_ASSERT(iterator->upstream_component);
bd14d768
PP
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 {
d94d92ac
PP
224 struct bt_component_class_source *src_comp_cls =
225 (void *) comp_class;
bd14d768 226
d94d92ac 227 method = (method_t) src_comp_cls->methods.notif_iter_finalize;
bd14d768
PP
228 break;
229 }
230 case BT_COMPONENT_CLASS_TYPE_FILTER:
231 {
d94d92ac
PP
232 struct bt_component_class_filter *flt_comp_cls =
233 (void *) comp_class;
bd14d768 234
d94d92ac 235 method = (method_t) flt_comp_cls->methods.notif_iter_finalize;
bd14d768
PP
236 break;
237 }
238 default:
239 /* Unreachable */
0fbb9a9f 240 abort();
bd14d768
PP
241 }
242
d94d92ac
PP
243 if (method) {
244 BT_LIB_LOGD("Calling user's finalization method: %!+i",
5af447e5 245 iterator);
d94d92ac 246 method(iterator);
bd14d768
PP
247 }
248
bd14d768
PP
249 iterator->upstream_component = NULL;
250 iterator->upstream_port = NULL;
d94d92ac 251 BT_LIB_LOGD("Finalized notification iterator: %!+i", iterator);
bd14d768
PP
252}
253
254BT_HIDDEN
d94d92ac
PP
255void bt_self_component_port_input_notification_iterator_set_connection(
256 struct bt_self_component_port_input_notification_iterator *iterator,
bd14d768
PP
257 struct bt_connection *connection)
258{
f6ccaed9 259 BT_ASSERT(iterator);
bd14d768 260 iterator->connection = connection;
d94d92ac
PP
261 BT_LIB_LOGV("Set notification iterator's connection: "
262 "%![iter-]+i, %![conn-]+x", iterator, connection);
bd14d768
PP
263}
264
90157d89 265static
d4393e08 266int init_notification_iterator(struct bt_notification_iterator *iterator,
90157d89
PP
267 enum bt_notification_iterator_type type,
268 bt_object_release_func destroy)
269{
d4393e08
PP
270 int ret = 0;
271
3fea54f6 272 bt_object_init_shared(&iterator->base, destroy);
90157d89 273 iterator->type = type;
d4393e08
PP
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;
90157d89
PP
285}
286
d94d92ac
PP
287static
288struct bt_self_component_port_input_notification_iterator *
289bt_self_component_port_input_notification_iterator_create_initial(
3230ee6b 290 struct bt_component *upstream_comp,
d94d92ac 291 struct bt_port *upstream_port)
47e5a032 292{
d4393e08 293 int ret;
d94d92ac 294 struct bt_self_component_port_input_notification_iterator *iterator = NULL;
47e5a032 295
f6ccaed9
PP
296 BT_ASSERT(upstream_comp);
297 BT_ASSERT(upstream_port);
f6ccaed9 298 BT_ASSERT(bt_port_is_connected(upstream_port));
d94d92ac
PP
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);
47e5a032 307 if (!iterator) {
d94d92ac
PP
308 BT_LOGE_STR("Failed to allocate one self component input port "
309 "notification iterator.");
73d5c1ad 310 goto end;
47e5a032
JG
311 }
312
d4393e08 313 ret = init_notification_iterator((void *) iterator,
d94d92ac
PP
314 BT_NOTIFICATION_ITERATOR_TYPE_SELF_COMPONENT_PORT_INPUT,
315 bt_self_component_port_input_notification_iterator_destroy);
d4393e08
PP
316 if (ret) {
317 /* init_notification_iterator() logs errors */
d94d92ac 318 BT_OBJECT_PUT_REF_AND_RESET(iterator);
d4393e08
PP
319 goto end;
320 }
3230ee6b
PP
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) {
5af447e5 325 BT_LOGE_STR("Failed to allocate a GHashTable.");
d94d92ac 326 BT_OBJECT_PUT_REF_AND_RESET(iterator);
73d5c1ad 327 goto end;
3230ee6b
PP
328 }
329
bd14d768
PP
330 iterator->upstream_component = upstream_comp;
331 iterator->upstream_port = upstream_port;
d94d92ac 332 iterator->connection = iterator->upstream_port->connection;
5c563278 333 iterator->graph = bt_component_borrow_graph(upstream_comp);
d94d92ac
PP
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);
3230ee6b 338
47e5a032 339end:
d94d92ac 340 return iterator;
47e5a032
JG
341}
342
d94d92ac
PP
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)
ea8d3e58 346{
d94d92ac
PP
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 }
890882ef 385
d94d92ac
PP
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;
ea8d3e58
JG
432}
433
d94d92ac
PP
434void *bt_self_notification_iterator_get_data(
435 struct bt_self_notification_iterator *self_iterator)
ea8d3e58 436{
d94d92ac
PP
437 struct bt_self_component_port_input_notification_iterator *iterator =
438 (void *) self_iterator;
ea8d3e58 439
f42867e2 440 BT_ASSERT_PRE_NON_NULL(iterator, "Notification iterator");
d94d92ac 441 return iterator->user_data;
8738a040 442}
413bc2c4 443
d94d92ac
PP
444void bt_self_notification_iterator_set_data(
445 struct bt_self_notification_iterator *self_iterator, void *data)
5c563278 446{
d94d92ac
PP
447 struct bt_self_component_port_input_notification_iterator *iterator =
448 (void *) self_iterator;
5c563278
PP
449
450 BT_ASSERT_PRE_NON_NULL(iterator, "Notification iterator");
d94d92ac
PP
451 iterator->user_data = data;
452 BT_LIB_LOGV("Set notification iterator's user data: "
453 "%!+i, user-data-addr=%p", iterator, data);
5c563278
PP
454}
455
f42867e2
PP
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)
fa054faf 460{
f42867e2 461 BT_ASSERT(notif);
fa054faf 462
f42867e2 463 switch (notif->type) {
fa054faf 464 case BT_NOTIFICATION_TYPE_EVENT:
f42867e2
PP
465 *packet = bt_event_borrow_packet(
466 bt_notification_event_borrow_event(notif));
467 *stream = bt_packet_borrow_stream(*packet);
fa054faf
PP
468 break;
469 case BT_NOTIFICATION_TYPE_STREAM_BEGIN:
f42867e2 470 *stream = bt_notification_stream_begin_borrow_stream(notif);
fa054faf
PP
471 break;
472 case BT_NOTIFICATION_TYPE_STREAM_END:
f42867e2 473 *stream = bt_notification_stream_end_borrow_stream(notif);
fa054faf
PP
474 break;
475 case BT_NOTIFICATION_TYPE_PACKET_BEGIN:
f42867e2
PP
476 *packet = bt_notification_packet_begin_borrow_packet(notif);
477 *stream = bt_packet_borrow_stream(*packet);
fa054faf
PP
478 break;
479 case BT_NOTIFICATION_TYPE_PACKET_END:
f42867e2
PP
480 *packet = bt_notification_packet_end_borrow_packet(notif);
481 *stream = bt_packet_borrow_stream(*packet);
2ec84d26 482 break;
fa054faf 483 default:
f42867e2 484 break;
fa054faf 485 }
fa054faf
PP
486}
487
f42867e2
PP
488BT_ASSERT_PRE_FUNC
489static inline
490bool validate_notification(
d94d92ac 491 struct bt_self_component_port_input_notification_iterator *iterator,
f42867e2 492 struct bt_notification *notif)
3230ee6b 493{
f42867e2 494 bool is_valid = true;
3230ee6b 495 struct stream_state *stream_state;
f42867e2
PP
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 }
3230ee6b 506
f42867e2
PP
507 stream_state = g_hash_table_lookup(iterator->stream_states, stream);
508 if (!stream_state) {
3230ee6b 509 /*
f42867e2
PP
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.
3230ee6b 513 */
f42867e2
PP
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;
3230ee6b
PP
520 goto end;
521 }
522
f42867e2
PP
523 if (notif->seq_num == -1ULL) {
524 notif->seq_num = 0;
3230ee6b
PP
525 }
526
f42867e2
PP
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;
3230ee6b 535 goto end;
3230ee6b 536 }
3230ee6b 537
f42867e2
PP
538 stream_state = create_stream_state(stream);
539 if (!stream_state) {
540 abort();
541 }
fa054faf 542
f42867e2
PP
543 g_hash_table_insert(iterator->stream_states, stream,
544 stream_state);
545 stream_state->expected_notif_seq_num++;
546 goto end;
fa054faf
PP
547 }
548
f42867e2
PP
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;
fa054faf
PP
560 goto end;
561 }
562
f42867e2
PP
563 if (notif->seq_num == -1ULL) {
564 notif->seq_num = stream_state->expected_notif_seq_num;
3230ee6b
PP
565 }
566
f42867e2
PP
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;
fa054faf
PP
575 goto end;
576 }
577
f42867e2
PP
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++;
65300d60 612 stream_state->cur_packet = bt_object_get_ref(packet);
f42867e2
PP
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++;
65300d60 627 BT_OBJECT_PUT_REF_AND_RESET(stream_state->cur_packet);
f42867e2
PP
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;
3230ee6b
PP
644 }
645
3230ee6b 646end:
f42867e2 647 return is_valid;
3230ee6b
PP
648}
649
d4393e08
PP
650BT_ASSERT_PRE_FUNC
651static inline
652bool validate_notifications(
d94d92ac 653 struct bt_self_component_port_input_notification_iterator *iterator,
d4393e08
PP
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
f42867e2
PP
670BT_ASSERT_PRE_FUNC
671static inline bool priv_conn_notif_iter_can_end(
d94d92ac 672 struct bt_self_component_port_input_notification_iterator *iterator)
3230ee6b 673{
f42867e2
PP
674 GHashTableIter iter;
675 gpointer stream_key, state_value;
676 bool ret = true;
3230ee6b 677
f42867e2
PP
678 /*
679 * Verify that this iterator received a
680 * BT_NOTIFICATION_TYPE_STREAM_END notification for each stream
681 * which has a state.
682 */
3230ee6b 683
f42867e2 684 g_hash_table_iter_init(&iter, iterator->stream_states);
3230ee6b 685
f42867e2
PP
686 while (g_hash_table_iter_next(&iter, &stream_key, &state_value)) {
687 struct stream_state *stream_state = (void *) state_value;
3230ee6b 688
f42867e2
PP
689 BT_ASSERT(stream_state);
690 BT_ASSERT(stream_key);
fa054faf 691
f42867e2
PP
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 }
3230ee6b
PP
699 }
700
3230ee6b 701end:
3230ee6b
PP
702 return ret;
703}
704
f42867e2 705enum bt_notification_iterator_status
d94d92ac
PP
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)
3230ee6b 709{
d94d92ac
PP
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)");
f42867e2 720 BT_ASSERT_PRE(iterator->state ==
d94d92ac 721 BT_SELF_COMPONENT_PORT_INPUT_NOTIFICATION_ITERATOR_STATE_ACTIVE,
f42867e2
PP
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);
d94d92ac
PP
726 BT_LIB_LOGD("Getting next self component input port "
727 "notification iterator's notifications: %!+i", iterator);
728 comp_cls = iterator->upstream_component->class;
d3eb6e8f 729
3230ee6b 730 /* Pick the appropriate "next" method */
d94d92ac 731 switch (comp_cls->type) {
d3eb6e8f
PP
732 case BT_COMPONENT_CLASS_TYPE_SOURCE:
733 {
d94d92ac
PP
734 struct bt_component_class_source *src_comp_cls =
735 (void *) comp_cls;
d3eb6e8f 736
d94d92ac 737 method = (method_t) src_comp_cls->methods.notif_iter_next;
d3eb6e8f
PP
738 break;
739 }
740 case BT_COMPONENT_CLASS_TYPE_FILTER:
741 {
d94d92ac
PP
742 struct bt_component_class_filter *flt_comp_cls =
743 (void *) comp_cls;
d3eb6e8f 744
d94d92ac 745 method = (method_t) flt_comp_cls->methods.notif_iter_next;
d3eb6e8f
PP
746 break;
747 }
748 default:
0fbb9a9f 749 abort();
d3eb6e8f
PP
750 }
751
3230ee6b 752 /*
d94d92ac 753 * Call the user's "next" method to get the next notifications
fa054faf 754 * and status.
3230ee6b 755 */
d94d92ac 756 BT_ASSERT(method);
f42867e2 757 BT_LOGD_STR("Calling user's \"next\" method.");
d94d92ac
PP
758 status = method(iterator,
759 (void *) iterator->base.notifs->pdata,
d4393e08 760 NOTIF_BATCH_SIZE, user_count);
f42867e2 761 BT_LOGD("User method returned: status=%s",
d4393e08
PP
762 bt_notification_iterator_status_string(status));
763 if (status < 0) {
f42867e2 764 BT_LOGW_STR("User method failed.");
f42867e2
PP
765 goto end;
766 }
3230ee6b 767
d94d92ac
PP
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) {
f42867e2
PP
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 *
65300d60 778 * Only bt_object_put_ref() the returned notification if
d94d92ac
PP
779 * the status is BT_NOTIFICATION_ITERATOR_STATUS_OK
780 * because otherwise this field could be garbage.
f42867e2 781 */
d4393e08
PP
782 if (status == BT_NOTIFICATION_ITERATOR_STATUS_OK) {
783 uint64_t i;
784 bt_notification_array notifs =
d94d92ac 785 (void *) iterator->base.notifs->pdata;
d4393e08
PP
786
787 for (i = 0; i < *user_count; i++) {
65300d60 788 bt_object_put_ref(notifs[i]);
d4393e08 789 }
3230ee6b
PP
790 }
791
f42867e2
PP
792 status = BT_NOTIFICATION_ITERATOR_STATUS_CANCELED;
793 goto end;
794 }
8cf27cc5 795
d4393e08
PP
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);
d94d92ac 802 *notifs = (void *) iterator->base.notifs->pdata;
d4393e08
PP
803 break;
804 case BT_NOTIFICATION_ITERATOR_STATUS_AGAIN:
d4393e08 805 goto end;
f42867e2
PP
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 ==
d94d92ac
PP
811 BT_SELF_COMPONENT_PORT_INPUT_NOTIFICATION_ITERATOR_STATE_ACTIVE);
812 iterator->state = BT_SELF_COMPONENT_PORT_INPUT_NOTIFICATION_ITERATOR_STATE_ENDED;
f42867e2
PP
813 BT_LOGD("Set new status: status=%s",
814 bt_notification_iterator_status_string(status));
815 goto end;
f42867e2
PP
816 default:
817 /* Unknown non-error status */
818 abort();
41a2b7ae
PP
819 }
820
821end:
3230ee6b
PP
822 return status;
823}
824
825enum bt_notification_iterator_status
d94d92ac
PP
826bt_port_output_notification_iterator_next(
827 struct bt_port_output_notification_iterator *iterator,
d4393e08
PP
828 bt_notification_array *notifs_to_user,
829 uint64_t *count_to_user)
3230ee6b
PP
830{
831 enum bt_notification_iterator_status status;
07245ac2 832 enum bt_graph_status graph_status;
3230ee6b 833
f42867e2 834 BT_ASSERT_PRE_NON_NULL(iterator, "Notification iterator");
d94d92ac
PP
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: "
07245ac2 838 "%!+i", iterator);
d4393e08 839
d94d92ac
PP
840 graph_status = bt_graph_consume_sink_no_check(iterator->graph,
841 iterator->colander);
07245ac2
PP
842 switch (graph_status) {
843 case BT_GRAPH_STATUS_CANCELED:
07245ac2 844 case BT_GRAPH_STATUS_AGAIN:
07245ac2 845 case BT_GRAPH_STATUS_END:
07245ac2 846 case BT_GRAPH_STATUS_NOMEM:
d94d92ac 847 status = (int) graph_status;
07245ac2
PP
848 break;
849 case BT_GRAPH_STATUS_OK:
07245ac2 850 status = BT_NOTIFICATION_ITERATOR_STATUS_OK;
d4393e08
PP
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 */
d94d92ac
PP
857 *notifs_to_user = (void *) iterator->base.notifs->pdata;
858 *count_to_user = iterator->count;
90157d89 859 break;
90157d89 860 default:
07245ac2
PP
861 /* Other errors */
862 status = BT_NOTIFICATION_ITERATOR_STATUS_ERROR;
90157d89 863 }
3230ee6b 864
3230ee6b 865 return status;
53d45b87
JG
866}
867
d94d92ac
PP
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)
413bc2c4 877{
d94d92ac
PP
878 struct bt_self_component_port_input_notification_iterator *iterator =
879 (void *) self_iterator;
90157d89 880
f42867e2 881 BT_ASSERT_PRE_NON_NULL(iterator, "Notification iterator");
d94d92ac 882 return (void *) iterator->upstream_component;
413bc2c4
JG
883}
884
d94d92ac
PP
885struct bt_self_port_output *bt_self_notification_iterator_borrow_port(
886 struct bt_self_notification_iterator *self_iterator)
91457551 887{
d94d92ac
PP
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;
91457551 893}
8ed535b5
PP
894
895static
d94d92ac 896void bt_port_output_notification_iterator_destroy(struct bt_object *obj)
8ed535b5 897{
d94d92ac 898 struct bt_port_output_notification_iterator *iterator = (void *) obj;
8ed535b5 899
d94d92ac 900 BT_LIB_LOGD("Destroying output port notification iterator object: %!+i",
8ed535b5
PP
901 iterator);
902 BT_LOGD_STR("Putting graph.");
d94d92ac 903 BT_OBJECT_PUT_REF_AND_RESET(iterator->graph);
8ed535b5 904 BT_LOGD_STR("Putting colander sink component.");
d94d92ac 905 BT_OBJECT_PUT_REF_AND_RESET(iterator->colander);
8ed535b5
PP
906 destroy_base_notification_iterator(obj);
907}
908
d94d92ac
PP
909struct bt_port_output_notification_iterator *
910bt_port_output_notification_iterator_create(
911 struct bt_private_graph *priv_graph,
5fd91d88 912 struct bt_port_output *output_port)
8ed535b5 913{
d94d92ac
PP
914 struct bt_port_output_notification_iterator *iterator = NULL;
915 struct bt_component_class_sink *colander_comp_cls = NULL;
8ed535b5 916 struct bt_component *output_port_comp = NULL;
d94d92ac
PP
917 struct bt_component_sink *colander_comp;
918 struct bt_graph *graph = (void *) priv_graph;
8ed535b5 919 enum bt_graph_status graph_status;
d94d92ac 920 struct bt_port_input *colander_in_port = NULL;
8ed535b5 921 struct bt_component_class_sink_colander_data colander_data;
d4393e08 922 int ret;
8ed535b5 923
d94d92ac 924 BT_ASSERT_PRE_NON_NULL(graph, "Graph");
f42867e2 925 BT_ASSERT_PRE_NON_NULL(output_port, "Output port");
d94d92ac 926 output_port_comp = bt_port_borrow_component((void *) output_port);
f42867e2
PP
927 BT_ASSERT_PRE(output_port_comp,
928 "Output port has no component: %!+p", output_port);
d94d92ac
PP
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);
8ed535b5
PP
933
934 /* Create notification iterator */
d94d92ac
PP
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);
8ed535b5
PP
938 if (!iterator) {
939 BT_LOGE_STR("Failed to allocate one output port notification iterator.");
940 goto error;
941 }
942
d4393e08 943 ret = init_notification_iterator((void *) iterator,
d94d92ac
PP
944 BT_NOTIFICATION_ITERATOR_TYPE_PORT_OUTPUT,
945 bt_port_output_notification_iterator_destroy);
d4393e08
PP
946 if (ret) {
947 /* init_notification_iterator() logs errors */
65300d60 948 BT_OBJECT_PUT_REF_AND_RESET(iterator);
d4393e08
PP
949 goto end;
950 }
8ed535b5
PP
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
d94d92ac 959 iterator->graph = bt_object_get_ref(graph);
d4393e08
PP
960 colander_data.notifs = (void *) iterator->base.notifs->pdata;
961 colander_data.count_addr = &iterator->count;
5fd91d88
PP
962
963 /* Hope that nobody uses this very unique name */
d94d92ac
PP
964 graph_status =
965 bt_private_graph_add_sink_component_with_init_method_data(
5fd91d88
PP
966 (void *) graph, colander_comp_cls,
967 "colander-36ac3409-b1a8-4d60-ab1f-4fdf341a8fb1",
d94d92ac 968 NULL, &colander_data, &iterator->colander);
8ed535b5 969 if (graph_status != BT_GRAPH_STATUS_OK) {
d94d92ac
PP
970 BT_LIB_LOGW("Cannot add colander sink component to graph: "
971 "%1[graph-]+g, status=%s", graph,
8ed535b5
PP
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 */
d94d92ac 980 colander_in_port = bt_component_sink_borrow_input_port_by_index(
8ed535b5 981 iterator->colander, 0);
f6ccaed9 982 BT_ASSERT(colander_in_port);
d94d92ac 983 graph_status = bt_private_graph_connect_ports(priv_graph,
8ed535b5
PP
984 output_port, colander_in_port, NULL);
985 if (graph_status != BT_GRAPH_STATUS_OK) {
d94d92ac
PP
986 BT_LIB_LOGW("Cannot add colander sink component to graph: "
987 "%![graph-]+g, %![comp-]+c, status=%s", graph,
988 iterator->colander,
8ed535b5
PP
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
07245ac2
PP
998 * sink and moved to the notification iterator's notification
999 * member.
8ed535b5 1000 */
d94d92ac 1001 bt_graph_set_can_consume(iterator->graph, false);
8ed535b5
PP
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;
65300d60 1010 BT_OBJECT_PUT_REF_AND_RESET(iterator->colander);
8ed535b5
PP
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,
d94d92ac 1024 (void *) colander_comp);
f6ccaed9 1025 BT_ASSERT(ret == 0);
8ed535b5
PP
1026 }
1027
65300d60 1028 BT_OBJECT_PUT_REF_AND_RESET(iterator);
8ed535b5
PP
1029
1030end:
65300d60 1031 bt_object_put_ref(colander_comp_cls);
8ed535b5
PP
1032 return (void *) iterator;
1033}
This page took 0.09496 seconds and 4 git commands to generate.