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
);
52 if (graph
->components
) {
53 g_ptr_array_free(graph
->components
, TRUE
);
55 if (graph
->connections
) {
56 g_ptr_array_free(graph
->connections
, TRUE
);
58 if (graph
->sinks_to_consume
) {
59 g_queue_free(graph
->sinks_to_consume
);
62 if (graph
->listeners
.port_added
) {
63 g_array_free(graph
->listeners
.port_added
, TRUE
);
66 if (graph
->listeners
.port_removed
) {
67 g_array_free(graph
->listeners
.port_removed
, TRUE
);
70 if (graph
->listeners
.ports_connected
) {
71 g_array_free(graph
->listeners
.ports_connected
, TRUE
);
74 if (graph
->listeners
.ports_disconnected
) {
75 g_array_free(graph
->listeners
.ports_disconnected
, TRUE
);
82 int init_listeners_array(GArray
**listeners
)
87 *listeners
= g_array_new(FALSE
, TRUE
, sizeof(struct bt_graph_listener
));
97 struct bt_graph
*bt_graph_create(void)
99 struct bt_graph
*graph
;
102 graph
= g_new0(struct bt_graph
, 1);
107 bt_object_init(graph
, bt_graph_destroy
);
109 graph
->connections
= g_ptr_array_new_with_free_func(bt_object_release
);
110 if (!graph
->connections
) {
113 graph
->components
= g_ptr_array_new_with_free_func(bt_object_release
);
114 if (!graph
->components
) {
117 graph
->sinks_to_consume
= g_queue_new();
118 if (!graph
->sinks_to_consume
) {
122 ret
= init_listeners_array(&graph
->listeners
.port_added
);
127 ret
= init_listeners_array(&graph
->listeners
.port_removed
);
132 ret
= init_listeners_array(&graph
->listeners
.ports_connected
);
137 ret
= init_listeners_array(&graph
->listeners
.ports_disconnected
);
149 struct bt_connection
*bt_graph_connect_ports(struct bt_graph
*graph
,
150 struct bt_port
*upstream_port
,
151 struct bt_port
*downstream_port
)
153 struct bt_connection
*connection
= NULL
;
154 struct bt_graph
*upstream_graph
= NULL
;
155 struct bt_graph
*downstream_graph
= NULL
;
156 struct bt_component
*upstream_component
= NULL
;
157 struct bt_component
*downstream_component
= NULL
;
158 enum bt_component_status component_status
;
159 bt_bool upstream_was_already_in_graph
;
160 bt_bool downstream_was_already_in_graph
;
162 if (!graph
|| !upstream_port
|| !downstream_port
) {
166 if (graph
->canceled
) {
170 /* Ensure appropriate types for upstream and downstream ports. */
171 if (bt_port_get_type(upstream_port
) != BT_PORT_TYPE_OUTPUT
) {
174 if (bt_port_get_type(downstream_port
) != BT_PORT_TYPE_INPUT
) {
178 /* Ensure that both ports are currently unconnected. */
179 if (bt_port_is_connected(upstream_port
)) {
180 fprintf(stderr
, "Upstream port is already connected\n");
184 if (bt_port_is_connected(downstream_port
)) {
185 fprintf(stderr
, "Downstream port is already connected\n");
190 * Ensure that both ports are still attached to their creating
193 upstream_component
= bt_port_get_component(upstream_port
);
194 if (!upstream_component
) {
195 fprintf(stderr
, "Upstream port does not belong to a component\n");
199 downstream_component
= bt_port_get_component(downstream_port
);
200 if (!downstream_component
) {
201 fprintf(stderr
, "Downstream port does not belong to a component\n");
205 /* Ensure the components are not already part of another graph. */
206 upstream_graph
= bt_component_get_graph(upstream_component
);
207 if (upstream_graph
&& (graph
!= upstream_graph
)) {
208 fprintf(stderr
, "Upstream component is already part of another graph\n");
211 upstream_was_already_in_graph
= (graph
== upstream_graph
);
212 downstream_graph
= bt_component_get_graph(downstream_component
);
213 if (downstream_graph
&& (graph
!= downstream_graph
)) {
214 fprintf(stderr
, "Downstream component is already part of another graph\n");
217 downstream_was_already_in_graph
= (graph
== downstream_graph
);
220 * At this point the ports are not connected yet. Both
221 * components need to accept an eventual connection to their
222 * port by the other port before we continue.
224 component_status
= bt_component_accept_port_connection(
225 upstream_component
, upstream_port
, downstream_port
);
226 if (component_status
!= BT_COMPONENT_STATUS_OK
) {
229 component_status
= bt_component_accept_port_connection(
230 downstream_component
, downstream_port
, upstream_port
);
231 if (component_status
!= BT_COMPONENT_STATUS_OK
) {
235 connection
= bt_connection_create(graph
, upstream_port
,
242 * Ownership of upstream_component/downstream_component and of
243 * the connection object is transferred to the graph.
245 g_ptr_array_add(graph
->connections
, connection
);
247 if (!upstream_was_already_in_graph
) {
248 g_ptr_array_add(graph
->components
, upstream_component
);
249 bt_component_set_graph(upstream_component
, graph
);
251 if (!downstream_was_already_in_graph
) {
252 g_ptr_array_add(graph
->components
, downstream_component
);
253 bt_component_set_graph(downstream_component
, graph
);
254 if (bt_component_get_class_type(downstream_component
) ==
255 BT_COMPONENT_CLASS_TYPE_SINK
) {
256 g_queue_push_tail(graph
->sinks_to_consume
,
257 downstream_component
);
262 * The graph is now the parent of these components which
263 * garantees their existence for the duration of the graph's
268 * Notify both components that their port is connected.
270 bt_component_port_connected(upstream_component
, upstream_port
,
272 bt_component_port_connected(downstream_component
, downstream_port
,
276 * Notify the graph's creator that both ports are connected.
278 bt_graph_notify_ports_connected(graph
, upstream_port
, downstream_port
);
281 bt_put(upstream_graph
);
282 bt_put(downstream_graph
);
283 bt_put(upstream_component
);
284 bt_put(downstream_component
);
288 BT_PUT(upstream_component
);
289 BT_PUT(downstream_component
);
294 enum bt_component_status
get_component_port_counts(
295 struct bt_component
*component
, int64_t *input_count
,
296 int64_t *output_count
)
298 enum bt_component_status ret
;
300 switch (bt_component_get_class_type(component
)) {
301 case BT_COMPONENT_CLASS_TYPE_SOURCE
:
303 bt_component_source_get_output_port_count(component
);
304 if (*output_count
< 0) {
305 ret
= BT_COMPONENT_STATUS_ERROR
;
309 case BT_COMPONENT_CLASS_TYPE_FILTER
:
311 bt_component_filter_get_output_port_count(component
);
312 if (*output_count
< 0) {
313 ret
= BT_COMPONENT_STATUS_ERROR
;
317 bt_component_filter_get_input_port_count(component
);
318 if (*input_count
< 0) {
319 ret
= BT_COMPONENT_STATUS_ERROR
;
323 case BT_COMPONENT_CLASS_TYPE_SINK
:
325 bt_component_sink_get_input_port_count(component
);
326 if (*input_count
< 0) {
327 ret
= BT_COMPONENT_STATUS_ERROR
;
335 ret
= BT_COMPONENT_STATUS_OK
;
340 enum bt_graph_status
bt_graph_add_component_as_sibling(struct bt_graph
*graph
,
341 struct bt_component
*origin
,
342 struct bt_component
*new_component
)
344 int64_t origin_input_port_count
= 0;
345 int64_t origin_output_port_count
= 0;
346 int64_t new_input_port_count
= 0;
347 int64_t new_output_port_count
= 0;
348 enum bt_graph_status status
= BT_GRAPH_STATUS_OK
;
349 struct bt_graph
*origin_graph
= NULL
;
350 struct bt_graph
*new_graph
= NULL
;
351 struct bt_port
*origin_port
= NULL
;
352 struct bt_port
*new_port
= NULL
;
353 struct bt_port
*upstream_port
= NULL
;
354 struct bt_port
*downstream_port
= NULL
;
355 struct bt_connection
*origin_connection
= NULL
;
356 struct bt_connection
*new_connection
= NULL
;
359 if (!graph
|| !origin
|| !new_component
) {
360 status
= BT_GRAPH_STATUS_INVALID
;
364 if (graph
->canceled
) {
365 status
= BT_GRAPH_STATUS_CANCELED
;
369 if (bt_component_get_class_type(origin
) !=
370 bt_component_get_class_type(new_component
)) {
371 status
= BT_GRAPH_STATUS_INVALID
;
375 origin_graph
= bt_component_get_graph(origin
);
376 if (!origin_graph
|| (origin_graph
!= graph
)) {
377 status
= BT_GRAPH_STATUS_INVALID
;
381 new_graph
= bt_component_get_graph(new_component
);
383 status
= BT_GRAPH_STATUS_ALREADY_IN_A_GRAPH
;
387 if (get_component_port_counts(origin
, &origin_input_port_count
,
388 &origin_output_port_count
) != BT_COMPONENT_STATUS_OK
) {
389 status
= BT_GRAPH_STATUS_INVALID
;
392 if (get_component_port_counts(new_component
, &new_input_port_count
,
393 &new_output_port_count
) != BT_COMPONENT_STATUS_OK
) {
394 status
= BT_GRAPH_STATUS_INVALID
;
398 if (origin_input_port_count
!= new_input_port_count
||
399 origin_output_port_count
!= new_output_port_count
) {
400 status
= BT_GRAPH_STATUS_INVALID
;
404 /* Replicate input connections. */
405 for (port_index
= 0; port_index
< origin_input_port_count
; port_index
++) {
406 origin_port
= bt_component_get_input_port_by_index(origin
,
409 status
= BT_GRAPH_STATUS_ERROR
;
410 goto error_disconnect
;
413 new_port
= bt_component_get_input_port_by_index(new_component
,
416 status
= BT_GRAPH_STATUS_ERROR
;
417 goto error_disconnect
;
420 origin_connection
= bt_port_get_connection(origin_port
);
421 if (origin_connection
) {
422 upstream_port
= bt_connection_get_upstream_port(
424 if (!upstream_port
) {
425 goto error_disconnect
;
428 new_connection
= bt_graph_connect_ports(graph
,
429 upstream_port
, new_port
);
430 if (!new_connection
) {
431 goto error_disconnect
;
435 BT_PUT(upstream_port
);
436 BT_PUT(origin_connection
);
437 BT_PUT(new_connection
);
442 /* Replicate output connections. */
443 for (port_index
= 0; port_index
< origin_output_port_count
; port_index
++) {
444 origin_port
= bt_component_get_output_port_by_index(origin
,
447 status
= BT_GRAPH_STATUS_ERROR
;
448 goto error_disconnect
;
450 new_port
= bt_component_get_output_port_by_index(new_component
,
453 status
= BT_GRAPH_STATUS_ERROR
;
454 goto error_disconnect
;
457 origin_connection
= bt_port_get_connection(origin_port
);
458 if (origin_connection
) {
459 downstream_port
= bt_connection_get_downstream_port(
461 if (!downstream_port
) {
462 goto error_disconnect
;
465 new_connection
= bt_graph_connect_ports(graph
,
466 new_port
, downstream_port
);
467 if (!new_connection
) {
468 goto error_disconnect
;
472 BT_PUT(downstream_port
);
473 BT_PUT(origin_connection
);
474 BT_PUT(new_connection
);
479 bt_put(origin_graph
);
483 bt_put(upstream_port
);
484 bt_put(downstream_port
);
485 bt_put(origin_connection
);
486 bt_put(new_connection
);
489 /* Destroy all connections of the new component. */
494 enum bt_graph_status
bt_graph_consume(struct bt_graph
*graph
)
496 struct bt_component
*sink
;
497 enum bt_graph_status status
= BT_GRAPH_STATUS_OK
;
498 enum bt_component_status comp_status
;
502 status
= BT_GRAPH_STATUS_INVALID
;
506 if (graph
->canceled
) {
507 status
= BT_GRAPH_STATUS_CANCELED
;
511 if (g_queue_is_empty(graph
->sinks_to_consume
)) {
512 status
= BT_GRAPH_STATUS_END
;
516 current_node
= g_queue_pop_head_link(graph
->sinks_to_consume
);
517 sink
= current_node
->data
;
518 comp_status
= bt_component_sink_consume(sink
);
519 switch (comp_status
) {
520 case BT_COMPONENT_STATUS_OK
:
522 case BT_COMPONENT_STATUS_END
:
523 status
= BT_GRAPH_STATUS_END
;
525 case BT_COMPONENT_STATUS_AGAIN
:
526 status
= BT_GRAPH_STATUS_AGAIN
;
528 case BT_COMPONENT_STATUS_INVALID
:
529 status
= BT_GRAPH_STATUS_INVALID
;
532 status
= BT_GRAPH_STATUS_ERROR
;
536 if (status
!= BT_GRAPH_STATUS_END
) {
537 g_queue_push_tail_link(graph
->sinks_to_consume
, current_node
);
541 /* End reached, the node is not added back to the queue and free'd. */
542 g_queue_delete_link(graph
->sinks_to_consume
, current_node
);
544 /* Don't forward an END status if there are sinks left to consume. */
545 if (!g_queue_is_empty(graph
->sinks_to_consume
)) {
546 status
= BT_GRAPH_STATUS_OK
;
553 enum bt_graph_status
bt_graph_run(struct bt_graph
*graph
)
555 enum bt_graph_status status
= BT_GRAPH_STATUS_OK
;
558 status
= BT_GRAPH_STATUS_INVALID
;
563 if (graph
->canceled
) {
564 status
= BT_GRAPH_STATUS_CANCELED
;
568 status
= bt_graph_consume(graph
);
569 if (status
== BT_GRAPH_STATUS_AGAIN
) {
571 * If AGAIN is received and there are multiple
572 * sinks, go ahead and consume from the next
575 * However, in the case where a single sink is
576 * left, the caller can decide to busy-wait and
577 * call bt_graph_run() continuously until the
578 * source is ready or it can decide to sleep for
579 * an arbitrary amount of time.
581 if (graph
->sinks_to_consume
->length
> 1) {
582 status
= BT_GRAPH_STATUS_OK
;
585 } while (status
== BT_GRAPH_STATUS_OK
);
587 if (g_queue_is_empty(graph
->sinks_to_consume
)) {
588 status
= BT_GRAPH_STATUS_END
;
595 void add_listener(GArray
*listeners
, void *func
, void *data
)
597 struct bt_graph_listener listener
= {
602 g_array_append_val(listeners
, listener
);
605 enum bt_graph_status
bt_graph_add_port_added_listener(
606 struct bt_graph
*graph
,
607 bt_graph_port_added_listener listener
, void *data
)
609 enum bt_graph_status status
= BT_GRAPH_STATUS_OK
;
611 if (!graph
|| !listener
) {
612 status
= BT_GRAPH_STATUS_INVALID
;
616 add_listener(graph
->listeners
.port_added
, listener
, data
);
622 enum bt_graph_status
bt_graph_add_port_removed_listener(
623 struct bt_graph
*graph
,
624 bt_graph_port_removed_listener listener
, void *data
)
626 enum bt_graph_status status
= BT_GRAPH_STATUS_OK
;
628 if (!graph
|| !listener
) {
629 status
= BT_GRAPH_STATUS_INVALID
;
633 add_listener(graph
->listeners
.port_removed
, listener
, data
);
639 enum bt_graph_status
bt_graph_add_ports_connected_listener(
640 struct bt_graph
*graph
,
641 bt_graph_ports_connected_listener listener
, void *data
)
643 enum bt_graph_status status
= BT_GRAPH_STATUS_OK
;
645 if (!graph
|| !listener
) {
646 status
= BT_GRAPH_STATUS_INVALID
;
650 add_listener(graph
->listeners
.ports_connected
, listener
, data
);
656 enum bt_graph_status
bt_graph_add_ports_disconnected_listener(
657 struct bt_graph
*graph
,
658 bt_graph_ports_disconnected_listener listener
, void *data
)
660 enum bt_graph_status status
= BT_GRAPH_STATUS_OK
;
662 if (!graph
|| !listener
) {
663 status
= BT_GRAPH_STATUS_INVALID
;
667 add_listener(graph
->listeners
.ports_disconnected
, listener
, data
);
674 void bt_graph_notify_port_added(struct bt_graph
*graph
, struct bt_port
*port
)
678 for (i
= 0; i
< graph
->listeners
.port_added
->len
; i
++) {
679 struct bt_graph_listener listener
=
680 g_array_index(graph
->listeners
.port_added
,
681 struct bt_graph_listener
, i
);
682 bt_graph_port_added_listener func
= listener
.func
;
685 func(port
, listener
.data
);
690 void bt_graph_notify_port_removed(struct bt_graph
*graph
,
691 struct bt_component
*comp
, struct bt_port
*port
)
695 for (i
= 0; i
< graph
->listeners
.port_removed
->len
; i
++) {
696 struct bt_graph_listener listener
=
697 g_array_index(graph
->listeners
.port_removed
,
698 struct bt_graph_listener
, i
);
699 bt_graph_port_removed_listener func
= listener
.func
;
702 func(comp
, port
, listener
.data
);
707 void bt_graph_notify_ports_connected(struct bt_graph
*graph
,
708 struct bt_port
*upstream_port
, struct bt_port
*downstream_port
)
712 for (i
= 0; i
< graph
->listeners
.ports_connected
->len
; i
++) {
713 struct bt_graph_listener listener
=
714 g_array_index(graph
->listeners
.ports_connected
,
715 struct bt_graph_listener
, i
);
716 bt_graph_ports_connected_listener func
= listener
.func
;
719 func(upstream_port
, downstream_port
, listener
.data
);
724 void bt_graph_notify_ports_disconnected(struct bt_graph
*graph
,
725 struct bt_component
*upstream_comp
,
726 struct bt_component
*downstream_comp
,
727 struct bt_port
*upstream_port
, struct bt_port
*downstream_port
)
731 for (i
= 0; i
< graph
->listeners
.ports_disconnected
->len
; i
++) {
732 struct bt_graph_listener listener
=
733 g_array_index(graph
->listeners
.ports_disconnected
,
734 struct bt_graph_listener
, i
);
735 bt_graph_ports_disconnected_listener func
= listener
.func
;
738 func(upstream_comp
, downstream_comp
, upstream_port
,
739 downstream_port
, listener
.data
);
743 extern enum bt_graph_status
bt_graph_cancel(struct bt_graph
*graph
)
745 enum bt_graph_status ret
= BT_GRAPH_STATUS_OK
;
748 ret
= BT_GRAPH_STATUS_INVALID
;
752 graph
->canceled
= BT_TRUE
;
758 extern bt_bool
bt_graph_is_canceled(struct bt_graph
*graph
)
760 return graph
? graph
->canceled
: BT_FALSE
;