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