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