Values API: split into private and public APIs
[babeltrace.git] / lib / graph / graph.c
1 /*
2 * graph.c
3 *
4 * Babeltrace Plugin Component Graph
5 *
6 * Copyright 2017 Jérémie Galarneau <jeremie.galarneau@efficios.com>
7 * Copyright 2017 Philippe Proulx <pproulx@efficios.com>
8 *
9 * Permission is hereby granted, free of charge, to any person obtaining a copy
10 * of this software and associated documentation files (the "Software"), to deal
11 * in the Software without restriction, including without limitation the rights
12 * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
13 * copies of the Software, and to permit persons to whom the Software is
14 * furnished to do so, subject to the following conditions:
15 *
16 * The above copyright notice and this permission notice shall be included in
17 * all copies or substantial portions of the Software.
18 *
19 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
20 * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
21 * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
22 * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
23 * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
24 * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
25 * SOFTWARE.
26 */
27
28 #define BT_LOG_TAG "GRAPH"
29 #include <babeltrace/lib-logging-internal.h>
30
31 #include <babeltrace/graph/component-internal.h>
32 #include <babeltrace/graph/graph-internal.h>
33 #include <babeltrace/graph/connection-internal.h>
34 #include <babeltrace/graph/component-sink-internal.h>
35 #include <babeltrace/graph/component-source.h>
36 #include <babeltrace/graph/component-filter.h>
37 #include <babeltrace/graph/port.h>
38 #include <babeltrace/graph/notification-internal.h>
39 #include <babeltrace/graph/notification-event-internal.h>
40 #include <babeltrace/graph/notification-packet-internal.h>
41 #include <babeltrace/compiler-internal.h>
42 #include <babeltrace/common-internal.h>
43 #include <babeltrace/types.h>
44 #include <babeltrace/values.h>
45 #include <babeltrace/private-values.h>
46 #include <babeltrace/values-internal.h>
47 #include <babeltrace/assert-internal.h>
48 #include <babeltrace/assert-pre-internal.h>
49 #include <unistd.h>
50 #include <glib.h>
51
52 struct bt_graph_listener {
53 void *func;
54 bt_graph_listener_removed removed;
55 void *data;
56 };
57
58 static
59 int init_listeners_array(GArray **listeners)
60 {
61 int ret = 0;
62
63 BT_ASSERT(listeners);
64 *listeners = g_array_new(FALSE, TRUE, sizeof(struct bt_graph_listener));
65 if (!*listeners) {
66 BT_LOGE_STR("Failed to allocate one GArray.");
67 ret = -1;
68 goto end;
69 }
70
71 end:
72 return ret;
73 }
74
75 static
76 void call_remove_listeners(GArray *listeners)
77 {
78 size_t i;
79
80 for (i = 0; i < listeners->len; i++) {
81 struct bt_graph_listener listener =
82 g_array_index(listeners, struct bt_graph_listener, i);
83
84 if (listener.removed) {
85 listener.removed(listener.data);
86 }
87 }
88 }
89
90 static
91 void bt_graph_destroy(struct bt_object *obj)
92 {
93 struct bt_graph *graph = container_of(obj,
94 struct bt_graph, base);
95
96 /*
97 * The graph's reference count is 0 if we're here. Increment
98 * it to avoid a double-destroy (possibly infinitely recursive)
99 * in this situation:
100 *
101 * 1. We put and destroy a connection.
102 * 2. This connection's destructor finalizes its active
103 * notification iterators.
104 * 3. A notification iterator's finalization function gets a
105 * new reference on its component (reference count goes from
106 * 0 to 1).
107 * 4. Since this component's reference count goes to 1, it takes
108 * a reference on its parent (this graph). This graph's
109 * reference count goes from 0 to 1.
110 * 5. The notification iterator's finalization function puts its
111 * component reference (reference count goes from 1 to 0).
112 * 6. Since this component's reference count goes from 1 to 0,
113 * it puts its parent (this graph). This graph's reference
114 * count goes from 1 to 0.
115 * 7. Since this graph's reference count goes from 1 to 0,
116 * its destructor is called (this function).
117 *
118 * With the incrementation below, the graph's reference count at
119 * step 4 goes from 1 to 2, and from 2 to 1 at step 6. This
120 * ensures that this function is not called two times.
121 */
122 BT_LOGD("Destroying graph: addr=%p", graph);
123 obj->ref_count++;
124
125 /*
126 * Cancel the graph to disallow some operations, like creating
127 * notification iterators and adding ports to components.
128 */
129 (void) bt_graph_cancel(graph);
130
131 /* Call all remove listeners */
132 call_remove_listeners(graph->listeners.port_added);
133 call_remove_listeners(graph->listeners.port_removed);
134 call_remove_listeners(graph->listeners.ports_connected);
135 call_remove_listeners(graph->listeners.ports_disconnected);
136
137 if (graph->notifications) {
138 g_ptr_array_free(graph->notifications, TRUE);
139 }
140
141 if (graph->connections) {
142 BT_LOGD_STR("Destroying connections.");
143 g_ptr_array_free(graph->connections, TRUE);
144 }
145
146 if (graph->components) {
147 BT_LOGD_STR("Destroying components.");
148 g_ptr_array_free(graph->components, TRUE);
149 }
150
151 if (graph->sinks_to_consume) {
152 g_queue_free(graph->sinks_to_consume);
153 }
154
155 if (graph->listeners.port_added) {
156 g_array_free(graph->listeners.port_added, TRUE);
157 }
158
159 if (graph->listeners.port_removed) {
160 g_array_free(graph->listeners.port_removed, TRUE);
161 }
162
163 if (graph->listeners.ports_connected) {
164 g_array_free(graph->listeners.ports_connected, TRUE);
165 }
166
167 if (graph->listeners.ports_disconnected) {
168 g_array_free(graph->listeners.ports_disconnected, TRUE);
169 }
170
171 bt_object_pool_finalize(&graph->event_notif_pool);
172 bt_object_pool_finalize(&graph->packet_begin_notif_pool);
173 bt_object_pool_finalize(&graph->packet_end_notif_pool);
174 g_free(graph);
175 }
176
177 static
178 void destroy_notification_event(struct bt_notification *notif,
179 struct bt_graph *graph)
180 {
181 bt_notification_event_destroy(notif);
182 }
183
184 static
185 void destroy_notification_packet_begin(struct bt_notification *notif,
186 struct bt_graph *graph)
187 {
188 bt_notification_packet_begin_destroy(notif);
189 }
190
191 static
192 void destroy_notification_packet_end(struct bt_notification *notif,
193 struct bt_graph *graph)
194 {
195 bt_notification_packet_end_destroy(notif);
196 }
197
198 static
199 void notify_notification_graph_is_destroyed(struct bt_notification *notif)
200 {
201 bt_notification_unlink_graph(notif);
202 }
203
204 struct bt_graph *bt_graph_create(void)
205 {
206 struct bt_graph *graph;
207 int ret;
208
209 BT_LOGD_STR("Creating graph object.");
210 graph = g_new0(struct bt_graph, 1);
211 if (!graph) {
212 BT_LOGE_STR("Failed to allocate one graph.");
213 goto end;
214 }
215
216 bt_object_init_shared(&graph->base, bt_graph_destroy);
217 graph->connections = g_ptr_array_new_with_free_func(
218 (GDestroyNotify) bt_object_try_spec_release);
219 if (!graph->connections) {
220 BT_LOGE_STR("Failed to allocate one GPtrArray.");
221 goto error;
222 }
223 graph->components = g_ptr_array_new_with_free_func(
224 (GDestroyNotify) bt_object_try_spec_release);
225 if (!graph->components) {
226 BT_LOGE_STR("Failed to allocate one GPtrArray.");
227 goto error;
228 }
229 graph->sinks_to_consume = g_queue_new();
230 if (!graph->sinks_to_consume) {
231 BT_LOGE_STR("Failed to allocate one GQueue.");
232 goto error;
233 }
234
235 bt_graph_set_can_consume(graph, BT_TRUE);
236 ret = init_listeners_array(&graph->listeners.port_added);
237 if (ret) {
238 BT_LOGE_STR("Cannot create the \"port added\" listener array.");
239 goto error;
240 }
241
242 ret = init_listeners_array(&graph->listeners.port_removed);
243 if (ret) {
244 BT_LOGE_STR("Cannot create the \"port removed\" listener array.");
245 goto error;
246 }
247
248 ret = init_listeners_array(&graph->listeners.ports_connected);
249 if (ret) {
250 BT_LOGE_STR("Cannot create the \"port connected\" listener array.");
251 goto error;
252 }
253
254 ret = init_listeners_array(&graph->listeners.ports_disconnected);
255 if (ret) {
256 BT_LOGE_STR("Cannot create the \"port disconneted\" listener array.");
257 goto error;
258 }
259
260 ret = bt_object_pool_initialize(&graph->event_notif_pool,
261 (bt_object_pool_new_object_func) bt_notification_event_new,
262 (bt_object_pool_destroy_object_func) destroy_notification_event,
263 graph);
264 if (ret) {
265 BT_LOGE("Failed to initialize event notification pool: ret=%d",
266 ret);
267 goto error;
268 }
269
270 ret = bt_object_pool_initialize(&graph->packet_begin_notif_pool,
271 (bt_object_pool_new_object_func) bt_notification_packet_begin_new,
272 (bt_object_pool_destroy_object_func) destroy_notification_packet_begin,
273 graph);
274 if (ret) {
275 BT_LOGE("Failed to initialize packet beginning notification pool: ret=%d",
276 ret);
277 goto error;
278 }
279
280 ret = bt_object_pool_initialize(&graph->packet_end_notif_pool,
281 (bt_object_pool_new_object_func) bt_notification_packet_end_new,
282 (bt_object_pool_destroy_object_func) destroy_notification_packet_end,
283 graph);
284 if (ret) {
285 BT_LOGE("Failed to initialize packet end notification pool: ret=%d",
286 ret);
287 goto error;
288 }
289
290 graph->notifications = g_ptr_array_new_with_free_func(
291 (GDestroyNotify) notify_notification_graph_is_destroyed);
292 BT_LOGD("Created graph object: addr=%p", graph);
293
294 end:
295 return graph;
296 error:
297 BT_OBJECT_PUT_REF_AND_RESET(graph);
298 goto end;
299 }
300
301 enum bt_graph_status bt_graph_connect_ports(struct bt_graph *graph,
302 struct bt_port *upstream_port, struct bt_port *downstream_port,
303 struct bt_connection **user_connection)
304 {
305 enum bt_graph_status status = BT_GRAPH_STATUS_OK;
306 struct bt_connection *connection = NULL;
307 struct bt_graph *upstream_graph = NULL;
308 struct bt_graph *downstream_graph = NULL;
309 struct bt_component *upstream_component = NULL;
310 struct bt_component *downstream_component = NULL;
311 enum bt_component_status component_status;
312 bt_bool init_can_consume;
313
314 if (!graph) {
315 BT_LOGW_STR("Invalid parameter: graph is NULL.");
316 status = BT_GRAPH_STATUS_INVALID;
317 goto end;
318 }
319 init_can_consume = graph->can_consume;
320
321 if (!upstream_port) {
322 BT_LOGW_STR("Invalid parameter: upstream port is NULL.");
323 status = BT_GRAPH_STATUS_INVALID;
324 goto end;
325 }
326
327 if (!downstream_port) {
328 BT_LOGW_STR("Invalid parameter: downstream port is NULL.");
329 status = BT_GRAPH_STATUS_INVALID;
330 goto end;
331 }
332
333 BT_LOGD("Connecting component ports within graph: "
334 "graph-addr=%p, "
335 "upstream-port-addr=%p, upstream-port-name=\"%s\", "
336 "downstream-port-addr=%p, downstream-port-name=\"%s\"",
337 graph, upstream_port, bt_port_get_name(upstream_port),
338 downstream_port, bt_port_get_name(downstream_port));
339
340 if (graph->canceled) {
341 BT_LOGW_STR("Invalid parameter: graph is canceled.");
342 status = BT_GRAPH_STATUS_CANCELED;
343 goto end;
344 }
345
346 bt_graph_set_can_consume(graph, BT_FALSE);
347
348 /* Ensure appropriate types for upstream and downstream ports. */
349 if (bt_port_get_type(upstream_port) != BT_PORT_TYPE_OUTPUT) {
350 BT_LOGW_STR("Invalid parameter: upstream port is not an output port.");
351 status = BT_GRAPH_STATUS_INVALID;
352 goto end;
353 }
354 if (bt_port_get_type(downstream_port) != BT_PORT_TYPE_INPUT) {
355 BT_LOGW_STR("Invalid parameter: downstream port is not an input port.");
356 status = BT_GRAPH_STATUS_INVALID;
357 goto end;
358 }
359
360 /* Ensure that both ports are currently unconnected. */
361 if (bt_port_is_connected(upstream_port)) {
362 BT_LOGW_STR("Invalid parameter: upstream port is already connected.");
363 status = BT_GRAPH_STATUS_INVALID;
364 goto end;
365 }
366
367 if (bt_port_is_connected(downstream_port)) {
368 BT_LOGW_STR("Invalid parameter: downstream port is already connected.");
369 status = BT_GRAPH_STATUS_INVALID;
370 goto end;
371 }
372
373 /*
374 * Ensure that both ports are still attached to their creating
375 * component.
376 */
377 upstream_component = bt_port_get_component(upstream_port);
378 if (!upstream_component) {
379 BT_LOGW_STR("Invalid parameter: upstream port is loose (does not belong to a component)");
380 status = BT_GRAPH_STATUS_INVALID;
381 goto end;
382 }
383
384 downstream_component = bt_port_get_component(downstream_port);
385 if (!downstream_component) {
386 BT_LOGW_STR("Invalid parameter: downstream port is loose (does not belong to a component)");
387 status = BT_GRAPH_STATUS_INVALID;
388 goto end;
389 }
390
391 BT_LOGD("Connecting component ports: "
392 "upstream-comp-addr=%p, upstream-comp-name=\"%s\", "
393 "downstream-comp-addr=%p, downstream-comp-name=\"%s\"",
394 upstream_component, bt_component_get_name(upstream_component),
395 downstream_component, bt_component_get_name(downstream_component));
396
397 /*
398 * At this point the ports are not connected yet. Both
399 * components need to accept an eventual connection to their
400 * port by the other port before we continue.
401 */
402 BT_LOGD_STR("Asking upstream component to accept the connection.");
403 component_status = bt_component_accept_port_connection(
404 upstream_component, upstream_port, downstream_port);
405 if (component_status != BT_COMPONENT_STATUS_OK) {
406 if (component_status == BT_COMPONENT_STATUS_REFUSE_PORT_CONNECTION) {
407 BT_LOGD_STR("Upstream component refused the connection.");
408 } else {
409 BT_LOGW("Cannot ask upstream component to accept the connection: "
410 "status=%s", bt_component_status_string(component_status));
411 }
412
413 status = bt_graph_status_from_component_status(
414 component_status);
415 goto end;
416 }
417
418 BT_LOGD_STR("Asking downstream component to accept the connection.");
419 component_status = bt_component_accept_port_connection(
420 downstream_component, downstream_port, upstream_port);
421 if (component_status != BT_COMPONENT_STATUS_OK) {
422 if (component_status == BT_COMPONENT_STATUS_REFUSE_PORT_CONNECTION) {
423 BT_LOGD_STR("Downstream component refused the connection.");
424 } else {
425 BT_LOGW("Cannot ask downstream component to accept the connection: "
426 "status=%s", bt_component_status_string(component_status));
427 }
428
429 status = bt_graph_status_from_component_status(
430 component_status);
431 goto end;
432 }
433
434 BT_LOGD_STR("Creating connection.");
435 connection = bt_connection_create(graph, upstream_port,
436 downstream_port);
437 if (!connection) {
438 BT_LOGW("Cannot create connection object.");
439 status = BT_GRAPH_STATUS_NOMEM;
440 goto end;
441 }
442
443 BT_LOGD("Connection object created: conn-addr=%p", connection);
444
445 /*
446 * Ownership of upstream_component/downstream_component and of
447 * the connection object is transferred to the graph.
448 */
449 g_ptr_array_add(graph->connections, connection);
450
451 /*
452 * Notify both components that their port is connected.
453 */
454 BT_LOGD_STR("Notifying upstream component that its port is connected.");
455 component_status = bt_component_port_connected(upstream_component,
456 upstream_port, downstream_port);
457 if (component_status != BT_COMPONENT_STATUS_OK) {
458 BT_LOGW("Error while notifying upstream component that its port is connected: "
459 "status=%s, graph-addr=%p, "
460 "upstream-comp-addr=%p, upstream-comp-name=\"%s\", "
461 "downstream-comp-addr=%p, downstream-comp-name=\"%s\", "
462 "upstream-port-addr=%p, upstream-port-name=\"%s\", "
463 "downstream-port-addr=%p, downstream-port-name=\"%s\"",
464 bt_component_status_string(component_status), graph,
465 upstream_component, bt_component_get_name(upstream_component),
466 downstream_component, bt_component_get_name(downstream_component),
467 upstream_port, bt_port_get_name(upstream_port),
468 downstream_port, bt_port_get_name(downstream_port));
469 bt_connection_end(connection, true);
470 status = bt_graph_status_from_component_status(
471 component_status);
472 goto end;
473 }
474
475 connection->notified_upstream_port_connected = true;
476 BT_LOGD_STR("Notifying downstream component that its port is connected.");
477 component_status = bt_component_port_connected(downstream_component,
478 downstream_port, upstream_port);
479 if (component_status != BT_COMPONENT_STATUS_OK) {
480 BT_LOGW("Error while notifying downstream component that its port is connected: "
481 "status=%s, graph-addr=%p, "
482 "upstream-comp-addr=%p, upstream-comp-name=\"%s\", "
483 "downstream-comp-addr=%p, downstream-comp-name=\"%s\", "
484 "upstream-port-addr=%p, upstream-port-name=\"%s\", "
485 "downstream-port-addr=%p, downstream-port-name=\"%s\"",
486 bt_component_status_string(component_status), graph,
487 upstream_component, bt_component_get_name(upstream_component),
488 downstream_component, bt_component_get_name(downstream_component),
489 upstream_port, bt_port_get_name(upstream_port),
490 downstream_port, bt_port_get_name(downstream_port));
491 bt_connection_end(connection, true);
492 status = bt_graph_status_from_component_status(
493 component_status);
494 goto end;
495 }
496
497 connection->notified_downstream_port_connected = true;
498
499 /*
500 * Notify the graph's creator that both ports are connected.
501 */
502 BT_LOGD_STR("Notifying graph's user that new component ports are connected.");
503 bt_graph_notify_ports_connected(graph, upstream_port, downstream_port);
504 connection->notified_graph_ports_connected = true;
505 BT_LOGD("Connected component ports within graph: "
506 "graph-addr=%p, "
507 "upstream-comp-addr=%p, upstream-comp-name=\"%s\", "
508 "downstream-comp-addr=%p, downstream-comp-name=\"%s\", "
509 "upstream-port-addr=%p, upstream-port-name=\"%s\", "
510 "downstream-port-addr=%p, downstream-port-name=\"%s\"",
511 graph,
512 upstream_component, bt_component_get_name(upstream_component),
513 downstream_component, bt_component_get_name(downstream_component),
514 upstream_port, bt_port_get_name(upstream_port),
515 downstream_port, bt_port_get_name(downstream_port));
516
517 if (user_connection) {
518 /* Move reference to user */
519 *user_connection = connection;
520 connection = NULL;
521 }
522
523 end:
524 bt_object_put_ref(upstream_graph);
525 bt_object_put_ref(downstream_graph);
526 bt_object_put_ref(upstream_component);
527 bt_object_put_ref(downstream_component);
528 bt_object_put_ref(connection);
529 if (graph) {
530 (void) init_can_consume;
531 bt_graph_set_can_consume(graph, init_can_consume);
532 }
533 return status;
534 }
535
536 static inline
537 enum bt_graph_status consume_graph_sink(struct bt_component *sink)
538 {
539 enum bt_component_status comp_status;
540
541 BT_ASSERT(sink);
542 comp_status = bt_component_sink_consume(sink);
543 BT_LOGV("Consumed from sink: addr=%p, name=\"%s\", status=%s",
544 sink, bt_component_get_name(sink),
545 bt_component_status_string(comp_status));
546 BT_ASSERT_PRE(comp_status == BT_COMPONENT_STATUS_OK ||
547 comp_status == BT_COMPONENT_STATUS_END ||
548 comp_status == BT_COMPONENT_STATUS_AGAIN ||
549 comp_status == BT_COMPONENT_STATUS_ERROR ||
550 comp_status == BT_COMPONENT_STATUS_NOMEM,
551 "Invalid component status returned by consuming function: "
552 "status=%s", bt_component_status_string(comp_status));
553 return (enum bt_graph_status) comp_status;
554 }
555
556 /*
557 * `node` is removed from the queue of sinks to consume when passed to
558 * this function. This function adds it back to the queue if there's
559 * still something to consume afterwards.
560 */
561 static inline
562 enum bt_graph_status consume_sink_node(struct bt_graph *graph,
563 GList *node)
564 {
565 enum bt_graph_status status;
566 struct bt_component *sink;
567
568 sink = node->data;
569 status = consume_graph_sink(sink);
570 if (unlikely(status != BT_GRAPH_STATUS_END)) {
571 g_queue_push_tail_link(graph->sinks_to_consume, node);
572 goto end;
573 }
574
575 /* End reached, the node is not added back to the queue and free'd. */
576 g_queue_delete_link(graph->sinks_to_consume, node);
577
578 /* Don't forward an END status if there are sinks left to consume. */
579 if (!g_queue_is_empty(graph->sinks_to_consume)) {
580 status = BT_GRAPH_STATUS_OK;
581 goto end;
582 }
583
584 end:
585 BT_LOGV("Consumed sink node: status=%s", bt_graph_status_string(status));
586 return status;
587 }
588
589 BT_HIDDEN
590 enum bt_graph_status bt_graph_consume_sink_no_check(struct bt_graph *graph,
591 struct bt_component *sink)
592 {
593 enum bt_graph_status status;
594 GList *sink_node;
595 int index;
596
597 BT_LOGV("Making specific sink consume: addr=%p, "
598 "comp-addr=%p, comp-name=\"%s\"",
599 graph, sink, bt_component_get_name(sink));
600
601 BT_ASSERT(bt_component_borrow_graph(sink) == graph);
602
603 if (g_queue_is_empty(graph->sinks_to_consume)) {
604 BT_LOGV_STR("Graph's sink queue is empty: end of graph.");
605 status = BT_GRAPH_STATUS_END;
606 goto end;
607 }
608
609 index = g_queue_index(graph->sinks_to_consume, sink);
610 if (index < 0) {
611 BT_LOGV_STR("Sink is not marked as consumable: sink is ended.");
612 status = BT_GRAPH_STATUS_END;
613 goto end;
614 }
615
616 sink_node = g_queue_pop_nth_link(graph->sinks_to_consume, index);
617 BT_ASSERT(sink_node);
618 status = consume_sink_node(graph, sink_node);
619
620 end:
621 return status;
622 }
623
624 static inline
625 enum bt_graph_status bt_graph_consume_no_check(struct bt_graph *graph)
626 {
627 enum bt_graph_status status = BT_GRAPH_STATUS_OK;
628 struct bt_component *sink;
629 GList *current_node;
630
631 BT_LOGV("Making next sink consume: addr=%p", graph);
632 BT_ASSERT_PRE(graph->has_sink,
633 "Graph has no sink component: %!+g", graph);
634
635 if (unlikely(g_queue_is_empty(graph->sinks_to_consume))) {
636 BT_LOGV_STR("Graph's sink queue is empty: end of graph.");
637 status = BT_GRAPH_STATUS_END;
638 goto end;
639 }
640
641 current_node = g_queue_pop_head_link(graph->sinks_to_consume);
642 sink = current_node->data;
643 BT_LOGV("Chose next sink to consume: comp-addr=%p, comp-name=\"%s\"",
644 sink, bt_component_get_name(sink));
645 status = consume_sink_node(graph, current_node);
646
647 end:
648 return status;
649 }
650
651 enum bt_graph_status bt_graph_consume(struct bt_graph *graph)
652 {
653 enum bt_graph_status status;
654
655 BT_ASSERT_PRE_NON_NULL(graph, "Graph");
656 BT_ASSERT_PRE(!graph->canceled, "Graph is canceled: %!+g", graph);
657 BT_ASSERT_PRE(graph->can_consume,
658 "Cannot consume graph in its current state: %!+g", graph);
659 bt_graph_set_can_consume(graph, BT_FALSE);
660 status = bt_graph_consume_no_check(graph);
661 bt_graph_set_can_consume(graph, BT_TRUE);
662 return status;
663 }
664
665 enum bt_graph_status bt_graph_run(struct bt_graph *graph)
666 {
667 enum bt_graph_status status = BT_GRAPH_STATUS_OK;
668
669 if (!graph) {
670 BT_LOGW_STR("Invalid parameter: graph is NULL.");
671 status = BT_GRAPH_STATUS_INVALID;
672 goto end;
673 }
674
675 if (graph->canceled) {
676 BT_LOGW("Invalid parameter: graph is canceled: "
677 "graph-addr=%p", graph);
678 status = BT_GRAPH_STATUS_CANCELED;
679 goto end;
680 }
681
682 BT_ASSERT_PRE(graph->can_consume,
683 "Cannot consume graph in its current state: %!+g", graph);
684 bt_graph_set_can_consume(graph, BT_FALSE);
685 BT_LOGV("Running graph: addr=%p", graph);
686
687 do {
688 /*
689 * Check if the graph is canceled at each iteration. If
690 * the graph was canceled by another thread or by a
691 * signal, this is not a warning nor an error, it was
692 * intentional: log with a DEBUG level only.
693 */
694 if (unlikely(graph->canceled)) {
695 BT_LOGD("Stopping the graph: graph is canceled: "
696 "graph-addr=%p", graph);
697 status = BT_GRAPH_STATUS_CANCELED;
698 goto end;
699 }
700
701 status = bt_graph_consume_no_check(graph);
702 if (unlikely(status == BT_GRAPH_STATUS_AGAIN)) {
703 /*
704 * If AGAIN is received and there are multiple
705 * sinks, go ahead and consume from the next
706 * sink.
707 *
708 * However, in the case where a single sink is
709 * left, the caller can decide to busy-wait and
710 * call bt_graph_run() continuously until the
711 * source is ready or it can decide to sleep for
712 * an arbitrary amount of time.
713 */
714 if (graph->sinks_to_consume->length > 1) {
715 status = BT_GRAPH_STATUS_OK;
716 }
717 } else if (status == BT_GRAPH_STATUS_NO_SINK) {
718 goto end;
719 }
720 } while (status == BT_GRAPH_STATUS_OK);
721
722 if (g_queue_is_empty(graph->sinks_to_consume)) {
723 status = BT_GRAPH_STATUS_END;
724 }
725
726 end:
727 BT_LOGV("Graph ran: status=%s", bt_graph_status_string(status));
728 if (graph) {
729 bt_graph_set_can_consume(graph, BT_TRUE);
730 }
731 return status;
732 }
733
734 static
735 int add_listener(GArray *listeners, void *func, void *removed, void *data)
736 {
737 struct bt_graph_listener listener = {
738 .func = func,
739 .removed = removed,
740 .data = data,
741 };
742
743 g_array_append_val(listeners, listener);
744 return listeners->len - 1;
745 }
746
747 int bt_graph_add_port_added_listener(
748 struct bt_graph *graph,
749 bt_graph_port_added_listener listener,
750 bt_graph_listener_removed listener_removed, void *data)
751 {
752 int ret;
753
754 if (!graph) {
755 BT_LOGW_STR("Invalid parameter: graph is NULL.");
756 ret = -1;
757 goto end;
758 }
759
760 if (graph->in_remove_listener) {
761 BT_LOGW("Cannot call this function during the execution of a remove listener: "
762 "addr=%p", graph);
763 ret = -1;
764 goto end;
765 }
766
767 if (!listener) {
768 BT_LOGW_STR("Invalid parameter: listener is NULL.");
769 ret = -1;
770 goto end;
771 }
772
773 ret = add_listener(graph->listeners.port_added, listener,
774 listener_removed, data);
775 BT_LOGV("Added \"port added\" listener to graph: "
776 "graph-addr=%p, listener-addr=%p, pos=%d",
777 graph, listener, ret);
778
779 end:
780 return ret;
781 }
782
783 int bt_graph_add_port_removed_listener(
784 struct bt_graph *graph,
785 bt_graph_port_removed_listener listener,
786 bt_graph_listener_removed listener_removed, void *data)
787 {
788 int ret;
789
790 if (!graph) {
791 BT_LOGW_STR("Invalid parameter: graph is NULL.");
792 ret = -1;
793 goto end;
794 }
795
796 if (graph->in_remove_listener) {
797 BT_LOGW("Cannot call this function during the execution of a remove listener: "
798 "addr=%p", graph);
799 ret = -1;
800 goto end;
801 }
802
803 if (!listener) {
804 BT_LOGW_STR("Invalid parameter: listener is NULL.");
805 ret = -1;
806 goto end;
807 }
808
809 ret = add_listener(graph->listeners.port_removed, listener,
810 listener_removed, data);
811 BT_LOGV("Added \"port removed\" listener to graph: "
812 "graph-addr=%p, listener-addr=%p, pos=%d",
813 graph, listener, ret);
814
815 end:
816 return ret;
817 }
818
819 int bt_graph_add_ports_connected_listener(
820 struct bt_graph *graph,
821 bt_graph_ports_connected_listener listener,
822 bt_graph_listener_removed listener_removed, void *data)
823 {
824 int ret;
825
826 if (!graph) {
827 BT_LOGW_STR("Invalid parameter: graph is NULL.");
828 ret = -1;
829 goto end;
830 }
831
832 if (graph->in_remove_listener) {
833 BT_LOGW("Cannot call this function during the execution of a remove listener: "
834 "addr=%p", graph);
835 ret = -1;
836 goto end;
837 }
838
839 if (!listener) {
840 BT_LOGW_STR("Invalid parameter: listener is NULL.");
841 ret = -1;
842 goto end;
843 }
844
845 ret = add_listener(graph->listeners.ports_connected, listener,
846 listener_removed, data);
847 BT_LOGV("Added \"port connected\" listener to graph: "
848 "graph-addr=%p, listener-addr=%p, pos=%d",
849 graph, listener, ret);
850
851 end:
852 return ret;
853 }
854
855 int bt_graph_add_ports_disconnected_listener(
856 struct bt_graph *graph,
857 bt_graph_ports_disconnected_listener listener,
858 bt_graph_listener_removed listener_removed, void *data)
859 {
860 int ret;
861
862 if (!graph) {
863 BT_LOGW_STR("Invalid parameter: graph is NULL.");
864 ret = -1;
865 goto end;
866 }
867
868 if (graph->in_remove_listener) {
869 BT_LOGW("Cannot call this function during the execution of a remove listener: "
870 "addr=%p", graph);
871 ret = -1;
872 goto end;
873 }
874
875 if (!listener) {
876 BT_LOGW_STR("Invalid parameter: listener is NULL.");
877 ret = -1;
878 goto end;
879 }
880
881 ret = add_listener(graph->listeners.ports_disconnected, listener,
882 listener_removed, data);
883 BT_LOGV("Added \"port disconnected\" listener to graph: "
884 "graph-addr=%p, listener-addr=%p, pos=%d",
885 graph, listener, ret);
886
887 end:
888 return ret;
889 }
890
891 BT_HIDDEN
892 void bt_graph_notify_port_added(struct bt_graph *graph, struct bt_port *port)
893 {
894 size_t i;
895
896 BT_LOGV("Notifying graph listeners that a port was added: "
897 "graph-addr=%p, port-addr=%p, port-name=\"%s\"",
898 graph, port, bt_port_get_name(port));
899
900 for (i = 0; i < graph->listeners.port_added->len; i++) {
901 struct bt_graph_listener listener =
902 g_array_index(graph->listeners.port_added,
903 struct bt_graph_listener, i);
904 bt_graph_port_added_listener func = listener.func;
905
906 BT_ASSERT(func);
907 func(port, listener.data);
908 }
909 }
910
911 BT_HIDDEN
912 void bt_graph_notify_port_removed(struct bt_graph *graph,
913 struct bt_component *comp, struct bt_port *port)
914 {
915 size_t i;
916
917 BT_LOGV("Notifying graph listeners that a port was removed: "
918 "graph-addr=%p, port-addr=%p, port-name=\"%s\"",
919 graph, port, bt_port_get_name(port));
920
921 for (i = 0; i < graph->listeners.port_removed->len; i++) {
922 struct bt_graph_listener listener =
923 g_array_index(graph->listeners.port_removed,
924 struct bt_graph_listener, i);
925 bt_graph_port_removed_listener func = listener.func;
926
927 BT_ASSERT(func);
928 func(comp, port, listener.data);
929 }
930 }
931
932 BT_HIDDEN
933 void bt_graph_notify_ports_connected(struct bt_graph *graph,
934 struct bt_port *upstream_port, struct bt_port *downstream_port)
935 {
936 size_t i;
937
938 BT_LOGV("Notifying graph listeners that two ports were connected: "
939 "graph-addr=%p, "
940 "upstream-port-addr=%p, upstream-port-name=\"%s\", "
941 "downstream-port-addr=%p, downstream-port-name=\"%s\"",
942 graph, upstream_port, bt_port_get_name(upstream_port),
943 downstream_port, bt_port_get_name(downstream_port));
944
945 for (i = 0; i < graph->listeners.ports_connected->len; i++) {
946 struct bt_graph_listener listener =
947 g_array_index(graph->listeners.ports_connected,
948 struct bt_graph_listener, i);
949 bt_graph_ports_connected_listener func = listener.func;
950
951 BT_ASSERT(func);
952 func(upstream_port, downstream_port, listener.data);
953 }
954 }
955
956 BT_HIDDEN
957 void bt_graph_notify_ports_disconnected(struct bt_graph *graph,
958 struct bt_component *upstream_comp,
959 struct bt_component *downstream_comp,
960 struct bt_port *upstream_port, struct bt_port *downstream_port)
961 {
962 size_t i;
963
964 BT_LOGV("Notifying graph listeners that two ports were disconnected: "
965 "graph-addr=%p, "
966 "upstream-port-addr=%p, upstream-port-name=\"%s\", "
967 "downstream-port-addr=%p, downstream-port-name=\"%s\"",
968 graph, upstream_port, bt_port_get_name(upstream_port),
969 downstream_port, bt_port_get_name(downstream_port));
970
971 for (i = 0; i < graph->listeners.ports_disconnected->len; i++) {
972 struct bt_graph_listener listener =
973 g_array_index(graph->listeners.ports_disconnected,
974 struct bt_graph_listener, i);
975 bt_graph_ports_disconnected_listener func = listener.func;
976
977 BT_ASSERT(func);
978 func(upstream_comp, downstream_comp, upstream_port,
979 downstream_port, listener.data);
980 }
981 }
982
983 enum bt_graph_status bt_graph_cancel(struct bt_graph *graph)
984 {
985 enum bt_graph_status ret = BT_GRAPH_STATUS_OK;
986
987 if (!graph) {
988 BT_LOGW_STR("Invalid parameter: graph is NULL.");
989 ret = BT_GRAPH_STATUS_INVALID;
990 goto end;
991 }
992
993 graph->canceled = BT_TRUE;
994 BT_LOGV("Canceled graph: addr=%p", graph);
995
996 end:
997 return ret;
998 }
999
1000 bt_bool bt_graph_is_canceled(struct bt_graph *graph)
1001 {
1002 bt_bool canceled = BT_FALSE;
1003
1004 if (!graph) {
1005 BT_LOGW_STR("Invalid parameter: graph is NULL.");
1006 goto end;
1007 }
1008
1009 canceled = graph->canceled;
1010
1011 end:
1012 return canceled;
1013 }
1014
1015 BT_HIDDEN
1016 void bt_graph_remove_connection(struct bt_graph *graph,
1017 struct bt_connection *connection)
1018 {
1019 BT_ASSERT(graph);
1020 BT_ASSERT(connection);
1021 BT_LOGV("Removing graph's connection: graph-addr=%p, conn-addr=%p",
1022 graph, connection);
1023 g_ptr_array_remove(graph->connections, connection);
1024 }
1025
1026 enum bt_graph_status bt_graph_add_component_with_init_method_data(
1027 struct bt_graph *graph,
1028 struct bt_component_class *component_class,
1029 const char *name, struct bt_value *params,
1030 void *init_method_data,
1031 struct bt_component **user_component)
1032 {
1033 enum bt_graph_status graph_status = BT_GRAPH_STATUS_OK;
1034 enum bt_component_status comp_status;
1035 struct bt_component *component = NULL;
1036 enum bt_component_class_type type;
1037 size_t i;
1038 bt_bool init_can_consume;
1039
1040 bt_object_get_ref(params);
1041
1042 if (!graph) {
1043 BT_LOGW_STR("Invalid parameter: graph is NULL.");
1044 graph_status = BT_GRAPH_STATUS_INVALID;
1045 goto end;
1046 }
1047 init_can_consume = graph->can_consume;
1048
1049 if (!component_class) {
1050 BT_LOGW_STR("Invalid parameter: component class is NULL.");
1051 graph_status = BT_GRAPH_STATUS_INVALID;
1052 goto end;
1053 }
1054
1055 graph->can_consume = BT_FALSE;
1056 type = bt_component_class_get_type(component_class);
1057 BT_LOGD("Adding component to graph: "
1058 "graph-addr=%p, comp-cls-addr=%p, "
1059 "comp-cls-type=%s, name=\"%s\", params-addr=%p, "
1060 "init-method-data-addr=%p",
1061 graph, component_class, bt_component_class_type_string(type),
1062 name, params, init_method_data);
1063
1064 if (!name) {
1065 BT_LOGW_STR("Invalid parameter: name is NULL.");
1066 graph_status = BT_GRAPH_STATUS_INVALID;
1067 goto end;
1068 }
1069
1070 if (graph->canceled) {
1071 BT_LOGW_STR("Invalid parameter: graph is canceled.");
1072 graph_status = BT_GRAPH_STATUS_CANCELED;
1073 goto end;
1074 }
1075
1076 if (type != BT_COMPONENT_CLASS_TYPE_SOURCE &&
1077 type != BT_COMPONENT_CLASS_TYPE_FILTER &&
1078 type != BT_COMPONENT_CLASS_TYPE_SINK) {
1079 BT_LOGW("Invalid parameter: unknown component class type: "
1080 "type=%d", type);
1081 graph_status = BT_GRAPH_STATUS_INVALID;
1082 goto end;
1083 }
1084
1085 for (i = 0; i < graph->components->len; i++) {
1086 void *other_comp = graph->components->pdata[i];
1087
1088 if (strcmp(name, bt_component_get_name(other_comp)) == 0) {
1089 BT_LOGW("Invalid parameter: another component with the same name already exists in the graph: "
1090 "other-comp-addr=%p, name=\"%s\"",
1091 other_comp, name);
1092 graph_status = BT_GRAPH_STATUS_INVALID;
1093 goto end;
1094 }
1095 }
1096
1097 /*
1098 * Parameters must be a map value, but we create a convenient
1099 * empty one if it's NULL.
1100 */
1101 if (params) {
1102 if (!bt_value_is_map(params)) {
1103 BT_LOGW("Invalid parameter: initialization parameters must be a map value: "
1104 "type=%s",
1105 bt_common_value_type_string(
1106 bt_value_get_type(params)));
1107 graph_status = BT_GRAPH_STATUS_INVALID;
1108 goto end;
1109 }
1110 } else {
1111 params = bt_value_borrow_from_private(
1112 bt_private_value_map_create());
1113 if (!params) {
1114 BT_LOGE_STR("Cannot create map value object.");
1115 graph_status = BT_GRAPH_STATUS_NOMEM;
1116 goto end;
1117 }
1118 }
1119
1120 comp_status = bt_component_create(component_class, name, &component);
1121 if (comp_status != BT_COMPONENT_STATUS_OK) {
1122 BT_LOGE("Cannot create empty component object: status=%s",
1123 bt_component_status_string(comp_status));
1124 graph_status = bt_graph_status_from_component_status(
1125 comp_status);
1126 goto end;
1127 }
1128
1129 /*
1130 * The user's initialization method needs to see that this
1131 * component is part of the graph. If the user method fails, we
1132 * immediately remove the component from the graph's components.
1133 */
1134 g_ptr_array_add(graph->components, component);
1135 bt_component_set_graph(component, graph);
1136
1137 if (component_class->methods.init) {
1138 BT_LOGD_STR("Calling user's initialization method.");
1139 comp_status = component_class->methods.init(
1140 bt_private_component_from_component(component), params,
1141 init_method_data);
1142 BT_LOGD("User method returned: status=%s",
1143 bt_component_status_string(comp_status));
1144 if (comp_status != BT_COMPONENT_STATUS_OK) {
1145 BT_LOGW_STR("Initialization method failed.");
1146 graph_status = bt_graph_status_from_component_status(
1147 comp_status);
1148 bt_component_set_graph(component, NULL);
1149 g_ptr_array_remove_fast(graph->components, component);
1150 goto end;
1151 }
1152 }
1153
1154 /*
1155 * Mark the component as initialized so that its finalization
1156 * method is called when it is destroyed.
1157 */
1158 component->initialized = true;
1159
1160 /*
1161 * If it's a sink component, it needs to be part of the graph's
1162 * sink queue to be consumed by bt_graph_consume().
1163 */
1164 if (bt_component_is_sink(component)) {
1165 graph->has_sink = BT_TRUE;
1166 g_queue_push_tail(graph->sinks_to_consume, component);
1167 }
1168
1169 /*
1170 * Freeze the component class now that it's instantiated at
1171 * least once.
1172 */
1173 BT_LOGD_STR("Freezing component class.");
1174 bt_component_class_freeze(component->class);
1175 BT_LOGD("Added component to graph: "
1176 "graph-addr=%p, comp-cls-addr=%p, "
1177 "comp-cls-type=%s, name=\"%s\", params-addr=%p, "
1178 "init-method-data-addr=%p, comp-addr=%p",
1179 graph, component_class, bt_component_class_type_string(type),
1180 name, params, init_method_data, component);
1181
1182 if (user_component) {
1183 /* Move reference to user */
1184 *user_component = component;
1185 component = NULL;
1186 }
1187
1188 end:
1189 bt_object_put_ref(component);
1190 bt_object_put_ref(params);
1191 if (graph) {
1192 graph->can_consume = init_can_consume;
1193 }
1194 return graph_status;
1195 }
1196
1197 enum bt_graph_status bt_graph_add_component(
1198 struct bt_graph *graph,
1199 struct bt_component_class *component_class,
1200 const char *name, struct bt_value *params,
1201 struct bt_component **component)
1202 {
1203 return bt_graph_add_component_with_init_method_data(graph,
1204 component_class, name, params, NULL, component);
1205 }
1206
1207 BT_HIDDEN
1208 int bt_graph_remove_unconnected_component(struct bt_graph *graph,
1209 struct bt_component *component)
1210 {
1211 bt_bool init_can_consume;
1212 int64_t count;
1213 uint64_t i;
1214 int ret = 0;
1215
1216 BT_ASSERT(graph);
1217 BT_ASSERT(component);
1218 BT_ASSERT(component->base.ref_count == 0);
1219 BT_ASSERT(bt_component_borrow_graph(component) == graph);
1220
1221 init_can_consume = graph->can_consume;
1222 count = bt_component_get_input_port_count(component);
1223
1224 for (i = 0; i < count; i++) {
1225 struct bt_port *port =
1226 bt_component_get_input_port_by_index(component, i);
1227
1228 BT_ASSERT(port);
1229 bt_object_put_ref(port);
1230
1231 if (bt_port_is_connected(port)) {
1232 BT_LOGW("Cannot remove component from graph: "
1233 "an input port is connected: "
1234 "graph-addr=%p, comp-addr=%p, "
1235 "comp-name=\"%s\", connected-port-addr=%p, "
1236 "connected-port-name=\"%s\"",
1237 graph, component,
1238 bt_component_get_name(component),
1239 port, bt_port_get_name(port));
1240 goto error;
1241 }
1242 }
1243
1244 count = bt_component_get_output_port_count(component);
1245
1246 for (i = 0; i < count; i++) {
1247 struct bt_port *port =
1248 bt_component_get_output_port_by_index(component, i);
1249
1250 BT_ASSERT(port);
1251 bt_object_put_ref(port);
1252
1253 if (bt_port_is_connected(port)) {
1254 BT_LOGW("Cannot remove component from graph: "
1255 "an output port is connected: "
1256 "graph-addr=%p, comp-addr=%p, "
1257 "comp-name=\"%s\", connected-port-addr=%p, "
1258 "connected-port-name=\"%s\"",
1259 graph, component,
1260 bt_component_get_name(component),
1261 port, bt_port_get_name(port));
1262 goto error;
1263 }
1264 }
1265
1266 graph->can_consume = BT_FALSE;
1267
1268 /* Possibly remove from sinks to consume */
1269 (void) g_queue_remove(graph->sinks_to_consume, component);
1270
1271 if (graph->sinks_to_consume->length == 0) {
1272 graph->has_sink = BT_FALSE;
1273 }
1274
1275 /*
1276 * This calls bt_object_try_spec_release() on the component, and
1277 * since its reference count is 0, its destructor is called. Its
1278 * destructor calls the user's finalization method (if set).
1279 */
1280 g_ptr_array_remove(graph->components, component);
1281 goto end;
1282
1283 error:
1284 ret = -1;
1285
1286 end:
1287 graph->can_consume = init_can_consume;
1288 return ret;
1289 }
1290
1291 BT_HIDDEN
1292 void bt_graph_add_notification(struct bt_graph *graph,
1293 struct bt_notification *notif)
1294 {
1295 BT_ASSERT(graph);
1296 BT_ASSERT(notif);
1297
1298 /*
1299 * It's okay not to take a reference because, when a
1300 * notification's reference count drops to 0, either:
1301 *
1302 * * It is recycled back to one of this graph's pool.
1303 * * It is destroyed because it doesn't have any link to any
1304 * graph, which means the original graph is already destroyed.
1305 */
1306 g_ptr_array_add(graph->notifications, notif);
1307 }
This page took 0.087851 seconds and 4 git commands to generate.