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