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