projects / babeltrace.git / blob
? search:
summary | shortlog | log | commit | commitdiff | tree
blame | history | raw | HEAD
lib: strictly type function return status enumerations
[babeltrace.git] / src / lib / graph / graph.c
1 /*
2 * Copyright 2017-2018 Philippe Proulx <pproulx@efficios.com>
3 * Copyright 2017 Jérémie Galarneau <jeremie.galarneau@efficios.com>
4 *
5 * Permission is hereby granted, free of charge, to any person obtaining a copy
6 * of this software and associated documentation files (the "Software"), to deal
7 * in the Software without restriction, including without limitation the rights
8 * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
9 * copies of the Software, and to permit persons to whom the Software is
10 * furnished to do so, subject to the following conditions:
11 *
12 * The above copyright notice and this permission notice shall be included in
13 * all copies or substantial portions of the Software.
14 *
15 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
16 * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
17 * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
18 * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
19 * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
20 * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
21 * SOFTWARE.
22 */
23
24 #define BT_LOG_TAG "LIB/GRAPH"
25 #include "lib/logging.h"
26
27 #include "common/assert.h"
28 #include "lib/assert-pre.h"
29 #include "lib/assert-post.h"
30 #include <babeltrace2/graph/graph.h>
31 #include <babeltrace2/graph/graph-const.h>
32 #include <babeltrace2/graph/component-source-const.h>
33 #include <babeltrace2/graph/component-filter-const.h>
34 #include <babeltrace2/graph/port-const.h>
35 #include "lib/graph/message/message.h"
36 #include "compat/compiler.h"
37 #include "common/common.h"
38 #include <babeltrace2/types.h>
39 #include <babeltrace2/value.h>
40 #include <babeltrace2/value-const.h>
41 #include "lib/value.h"
42 #include <unistd.h>
43 #include <glib.h>
44
45 #include "component.h"
46 #include "component-sink.h"
47 #include "connection.h"
48 #include "graph.h"
49 #include "message/event.h"
50 #include "message/packet.h"
51
52 typedef enum bt_graph_listener_func_status
53 (*port_added_func_t)(const void *, const void *, void *);
54
55 typedef enum bt_graph_listener_func_status
56 (*ports_connected_func_t)(const void *, const void *, const void *,
57 const void *, void *);
58
59 typedef enum bt_component_class_init_method_status
60 (*comp_init_method_t)(const void *, const void *, void *);
61
62 struct bt_graph_listener {
63 bt_graph_listener_removed_func removed;
64 void *data;
65 };
66
67 struct bt_graph_listener_port_added {
68 struct bt_graph_listener base;
69 port_added_func_t func;
70 };
71
72 struct bt_graph_listener_ports_connected {
73 struct bt_graph_listener base;
74 ports_connected_func_t func;
75 };
76
77 #define INIT_LISTENERS_ARRAY(_type, _listeners) \
78 do { \
79 _listeners = g_array_new(FALSE, TRUE, sizeof(_type)); \
80 if (!(_listeners)) { \
81 BT_LOGE_STR("Failed to allocate one GArray."); \
82 } \
83 } while (0)
84
85 #define CALL_REMOVE_LISTENERS(_type, _listeners) \
86 do { \
87 size_t i; \
88 \
89 if (!_listeners) { \
90 break; \
91 } \
92 for (i = 0; i < (_listeners)->len; i++) { \
93 _type *listener = \
94 &g_array_index((_listeners), _type, i); \
95 \
96 if (listener->base.removed) { \
97 listener->base.removed(listener->base.data); \
98 } \
99 } \
100 } while (0)
101
102 static
103 void destroy_graph(struct bt_object *obj)
104 {
105 struct bt_graph *graph = container_of(obj, struct bt_graph, base);
106
107 /*
108 * The graph's reference count is 0 if we're here. Increment
109 * it to avoid a double-destroy (possibly infinitely recursive)
110 * in this situation:
111 *
112 * 1. We put and destroy a connection.
113 * 2. This connection's destructor finalizes its active message
114 * iterators.
115 * 3. A message iterator's finalization function gets a new
116 * reference on its component (reference count goes from 0 to
117 * 1).
118 * 4. Since this component's reference count goes to 1, it takes
119 * a reference on its parent (this graph). This graph's
120 * reference count goes from 0 to 1.
121 * 5. The message iterator's finalization function puts its
122 * component reference (reference count goes from 1 to 0).
123 * 6. Since this component's reference count goes from 1 to 0,
124 * it puts its parent (this graph). This graph's reference
125 * count goes from 1 to 0.
126 * 7. Since this graph's reference count goes from 1 to 0, its
127 * destructor is called (this function).
128 *
129 * With the incrementation below, the graph's reference count at
130 * step 4 goes from 1 to 2, and from 2 to 1 at step 6. This
131 * ensures that this function is not called two times.
132 */
133 BT_LIB_LOGI("Destroying graph: %!+g", graph);
134 obj->ref_count++;
135
136 /*
137 * Cancel the graph to disallow some operations, like creating
138 * message iterators and adding ports to components.
139 */
140 (void) bt_graph_cancel((void *) graph);
141
142 /* Call all remove listeners */
143 CALL_REMOVE_LISTENERS(struct bt_graph_listener_port_added,
144 graph->listeners.source_output_port_added);
145 CALL_REMOVE_LISTENERS(struct bt_graph_listener_port_added,
146 graph->listeners.filter_output_port_added);
147 CALL_REMOVE_LISTENERS(struct bt_graph_listener_port_added,
148 graph->listeners.filter_input_port_added);
149 CALL_REMOVE_LISTENERS(struct bt_graph_listener_port_added,
150 graph->listeners.sink_input_port_added);
151 CALL_REMOVE_LISTENERS(struct bt_graph_listener_ports_connected,
152 graph->listeners.source_filter_ports_connected);
153 CALL_REMOVE_LISTENERS(struct bt_graph_listener_ports_connected,
154 graph->listeners.filter_filter_ports_connected);
155 CALL_REMOVE_LISTENERS(struct bt_graph_listener_ports_connected,
156 graph->listeners.source_sink_ports_connected);
157 CALL_REMOVE_LISTENERS(struct bt_graph_listener_ports_connected,
158 graph->listeners.filter_sink_ports_connected);
159
160 if (graph->messages) {
161 g_ptr_array_free(graph->messages, TRUE);
162 graph->messages = NULL;
163 }
164
165 if (graph->connections) {
166 BT_LOGD_STR("Destroying connections.");
167 g_ptr_array_free(graph->connections, TRUE);
168 graph->connections = NULL;
169 }
170
171 if (graph->components) {
172 BT_LOGD_STR("Destroying components.");
173 g_ptr_array_free(graph->components, TRUE);
174 graph->components = NULL;
175 }
176
177 if (graph->sinks_to_consume) {
178 g_queue_free(graph->sinks_to_consume);
179 graph->sinks_to_consume = NULL;
180 }
181
182 if (graph->listeners.source_output_port_added) {
183 g_array_free(graph->listeners.source_output_port_added, TRUE);
184 graph->listeners.source_output_port_added = NULL;
185 }
186
187 if (graph->listeners.filter_output_port_added) {
188 g_array_free(graph->listeners.filter_output_port_added, TRUE);
189 graph->listeners.filter_output_port_added = NULL;
190 }
191
192 if (graph->listeners.filter_input_port_added) {
193 g_array_free(graph->listeners.filter_input_port_added, TRUE);
194 graph->listeners.filter_input_port_added = NULL;
195 }
196
197 if (graph->listeners.sink_input_port_added) {
198 g_array_free(graph->listeners.sink_input_port_added, TRUE);
199 graph->listeners.sink_input_port_added = NULL;
200 }
201
202 if (graph->listeners.source_filter_ports_connected) {
203 g_array_free(graph->listeners.source_filter_ports_connected,
204 TRUE);
205 graph->listeners.source_filter_ports_connected = NULL;
206 }
207
208 if (graph->listeners.filter_filter_ports_connected) {
209 g_array_free(graph->listeners.filter_filter_ports_connected,
210 TRUE);
211 graph->listeners.filter_filter_ports_connected = NULL;
212 }
213
214 if (graph->listeners.source_sink_ports_connected) {
215 g_array_free(graph->listeners.source_sink_ports_connected,
216 TRUE);
217 graph->listeners.source_sink_ports_connected = NULL;
218 }
219
220 if (graph->listeners.filter_sink_ports_connected) {
221 g_array_free(graph->listeners.filter_sink_ports_connected,
222 TRUE);
223 graph->listeners.filter_sink_ports_connected = NULL;
224 }
225
226 bt_object_pool_finalize(&graph->event_msg_pool);
227 bt_object_pool_finalize(&graph->packet_begin_msg_pool);
228 bt_object_pool_finalize(&graph->packet_end_msg_pool);
229 g_free(graph);
230 }
231
232 static
233 void destroy_message_event(struct bt_message *msg,
234 struct bt_graph *graph)
235 {
236 bt_message_event_destroy(msg);
237 }
238
239 static
240 void destroy_message_packet_begin(struct bt_message *msg,
241 struct bt_graph *graph)
242 {
243 bt_message_packet_destroy(msg);
244 }
245
246 static
247 void destroy_message_packet_end(struct bt_message *msg,
248 struct bt_graph *graph)
249 {
250 bt_message_packet_destroy(msg);
251 }
252
253 static
254 void notify_message_graph_is_destroyed(struct bt_message *msg)
255 {
256 bt_message_unlink_graph(msg);
257 }
258
259 struct bt_graph *bt_graph_create(void)
260 {
261 struct bt_graph *graph;
262 int ret;
263
264 BT_LOGI_STR("Creating graph object.");
265 graph = g_new0(struct bt_graph, 1);
266 if (!graph) {
267 BT_LOGE_STR("Failed to allocate one graph.");
268 goto end;
269 }
270
271 bt_object_init_shared(&graph->base, destroy_graph);
272 graph->connections = g_ptr_array_new_with_free_func(
273 (GDestroyNotify) bt_object_try_spec_release);
274 if (!graph->connections) {
275 BT_LOGE_STR("Failed to allocate one GPtrArray.");
276 goto error;
277 }
278 graph->components = g_ptr_array_new_with_free_func(
279 (GDestroyNotify) bt_object_try_spec_release);
280 if (!graph->components) {
281 BT_LOGE_STR("Failed to allocate one GPtrArray.");
282 goto error;
283 }
284 graph->sinks_to_consume = g_queue_new();
285 if (!graph->sinks_to_consume) {
286 BT_LOGE_STR("Failed to allocate one GQueue.");
287 goto error;
288 }
289
290 bt_graph_set_can_consume(graph, true);
291 INIT_LISTENERS_ARRAY(struct bt_graph_listener_port_added,
292 graph->listeners.source_output_port_added);
293
294 if (!graph->listeners.source_output_port_added) {
295 ret = -1;
296 goto error;
297 }
298
299 INIT_LISTENERS_ARRAY(struct bt_graph_listener_port_added,
300 graph->listeners.filter_output_port_added);
301
302 if (!graph->listeners.filter_output_port_added) {
303 ret = -1;
304 goto error;
305 }
306
307 INIT_LISTENERS_ARRAY(struct bt_graph_listener_port_added,
308 graph->listeners.filter_input_port_added);
309
310 if (!graph->listeners.filter_input_port_added) {
311 ret = -1;
312 goto error;
313 }
314
315 INIT_LISTENERS_ARRAY(struct bt_graph_listener_port_added,
316 graph->listeners.sink_input_port_added);
317
318 if (!graph->listeners.sink_input_port_added) {
319 ret = -1;
320 goto error;
321 }
322
323 INIT_LISTENERS_ARRAY(struct bt_graph_listener_ports_connected,
324 graph->listeners.source_filter_ports_connected);
325
326 if (!graph->listeners.source_filter_ports_connected) {
327 ret = -1;
328 goto error;
329 }
330
331 INIT_LISTENERS_ARRAY(struct bt_graph_listener_ports_connected,
332 graph->listeners.source_sink_ports_connected);
333
334 if (!graph->listeners.source_sink_ports_connected) {
335 ret = -1;
336 goto error;
337 }
338
339 INIT_LISTENERS_ARRAY(struct bt_graph_listener_ports_connected,
340 graph->listeners.filter_filter_ports_connected);
341
342 if (!graph->listeners.filter_filter_ports_connected) {
343 ret = -1;
344 goto error;
345 }
346
347 INIT_LISTENERS_ARRAY(struct bt_graph_listener_ports_connected,
348 graph->listeners.filter_sink_ports_connected);
349
350 if (!graph->listeners.filter_sink_ports_connected) {
351 ret = -1;
352 goto error;
353 }
354
355 ret = bt_object_pool_initialize(&graph->event_msg_pool,
356 (bt_object_pool_new_object_func) bt_message_event_new,
357 (bt_object_pool_destroy_object_func) destroy_message_event,
358 graph);
359 if (ret) {
360 BT_LOGE("Failed to initialize event message pool: ret=%d",
361 ret);
362 goto error;
363 }
364
365 ret = bt_object_pool_initialize(&graph->packet_begin_msg_pool,
366 (bt_object_pool_new_object_func) bt_message_packet_beginning_new,
367 (bt_object_pool_destroy_object_func) destroy_message_packet_begin,
368 graph);
369 if (ret) {
370 BT_LOGE("Failed to initialize packet beginning message pool: ret=%d",
371 ret);
372 goto error;
373 }
374
375 ret = bt_object_pool_initialize(&graph->packet_end_msg_pool,
376 (bt_object_pool_new_object_func) bt_message_packet_end_new,
377 (bt_object_pool_destroy_object_func) destroy_message_packet_end,
378 graph);
379 if (ret) {
380 BT_LOGE("Failed to initialize packet end message pool: ret=%d",
381 ret);
382 goto error;
383 }
384
385 graph->messages = g_ptr_array_new_with_free_func(
386 (GDestroyNotify) notify_message_graph_is_destroyed);
387 BT_LIB_LOGI("Created graph object: %!+g", graph);
388
389 end:
390 return (void *) graph;
391
392 error:
393 BT_OBJECT_PUT_REF_AND_RESET(graph);
394 goto end;
395 }
396
397 enum bt_graph_connect_ports_status bt_graph_connect_ports(
398 struct bt_graph *graph,
399 const struct bt_port_output *upstream_port_out,
400 const struct bt_port_input *downstream_port_in,
401 const struct bt_connection **user_connection)
402 {
403 enum bt_graph_connect_ports_status status = BT_FUNC_STATUS_OK;
404 enum bt_graph_listener_func_status listener_status;
405 struct bt_connection *connection = NULL;
406 struct bt_port *upstream_port = (void *) upstream_port_out;
407 struct bt_port *downstream_port = (void *) downstream_port_in;
408 struct bt_component *upstream_component = NULL;
409 struct bt_component *downstream_component = NULL;
410 enum bt_component_class_port_connected_method_status port_connected_status;
411 bool init_can_consume;
412
413 BT_ASSERT_PRE_NON_NULL(graph, "Graph");
414 BT_ASSERT_PRE_NON_NULL(upstream_port, "Upstream port");
415 BT_ASSERT_PRE_NON_NULL(downstream_port, "Downstream port port");
416 BT_ASSERT_PRE(!graph->canceled, "Graph is canceled: %!+g", graph);
417 BT_ASSERT_PRE(
418 graph->config_state == BT_GRAPH_CONFIGURATION_STATE_CONFIGURING,
419 "Graph is not in the \"configuring\" state: %!+g", graph);
420 BT_ASSERT_PRE(!bt_port_is_connected(upstream_port),
421 "Upstream port is already connected: %!+p", upstream_port);
422 BT_ASSERT_PRE(!bt_port_is_connected(downstream_port),
423 "Downstream port is already connected: %!+p", downstream_port);
424 BT_ASSERT_PRE(bt_port_borrow_component_inline((void *) upstream_port),
425 "Upstream port does not belong to a component: %!+p",
426 upstream_port);
427 BT_ASSERT_PRE(bt_port_borrow_component_inline((void *) downstream_port),
428 "Downstream port does not belong to a component: %!+p",
429 downstream_port);
430 init_can_consume = graph->can_consume;
431 BT_LIB_LOGI("Connecting component ports within graph: "
432 "%![graph-]+g, %![up-port-]+p, %![down-port-]+p",
433 graph, upstream_port, downstream_port);
434 bt_graph_set_can_consume(graph, false);
435 upstream_component = bt_port_borrow_component_inline(
436 (void *) upstream_port);
437 downstream_component = bt_port_borrow_component_inline(
438 (void *) downstream_port);
439
440 BT_LOGD_STR("Creating connection.");
441 connection = bt_connection_create(graph, (void *) upstream_port,
442 (void *) downstream_port);
443 if (!connection) {
444 BT_LOGW("Cannot create connection object.");
445 status = BT_FUNC_STATUS_MEMORY_ERROR;
446 goto end;
447 }
448
449 BT_LIB_LOGD("Connection object created: %!+x", connection);
450
451 /*
452 * Ownership of upstream_component/downstream_component and of
453 * the connection object is transferred to the graph.
454 */
455 g_ptr_array_add(graph->connections, connection);
456
457 /*
458 * Notify both components that their port is connected.
459 */
460 BT_LIB_LOGD("Notifying upstream component that its port is connected: "
461 "%![comp-]+c, %![port-]+p", upstream_component, upstream_port);
462 port_connected_status = bt_component_port_connected(upstream_component,
463 (void *) upstream_port, (void *) downstream_port);
464 if (port_connected_status != BT_FUNC_STATUS_OK) {
465 BT_LIB_LOGW("Error while notifying upstream component that its port is connected: "
466 "status=%s, %![graph-]+g, %![up-comp-]+c, "
467 "%![down-comp-]+c, %![up-port-]+p, %![down-port-]+p",
468 bt_common_func_status_string(port_connected_status),
469 graph, upstream_component, downstream_component,
470 upstream_port, downstream_port);
471 bt_connection_end(connection, true);
472 status = (int) port_connected_status;
473 goto end;
474 }
475
476 connection->notified_upstream_port_connected = true;
477 BT_LIB_LOGD("Notifying downstream component that its port is connected: "
478 "%![comp-]+c, %![port-]+p", downstream_component,
479 downstream_port);
480 port_connected_status = bt_component_port_connected(downstream_component,
481 (void *) downstream_port, (void *) upstream_port);
482 if (port_connected_status != BT_FUNC_STATUS_OK) {
483 BT_LIB_LOGW("Error while notifying downstream component that its port is connected: "
484 "status=%s, %![graph-]+g, %![up-comp-]+c, "
485 "%![down-comp-]+c, %![up-port-]+p, %![down-port-]+p",
486 bt_common_func_status_string(port_connected_status),
487 graph, upstream_component, downstream_component,
488 upstream_port, downstream_port);
489 bt_connection_end(connection, true);
490 status = (int) port_connected_status;
491 goto end;
492 }
493
494 connection->notified_downstream_port_connected = true;
495
496 /*
497 * Notify the graph's creator that both ports are connected.
498 */
499 BT_LOGD_STR("Notifying graph's user that new component ports are connected.");
500 listener_status = bt_graph_notify_ports_connected(graph, upstream_port, downstream_port);
501 if (listener_status != BT_FUNC_STATUS_OK) {
502 status = (int) listener_status;
503 goto end;
504 }
505
506 connection->notified_graph_ports_connected = true;
507 BT_LIB_LOGI("Connected component ports within graph: "
508 "%![graph-]+g, %![up-comp-]+c, %![down-comp-]+c, "
509 "%![up-port-]+p, %![down-port-]+p",
510 graph, upstream_component, downstream_component,
511 upstream_port, downstream_port);
512
513 if (user_connection) {
514 /* Move reference to user */
515 *user_connection = connection;
516 connection = NULL;
517 }
518
519 end:
520 if (status != BT_FUNC_STATUS_OK) {
521 bt_graph_make_faulty(graph);
522 }
523
524 bt_object_put_ref(connection);
525 (void) init_can_consume;
526 bt_graph_set_can_consume(graph, init_can_consume);
527 return status;
528 }
529
530 static inline
531 int consume_graph_sink(struct bt_component_sink *comp)
532 {
533 enum bt_component_class_sink_consume_method_status consume_status;
534 struct bt_component_class_sink *sink_class = NULL;
535
536 BT_ASSERT(comp);
537 sink_class = (void *) comp->parent.class;
538 BT_ASSERT(sink_class->methods.consume);
539 BT_LIB_LOGD("Calling user's consume method: %!+c", comp);
540 consume_status = sink_class->methods.consume((void *) comp);
541 BT_LOGD("User method returned: status=%s",
542 bt_common_func_status_string(consume_status));
543 BT_ASSERT_POST(consume_status == BT_FUNC_STATUS_OK ||
544 consume_status == BT_FUNC_STATUS_END ||
545 consume_status == BT_FUNC_STATUS_AGAIN ||
546 consume_status == BT_FUNC_STATUS_ERROR ||
547 consume_status == BT_FUNC_STATUS_MEMORY_ERROR,
548 "Invalid component status returned by consuming method: "
549 "status=%s", bt_common_func_status_string(consume_status));
550 if (consume_status < 0) {
551 BT_LOGW_STR("Consume method failed.");
552 goto end;
553 }
554
555 BT_LIB_LOGD("Consumed from sink: %![comp-]+c, status=%s",
556 comp, bt_common_func_status_string(consume_status));
557
558 end:
559 return consume_status;
560 }
561
562 /*
563 * `node` is removed from the queue of sinks to consume when passed to
564 * this function. This function adds it back to the queue if there's
565 * still something to consume afterwards.
566 */
567 static inline
568 int consume_sink_node(struct bt_graph *graph, GList *node)
569 {
570 int status;
571 struct bt_component_sink *sink;
572
573 sink = node->data;
574 status = consume_graph_sink(sink);
575 if (G_UNLIKELY(status != BT_FUNC_STATUS_END)) {
576 g_queue_push_tail_link(graph->sinks_to_consume, node);
577 goto end;
578 }
579
580 /* End reached, the node is not added back to the queue and free'd. */
581 g_queue_delete_link(graph->sinks_to_consume, node);
582
583 /* Don't forward an END status if there are sinks left to consume. */
584 if (!g_queue_is_empty(graph->sinks_to_consume)) {
585 status = BT_FUNC_STATUS_OK;
586 goto end;
587 }
588
589 end:
590 BT_LIB_LOGD("Consumed sink node: %![comp-]+c, status=%s",
591 sink, bt_common_func_status_string(status));
592 return status;
593 }
594
595 BT_HIDDEN
596 int bt_graph_consume_sink_no_check(struct bt_graph *graph,
597 struct bt_component_sink *sink)
598 {
599 int status;
600 GList *sink_node;
601 int index;
602
603 BT_LIB_LOGD("Making specific sink consume: %![comp-]+c", sink);
604 BT_ASSERT(bt_component_borrow_graph((void *) sink) == graph);
605
606 if (g_queue_is_empty(graph->sinks_to_consume)) {
607 BT_LOGD_STR("Graph's sink queue is empty: end of graph.");
608 status = BT_FUNC_STATUS_END;
609 goto end;
610 }
611
612 index = g_queue_index(graph->sinks_to_consume, sink);
613 if (index < 0) {
614 BT_LIB_LOGD("Sink component is not marked as consumable: "
615 "component sink is ended: %![comp-]+c", sink);
616 status = BT_FUNC_STATUS_END;
617 goto end;
618 }
619
620 sink_node = g_queue_pop_nth_link(graph->sinks_to_consume, index);
621 BT_ASSERT(sink_node);
622 status = consume_sink_node(graph, sink_node);
623
624 end:
625 return status;
626 }
627
628 static inline
629 int consume_no_check(struct bt_graph *graph)
630 {
631 int status = BT_FUNC_STATUS_OK;
632 struct bt_component *sink;
633 GList *current_node;
634
635 BT_ASSERT_PRE(graph->has_sink,
636 "Graph has no sink component: %!+g", graph);
637 BT_LIB_LOGD("Making next sink component consume: %![graph-]+g", graph);
638
639 if (G_UNLIKELY(g_queue_is_empty(graph->sinks_to_consume))) {
640 BT_LOGD_STR("Graph's sink queue is empty: end of graph.");
641 status = BT_FUNC_STATUS_END;
642 goto end;
643 }
644
645 current_node = g_queue_pop_head_link(graph->sinks_to_consume);
646 sink = current_node->data;
647 BT_LIB_LOGD("Chose next sink to consume: %!+c", sink);
648 status = consume_sink_node(graph, current_node);
649
650 end:
651 return status;
652 }
653
654 enum bt_graph_consume_status bt_graph_consume(struct bt_graph *graph)
655 {
656 enum bt_graph_consume_status status;
657
658 BT_ASSERT_PRE_NON_NULL(graph, "Graph");
659 BT_ASSERT_PRE(!graph->canceled, "Graph is canceled: %!+g", graph);
660 BT_ASSERT_PRE(graph->can_consume,
661 "Cannot consume graph in its current state: %!+g", graph);
662 BT_ASSERT_PRE(graph->config_state != BT_GRAPH_CONFIGURATION_STATE_FAULTY,
663 "Graph is in a faulty state: %!+g", graph);
664 bt_graph_set_can_consume(graph, false);
665 status = bt_graph_configure(graph);
666 if (G_UNLIKELY(status)) {
667 /* bt_graph_configure() logs errors */
668 goto end;
669 }
670
671 status = consume_no_check(graph);
672 bt_graph_set_can_consume(graph, true);
673
674 end:
675 return status;
676 }
677
678 enum bt_graph_run_status bt_graph_run(struct bt_graph *graph)
679 {
680 enum bt_graph_run_status status;
681
682 BT_ASSERT_PRE_NON_NULL(graph, "Graph");
683 BT_ASSERT_PRE(!graph->canceled, "Graph is canceled: %!+g", graph);
684 BT_ASSERT_PRE(graph->can_consume,
685 "Cannot consume graph in its current state: %!+g", graph);
686 BT_ASSERT_PRE(graph->config_state != BT_GRAPH_CONFIGURATION_STATE_FAULTY,
687 "Graph is in a faulty state: %!+g", graph);
688 bt_graph_set_can_consume(graph, false);
689 status = bt_graph_configure(graph);
690 if (G_UNLIKELY(status)) {
691 /* bt_graph_configure() logs errors */
692 goto end;
693 }
694
695 BT_LIB_LOGI("Running graph: %!+g", graph);
696
697 do {
698 /*
699 * Check if the graph is canceled at each iteration. If
700 * the graph was canceled by another thread or by a
701 * signal handler, this is not a warning nor an error,
702 * it was intentional: log with a DEBUG level only.
703 */
704 if (G_UNLIKELY(graph->canceled)) {
705 BT_LIB_LOGI("Stopping the graph: graph is canceled: "
706 "%!+g", graph);
707 status = BT_FUNC_STATUS_CANCELED;
708 goto end;
709 }
710
711 status = consume_no_check(graph);
712 if (G_UNLIKELY(status == BT_FUNC_STATUS_AGAIN)) {
713 /*
714 * If AGAIN is received and there are multiple
715 * sinks, go ahead and consume from the next
716 * sink.
717 *
718 * However, in the case where a single sink is
719 * left, the caller can decide to busy-wait and
720 * call bt_graph_run() continuously
721 * until the source is ready or it can decide to
722 * sleep for an arbitrary amount of time.
723 */
724 if (graph->sinks_to_consume->length > 1) {
725 status = BT_FUNC_STATUS_OK;
726 }
727 }
728 } while (status == BT_FUNC_STATUS_OK);
729
730 if (g_queue_is_empty(graph->sinks_to_consume)) {
731 status = BT_FUNC_STATUS_END;
732 }
733
734 end:
735 BT_LIB_LOGI("Graph ran: %![graph-]+g, status=%s", graph,
736 bt_common_func_status_string(status));
737 bt_graph_set_can_consume(graph, true);
738 return status;
739 }
740
741 enum bt_graph_add_listener_status
742 bt_graph_add_source_component_output_port_added_listener(
743 struct bt_graph *graph,
744 bt_graph_source_component_output_port_added_listener_func func,
745 bt_graph_listener_removed_func listener_removed, void *data,
746 int *out_listener_id)
747 {
748 struct bt_graph_listener_port_added listener = {
749 .base = {
750 .removed = listener_removed,
751 .data = data,
752 },
753 .func = (port_added_func_t) func,
754 };
755 int listener_id;
756
757 BT_ASSERT_PRE_NON_NULL(graph, "Graph");
758 BT_ASSERT_PRE_NON_NULL(func, "Listener");
759 BT_ASSERT_PRE_NON_NULL(func, "\"Listener removed\" listener");
760 BT_ASSERT_PRE(!graph->in_remove_listener,
761 "Graph currently executing a \"listener removed\" listener: "
762 "%!+g", graph);
763 g_array_append_val(graph->listeners.source_output_port_added, listener);
764 listener_id = graph->listeners.source_output_port_added->len - 1;
765 BT_LIB_LOGD("Added \"source component output port added\" listener to graph: "
766 "%![graph-]+g, listener-addr=%p, id=%d", graph, listener,
767 listener_id);
768
769 if (listener_id) {
770 *out_listener_id = listener_id;
771 }
772
773 return BT_FUNC_STATUS_OK;
774 }
775
776 enum bt_graph_add_listener_status
777 bt_graph_add_filter_component_output_port_added_listener(
778 struct bt_graph *graph,
779 bt_graph_filter_component_output_port_added_listener_func func,
780 bt_graph_listener_removed_func listener_removed, void *data,
781 int *out_listener_id)
782 {
783 struct bt_graph_listener_port_added listener = {
784 .base = {
785 .removed = listener_removed,
786 .data = data,
787 },
788 .func = (port_added_func_t) func,
789 };
790 int listener_id;
791
792 BT_ASSERT_PRE_NON_NULL(graph, "Graph");
793 BT_ASSERT_PRE_NON_NULL(func, "Listener");
794 BT_ASSERT_PRE_NON_NULL(func, "\"Listener removed\" listener");
795 BT_ASSERT_PRE(!graph->in_remove_listener,
796 "Graph currently executing a \"listener removed\" listener: "
797 "%!+g", graph);
798 g_array_append_val(graph->listeners.filter_output_port_added, listener);
799 listener_id = graph->listeners.filter_output_port_added->len - 1;
800 BT_LIB_LOGD("Added \"filter component output port added\" listener to graph: "
801 "%![graph-]+g, listener-addr=%p, id=%d", graph, listener,
802 listener_id);
803
804 if (listener_id) {
805 *out_listener_id = listener_id;
806 }
807
808 return BT_FUNC_STATUS_OK;
809 }
810
811 enum bt_graph_add_listener_status
812 bt_graph_add_filter_component_input_port_added_listener(
813 struct bt_graph *graph,
814 bt_graph_filter_component_input_port_added_listener_func func,
815 bt_graph_listener_removed_func listener_removed, void *data,
816 int *out_listener_id)
817 {
818 struct bt_graph_listener_port_added listener = {
819 .base = {
820 .removed = listener_removed,
821 .data = data,
822 },
823 .func = (port_added_func_t) func,
824 };
825 int listener_id;
826
827 BT_ASSERT_PRE_NON_NULL(graph, "Graph");
828 BT_ASSERT_PRE_NON_NULL(func, "Listener");
829 BT_ASSERT_PRE_NON_NULL(func, "\"Listener removed\" listener");
830 BT_ASSERT_PRE(!graph->in_remove_listener,
831 "Graph currently executing a \"listener removed\" listener: "
832 "%!+g", graph);
833 g_array_append_val(graph->listeners.filter_input_port_added, listener);
834 listener_id = graph->listeners.filter_input_port_added->len - 1;
835 BT_LIB_LOGD("Added \"filter component input port added\" listener to graph: "
836 "%![graph-]+g, listener-addr=%p, id=%d", graph, listener,
837 listener_id);
838
839 if (listener_id) {
840 *out_listener_id = listener_id;
841 }
842
843 return BT_FUNC_STATUS_OK;
844 }
845
846 enum bt_graph_add_listener_status
847 bt_graph_add_sink_component_input_port_added_listener(
848 struct bt_graph *graph,
849 bt_graph_sink_component_input_port_added_listener_func func,
850 bt_graph_listener_removed_func listener_removed, void *data,
851 int *out_listener_id)
852 {
853 struct bt_graph_listener_port_added listener = {
854 .base = {
855 .removed = listener_removed,
856 .data = data,
857 },
858 .func = (port_added_func_t) func,
859 };
860 int listener_id;
861
862 BT_ASSERT_PRE_NON_NULL(graph, "Graph");
863 BT_ASSERT_PRE_NON_NULL(func, "Listener");
864 BT_ASSERT_PRE_NON_NULL(func, "\"Listener removed\" listener");
865 BT_ASSERT_PRE(!graph->in_remove_listener,
866 "Graph currently executing a \"listener removed\" listener: "
867 "%!+g", graph);
868 g_array_append_val(graph->listeners.sink_input_port_added, listener);
869 listener_id = graph->listeners.sink_input_port_added->len - 1;
870 BT_LIB_LOGD("Added \"sink component input port added\" listener to graph: "
871 "%![graph-]+g, listener-addr=%p, id=%d", graph, listener,
872 listener_id);
873
874 if (listener_id) {
875 *out_listener_id = listener_id;
876 }
877
878 return BT_FUNC_STATUS_OK;
879 }
880
881 enum bt_graph_add_listener_status
882 bt_graph_add_source_filter_component_ports_connected_listener(
883 struct bt_graph *graph,
884 bt_graph_source_filter_component_ports_connected_listener_func func,
885 bt_graph_listener_removed_func listener_removed, void *data,
886 int *out_listener_id)
887 {
888 struct bt_graph_listener_ports_connected listener = {
889 .base = {
890 .removed = listener_removed,
891 .data = data,
892 },
893 .func = (ports_connected_func_t) func,
894 };
895 int listener_id;
896
897 BT_ASSERT_PRE_NON_NULL(graph, "Graph");
898 BT_ASSERT_PRE_NON_NULL(func, "Listener");
899 BT_ASSERT_PRE_NON_NULL(func, "\"Listener removed\" listener");
900 BT_ASSERT_PRE(!graph->in_remove_listener,
901 "Graph currently executing a \"listener removed\" listener: "
902 "%!+g", graph);
903 g_array_append_val(graph->listeners.source_filter_ports_connected,
904 listener);
905 listener_id = graph->listeners.source_filter_ports_connected->len - 1;
906 BT_LIB_LOGD("Added \"source to filter component ports connected\" listener to graph: "
907 "%![graph-]+g, listener-addr=%p, id=%d", graph, listener,
908 listener_id);
909
910 if (listener_id) {
911 *out_listener_id = listener_id;
912 }
913
914 return BT_FUNC_STATUS_OK;
915 }
916
917 enum bt_graph_add_listener_status
918 bt_graph_add_source_sink_component_ports_connected_listener(
919 struct bt_graph *graph,
920 bt_graph_source_sink_component_ports_connected_listener_func func,
921 bt_graph_listener_removed_func listener_removed, void *data,
922 int *out_listener_id)
923 {
924 struct bt_graph_listener_ports_connected listener = {
925 .base = {
926 .removed = listener_removed,
927 .data = data,
928 },
929 .func = (ports_connected_func_t) func,
930 };
931 int listener_id;
932
933 BT_ASSERT_PRE_NON_NULL(graph, "Graph");
934 BT_ASSERT_PRE_NON_NULL(func, "Listener");
935 BT_ASSERT_PRE_NON_NULL(func, "\"Listener removed\" listener");
936 BT_ASSERT_PRE(!graph->in_remove_listener,
937 "Graph currently executing a \"listener removed\" listener: "
938 "%!+g", graph);
939 g_array_append_val(graph->listeners.source_sink_ports_connected,
940 listener);
941 listener_id = graph->listeners.source_sink_ports_connected->len - 1;
942 BT_LIB_LOGD("Added \"source to sink component ports connected\" listener to graph: "
943 "%![graph-]+g, listener-addr=%p, id=%d", graph, listener,
944 listener_id);
945
946 if (listener_id) {
947 *out_listener_id = listener_id;
948 }
949
950 return BT_FUNC_STATUS_OK;
951 }
952
953 enum bt_graph_add_listener_status
954 bt_graph_add_filter_filter_component_ports_connected_listener(
955 struct bt_graph *graph,
956 bt_graph_filter_filter_component_ports_connected_listener_func func,
957 bt_graph_listener_removed_func listener_removed, void *data,
958 int *out_listener_id)
959 {
960 struct bt_graph_listener_ports_connected listener = {
961 .base = {
962 .removed = listener_removed,
963 .data = data,
964 },
965 .func = (ports_connected_func_t) func,
966 };
967 int listener_id;
968
969 BT_ASSERT_PRE_NON_NULL(graph, "Graph");
970 BT_ASSERT_PRE_NON_NULL(func, "Listener");
971 BT_ASSERT_PRE_NON_NULL(func, "\"Listener removed\" listener");
972 BT_ASSERT_PRE(!graph->in_remove_listener,
973 "Graph currently executing a \"listener removed\" listener: "
974 "%!+g", graph);
975 g_array_append_val(graph->listeners.filter_filter_ports_connected,
976 listener);
977 listener_id = graph->listeners.filter_filter_ports_connected->len - 1;
978 BT_LIB_LOGD("Added \"filter to filter component ports connected\" listener to graph: "
979 "%![graph-]+g, listener-addr=%p, id=%d", graph, listener,
980 listener_id);
981
982 if (listener_id) {
983 *out_listener_id = listener_id;
984 }
985
986 return BT_FUNC_STATUS_OK;
987 }
988
989 enum bt_graph_add_listener_status
990 bt_graph_add_filter_sink_component_ports_connected_listener(
991 struct bt_graph *graph,
992 bt_graph_filter_sink_component_ports_connected_listener_func func,
993 bt_graph_listener_removed_func listener_removed, void *data,
994 int *out_listener_id)
995 {
996 struct bt_graph_listener_ports_connected listener = {
997 .base = {
998 .removed = listener_removed,
999 .data = data,
1000 },
1001 .func = (ports_connected_func_t) func,
1002 };
1003 int listener_id;
1004
1005 BT_ASSERT_PRE_NON_NULL(graph, "Graph");
1006 BT_ASSERT_PRE_NON_NULL(func, "Listener");
1007 BT_ASSERT_PRE_NON_NULL(func, "\"Listener removed\" listener");
1008 BT_ASSERT_PRE(!graph->in_remove_listener,
1009 "Graph currently executing a \"listener removed\" listener: "
1010 "%!+g", graph);
1011 g_array_append_val(graph->listeners.filter_sink_ports_connected,
1012 listener);
1013 listener_id = graph->listeners.filter_sink_ports_connected->len - 1;
1014 BT_LIB_LOGD("Added \"filter to sink component ports connected\" listener to graph: "
1015 "%![graph-]+g, listener-addr=%p, id=%d", graph, listener,
1016 listener_id);
1017
1018 if (listener_id) {
1019 *out_listener_id = listener_id;
1020 }
1021
1022 return BT_FUNC_STATUS_OK;
1023 }
1024
1025 BT_HIDDEN
1026 enum bt_graph_listener_func_status bt_graph_notify_port_added(
1027 struct bt_graph *graph, struct bt_port *port)
1028 {
1029 uint64_t i;
1030 GArray *listeners;
1031 struct bt_component *comp;
1032 enum bt_graph_listener_func_status status = BT_FUNC_STATUS_OK;
1033
1034 BT_ASSERT(graph);
1035 BT_ASSERT(port);
1036 BT_LIB_LOGD("Notifying graph listeners that a port was added: "
1037 "%![graph-]+g, %![port-]+p", graph, port);
1038 comp = bt_port_borrow_component_inline(port);
1039 BT_ASSERT(comp);
1040
1041 switch (comp->class->type) {
1042 case BT_COMPONENT_CLASS_TYPE_SOURCE:
1043 {
1044 switch (port->type) {
1045 case BT_PORT_TYPE_OUTPUT:
1046 listeners = graph->listeners.source_output_port_added;
1047 break;
1048 default:
1049 abort();
1050 }
1051
1052 break;
1053 }
1054 case BT_COMPONENT_CLASS_TYPE_FILTER:
1055 {
1056 switch (port->type) {
1057 case BT_PORT_TYPE_INPUT:
1058 listeners = graph->listeners.filter_input_port_added;
1059 break;
1060 case BT_PORT_TYPE_OUTPUT:
1061 listeners = graph->listeners.filter_output_port_added;
1062 break;
1063 default:
1064 abort();
1065 }
1066
1067 break;
1068 }
1069 case BT_COMPONENT_CLASS_TYPE_SINK:
1070 {
1071 switch (port->type) {
1072 case BT_PORT_TYPE_INPUT:
1073 listeners = graph->listeners.sink_input_port_added;
1074 break;
1075 default:
1076 abort();
1077 }
1078
1079 break;
1080 }
1081 default:
1082 abort();
1083 }
1084
1085 for (i = 0; i < listeners->len; i++) {
1086 struct bt_graph_listener_port_added *listener =
1087 &g_array_index(listeners,
1088 struct bt_graph_listener_port_added, i);
1089
1090
1091 BT_ASSERT(listener->func);
1092 status = listener->func(comp, port, listener->base.data);
1093 if (status != BT_FUNC_STATUS_OK) {
1094 goto end;
1095 }
1096 }
1097
1098 end:
1099 return status;
1100 }
1101
1102 BT_HIDDEN
1103 enum bt_graph_listener_func_status bt_graph_notify_ports_connected(
1104 struct bt_graph *graph, struct bt_port *upstream_port,
1105 struct bt_port *downstream_port)
1106 {
1107 uint64_t i;
1108 GArray *listeners;
1109 struct bt_component *upstream_comp;
1110 struct bt_component *downstream_comp;
1111 enum bt_graph_listener_func_status status = BT_FUNC_STATUS_OK;
1112
1113 BT_ASSERT(graph);
1114 BT_ASSERT(upstream_port);
1115 BT_ASSERT(downstream_port);
1116 BT_LIB_LOGD("Notifying graph listeners that ports were connected: "
1117 "%![graph-]+g, %![up-port-]+p, %![down-port-]+p",
1118 graph, upstream_port, downstream_port);
1119 upstream_comp = bt_port_borrow_component_inline(upstream_port);
1120 BT_ASSERT(upstream_comp);
1121 downstream_comp = bt_port_borrow_component_inline(downstream_port);
1122 BT_ASSERT(downstream_comp);
1123
1124 switch (upstream_comp->class->type) {
1125 case BT_COMPONENT_CLASS_TYPE_SOURCE:
1126 {
1127 switch (downstream_comp->class->type) {
1128 case BT_COMPONENT_CLASS_TYPE_FILTER:
1129 listeners =
1130 graph->listeners.source_filter_ports_connected;
1131 break;
1132 case BT_COMPONENT_CLASS_TYPE_SINK:
1133 listeners =
1134 graph->listeners.source_sink_ports_connected;
1135 break;
1136 default:
1137 abort();
1138 }
1139
1140 break;
1141 }
1142 case BT_COMPONENT_CLASS_TYPE_FILTER:
1143 {
1144 switch (downstream_comp->class->type) {
1145 case BT_COMPONENT_CLASS_TYPE_FILTER:
1146 listeners =
1147 graph->listeners.filter_filter_ports_connected;
1148 break;
1149 case BT_COMPONENT_CLASS_TYPE_SINK:
1150 listeners =
1151 graph->listeners.filter_sink_ports_connected;
1152 break;
1153 default:
1154 abort();
1155 }
1156
1157 break;
1158 }
1159 default:
1160 abort();
1161 }
1162
1163 for (i = 0; i < listeners->len; i++) {
1164 struct bt_graph_listener_ports_connected *listener =
1165 &g_array_index(listeners,
1166 struct bt_graph_listener_ports_connected, i);
1167
1168 BT_ASSERT(listener->func);
1169 status = listener->func(upstream_comp, downstream_comp,
1170 upstream_port, downstream_port, listener->base.data);
1171 if (status != BT_FUNC_STATUS_OK) {
1172 goto end;
1173 }
1174 }
1175
1176 end:
1177 return status;
1178 }
1179
1180 enum bt_graph_cancel_status bt_graph_cancel(struct bt_graph *graph)
1181 {
1182 BT_ASSERT_PRE_NON_NULL(graph, "Graph");
1183 graph->canceled = true;
1184 BT_LIB_LOGI("Canceled graph: %!+i", graph);
1185 return BT_FUNC_STATUS_OK;
1186 }
1187
1188 bt_bool bt_graph_is_canceled(const struct bt_graph *graph)
1189 {
1190 BT_ASSERT_PRE_NON_NULL(graph, "Graph");
1191 return graph->canceled ? BT_TRUE : BT_FALSE;
1192 }
1193
1194 BT_HIDDEN
1195 void bt_graph_remove_connection(struct bt_graph *graph,
1196 struct bt_connection *connection)
1197 {
1198 BT_ASSERT(graph);
1199 BT_ASSERT(connection);
1200 BT_LIB_LOGD("Removing graph's connection: %![graph-]+g, %![conn-]+x",
1201 graph, connection);
1202 g_ptr_array_remove(graph->connections, connection);
1203 }
1204
1205 BT_ASSERT_PRE_FUNC
1206 static inline
1207 bool component_name_exists(struct bt_graph *graph, const char *name)
1208 {
1209 bool exists = false;
1210 uint64_t i;
1211
1212 for (i = 0; i < graph->components->len; i++) {
1213 struct bt_component *other_comp = graph->components->pdata[i];
1214
1215 if (strcmp(name, bt_component_get_name(other_comp)) == 0) {
1216 BT_ASSERT_PRE_MSG("Another component with the same name already exists in the graph: "
1217 "%![other-comp-]+c, name=\"%s\"",
1218 other_comp, name);
1219 exists = true;
1220 goto end;
1221 }
1222 }
1223
1224 end:
1225 return exists;
1226 }
1227
1228 static
1229 int add_component_with_init_method_data(
1230 struct bt_graph *graph,
1231 struct bt_component_class *comp_cls,
1232 comp_init_method_t init_method,
1233 const char *name, const struct bt_value *params,
1234 void *init_method_data, bt_logging_level log_level,
1235 struct bt_component **user_component)
1236 {
1237 int status = BT_FUNC_STATUS_OK;
1238 enum bt_component_class_init_method_status init_status;
1239 struct bt_component *component = NULL;
1240 int ret;
1241 bool init_can_consume;
1242 struct bt_value *new_params = NULL;
1243
1244 BT_ASSERT(comp_cls);
1245 BT_ASSERT_PRE_NON_NULL(graph, "Graph");
1246 BT_ASSERT_PRE_NON_NULL(name, "Name");
1247 BT_ASSERT_PRE(!graph->canceled, "Graph is canceled: %!+g", graph);
1248 BT_ASSERT_PRE(
1249 graph->config_state == BT_GRAPH_CONFIGURATION_STATE_CONFIGURING,
1250 "Graph is not in the \"configuring\" state: %!+g", graph);
1251 BT_ASSERT_PRE(!component_name_exists(graph, name),
1252 "Duplicate component name: %!+g, name=\"%s\"", graph, name);
1253 BT_ASSERT_PRE(!params || bt_value_is_map(params),
1254 "Parameter value is not a map value: %!+v", params);
1255 init_can_consume = graph->can_consume;
1256 bt_graph_set_can_consume(graph, false);
1257 BT_LIB_LOGI("Adding component to graph: "
1258 "%![graph-]+g, %![cc-]+C, name=\"%s\", log-level=%s, "
1259 "%![params-]+v, init-method-data-addr=%p",
1260 graph, comp_cls, name,
1261 bt_common_logging_level_string(log_level), params,
1262 init_method_data);
1263
1264 if (!params) {
1265 new_params = bt_value_map_create();
1266 if (!new_params) {
1267 BT_LOGE_STR("Cannot create empty map value object.");
1268 status = BT_FUNC_STATUS_MEMORY_ERROR;
1269 goto end;
1270 }
1271
1272 params = new_params;
1273 }
1274
1275 ret = bt_component_create(comp_cls, name, log_level, &component);
1276 if (ret) {
1277 BT_LOGE("Cannot create empty component object: ret=%d",
1278 ret);
1279 status = BT_FUNC_STATUS_MEMORY_ERROR;
1280 goto end;
1281 }
1282
1283 /*
1284 * The user's initialization method needs to see that this
1285 * component is part of the graph. If the user method fails, we
1286 * immediately remove the component from the graph's components.
1287 */
1288 g_ptr_array_add(graph->components, component);
1289 bt_component_set_graph(component, graph);
1290 bt_value_freeze(params);
1291
1292 if (init_method) {
1293 BT_LOGD_STR("Calling user's initialization method.");
1294 init_status = init_method(component, params, init_method_data);
1295 BT_LOGD("User method returned: status=%s",
1296 bt_common_func_status_string(init_status));
1297 if (init_status != BT_FUNC_STATUS_OK) {
1298 BT_LIB_LOGW("Component initialization method failed: "
1299 "%!+c", component);
1300 status = init_status;
1301 bt_component_set_graph(component, NULL);
1302 g_ptr_array_remove_fast(graph->components, component);
1303 goto end;
1304 }
1305 }
1306
1307 /*
1308 * Mark the component as initialized so that its finalization
1309 * method is called when it is destroyed.
1310 */
1311 component->initialized = true;
1312
1313 /*
1314 * If it's a sink component, it needs to be part of the graph's
1315 * sink queue to be consumed by bt_graph_consume().
1316 */
1317 if (bt_component_is_sink(component)) {
1318 graph->has_sink = true;
1319 g_queue_push_tail(graph->sinks_to_consume, component);
1320 }
1321
1322 /*
1323 * Freeze the component class now that it's instantiated at
1324 * least once.
1325 */
1326 BT_LOGD_STR("Freezing component class.");
1327 bt_component_class_freeze(comp_cls);
1328 BT_LIB_LOGI("Added component to graph: "
1329 "%![graph-]+g, %![cc-]+C, name=\"%s\", log-level=%s, "
1330 "%![params-]+v, init-method-data-addr=%p, %![comp-]+c",
1331 graph, comp_cls, name,
1332 bt_common_logging_level_string(log_level), params,
1333 init_method_data, component);
1334
1335 if (user_component) {
1336 /* Move reference to user */
1337 *user_component = component;
1338 component = NULL;
1339 }
1340
1341 end:
1342 if (status != BT_FUNC_STATUS_OK) {
1343 bt_graph_make_faulty(graph);
1344 }
1345
1346 bt_object_put_ref(component);
1347 bt_object_put_ref(new_params);
1348 (void) init_can_consume;
1349 bt_graph_set_can_consume(graph, init_can_consume);
1350 return status;
1351 }
1352
1353 enum bt_graph_add_component_status
1354 bt_graph_add_source_component_with_init_method_data(
1355 struct bt_graph *graph,
1356 const struct bt_component_class_source *comp_cls,
1357 const char *name, const struct bt_value *params,
1358 void *init_method_data, bt_logging_level log_level,
1359 const struct bt_component_source **component)
1360 {
1361 BT_ASSERT_PRE_NON_NULL(comp_cls, "Component class");
1362 return add_component_with_init_method_data(graph,
1363 (void *) comp_cls, (comp_init_method_t) comp_cls->methods.init,
1364 name, params, init_method_data, log_level, (void *) component);
1365 }
1366
1367 enum bt_graph_add_component_status bt_graph_add_source_component(
1368 struct bt_graph *graph,
1369 const struct bt_component_class_source *comp_cls,
1370 const char *name, const struct bt_value *params,
1371 bt_logging_level log_level,
1372 const struct bt_component_source **component)
1373 {
1374 return bt_graph_add_source_component_with_init_method_data(
1375 graph, comp_cls, name, params, NULL, log_level, component);
1376 }
1377
1378 enum bt_graph_add_component_status
1379 bt_graph_add_filter_component_with_init_method_data(
1380 struct bt_graph *graph,
1381 const struct bt_component_class_filter *comp_cls,
1382 const char *name, const struct bt_value *params,
1383 void *init_method_data, bt_logging_level log_level,
1384 const struct bt_component_filter **component)
1385 {
1386 BT_ASSERT_PRE_NON_NULL(comp_cls, "Component class");
1387 return add_component_with_init_method_data(graph,
1388 (void *) comp_cls, (comp_init_method_t) comp_cls->methods.init,
1389 name, params, init_method_data, log_level, (void *) component);
1390 }
1391
1392 enum bt_graph_add_component_status bt_graph_add_filter_component(
1393 struct bt_graph *graph,
1394 const struct bt_component_class_filter *comp_cls,
1395 const char *name, const struct bt_value *params,
1396 bt_logging_level log_level,
1397 const struct bt_component_filter **component)
1398 {
1399 return bt_graph_add_filter_component_with_init_method_data(
1400 graph, comp_cls, name, params, NULL, log_level, component);
1401 }
1402
1403 enum bt_graph_add_component_status
1404 bt_graph_add_sink_component_with_init_method_data(
1405 struct bt_graph *graph,
1406 const struct bt_component_class_sink *comp_cls,
1407 const char *name, const struct bt_value *params,
1408 void *init_method_data, bt_logging_level log_level,
1409 const struct bt_component_sink **component)
1410 {
1411 BT_ASSERT_PRE_NON_NULL(comp_cls, "Component class");
1412 return add_component_with_init_method_data(graph,
1413 (void *) comp_cls, (comp_init_method_t) comp_cls->methods.init,
1414 name, params, init_method_data, log_level, (void *) component);
1415 }
1416
1417 enum bt_graph_add_component_status bt_graph_add_sink_component(
1418 struct bt_graph *graph,
1419 const struct bt_component_class_sink *comp_cls,
1420 const char *name, const struct bt_value *params,
1421 bt_logging_level log_level,
1422 const struct bt_component_sink **component)
1423 {
1424 return bt_graph_add_sink_component_with_init_method_data(
1425 graph, comp_cls, name, params, NULL, log_level, component);
1426 }
1427
1428 BT_HIDDEN
1429 int bt_graph_remove_unconnected_component(struct bt_graph *graph,
1430 struct bt_component *component)
1431 {
1432 bool init_can_consume;
1433 uint64_t count;
1434 uint64_t i;
1435 int ret = 0;
1436
1437 BT_ASSERT(graph);
1438 BT_ASSERT(component);
1439 BT_ASSERT(component->base.ref_count == 0);
1440 BT_ASSERT(bt_component_borrow_graph(component) == graph);
1441
1442 init_can_consume = graph->can_consume;
1443 count = bt_component_get_input_port_count(component);
1444
1445 for (i = 0; i < count; i++) {
1446 struct bt_port *port = (void *)
1447 bt_component_borrow_input_port_by_index(component, i);
1448
1449 BT_ASSERT(port);
1450
1451 if (bt_port_is_connected(port)) {
1452 BT_LIB_LOGW("Cannot remove component from graph: "
1453 "an input port is connected: "
1454 "%![graph-]+g, %![comp-]+c, %![port-]+p",
1455 graph, component, port);
1456 goto error;
1457 }
1458 }
1459
1460 count = bt_component_get_output_port_count(component);
1461
1462 for (i = 0; i < count; i++) {
1463 struct bt_port *port = (void *)
1464 bt_component_borrow_output_port_by_index(component, i);
1465
1466 BT_ASSERT(port);
1467
1468 if (bt_port_is_connected(port)) {
1469 BT_LIB_LOGW("Cannot remove component from graph: "
1470 "an output port is connected: "
1471 "%![graph-]+g, %![comp-]+c, %![port-]+p",
1472 graph, component, port);
1473 goto error;
1474 }
1475 }
1476
1477 bt_graph_set_can_consume(graph, false);
1478
1479 /* Possibly remove from sinks to consume */
1480 (void) g_queue_remove(graph->sinks_to_consume, component);
1481
1482 if (graph->sinks_to_consume->length == 0) {
1483 graph->has_sink = false;
1484 }
1485
1486 /*
1487 * This calls bt_object_try_spec_release() on the component, and
1488 * since its reference count is 0, its destructor is called. Its
1489 * destructor calls the user's finalization method (if set).
1490 */
1491 g_ptr_array_remove(graph->components, component);
1492 goto end;
1493
1494 error:
1495 ret = -1;
1496
1497 end:
1498 (void) init_can_consume;
1499 bt_graph_set_can_consume(graph, init_can_consume);
1500 return ret;
1501 }
1502
1503 BT_HIDDEN
1504 void bt_graph_add_message(struct bt_graph *graph,
1505 struct bt_message *msg)
1506 {
1507 BT_ASSERT(graph);
1508 BT_ASSERT(msg);
1509
1510 /*
1511 * It's okay not to take a reference because, when a
1512 * message's reference count drops to 0, either:
1513 *
1514 * * It is recycled back to one of this graph's pool.
1515 * * It is destroyed because it doesn't have any link to any
1516 * graph, which means the original graph is already destroyed.
1517 */
1518 g_ptr_array_add(graph->messages, msg);
1519 }
1520
1521 void bt_graph_get_ref(const struct bt_graph *graph)
1522 {
1523 bt_object_get_ref(graph);
1524 }
1525
1526 void bt_graph_put_ref(const struct bt_graph *graph)
1527 {
1528 bt_object_put_ref(graph);
1529 }
BabelTrace - Trace Format Babel Tower
This page took 0.0621 seconds and 4 git commands to generate.