Make bt_private_connection_create_notification_iterator() return a status code
[babeltrace.git] / plugins / utils / muxer / muxer.c
1 /*
2 * Copyright 2017 Philippe Proulx <pproulx@efficios.com>
3 *
4 * Permission is hereby granted, free of charge, to any person obtaining a copy
5 * of this software and associated documentation files (the "Software"), to deal
6 * in the Software without restriction, including without limitation the rights
7 * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
8 * copies of the Software, and to permit persons to whom the Software is
9 * furnished to do so, subject to the following conditions:
10 *
11 * The above copyright notice and this permission notice shall be included in
12 * all copies or substantial portions of the Software.
13 *
14 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
15 * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
16 * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
17 * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
18 * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
19 * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
20 * SOFTWARE.
21 */
22
23 #include <babeltrace/babeltrace-internal.h>
24 #include <babeltrace/ctf-ir/clock-class.h>
25 #include <babeltrace/ctf-ir/event.h>
26 #include <babeltrace/graph/clock-class-priority-map.h>
27 #include <babeltrace/graph/component-filter.h>
28 #include <babeltrace/graph/component.h>
29 #include <babeltrace/graph/notification-event.h>
30 #include <babeltrace/graph/notification-inactivity.h>
31 #include <babeltrace/graph/notification-iterator.h>
32 #include <babeltrace/graph/notification.h>
33 #include <babeltrace/graph/port.h>
34 #include <babeltrace/graph/private-component-filter.h>
35 #include <babeltrace/graph/private-component.h>
36 #include <babeltrace/graph/private-component.h>
37 #include <babeltrace/graph/private-connection.h>
38 #include <babeltrace/graph/private-notification-iterator.h>
39 #include <babeltrace/graph/private-port.h>
40 #include <babeltrace/graph/connection.h>
41 #include <plugins-common.h>
42 #include <glib.h>
43 #include <stdbool.h>
44 #include <assert.h>
45 #include <stdlib.h>
46
47 #define ASSUME_ABSOLUTE_CLOCK_CLASSES_PARAM_NAME "assume-absolute-clock-classes"
48
49 struct muxer_comp {
50 /* Array of struct bt_private_notification_iterator * (weak refs) */
51 GPtrArray *muxer_notif_iters;
52
53 /* Weak ref */
54 struct bt_private_component *priv_comp;
55 unsigned int next_port_num;
56 size_t available_input_ports;
57 bool error;
58 bool initializing_muxer_notif_iter;
59 bool ignore_absolute;
60 };
61
62 struct muxer_upstream_notif_iter {
63 /* Owned by this, NULL if ended */
64 struct bt_notification_iterator *notif_iter;
65
66 /*
67 * This flag is true if the upstream notification iterator's
68 * current notification must be considered for the multiplexing
69 * operations. If the upstream iterator returns
70 * BT_NOTIFICATION_ITERATOR_STATUS_AGAIN, then this object
71 * is considered invalid, because its current notification is
72 * still the previous one, but we already took it into account.
73 *
74 * The value of this flag is not important if notif_iter above
75 * is NULL (which means the upstream iterator is finished).
76 */
77 bool is_valid;
78 };
79
80 struct muxer_notif_iter {
81 /*
82 * Array of struct muxer_upstream_notif_iter * (owned by this).
83 *
84 * NOTE: This array is searched in linearly to find the youngest
85 * current notification. Keep this until benchmarks confirm that
86 * another data structure is faster than this for our typical
87 * use cases.
88 */
89 GPtrArray *muxer_upstream_notif_iters;
90
91 /*
92 * List of "recently" connected input ports (weak) to
93 * handle by this muxer notification iterator.
94 * muxer_port_connected() adds entries to this list, and the
95 * entries are removed when a notification iterator is created
96 * on the port's connection and put into
97 * muxer_upstream_notif_iters above by
98 * muxer_notif_iter_handle_newly_connected_ports().
99 */
100 GList *newly_connected_priv_ports;
101
102 /* Next thing to return by the "next" method */
103 struct bt_notification_iterator_next_return next_next_return;
104
105 /* Last time returned in a notification */
106 int64_t last_returned_ts_ns;
107 };
108
109 static
110 void destroy_muxer_upstream_notif_iter(
111 struct muxer_upstream_notif_iter *muxer_upstream_notif_iter)
112 {
113 if (!muxer_upstream_notif_iter) {
114 return;
115 }
116
117 bt_put(muxer_upstream_notif_iter->notif_iter);
118 g_free(muxer_upstream_notif_iter);
119 }
120
121 static
122 struct muxer_upstream_notif_iter *muxer_notif_iter_add_upstream_notif_iter(
123 struct muxer_notif_iter *muxer_notif_iter,
124 struct bt_notification_iterator *notif_iter,
125 struct bt_private_port *priv_port)
126 {
127 struct muxer_upstream_notif_iter *muxer_upstream_notif_iter =
128 g_new0(struct muxer_upstream_notif_iter, 1);
129
130 if (!muxer_upstream_notif_iter) {
131 goto end;
132 }
133
134 muxer_upstream_notif_iter->notif_iter = bt_get(notif_iter);
135 muxer_upstream_notif_iter->is_valid = false;
136 g_ptr_array_add(muxer_notif_iter->muxer_upstream_notif_iters,
137 muxer_upstream_notif_iter);
138
139 end:
140 return muxer_upstream_notif_iter;
141 }
142
143 static
144 enum bt_component_status ensure_available_input_port(
145 struct bt_private_component *priv_comp)
146 {
147 struct muxer_comp *muxer_comp =
148 bt_private_component_get_user_data(priv_comp);
149 enum bt_component_status status = BT_COMPONENT_STATUS_OK;
150 GString *port_name = NULL;
151
152 assert(muxer_comp);
153
154 if (muxer_comp->available_input_ports >= 1) {
155 goto end;
156 }
157
158 port_name = g_string_new("in");
159 if (!port_name) {
160 status = BT_COMPONENT_STATUS_NOMEM;
161 goto end;
162 }
163
164 g_string_append_printf(port_name, "%u", muxer_comp->next_port_num);
165 status = bt_private_component_filter_add_input_private_port(
166 priv_comp, port_name->str, NULL, NULL);
167 if (status != BT_COMPONENT_STATUS_OK) {
168 goto end;
169 }
170
171 muxer_comp->available_input_ports++;
172 muxer_comp->next_port_num++;
173
174 end:
175 if (port_name) {
176 g_string_free(port_name, TRUE);
177 }
178
179 return status;
180 }
181
182 static
183 enum bt_component_status create_output_port(
184 struct bt_private_component *priv_comp)
185 {
186 return bt_private_component_filter_add_output_private_port(
187 priv_comp, "out", NULL, NULL);
188 }
189
190 static
191 void destroy_muxer_comp(struct muxer_comp *muxer_comp)
192 {
193 if (!muxer_comp) {
194 return;
195 }
196
197 if (muxer_comp->muxer_notif_iters) {
198 g_ptr_array_free(muxer_comp->muxer_notif_iters, TRUE);
199 }
200
201 g_free(muxer_comp);
202 }
203
204 static
205 struct bt_value *get_default_params(void)
206 {
207 struct bt_value *params;
208 int ret;
209
210 params = bt_value_map_create();
211 if (!params) {
212 goto error;
213 }
214
215 ret = bt_value_map_insert_bool(params,
216 ASSUME_ABSOLUTE_CLOCK_CLASSES_PARAM_NAME, false);
217 if (ret) {
218 goto error;
219 }
220
221 goto end;
222
223 error:
224 BT_PUT(params);
225
226 end:
227 return params;
228 }
229
230 static
231 int configure_muxer_comp(struct muxer_comp *muxer_comp, struct bt_value *params)
232 {
233 struct bt_value *default_params = NULL;
234 struct bt_value *real_params = NULL;
235 struct bt_value *ignore_absolute = NULL;
236 int ret = 0;
237 bt_bool bool_val;
238
239 default_params = get_default_params();
240 if (!default_params) {
241 goto error;
242 }
243
244 real_params = bt_value_map_extend(default_params, params);
245 if (!real_params) {
246 goto error;
247 }
248
249 ignore_absolute = bt_value_map_get(real_params,
250 ASSUME_ABSOLUTE_CLOCK_CLASSES_PARAM_NAME);
251 if (!bt_value_is_bool(ignore_absolute)) {
252 goto error;
253 }
254
255 if (bt_value_bool_get(ignore_absolute, &bool_val)) {
256 goto error;
257 }
258
259 muxer_comp->ignore_absolute = (bool) bool_val;
260
261 goto end;
262
263 error:
264 ret = -1;
265
266 end:
267 bt_put(default_params);
268 bt_put(real_params);
269 bt_put(ignore_absolute);
270 return ret;
271 }
272
273 BT_HIDDEN
274 enum bt_component_status muxer_init(
275 struct bt_private_component *priv_comp,
276 struct bt_value *params, void *init_data)
277 {
278 int ret;
279 enum bt_component_status status = BT_COMPONENT_STATUS_OK;
280 struct muxer_comp *muxer_comp = g_new0(struct muxer_comp, 1);
281
282 if (!muxer_comp) {
283 goto error;
284 }
285
286 ret = configure_muxer_comp(muxer_comp, params);
287 if (ret) {
288 goto error;
289 }
290
291 muxer_comp->muxer_notif_iters = g_ptr_array_new();
292 if (!muxer_comp->muxer_notif_iters) {
293 goto error;
294 }
295
296 muxer_comp->priv_comp = priv_comp;
297 ret = bt_private_component_set_user_data(priv_comp, muxer_comp);
298 assert(ret == 0);
299 status = ensure_available_input_port(priv_comp);
300 if (status != BT_COMPONENT_STATUS_OK) {
301 goto error;
302 }
303
304 ret = create_output_port(priv_comp);
305 if (ret) {
306 goto error;
307 }
308
309 goto end;
310
311 error:
312 destroy_muxer_comp(muxer_comp);
313 ret = bt_private_component_set_user_data(priv_comp, NULL);
314 assert(ret == 0);
315
316 if (status == BT_COMPONENT_STATUS_OK) {
317 status = BT_COMPONENT_STATUS_ERROR;
318 }
319
320 end:
321 return status;
322 }
323
324 BT_HIDDEN
325 void muxer_finalize(struct bt_private_component *priv_comp)
326 {
327 struct muxer_comp *muxer_comp =
328 bt_private_component_get_user_data(priv_comp);
329
330 destroy_muxer_comp(muxer_comp);
331 }
332
333 static
334 struct bt_notification_iterator *create_notif_iter_on_input_port(
335 struct bt_private_port *priv_port, int *ret)
336 {
337 struct bt_port *port = bt_port_from_private_port(priv_port);
338 struct bt_notification_iterator *notif_iter = NULL;
339 struct bt_private_connection *priv_conn = NULL;
340 enum bt_connection_status conn_status;
341
342 assert(ret);
343 *ret = 0;
344 assert(port);
345
346 assert(bt_port_is_connected(port));
347 priv_conn = bt_private_port_get_private_connection(priv_port);
348 if (!priv_conn) {
349 *ret = -1;
350 goto end;
351 }
352
353 // TODO: Advance the iterator to >= the time of the latest
354 // returned notification by the muxer notification
355 // iterator which creates it.
356 conn_status = bt_private_connection_create_notification_iterator(
357 priv_conn, NULL, &notif_iter);
358 if (conn_status != BT_CONNECTION_STATUS_OK) {
359 *ret = -1;
360 goto end;
361 }
362
363 end:
364 bt_put(port);
365 bt_put(priv_conn);
366 return notif_iter;
367 }
368
369 static
370 enum bt_notification_iterator_status muxer_upstream_notif_iter_next(
371 struct muxer_upstream_notif_iter *muxer_upstream_notif_iter)
372 {
373 enum bt_notification_iterator_status status;
374
375 status = bt_notification_iterator_next(
376 muxer_upstream_notif_iter->notif_iter);
377
378 switch (status) {
379 case BT_NOTIFICATION_ITERATOR_STATUS_OK:
380 /*
381 * Notification iterator's current notification is valid:
382 * it must be considered for muxing operations.
383 */
384 muxer_upstream_notif_iter->is_valid = true;
385 break;
386 case BT_NOTIFICATION_ITERATOR_STATUS_AGAIN:
387 /*
388 * Notification iterator's current notification is not
389 * valid anymore. Return
390 * BT_NOTIFICATION_ITERATOR_STATUS_AGAIN
391 * immediately.
392 */
393 muxer_upstream_notif_iter->is_valid = false;
394 break;
395 case BT_NOTIFICATION_ITERATOR_STATUS_END: /* Fall-through. */
396 case BT_NOTIFICATION_ITERATOR_STATUS_CANCELED:
397 /*
398 * Notification iterator reached the end: release it. It
399 * won't be considered again to find the youngest
400 * notification.
401 */
402 BT_PUT(muxer_upstream_notif_iter->notif_iter);
403 muxer_upstream_notif_iter->is_valid = false;
404 status = BT_NOTIFICATION_ITERATOR_STATUS_OK;
405 break;
406 default:
407 /* Error or unsupported status code */
408 status = BT_NOTIFICATION_ITERATOR_STATUS_ERROR;
409 break;
410 }
411
412 return status;
413 }
414
415 static
416 int muxer_notif_iter_handle_newly_connected_ports(
417 struct muxer_notif_iter *muxer_notif_iter)
418 {
419 int ret = 0;
420
421 /*
422 * Here we create one upstream notification iterator for each
423 * newly connected port. We do not perform an initial "next" on
424 * those new upstream notification iterators: they are
425 * invalidated, to be validated later. The list of newly
426 * connected ports to handle here is updated by
427 * muxer_port_connected().
428 */
429 while (true) {
430 GList *node = muxer_notif_iter->newly_connected_priv_ports;
431 struct bt_private_port *priv_port;
432 struct bt_port *port;
433 struct bt_notification_iterator *upstream_notif_iter = NULL;
434 struct muxer_upstream_notif_iter *muxer_upstream_notif_iter;
435
436 if (!node) {
437 break;
438 }
439
440 priv_port = node->data;
441 port = bt_port_from_private_port(priv_port);
442 assert(port);
443
444 if (!bt_port_is_connected(port)) {
445 /*
446 * Looks like this port is not connected
447 * anymore: we can't create an upstream
448 * notification iterator on its (non-existing)
449 * connection in this case.
450 */
451 goto remove_node;
452 }
453
454 BT_PUT(port);
455 upstream_notif_iter = create_notif_iter_on_input_port(priv_port,
456 &ret);
457 if (ret) {
458 assert(!upstream_notif_iter);
459 goto error;
460 }
461
462 muxer_upstream_notif_iter =
463 muxer_notif_iter_add_upstream_notif_iter(
464 muxer_notif_iter, upstream_notif_iter,
465 priv_port);
466 BT_PUT(upstream_notif_iter);
467 if (!muxer_upstream_notif_iter) {
468 goto error;
469 }
470
471 remove_node:
472 bt_put(upstream_notif_iter);
473 bt_put(port);
474 muxer_notif_iter->newly_connected_priv_ports =
475 g_list_delete_link(
476 muxer_notif_iter->newly_connected_priv_ports,
477 node);
478 }
479
480 goto end;
481
482 error:
483 if (ret >= 0) {
484 ret = -1;
485 }
486
487 end:
488 return ret;
489 }
490
491 static
492 int get_notif_ts_ns(struct muxer_comp *muxer_comp,
493 struct bt_notification *notif, int64_t last_returned_ts_ns,
494 int64_t *ts_ns)
495 {
496 struct bt_clock_class_priority_map *cc_prio_map = NULL;
497 struct bt_ctf_clock_class *clock_class = NULL;
498 struct bt_ctf_clock_value *clock_value = NULL;
499 struct bt_ctf_event *event = NULL;
500 int ret = 0;
501
502 assert(notif);
503 assert(ts_ns);
504
505 switch (bt_notification_get_type(notif)) {
506 case BT_NOTIFICATION_TYPE_EVENT:
507 cc_prio_map =
508 bt_notification_event_get_clock_class_priority_map(
509 notif);
510 break;
511
512 case BT_NOTIFICATION_TYPE_INACTIVITY:
513 cc_prio_map =
514 bt_notification_inactivity_get_clock_class_priority_map(
515 notif);
516 break;
517 default:
518 /* All the other notifications have a higher priority */
519 *ts_ns = last_returned_ts_ns;
520 goto end;
521 }
522
523 if (!cc_prio_map) {
524 goto error;
525 }
526
527 /*
528 * If the clock class priority map is empty, then we consider
529 * that this notification has no time. In this case it's always
530 * the youngest.
531 */
532 if (bt_clock_class_priority_map_get_clock_class_count(cc_prio_map) == 0) {
533 *ts_ns = last_returned_ts_ns;
534 goto end;
535 }
536
537 clock_class =
538 bt_clock_class_priority_map_get_highest_priority_clock_class(
539 cc_prio_map);
540 if (!clock_class) {
541 goto error;
542 }
543
544 if (!muxer_comp->ignore_absolute &&
545 !bt_ctf_clock_class_is_absolute(clock_class)) {
546 goto error;
547 }
548
549 switch (bt_notification_get_type(notif)) {
550 case BT_NOTIFICATION_TYPE_EVENT:
551 event = bt_notification_event_get_event(notif);
552 assert(event);
553 clock_value = bt_ctf_event_get_clock_value(event,
554 clock_class);
555 break;
556 case BT_NOTIFICATION_TYPE_INACTIVITY:
557 clock_value = bt_notification_inactivity_get_clock_value(
558 notif, clock_class);
559 break;
560 default:
561 abort();
562 }
563
564 if (!clock_value) {
565 goto error;
566 }
567
568 ret = bt_ctf_clock_value_get_value_ns_from_epoch(clock_value, ts_ns);
569 if (ret) {
570 goto error;
571 }
572
573 goto end;
574
575 error:
576 ret = -1;
577
578 end:
579 bt_put(cc_prio_map);
580 bt_put(event);
581 bt_put(clock_class);
582 bt_put(clock_value);
583 return ret;
584 }
585
586 /*
587 * This function finds the youngest available notification amongst the
588 * non-ended upstream notification iterators and returns the upstream
589 * notification iterator which has it, or
590 * BT_NOTIFICATION_ITERATOR_STATUS_END if there's no available
591 * notification.
592 *
593 * This function does NOT:
594 *
595 * * Update any upstream notification iterator.
596 * * Check for newly connected ports.
597 * * Check the upstream notification iterators to retry.
598 *
599 * On sucess, this function sets *muxer_upstream_notif_iter to the
600 * upstream notification iterator of which the current notification is
601 * the youngest, and sets *ts_ns to its time.
602 */
603 static
604 enum bt_notification_iterator_status
605 muxer_notif_iter_youngest_upstream_notif_iter(
606 struct muxer_comp *muxer_comp,
607 struct muxer_notif_iter *muxer_notif_iter,
608 struct muxer_upstream_notif_iter **muxer_upstream_notif_iter,
609 int64_t *ts_ns)
610 {
611 size_t i;
612 int ret;
613 int64_t youngest_ts_ns = INT64_MAX;
614 enum bt_notification_iterator_status status =
615 BT_NOTIFICATION_ITERATOR_STATUS_OK;
616
617 assert(muxer_comp);
618 assert(muxer_notif_iter);
619 assert(muxer_upstream_notif_iter);
620 *muxer_upstream_notif_iter = NULL;
621
622 for (i = 0; i < muxer_notif_iter->muxer_upstream_notif_iters->len; i++) {
623 struct bt_notification *notif;
624 struct muxer_upstream_notif_iter *cur_muxer_upstream_notif_iter =
625 g_ptr_array_index(muxer_notif_iter->muxer_upstream_notif_iters, i);
626 int64_t notif_ts_ns;
627
628 if (!cur_muxer_upstream_notif_iter->notif_iter) {
629 /* This upstream notification iterator is ended */
630 continue;
631 }
632
633 assert(cur_muxer_upstream_notif_iter->is_valid);
634 notif = bt_notification_iterator_get_notification(
635 cur_muxer_upstream_notif_iter->notif_iter);
636 assert(notif);
637 ret = get_notif_ts_ns(muxer_comp, notif,
638 muxer_notif_iter->last_returned_ts_ns, &notif_ts_ns);
639 bt_put(notif);
640 if (ret) {
641 *muxer_upstream_notif_iter = NULL;
642 status = BT_NOTIFICATION_ITERATOR_STATUS_ERROR;
643 goto end;
644 }
645
646 if (notif_ts_ns <= youngest_ts_ns) {
647 *muxer_upstream_notif_iter =
648 cur_muxer_upstream_notif_iter;
649 youngest_ts_ns = notif_ts_ns;
650 *ts_ns = youngest_ts_ns;
651 }
652 }
653
654 if (!*muxer_upstream_notif_iter) {
655 status = BT_NOTIFICATION_ITERATOR_STATUS_END;
656 *ts_ns = INT64_MIN;
657 }
658
659 end:
660 return status;
661 }
662
663 static
664 enum bt_notification_iterator_status validate_muxer_upstream_notif_iter(
665 struct muxer_upstream_notif_iter *muxer_upstream_notif_iter)
666 {
667 enum bt_notification_iterator_status status =
668 BT_NOTIFICATION_ITERATOR_STATUS_OK;
669
670 if (muxer_upstream_notif_iter->is_valid ||
671 !muxer_upstream_notif_iter->notif_iter) {
672 goto end;
673 }
674
675 status = muxer_upstream_notif_iter_next(muxer_upstream_notif_iter);
676
677 end:
678 return status;
679 }
680
681 static
682 enum bt_notification_iterator_status validate_muxer_upstream_notif_iters(
683 struct muxer_notif_iter *muxer_notif_iter)
684 {
685 enum bt_notification_iterator_status status =
686 BT_NOTIFICATION_ITERATOR_STATUS_OK;
687 size_t i;
688
689 for (i = 0; i < muxer_notif_iter->muxer_upstream_notif_iters->len; i++) {
690 struct muxer_upstream_notif_iter *muxer_upstream_notif_iter =
691 g_ptr_array_index(
692 muxer_notif_iter->muxer_upstream_notif_iters,
693 i);
694
695 status = validate_muxer_upstream_notif_iter(
696 muxer_upstream_notif_iter);
697 if (status != BT_NOTIFICATION_ITERATOR_STATUS_OK) {
698 goto end;
699 }
700
701 /*
702 * Remove this muxer upstream notification iterator
703 * if it's ended or canceled.
704 */
705 if (!muxer_upstream_notif_iter->notif_iter) {
706 /*
707 * Use g_ptr_array_remove_fast() because the
708 * order of those elements is not important.
709 */
710 g_ptr_array_remove_index_fast(
711 muxer_notif_iter->muxer_upstream_notif_iters,
712 i);
713 i--;
714 }
715 }
716
717 end:
718 return status;
719 }
720
721 static
722 struct bt_notification_iterator_next_return muxer_notif_iter_do_next(
723 struct muxer_comp *muxer_comp,
724 struct muxer_notif_iter *muxer_notif_iter)
725 {
726 struct muxer_upstream_notif_iter *muxer_upstream_notif_iter = NULL;
727 struct bt_notification_iterator_next_return next_return = {
728 .notification = NULL,
729 .status = BT_NOTIFICATION_ITERATOR_STATUS_OK,
730 };
731 int64_t next_return_ts;
732
733 while (true) {
734 int ret = muxer_notif_iter_handle_newly_connected_ports(
735 muxer_notif_iter);
736
737 if (ret) {
738 next_return.status =
739 BT_NOTIFICATION_ITERATOR_STATUS_ERROR;
740 goto end;
741 }
742
743 next_return.status =
744 validate_muxer_upstream_notif_iters(muxer_notif_iter);
745 if (next_return.status != BT_NOTIFICATION_ITERATOR_STATUS_OK) {
746 goto end;
747 }
748
749 /*
750 * At this point, we know that all the existing upstream
751 * notification iterators are valid. However the
752 * operations to validate them (during
753 * validate_muxer_upstream_notif_iters()) may have
754 * connected new ports. If no ports were connected
755 * during this operation, exit the loop.
756 */
757 if (!muxer_notif_iter->newly_connected_priv_ports) {
758 break;
759 }
760 }
761
762 assert(!muxer_notif_iter->newly_connected_priv_ports);
763
764 /*
765 * At this point we know that all the existing upstream
766 * notification iterators are valid. We can find the one,
767 * amongst those, of which the current notification is the
768 * youngest.
769 */
770 next_return.status =
771 muxer_notif_iter_youngest_upstream_notif_iter(muxer_comp,
772 muxer_notif_iter, &muxer_upstream_notif_iter,
773 &next_return_ts);
774 if (next_return.status < 0 ||
775 next_return.status == BT_NOTIFICATION_ITERATOR_STATUS_END ||
776 next_return.status == BT_NOTIFICATION_ITERATOR_STATUS_CANCELED) {
777 goto end;
778 }
779
780 if (next_return_ts < muxer_notif_iter->last_returned_ts_ns) {
781 next_return.status = BT_NOTIFICATION_ITERATOR_STATUS_ERROR;
782 goto end;
783 }
784
785 assert(next_return.status == BT_NOTIFICATION_ITERATOR_STATUS_OK);
786 assert(muxer_upstream_notif_iter);
787 next_return.notification = bt_notification_iterator_get_notification(
788 muxer_upstream_notif_iter->notif_iter);
789 assert(next_return.notification);
790
791 /*
792 * We invalidate the upstream notification iterator so that, the
793 * next time this function is called,
794 * validate_muxer_upstream_notif_iters() will make it valid.
795 */
796 muxer_upstream_notif_iter->is_valid = false;
797 muxer_notif_iter->last_returned_ts_ns = next_return_ts;
798
799 end:
800 return next_return;
801 }
802
803 static
804 void destroy_muxer_notif_iter(struct muxer_notif_iter *muxer_notif_iter)
805 {
806 if (!muxer_notif_iter) {
807 return;
808 }
809
810 if (muxer_notif_iter->muxer_upstream_notif_iters) {
811 g_ptr_array_free(
812 muxer_notif_iter->muxer_upstream_notif_iters, TRUE);
813 }
814
815 g_list_free(muxer_notif_iter->newly_connected_priv_ports);
816 g_free(muxer_notif_iter);
817 }
818
819 static
820 int muxer_notif_iter_init_newly_connected_ports(struct muxer_comp *muxer_comp,
821 struct muxer_notif_iter *muxer_notif_iter)
822 {
823 struct bt_component *comp;
824 int64_t count;
825 int64_t i;
826 int ret = 0;
827
828 /*
829 * Add the connected input ports to this muxer notification
830 * iterator's list of newly connected ports. They will be
831 * handled by muxer_notif_iter_handle_newly_connected_ports().
832 */
833 comp = bt_component_from_private_component(muxer_comp->priv_comp);
834 assert(comp);
835 count = bt_component_filter_get_input_port_count(comp);
836 if (count < 0) {
837 goto end;
838 }
839
840 for (i = 0; i < count; i++) {
841 struct bt_private_port *priv_port =
842 bt_private_component_filter_get_input_private_port_by_index(
843 muxer_comp->priv_comp, i);
844 struct bt_port *port;
845
846 assert(priv_port);
847 port = bt_port_from_private_port(priv_port);
848 assert(port);
849
850 if (!bt_port_is_connected(port)) {
851 bt_put(priv_port);
852 bt_put(port);
853 continue;
854 }
855
856 bt_put(port);
857 bt_put(priv_port);
858 muxer_notif_iter->newly_connected_priv_ports =
859 g_list_append(
860 muxer_notif_iter->newly_connected_priv_ports,
861 priv_port);
862 if (!muxer_notif_iter->newly_connected_priv_ports) {
863 ret = -1;
864 goto end;
865 }
866 }
867
868 end:
869 bt_put(comp);
870 return ret;
871 }
872
873 BT_HIDDEN
874 enum bt_notification_iterator_status muxer_notif_iter_init(
875 struct bt_private_notification_iterator *priv_notif_iter,
876 struct bt_private_port *output_priv_port)
877 {
878 struct muxer_comp *muxer_comp = NULL;
879 struct muxer_notif_iter *muxer_notif_iter = NULL;
880 struct bt_private_component *priv_comp = NULL;
881 enum bt_notification_iterator_status status =
882 BT_NOTIFICATION_ITERATOR_STATUS_OK;
883 int ret;
884
885 priv_comp = bt_private_notification_iterator_get_private_component(
886 priv_notif_iter);
887 assert(priv_comp);
888 muxer_comp = bt_private_component_get_user_data(priv_comp);
889 assert(muxer_comp);
890
891 if (muxer_comp->initializing_muxer_notif_iter) {
892 /*
893 * Weird, unhandled situation detected: downstream
894 * creates a muxer notification iterator while creating
895 * another muxer notification iterator (same component).
896 */
897 goto error;
898 }
899
900 muxer_comp->initializing_muxer_notif_iter = true;
901 muxer_notif_iter = g_new0(struct muxer_notif_iter, 1);
902 if (!muxer_notif_iter) {
903 goto error;
904 }
905
906 muxer_notif_iter->last_returned_ts_ns = INT64_MIN;
907 muxer_notif_iter->muxer_upstream_notif_iters =
908 g_ptr_array_new_with_free_func(
909 (GDestroyNotify) destroy_muxer_upstream_notif_iter);
910 if (!muxer_notif_iter->muxer_upstream_notif_iters) {
911 goto error;
912 }
913
914 /*
915 * Add the muxer notification iterator to the component's array
916 * of muxer notification iterators here because
917 * muxer_notif_iter_init_newly_connected_ports() can cause
918 * muxer_port_connected() to be called, which adds the newly
919 * connected port to each muxer notification iterator's list of
920 * newly connected ports.
921 */
922 g_ptr_array_add(muxer_comp->muxer_notif_iters, muxer_notif_iter);
923 ret = muxer_notif_iter_init_newly_connected_ports(muxer_comp,
924 muxer_notif_iter);
925 if (ret) {
926 goto error;
927 }
928
929 ret = bt_private_notification_iterator_set_user_data(priv_notif_iter,
930 muxer_notif_iter);
931 assert(ret == 0);
932 goto end;
933
934 error:
935 if (g_ptr_array_index(muxer_comp->muxer_notif_iters,
936 muxer_comp->muxer_notif_iters->len - 1) == muxer_notif_iter) {
937 g_ptr_array_remove_index(muxer_comp->muxer_notif_iters,
938 muxer_comp->muxer_notif_iters->len - 1);
939 }
940
941 destroy_muxer_notif_iter(muxer_notif_iter);
942 ret = bt_private_notification_iterator_set_user_data(priv_notif_iter,
943 NULL);
944 assert(ret == 0);
945 status = BT_NOTIFICATION_ITERATOR_STATUS_ERROR;
946
947 end:
948 muxer_comp->initializing_muxer_notif_iter = false;
949 bt_put(priv_comp);
950 return status;
951 }
952
953 BT_HIDDEN
954 void muxer_notif_iter_finalize(
955 struct bt_private_notification_iterator *priv_notif_iter)
956 {
957 struct muxer_notif_iter *muxer_notif_iter =
958 bt_private_notification_iterator_get_user_data(priv_notif_iter);
959 struct bt_private_component *priv_comp = NULL;
960 struct muxer_comp *muxer_comp = NULL;
961
962 priv_comp = bt_private_notification_iterator_get_private_component(
963 priv_notif_iter);
964 assert(priv_comp);
965 muxer_comp = bt_private_component_get_user_data(priv_comp);
966
967 if (muxer_comp) {
968 (void) g_ptr_array_remove_fast(muxer_comp->muxer_notif_iters,
969 muxer_notif_iter);
970 destroy_muxer_notif_iter(muxer_notif_iter);
971 }
972
973 bt_put(priv_comp);
974 }
975
976 BT_HIDDEN
977 struct bt_notification_iterator_next_return muxer_notif_iter_next(
978 struct bt_private_notification_iterator *priv_notif_iter)
979 {
980 struct bt_notification_iterator_next_return next_ret;
981 struct muxer_notif_iter *muxer_notif_iter =
982 bt_private_notification_iterator_get_user_data(priv_notif_iter);
983 struct bt_private_component *priv_comp = NULL;
984 struct muxer_comp *muxer_comp = NULL;
985
986 assert(muxer_notif_iter);
987 priv_comp = bt_private_notification_iterator_get_private_component(
988 priv_notif_iter);
989 assert(priv_comp);
990 muxer_comp = bt_private_component_get_user_data(priv_comp);
991 assert(muxer_comp);
992
993 /* Are we in an error state set elsewhere? */
994 if (unlikely(muxer_comp->error)) {
995 next_ret.notification = NULL;
996 next_ret.status = BT_NOTIFICATION_ITERATOR_STATUS_ERROR;
997 goto end;
998 }
999
1000 next_ret = muxer_notif_iter_do_next(muxer_comp, muxer_notif_iter);
1001
1002 end:
1003 bt_put(priv_comp);
1004 return next_ret;
1005 }
1006
1007 BT_HIDDEN
1008 void muxer_port_connected(
1009 struct bt_private_component *priv_comp,
1010 struct bt_private_port *self_private_port,
1011 struct bt_port *other_port)
1012 {
1013 struct bt_port *self_port =
1014 bt_port_from_private_port(self_private_port);
1015 struct muxer_comp *muxer_comp =
1016 bt_private_component_get_user_data(priv_comp);
1017 size_t i;
1018 int ret;
1019
1020 assert(self_port);
1021 assert(muxer_comp);
1022
1023 if (bt_port_get_type(self_port) == BT_PORT_TYPE_OUTPUT) {
1024 goto end;
1025 }
1026
1027 for (i = 0; i < muxer_comp->muxer_notif_iters->len; i++) {
1028 struct muxer_notif_iter *muxer_notif_iter =
1029 g_ptr_array_index(muxer_comp->muxer_notif_iters, i);
1030
1031 /*
1032 * Add this port to the list of newly connected ports
1033 * for this muxer notification iterator. We append at
1034 * the end of this list while
1035 * muxer_notif_iter_handle_newly_connected_ports()
1036 * removes the nodes from the beginning.
1037 *
1038 * The list node owns the private port.
1039 */
1040 muxer_notif_iter->newly_connected_priv_ports =
1041 g_list_append(
1042 muxer_notif_iter->newly_connected_priv_ports,
1043 self_private_port);
1044 if (!muxer_notif_iter->newly_connected_priv_ports) {
1045 /* Put reference taken by bt_get() above */
1046 muxer_comp->error = true;
1047 goto end;
1048 }
1049 }
1050
1051 /* One less available input port */
1052 muxer_comp->available_input_ports--;
1053 ret = ensure_available_input_port(priv_comp);
1054 if (ret) {
1055 /*
1056 * Only way to report an error later since this
1057 * method does not return anything.
1058 */
1059 muxer_comp->error = true;
1060 goto end;
1061 }
1062
1063 end:
1064 bt_put(self_port);
1065 }
1066
1067 BT_HIDDEN
1068 void muxer_port_disconnected(struct bt_private_component *priv_comp,
1069 struct bt_private_port *priv_port)
1070 {
1071 struct bt_port *port = bt_port_from_private_port(priv_port);
1072 struct muxer_comp *muxer_comp =
1073 bt_private_component_get_user_data(priv_comp);
1074
1075 assert(port);
1076 assert(muxer_comp);
1077
1078 /*
1079 * There's nothing special to do when a port is disconnected
1080 * because this component deals with upstream notification
1081 * iterators which were already created thanks to connected
1082 * ports. The fact that the port is disconnected does not cancel
1083 * the upstream notification iterators created using its
1084 * connection: they still exist, even if the connection is dead.
1085 * The only way to remove an upstream notification iterator is
1086 * for its "next" operation to return
1087 * BT_NOTIFICATION_ITERATOR_STATUS_END.
1088 */
1089 if (bt_port_get_type(port) == BT_PORT_TYPE_INPUT) {
1090 /* One more available input port */
1091 muxer_comp->available_input_ports++;
1092 }
1093
1094 bt_put(port);
1095 }
This page took 0.054938 seconds and 4 git commands to generate.