4 * Babeltrace Plugin Component Graph
6 * Copyright 2017 Jérémie Galarneau <jeremie.galarneau@efficios.com>
8 * Author: Jérémie Galarneau <jeremie.galarneau@efficios.com>
10 * Permission is hereby granted, free of charge, to any person obtaining a copy
11 * of this software and associated documentation files (the "Software"), to deal
12 * in the Software without restriction, including without limitation the rights
13 * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
14 * copies of the Software, and to permit persons to whom the Software is
15 * furnished to do so, subject to the following conditions:
17 * The above copyright notice and this permission notice shall be included in
18 * all copies or substantial portions of the Software.
20 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
21 * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
22 * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
23 * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
24 * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
25 * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
29 #include <babeltrace/graph/component-internal.h>
30 #include <babeltrace/graph/graph-internal.h>
31 #include <babeltrace/graph/connection-internal.h>
32 #include <babeltrace/graph/component-sink-internal.h>
33 #include <babeltrace/graph/component-source.h>
34 #include <babeltrace/graph/component-filter.h>
35 #include <babeltrace/graph/port.h>
36 #include <babeltrace/compiler-internal.h>
37 #include <babeltrace/types.h>
41 struct bt_graph_listener
{
47 void bt_graph_destroy(struct bt_object
*obj
)
49 struct bt_graph
*graph
= container_of(obj
,
50 struct bt_graph
, base
);
53 * The graph's reference count is 0 if we're here. Increment
54 * it to avoid a double-destroy (possibly infinitely recursive)
57 * 1. We put and destroy a connection.
58 * 2. This connection's destructor finalizes its active
59 * notification iterators.
60 * 3. A notification iterator's finalization function gets a
61 * new reference on its component (reference count goes from
63 * 4. Since this component's reference count goes to 1, it takes
64 * a reference on its parent (this graph). This graph's
65 * reference count goes from 0 to 1.
66 * 5. The notification iterator's finalization function puts its
67 * component reference (reference count goes from 1 to 0).
68 * 6. Since this component's reference count goes from 1 to 0,
69 * it puts its parent (this graph). This graph's reference
70 * count goes from 1 to 0.
71 * 7. Since this graph's reference count goes from 1 to 0,
72 * its destructor is called (this function).
74 * With the incrementation below, the graph's reference count at
75 * step 4 goes from 1 to 2, and from 2 to 1 at step 6. This
76 * ensures that this function is not called two times.
78 obj
->ref_count
.count
++;
80 if (graph
->connections
) {
81 g_ptr_array_free(graph
->connections
, TRUE
);
83 if (graph
->components
) {
84 g_ptr_array_free(graph
->components
, TRUE
);
86 if (graph
->sinks_to_consume
) {
87 g_queue_free(graph
->sinks_to_consume
);
90 if (graph
->listeners
.port_added
) {
91 g_array_free(graph
->listeners
.port_added
, TRUE
);
94 if (graph
->listeners
.port_removed
) {
95 g_array_free(graph
->listeners
.port_removed
, TRUE
);
98 if (graph
->listeners
.ports_connected
) {
99 g_array_free(graph
->listeners
.ports_connected
, TRUE
);
102 if (graph
->listeners
.ports_disconnected
) {
103 g_array_free(graph
->listeners
.ports_disconnected
, TRUE
);
110 int init_listeners_array(GArray
**listeners
)
115 *listeners
= g_array_new(FALSE
, TRUE
, sizeof(struct bt_graph_listener
));
125 struct bt_graph
*bt_graph_create(void)
127 struct bt_graph
*graph
;
130 graph
= g_new0(struct bt_graph
, 1);
135 bt_object_init(graph
, bt_graph_destroy
);
137 graph
->connections
= g_ptr_array_new_with_free_func(bt_object_release
);
138 if (!graph
->connections
) {
141 graph
->components
= g_ptr_array_new_with_free_func(bt_object_release
);
142 if (!graph
->components
) {
145 graph
->sinks_to_consume
= g_queue_new();
146 if (!graph
->sinks_to_consume
) {
150 ret
= init_listeners_array(&graph
->listeners
.port_added
);
155 ret
= init_listeners_array(&graph
->listeners
.port_removed
);
160 ret
= init_listeners_array(&graph
->listeners
.ports_connected
);
165 ret
= init_listeners_array(&graph
->listeners
.ports_disconnected
);
177 struct bt_connection
*bt_graph_connect_ports(struct bt_graph
*graph
,
178 struct bt_port
*upstream_port
,
179 struct bt_port
*downstream_port
)
181 struct bt_connection
*connection
= NULL
;
182 struct bt_graph
*upstream_graph
= NULL
;
183 struct bt_graph
*downstream_graph
= NULL
;
184 struct bt_component
*upstream_component
= NULL
;
185 struct bt_component
*downstream_component
= NULL
;
186 enum bt_component_status component_status
;
187 bt_bool upstream_was_already_in_graph
;
188 bt_bool downstream_was_already_in_graph
;
190 if (!graph
|| !upstream_port
|| !downstream_port
) {
194 if (graph
->canceled
) {
198 /* Ensure appropriate types for upstream and downstream ports. */
199 if (bt_port_get_type(upstream_port
) != BT_PORT_TYPE_OUTPUT
) {
202 if (bt_port_get_type(downstream_port
) != BT_PORT_TYPE_INPUT
) {
206 /* Ensure that both ports are currently unconnected. */
207 if (bt_port_is_connected(upstream_port
)) {
208 fprintf(stderr
, "Upstream port is already connected\n");
212 if (bt_port_is_connected(downstream_port
)) {
213 fprintf(stderr
, "Downstream port is already connected\n");
218 * Ensure that both ports are still attached to their creating
221 upstream_component
= bt_port_get_component(upstream_port
);
222 if (!upstream_component
) {
223 fprintf(stderr
, "Upstream port does not belong to a component\n");
227 downstream_component
= bt_port_get_component(downstream_port
);
228 if (!downstream_component
) {
229 fprintf(stderr
, "Downstream port does not belong to a component\n");
233 /* Ensure the components are not already part of another graph. */
234 upstream_graph
= bt_component_get_graph(upstream_component
);
235 if (upstream_graph
&& (graph
!= upstream_graph
)) {
236 fprintf(stderr
, "Upstream component is already part of another graph\n");
239 upstream_was_already_in_graph
= (graph
== upstream_graph
);
240 downstream_graph
= bt_component_get_graph(downstream_component
);
241 if (downstream_graph
&& (graph
!= downstream_graph
)) {
242 fprintf(stderr
, "Downstream component is already part of another graph\n");
245 downstream_was_already_in_graph
= (graph
== downstream_graph
);
248 * At this point the ports are not connected yet. Both
249 * components need to accept an eventual connection to their
250 * port by the other port before we continue.
252 component_status
= bt_component_accept_port_connection(
253 upstream_component
, upstream_port
, downstream_port
);
254 if (component_status
!= BT_COMPONENT_STATUS_OK
) {
257 component_status
= bt_component_accept_port_connection(
258 downstream_component
, downstream_port
, upstream_port
);
259 if (component_status
!= BT_COMPONENT_STATUS_OK
) {
263 connection
= bt_connection_create(graph
, upstream_port
,
270 * Ownership of upstream_component/downstream_component and of
271 * the connection object is transferred to the graph.
273 g_ptr_array_add(graph
->connections
, connection
);
275 if (!upstream_was_already_in_graph
) {
276 g_ptr_array_add(graph
->components
, upstream_component
);
277 bt_component_set_graph(upstream_component
, graph
);
279 if (!downstream_was_already_in_graph
) {
280 g_ptr_array_add(graph
->components
, downstream_component
);
281 bt_component_set_graph(downstream_component
, graph
);
282 if (bt_component_get_class_type(downstream_component
) ==
283 BT_COMPONENT_CLASS_TYPE_SINK
) {
284 g_queue_push_tail(graph
->sinks_to_consume
,
285 downstream_component
);
290 * The graph is now the parent of these components which
291 * garantees their existence for the duration of the graph's
296 * Notify both components that their port is connected.
298 bt_component_port_connected(upstream_component
, upstream_port
,
300 bt_component_port_connected(downstream_component
, downstream_port
,
304 * Notify the graph's creator that both ports are connected.
306 bt_graph_notify_ports_connected(graph
, upstream_port
, downstream_port
);
309 bt_put(upstream_graph
);
310 bt_put(downstream_graph
);
311 bt_put(upstream_component
);
312 bt_put(downstream_component
);
316 BT_PUT(upstream_component
);
317 BT_PUT(downstream_component
);
322 enum bt_component_status
get_component_port_counts(
323 struct bt_component
*component
, int64_t *input_count
,
324 int64_t *output_count
)
326 enum bt_component_status ret
;
328 switch (bt_component_get_class_type(component
)) {
329 case BT_COMPONENT_CLASS_TYPE_SOURCE
:
331 bt_component_source_get_output_port_count(component
);
332 if (*output_count
< 0) {
333 ret
= BT_COMPONENT_STATUS_ERROR
;
337 case BT_COMPONENT_CLASS_TYPE_FILTER
:
339 bt_component_filter_get_output_port_count(component
);
340 if (*output_count
< 0) {
341 ret
= BT_COMPONENT_STATUS_ERROR
;
345 bt_component_filter_get_input_port_count(component
);
346 if (*input_count
< 0) {
347 ret
= BT_COMPONENT_STATUS_ERROR
;
351 case BT_COMPONENT_CLASS_TYPE_SINK
:
353 bt_component_sink_get_input_port_count(component
);
354 if (*input_count
< 0) {
355 ret
= BT_COMPONENT_STATUS_ERROR
;
363 ret
= BT_COMPONENT_STATUS_OK
;
368 enum bt_graph_status
bt_graph_add_component_as_sibling(struct bt_graph
*graph
,
369 struct bt_component
*origin
,
370 struct bt_component
*new_component
)
372 int64_t origin_input_port_count
= 0;
373 int64_t origin_output_port_count
= 0;
374 int64_t new_input_port_count
= 0;
375 int64_t new_output_port_count
= 0;
376 enum bt_graph_status status
= BT_GRAPH_STATUS_OK
;
377 struct bt_graph
*origin_graph
= NULL
;
378 struct bt_graph
*new_graph
= NULL
;
379 struct bt_port
*origin_port
= NULL
;
380 struct bt_port
*new_port
= NULL
;
381 struct bt_port
*upstream_port
= NULL
;
382 struct bt_port
*downstream_port
= NULL
;
383 struct bt_connection
*origin_connection
= NULL
;
384 struct bt_connection
*new_connection
= NULL
;
387 if (!graph
|| !origin
|| !new_component
) {
388 status
= BT_GRAPH_STATUS_INVALID
;
392 if (graph
->canceled
) {
393 status
= BT_GRAPH_STATUS_CANCELED
;
397 if (bt_component_get_class_type(origin
) !=
398 bt_component_get_class_type(new_component
)) {
399 status
= BT_GRAPH_STATUS_INVALID
;
403 origin_graph
= bt_component_get_graph(origin
);
404 if (!origin_graph
|| (origin_graph
!= graph
)) {
405 status
= BT_GRAPH_STATUS_INVALID
;
409 new_graph
= bt_component_get_graph(new_component
);
411 status
= BT_GRAPH_STATUS_ALREADY_IN_A_GRAPH
;
415 if (get_component_port_counts(origin
, &origin_input_port_count
,
416 &origin_output_port_count
) != BT_COMPONENT_STATUS_OK
) {
417 status
= BT_GRAPH_STATUS_INVALID
;
420 if (get_component_port_counts(new_component
, &new_input_port_count
,
421 &new_output_port_count
) != BT_COMPONENT_STATUS_OK
) {
422 status
= BT_GRAPH_STATUS_INVALID
;
426 if (origin_input_port_count
!= new_input_port_count
||
427 origin_output_port_count
!= new_output_port_count
) {
428 status
= BT_GRAPH_STATUS_INVALID
;
432 /* Replicate input connections. */
433 for (port_index
= 0; port_index
< origin_input_port_count
; port_index
++) {
434 origin_port
= bt_component_get_input_port_by_index(origin
,
437 status
= BT_GRAPH_STATUS_ERROR
;
438 goto error_disconnect
;
441 new_port
= bt_component_get_input_port_by_index(new_component
,
444 status
= BT_GRAPH_STATUS_ERROR
;
445 goto error_disconnect
;
448 origin_connection
= bt_port_get_connection(origin_port
);
449 if (origin_connection
) {
450 upstream_port
= bt_connection_get_upstream_port(
452 if (!upstream_port
) {
453 goto error_disconnect
;
456 new_connection
= bt_graph_connect_ports(graph
,
457 upstream_port
, new_port
);
458 if (!new_connection
) {
459 goto error_disconnect
;
463 BT_PUT(upstream_port
);
464 BT_PUT(origin_connection
);
465 BT_PUT(new_connection
);
470 /* Replicate output connections. */
471 for (port_index
= 0; port_index
< origin_output_port_count
; port_index
++) {
472 origin_port
= bt_component_get_output_port_by_index(origin
,
475 status
= BT_GRAPH_STATUS_ERROR
;
476 goto error_disconnect
;
478 new_port
= bt_component_get_output_port_by_index(new_component
,
481 status
= BT_GRAPH_STATUS_ERROR
;
482 goto error_disconnect
;
485 origin_connection
= bt_port_get_connection(origin_port
);
486 if (origin_connection
) {
487 downstream_port
= bt_connection_get_downstream_port(
489 if (!downstream_port
) {
490 goto error_disconnect
;
493 new_connection
= bt_graph_connect_ports(graph
,
494 new_port
, downstream_port
);
495 if (!new_connection
) {
496 goto error_disconnect
;
500 BT_PUT(downstream_port
);
501 BT_PUT(origin_connection
);
502 BT_PUT(new_connection
);
507 bt_put(origin_graph
);
511 bt_put(upstream_port
);
512 bt_put(downstream_port
);
513 bt_put(origin_connection
);
514 bt_put(new_connection
);
517 /* Destroy all connections of the new component. */
522 enum bt_graph_status
bt_graph_consume(struct bt_graph
*graph
)
524 struct bt_component
*sink
;
525 enum bt_graph_status status
= BT_GRAPH_STATUS_OK
;
526 enum bt_component_status comp_status
;
530 status
= BT_GRAPH_STATUS_INVALID
;
534 if (graph
->canceled
) {
535 status
= BT_GRAPH_STATUS_CANCELED
;
539 if (g_queue_is_empty(graph
->sinks_to_consume
)) {
540 status
= BT_GRAPH_STATUS_END
;
544 current_node
= g_queue_pop_head_link(graph
->sinks_to_consume
);
545 sink
= current_node
->data
;
546 comp_status
= bt_component_sink_consume(sink
);
547 switch (comp_status
) {
548 case BT_COMPONENT_STATUS_OK
:
550 case BT_COMPONENT_STATUS_END
:
551 status
= BT_GRAPH_STATUS_END
;
553 case BT_COMPONENT_STATUS_AGAIN
:
554 status
= BT_GRAPH_STATUS_AGAIN
;
556 case BT_COMPONENT_STATUS_INVALID
:
557 status
= BT_GRAPH_STATUS_INVALID
;
560 status
= BT_GRAPH_STATUS_ERROR
;
564 if (status
!= BT_GRAPH_STATUS_END
) {
565 g_queue_push_tail_link(graph
->sinks_to_consume
, current_node
);
569 /* End reached, the node is not added back to the queue and free'd. */
570 g_queue_delete_link(graph
->sinks_to_consume
, current_node
);
572 /* Don't forward an END status if there are sinks left to consume. */
573 if (!g_queue_is_empty(graph
->sinks_to_consume
)) {
574 status
= BT_GRAPH_STATUS_OK
;
581 enum bt_graph_status
bt_graph_run(struct bt_graph
*graph
)
583 enum bt_graph_status status
= BT_GRAPH_STATUS_OK
;
586 status
= BT_GRAPH_STATUS_INVALID
;
591 status
= bt_graph_consume(graph
);
592 if (status
== BT_GRAPH_STATUS_AGAIN
) {
594 * If AGAIN is received and there are multiple
595 * sinks, go ahead and consume from the next
598 * However, in the case where a single sink is
599 * left, the caller can decide to busy-wait and
600 * call bt_graph_run() continuously until the
601 * source is ready or it can decide to sleep for
602 * an arbitrary amount of time.
604 if (graph
->sinks_to_consume
->length
> 1) {
605 status
= BT_GRAPH_STATUS_OK
;
608 } while (status
== BT_GRAPH_STATUS_OK
);
610 if (g_queue_is_empty(graph
->sinks_to_consume
)) {
611 status
= BT_GRAPH_STATUS_END
;
618 void add_listener(GArray
*listeners
, void *func
, void *data
)
620 struct bt_graph_listener listener
= {
625 g_array_append_val(listeners
, listener
);
628 enum bt_graph_status
bt_graph_add_port_added_listener(
629 struct bt_graph
*graph
,
630 bt_graph_port_added_listener listener
, void *data
)
632 enum bt_graph_status status
= BT_GRAPH_STATUS_OK
;
634 if (!graph
|| !listener
) {
635 status
= BT_GRAPH_STATUS_INVALID
;
639 add_listener(graph
->listeners
.port_added
, listener
, data
);
645 enum bt_graph_status
bt_graph_add_port_removed_listener(
646 struct bt_graph
*graph
,
647 bt_graph_port_removed_listener listener
, void *data
)
649 enum bt_graph_status status
= BT_GRAPH_STATUS_OK
;
651 if (!graph
|| !listener
) {
652 status
= BT_GRAPH_STATUS_INVALID
;
656 add_listener(graph
->listeners
.port_removed
, listener
, data
);
662 enum bt_graph_status
bt_graph_add_ports_connected_listener(
663 struct bt_graph
*graph
,
664 bt_graph_ports_connected_listener listener
, void *data
)
666 enum bt_graph_status status
= BT_GRAPH_STATUS_OK
;
668 if (!graph
|| !listener
) {
669 status
= BT_GRAPH_STATUS_INVALID
;
673 add_listener(graph
->listeners
.ports_connected
, listener
, data
);
679 enum bt_graph_status
bt_graph_add_ports_disconnected_listener(
680 struct bt_graph
*graph
,
681 bt_graph_ports_disconnected_listener listener
, void *data
)
683 enum bt_graph_status status
= BT_GRAPH_STATUS_OK
;
685 if (!graph
|| !listener
) {
686 status
= BT_GRAPH_STATUS_INVALID
;
690 add_listener(graph
->listeners
.ports_disconnected
, listener
, data
);
697 void bt_graph_notify_port_added(struct bt_graph
*graph
, struct bt_port
*port
)
701 for (i
= 0; i
< graph
->listeners
.port_added
->len
; i
++) {
702 struct bt_graph_listener listener
=
703 g_array_index(graph
->listeners
.port_added
,
704 struct bt_graph_listener
, i
);
705 bt_graph_port_added_listener func
= listener
.func
;
708 func(port
, listener
.data
);
713 void bt_graph_notify_port_removed(struct bt_graph
*graph
,
714 struct bt_component
*comp
, struct bt_port
*port
)
718 for (i
= 0; i
< graph
->listeners
.port_removed
->len
; i
++) {
719 struct bt_graph_listener listener
=
720 g_array_index(graph
->listeners
.port_removed
,
721 struct bt_graph_listener
, i
);
722 bt_graph_port_removed_listener func
= listener
.func
;
725 func(comp
, port
, listener
.data
);
730 void bt_graph_notify_ports_connected(struct bt_graph
*graph
,
731 struct bt_port
*upstream_port
, struct bt_port
*downstream_port
)
735 for (i
= 0; i
< graph
->listeners
.ports_connected
->len
; i
++) {
736 struct bt_graph_listener listener
=
737 g_array_index(graph
->listeners
.ports_connected
,
738 struct bt_graph_listener
, i
);
739 bt_graph_ports_connected_listener func
= listener
.func
;
742 func(upstream_port
, downstream_port
, listener
.data
);
747 void bt_graph_notify_ports_disconnected(struct bt_graph
*graph
,
748 struct bt_component
*upstream_comp
,
749 struct bt_component
*downstream_comp
,
750 struct bt_port
*upstream_port
, struct bt_port
*downstream_port
)
754 for (i
= 0; i
< graph
->listeners
.ports_disconnected
->len
; i
++) {
755 struct bt_graph_listener listener
=
756 g_array_index(graph
->listeners
.ports_disconnected
,
757 struct bt_graph_listener
, i
);
758 bt_graph_ports_disconnected_listener func
= listener
.func
;
761 func(upstream_comp
, downstream_comp
, upstream_port
,
762 downstream_port
, listener
.data
);
766 extern enum bt_graph_status
bt_graph_cancel(struct bt_graph
*graph
)
768 enum bt_graph_status ret
= BT_GRAPH_STATUS_OK
;
771 ret
= BT_GRAPH_STATUS_INVALID
;
775 graph
->canceled
= BT_TRUE
;
781 extern bt_bool
bt_graph_is_canceled(struct bt_graph
*graph
)
783 return graph
? graph
->canceled
: BT_FALSE
;