Visibility hidden by default
[babeltrace.git] / src / plugins / utils / trimmer / trimmer.c
1 /*
2 * SPDX-License-Identifier: MIT
3 *
4 * Copyright 2016 Jérémie Galarneau <jeremie.galarneau@efficios.com>
5 * Copyright 2019 Philippe Proulx <pproulx@efficios.com>
6 */
7
8 #define BT_LOG_OUTPUT_LEVEL (trimmer_comp->log_level)
9 #define BT_LOG_TAG "PLUGIN/FLT.UTILS.TRIMMER"
10 #include "logging/comp-logging.h"
11
12 #include "compat/utc.h"
13 #include "compat/time.h"
14 #include <babeltrace2/babeltrace.h>
15 #include "common/common.h"
16 #include "common/assert.h"
17 #include <stdbool.h>
18 #include <stdint.h>
19 #include <inttypes.h>
20 #include <glib.h>
21 #include "compat/glib.h"
22 #include "plugins/common/param-validation/param-validation.h"
23
24 #include "trimmer.h"
25
26 #define NS_PER_S INT64_C(1000000000)
27
28 static const char * const in_port_name = "in";
29
30 struct trimmer_time {
31 unsigned int hour, minute, second, ns;
32 };
33
34 struct trimmer_bound {
35 /*
36 * Nanoseconds from origin, valid if `is_set` is set and
37 * `is_infinite` is false.
38 */
39 int64_t ns_from_origin;
40
41 /* True if this bound's full time (`ns_from_origin`) is set */
42 bool is_set;
43
44 /*
45 * True if this bound represents the infinity (negative or
46 * positive depending on which bound it is). If this is true,
47 * then we don't care about `ns_from_origin` above.
48 */
49 bool is_infinite;
50
51 /*
52 * This bound's time without the date; this time is used to set
53 * `ns_from_origin` once we know the date.
54 */
55 struct trimmer_time time;
56 };
57
58 struct trimmer_comp {
59 struct trimmer_bound begin, end;
60 bool is_gmt;
61 bt_logging_level log_level;
62 bt_self_component *self_comp;
63 bt_self_component_filter *self_comp_filter;
64 };
65
66 enum trimmer_iterator_state {
67 /*
68 * Find the first message's date and set the bounds's times
69 * accordingly.
70 */
71 TRIMMER_ITERATOR_STATE_SET_BOUNDS_NS_FROM_ORIGIN,
72
73 /*
74 * Initially seek to the trimming range's beginning time.
75 */
76 TRIMMER_ITERATOR_STATE_SEEK_INITIALLY,
77
78 /*
79 * Fill the output message queue for as long as received input
80 * messages are within the trimming time range.
81 */
82 TRIMMER_ITERATOR_STATE_TRIM,
83
84 /* Flush the remaining messages in the output message queue */
85 TRIMMER_ITERATOR_STATE_ENDING,
86
87 /* Trimming operation and message iterator is ended */
88 TRIMMER_ITERATOR_STATE_ENDED,
89 };
90
91 struct trimmer_iterator {
92 /* Weak */
93 struct trimmer_comp *trimmer_comp;
94
95 /* Weak */
96 bt_self_message_iterator *self_msg_iter;
97
98 enum trimmer_iterator_state state;
99
100 /* Owned by this */
101 bt_message_iterator *upstream_iter;
102 struct trimmer_bound begin, end;
103
104 /*
105 * Queue of `const bt_message *` (owned by the queue).
106 *
107 * This is where the trimming operation pushes the messages to
108 * output by this message iterator.
109 */
110 GQueue *output_messages;
111
112 /*
113 * Hash table of `bt_stream *` (weak) to
114 * `struct trimmer_iterator_stream_state *` (owned by the HT).
115 */
116 GHashTable *stream_states;
117 };
118
119 struct trimmer_iterator_stream_state {
120 /* Weak */
121 const bt_stream *stream;
122
123 /* Have we seen a message with clock_snapshot going through this stream? */
124 bool seen_clock_snapshot;
125
126 /* Owned by this (`NULL` initially and between packets) */
127 const bt_packet *cur_packet;
128 };
129
130 static
131 void destroy_trimmer_comp(struct trimmer_comp *trimmer_comp)
132 {
133 BT_ASSERT(trimmer_comp);
134 g_free(trimmer_comp);
135 }
136
137 static
138 struct trimmer_comp *create_trimmer_comp(void)
139 {
140 return g_new0(struct trimmer_comp, 1);
141 }
142
143 void trimmer_finalize(bt_self_component_filter *self_comp)
144 {
145 struct trimmer_comp *trimmer_comp =
146 bt_self_component_get_data(
147 bt_self_component_filter_as_self_component(self_comp));
148
149 if (trimmer_comp) {
150 destroy_trimmer_comp(trimmer_comp);
151 }
152 }
153
154 /*
155 * Compile regex in `pattern`, and try to match `string`. If there's a match,
156 * return true and set `*match_info` to the list of matches. The list of
157 * matches must be freed by the caller. If there's no match, return false and
158 * set `*match_info` to NULL;
159 */
160 static
161 bool compile_and_match(const char *pattern, const char *string, GMatchInfo **match_info) {
162 bool matches = false;
163 GError *regex_error = NULL;
164 GRegex *regex;
165
166 regex = g_regex_new(pattern, 0, 0, &regex_error);
167 if (!regex) {
168 goto end;
169 }
170
171 matches = g_regex_match(regex, string, 0, match_info);
172 if (!matches) {
173 /*
174 * g_regex_match allocates `*match_info` even if it returns
175 * FALSE. If there's no match, we have no use for it, so free
176 * it immediatly and don't return it to the caller.
177 */
178 g_match_info_free(*match_info);
179 *match_info = NULL;
180 }
181
182 g_regex_unref(regex);
183
184 end:
185
186 if (regex_error) {
187 g_error_free(regex_error);
188 }
189
190 return matches;
191 }
192
193 /*
194 * Convert the captured text in match number `match_num` in `match_info`
195 * to an unsigned integer.
196 */
197 static
198 guint64 match_to_uint(const GMatchInfo *match_info, gint match_num) {
199 gchar *text, *endptr;
200 guint64 result;
201
202 text = g_match_info_fetch(match_info, match_num);
203 BT_ASSERT(text);
204
205 /*
206 * Because the input is carefully sanitized with regexes by the caller,
207 * we assume that g_ascii_strtoull cannot fail.
208 */
209 errno = 0;
210 result = g_ascii_strtoull(text, &endptr, 10);
211 BT_ASSERT(endptr > text);
212 BT_ASSERT(errno == 0);
213
214 g_free(text);
215
216 return result;
217 }
218
219 /*
220 * When parsing the nanoseconds part, .512 means .512000000, not .000000512.
221 * This function is like match_to_uint, but multiplies the parsed number to get
222 * the expected result.
223 */
224 static
225 guint64 match_to_uint_ns(const GMatchInfo *match_info, gint match_num) {
226 guint64 nanoseconds;
227 gboolean ret;
228 gint start_pos, end_pos, power;
229 static int pow10[] = {
230 1, 10, 100, 1000, 10000, 100000, 1000000, 10000000, 100000000,
231 };
232
233 nanoseconds = match_to_uint(match_info, match_num);
234
235 /* Multiply by 10 as many times as there are omitted digits. */
236 ret = g_match_info_fetch_pos(match_info, match_num, &start_pos, &end_pos);
237 BT_ASSERT(ret);
238
239 power = 9 - (end_pos - start_pos);
240 BT_ASSERT(power >= 0 && power <= 8);
241
242 nanoseconds *= pow10[power];
243
244 return nanoseconds;
245 }
246
247 /*
248 * Sets the time (in ns from origin) of a trimmer bound from date and
249 * time components.
250 *
251 * Returns a negative value if anything goes wrong.
252 */
253 static
254 int set_bound_ns_from_origin(struct trimmer_bound *bound,
255 unsigned int year, unsigned int month, unsigned int day,
256 unsigned int hour, unsigned int minute, unsigned int second,
257 unsigned int ns, bool is_gmt)
258 {
259 int ret = 0;
260 time_t result;
261 struct tm tm = {
262 .tm_sec = second,
263 .tm_min = minute,
264 .tm_hour = hour,
265 .tm_mday = day,
266 .tm_mon = month - 1,
267 .tm_year = year - 1900,
268 .tm_isdst = -1,
269 };
270
271 if (is_gmt) {
272 result = bt_timegm(&tm);
273 } else {
274 result = mktime(&tm);
275 }
276
277 if (result < 0) {
278 ret = -1;
279 goto end;
280 }
281
282 BT_ASSERT(bound);
283 bound->ns_from_origin = (int64_t) result;
284 bound->ns_from_origin *= NS_PER_S;
285 bound->ns_from_origin += ns;
286 bound->is_set = true;
287
288 end:
289 return ret;
290 }
291
292 /*
293 * Parses a timestamp, figuring out its format.
294 *
295 * Returns a negative value if anything goes wrong.
296 *
297 * Expected formats:
298 *
299 * YYYY-MM-DD hh:mm[:ss[.ns]]
300 * [hh:mm:]ss[.ns]
301 * [-]s[.ns]
302 *
303 * TODO: Check overflows.
304 */
305 static
306 int set_bound_from_str(struct trimmer_comp *trimmer_comp,
307 const char *str, struct trimmer_bound *bound, bool is_gmt)
308 {
309 /* Matches YYYY-MM-DD */
310 #define DATE_RE "([0-9]{4})-([0-9]{2})-([0-9]{2})"
311
312 /* Matches HH:MM[:SS[.NS]] */
313 #define TIME_RE "([0-9]{2}):([0-9]{2})(?::([0-9]{2})(?:\\.([0-9]{1,9}))?)?"
314
315 /* Matches [-]SS[.NS] */
316 #define S_NS_RE "^(-?)([0-9]+)(?:\\.([0-9]{1,9}))?$"
317
318 GMatchInfo *match_info;
319 int ret = 0;
320
321 /* Try `YYYY-MM-DD hh:mm[:ss[.ns]]` format */
322 if (compile_and_match("^" DATE_RE " " TIME_RE "$", str, &match_info)) {
323 unsigned int year = 0, month = 0, day = 0, hours = 0, minutes = 0, seconds = 0, nanoseconds = 0;
324 gint match_count = g_match_info_get_match_count(match_info);
325
326 BT_ASSERT(match_count >= 6 && match_count <= 8);
327
328 year = match_to_uint(match_info, 1);
329 month = match_to_uint(match_info, 2);
330 day = match_to_uint(match_info, 3);
331 hours = match_to_uint(match_info, 4);
332 minutes = match_to_uint(match_info, 5);
333
334 if (match_count >= 7) {
335 seconds = match_to_uint(match_info, 6);
336 }
337
338 if (match_count >= 8) {
339 nanoseconds = match_to_uint_ns(match_info, 7);
340 }
341
342 set_bound_ns_from_origin(bound, year, month, day, hours, minutes, seconds, nanoseconds, is_gmt);
343
344 goto end;
345 }
346
347 if (compile_and_match("^" DATE_RE "$", str, &match_info)) {
348 unsigned int year = 0, month = 0, day = 0;
349
350 BT_ASSERT(g_match_info_get_match_count(match_info) == 4);
351
352 year = match_to_uint(match_info, 1);
353 month = match_to_uint(match_info, 2);
354 day = match_to_uint(match_info, 3);
355
356 set_bound_ns_from_origin(bound, year, month, day, 0, 0, 0, 0, is_gmt);
357
358 goto end;
359 }
360
361 /* Try `hh:mm[:ss[.ns]]` format */
362 if (compile_and_match("^" TIME_RE "$", str, &match_info)) {
363 gint match_count = g_match_info_get_match_count(match_info);
364 BT_ASSERT(match_count >= 3 && match_count <= 5);
365 bound->time.hour = match_to_uint(match_info, 1);
366 bound->time.minute = match_to_uint(match_info, 2);
367
368 if (match_count >= 4) {
369 bound->time.second = match_to_uint(match_info, 3);
370 }
371
372 if (match_count >= 5) {
373 bound->time.ns = match_to_uint_ns(match_info, 4);
374 }
375
376 goto end;
377 }
378
379 /* Try `[-]s[.ns]` format */
380 if (compile_and_match("^" S_NS_RE "$", str, &match_info)) {
381 gboolean is_neg, fetch_pos_ret;
382 gint start_pos, end_pos, match_count;
383 guint64 seconds, nanoseconds = 0;
384
385 match_count = g_match_info_get_match_count(match_info);
386 BT_ASSERT(match_count >= 3 && match_count <= 4);
387
388 /* Check for presence of negation sign. */
389 fetch_pos_ret = g_match_info_fetch_pos(match_info, 1, &start_pos, &end_pos);
390 BT_ASSERT(fetch_pos_ret);
391 is_neg = (end_pos - start_pos) > 0;
392
393 seconds = match_to_uint(match_info, 2);
394
395 if (match_count >= 4) {
396 nanoseconds = match_to_uint_ns(match_info, 3);
397 }
398
399 bound->ns_from_origin = seconds * NS_PER_S + nanoseconds;
400
401 if (is_neg) {
402 bound->ns_from_origin = -bound->ns_from_origin;
403 }
404
405 bound->is_set = true;
406
407 goto end;
408 }
409
410 BT_COMP_LOGE_APPEND_CAUSE(trimmer_comp->self_comp,
411 "Invalid date/time format: param=\"%s\"", str);
412 ret = -1;
413
414 end:
415 g_match_info_free(match_info);
416
417 return ret;
418 }
419
420 /*
421 * Sets a trimmer bound's properties from a parameter string/integer
422 * value.
423 *
424 * Returns a negative value if anything goes wrong.
425 */
426 static
427 int set_bound_from_param(struct trimmer_comp *trimmer_comp,
428 const char *param_name, const bt_value *param,
429 struct trimmer_bound *bound, bool is_gmt)
430 {
431 int ret;
432 const char *arg;
433 char tmp_arg[64];
434
435 if (bt_value_is_signed_integer(param)) {
436 int64_t value = bt_value_integer_signed_get(param);
437
438 /*
439 * Just convert it to a temporary string to handle
440 * everything the same way.
441 */
442 snprintf(tmp_arg, sizeof(tmp_arg), "%" PRId64, value);
443 arg = tmp_arg;
444 } else {
445 BT_ASSERT(bt_value_is_string(param));
446 arg = bt_value_string_get(param);
447 }
448
449 ret = set_bound_from_str(trimmer_comp, arg, bound, is_gmt);
450
451 return ret;
452 }
453
454 static
455 int validate_trimmer_bounds(struct trimmer_comp *trimmer_comp,
456 struct trimmer_bound *begin, struct trimmer_bound *end)
457 {
458 int ret = 0;
459
460 BT_ASSERT(begin->is_set);
461 BT_ASSERT(end->is_set);
462
463 if (!begin->is_infinite && !end->is_infinite &&
464 begin->ns_from_origin > end->ns_from_origin) {
465 BT_COMP_LOGE_APPEND_CAUSE(trimmer_comp->self_comp,
466 "Trimming time range's beginning time is greater than end time: "
467 "begin-ns-from-origin=%" PRId64 ", "
468 "end-ns-from-origin=%" PRId64,
469 begin->ns_from_origin,
470 end->ns_from_origin);
471 ret = -1;
472 goto end;
473 }
474
475 if (!begin->is_infinite && begin->ns_from_origin == INT64_MIN) {
476 BT_COMP_LOGE_APPEND_CAUSE(trimmer_comp->self_comp,
477 "Invalid trimming time range's beginning time: "
478 "ns-from-origin=%" PRId64,
479 begin->ns_from_origin);
480 ret = -1;
481 goto end;
482 }
483
484 if (!end->is_infinite && end->ns_from_origin == INT64_MIN) {
485 BT_COMP_LOGE_APPEND_CAUSE(trimmer_comp->self_comp,
486 "Invalid trimming time range's end time: "
487 "ns-from-origin=%" PRId64,
488 end->ns_from_origin);
489 ret = -1;
490 goto end;
491 }
492
493 end:
494 return ret;
495 }
496
497 static
498 enum bt_param_validation_status validate_bound_type(
499 const bt_value *value,
500 struct bt_param_validation_context *context)
501 {
502 enum bt_param_validation_status status = BT_PARAM_VALIDATION_STATUS_OK;
503
504 if (!bt_value_is_signed_integer(value) &&
505 !bt_value_is_string(value)) {
506 status = bt_param_validation_error(context,
507 "unexpected type: expected-types=[%s, %s], actual-type=%s",
508 bt_common_value_type_string(BT_VALUE_TYPE_SIGNED_INTEGER),
509 bt_common_value_type_string(BT_VALUE_TYPE_STRING),
510 bt_common_value_type_string(bt_value_get_type(value)));
511 }
512
513 return status;
514 }
515
516 static
517 struct bt_param_validation_map_value_entry_descr trimmer_params[] = {
518 { "gmt", BT_PARAM_VALIDATION_MAP_VALUE_ENTRY_OPTIONAL, { .type = BT_VALUE_TYPE_BOOL } },
519 { "begin", BT_PARAM_VALIDATION_MAP_VALUE_ENTRY_OPTIONAL, { .validation_func = validate_bound_type } },
520 { "end", BT_PARAM_VALIDATION_MAP_VALUE_ENTRY_OPTIONAL, { .validation_func = validate_bound_type } },
521 BT_PARAM_VALIDATION_MAP_VALUE_ENTRY_END
522 };
523
524 static
525 bt_component_class_initialize_method_status init_trimmer_comp_from_params(
526 struct trimmer_comp *trimmer_comp,
527 const bt_value *params)
528 {
529 const bt_value *value;
530 bt_component_class_initialize_method_status status;
531 enum bt_param_validation_status validation_status;
532 gchar *validate_error = NULL;
533
534 validation_status = bt_param_validation_validate(params,
535 trimmer_params, &validate_error);
536 if (validation_status == BT_PARAM_VALIDATION_STATUS_MEMORY_ERROR) {
537 status = BT_COMPONENT_CLASS_INITIALIZE_METHOD_STATUS_MEMORY_ERROR;
538 goto end;
539 } else if (validation_status == BT_PARAM_VALIDATION_STATUS_VALIDATION_ERROR) {
540 status = BT_COMPONENT_CLASS_INITIALIZE_METHOD_STATUS_ERROR;
541 BT_COMP_LOGE_APPEND_CAUSE(trimmer_comp->self_comp, "%s",
542 validate_error);
543 goto end;
544 }
545
546 BT_ASSERT(params);
547 value = bt_value_map_borrow_entry_value_const(params, "gmt");
548 if (value) {
549 trimmer_comp->is_gmt = (bool) bt_value_bool_get(value);
550 }
551
552 value = bt_value_map_borrow_entry_value_const(params, "begin");
553 if (value) {
554 if (set_bound_from_param(trimmer_comp, "begin", value,
555 &trimmer_comp->begin, trimmer_comp->is_gmt)) {
556 /* set_bound_from_param() logs errors */
557 status = BT_COMPONENT_CLASS_INITIALIZE_METHOD_STATUS_ERROR;
558 goto end;
559 }
560 } else {
561 trimmer_comp->begin.is_infinite = true;
562 trimmer_comp->begin.is_set = true;
563 }
564
565 value = bt_value_map_borrow_entry_value_const(params, "end");
566 if (value) {
567 if (set_bound_from_param(trimmer_comp, "end", value,
568 &trimmer_comp->end, trimmer_comp->is_gmt)) {
569 /* set_bound_from_param() logs errors */
570 status = BT_COMPONENT_CLASS_INITIALIZE_METHOD_STATUS_ERROR;
571 goto end;
572 }
573 } else {
574 trimmer_comp->end.is_infinite = true;
575 trimmer_comp->end.is_set = true;
576 }
577
578 if (trimmer_comp->begin.is_set && trimmer_comp->end.is_set) {
579 /* validate_trimmer_bounds() logs errors */
580 if (validate_trimmer_bounds(trimmer_comp,
581 &trimmer_comp->begin, &trimmer_comp->end)) {
582 status = BT_COMPONENT_CLASS_INITIALIZE_METHOD_STATUS_ERROR;
583 goto end;
584 }
585 }
586
587 status = BT_COMPONENT_CLASS_INITIALIZE_METHOD_STATUS_OK;
588
589 end:
590 g_free(validate_error);
591
592 return status;
593 }
594
595 bt_component_class_initialize_method_status trimmer_init(
596 bt_self_component_filter *self_comp_flt,
597 bt_self_component_filter_configuration *config,
598 const bt_value *params, void *init_data)
599 {
600 bt_component_class_initialize_method_status status;
601 bt_self_component_add_port_status add_port_status;
602 struct trimmer_comp *trimmer_comp = create_trimmer_comp();
603 bt_self_component *self_comp =
604 bt_self_component_filter_as_self_component(self_comp_flt);
605
606 if (!trimmer_comp) {
607 status = BT_COMPONENT_CLASS_INITIALIZE_METHOD_STATUS_MEMORY_ERROR;
608 goto error;
609 }
610
611 trimmer_comp->log_level = bt_component_get_logging_level(
612 bt_self_component_as_component(self_comp));
613 trimmer_comp->self_comp = self_comp;
614 trimmer_comp->self_comp_filter = self_comp_flt;
615
616 add_port_status = bt_self_component_filter_add_input_port(
617 self_comp_flt, in_port_name, NULL, NULL);
618 if (add_port_status != BT_SELF_COMPONENT_ADD_PORT_STATUS_OK) {
619 status = (int) add_port_status;
620 goto error;
621 }
622
623 add_port_status = bt_self_component_filter_add_output_port(
624 self_comp_flt, "out", NULL, NULL);
625 if (add_port_status != BT_SELF_COMPONENT_ADD_PORT_STATUS_OK) {
626 status = (int) add_port_status;
627 goto error;
628 }
629
630 status = init_trimmer_comp_from_params(trimmer_comp, params);
631 if (status != BT_COMPONENT_CLASS_INITIALIZE_METHOD_STATUS_OK) {
632 goto error;
633 }
634
635 bt_self_component_set_data(self_comp, trimmer_comp);
636
637 status = BT_COMPONENT_CLASS_INITIALIZE_METHOD_STATUS_OK;
638 goto end;
639
640 error:
641 if (trimmer_comp) {
642 destroy_trimmer_comp(trimmer_comp);
643 }
644
645 end:
646 return status;
647 }
648
649 static
650 void destroy_trimmer_iterator(struct trimmer_iterator *trimmer_it)
651 {
652 if (!trimmer_it) {
653 goto end;
654 }
655
656 bt_message_iterator_put_ref(
657 trimmer_it->upstream_iter);
658
659 if (trimmer_it->output_messages) {
660 g_queue_free(trimmer_it->output_messages);
661 }
662
663 if (trimmer_it->stream_states) {
664 g_hash_table_destroy(trimmer_it->stream_states);
665 }
666
667 g_free(trimmer_it);
668 end:
669 return;
670 }
671
672 static
673 void destroy_trimmer_iterator_stream_state(
674 struct trimmer_iterator_stream_state *sstate)
675 {
676 BT_ASSERT(sstate);
677 BT_PACKET_PUT_REF_AND_RESET(sstate->cur_packet);
678 g_free(sstate);
679 }
680
681 bt_message_iterator_class_initialize_method_status trimmer_msg_iter_init(
682 bt_self_message_iterator *self_msg_iter,
683 bt_self_message_iterator_configuration *config,
684 bt_self_component_port_output *port)
685 {
686 bt_message_iterator_class_initialize_method_status status;
687 bt_message_iterator_create_from_message_iterator_status
688 msg_iter_status;
689 struct trimmer_iterator *trimmer_it;
690 bt_self_component *self_comp =
691 bt_self_message_iterator_borrow_component(self_msg_iter);
692
693 trimmer_it = g_new0(struct trimmer_iterator, 1);
694 if (!trimmer_it) {
695 status = BT_MESSAGE_ITERATOR_CLASS_INITIALIZE_METHOD_STATUS_MEMORY_ERROR;
696 goto error;
697 }
698
699 trimmer_it->trimmer_comp = bt_self_component_get_data(self_comp);
700 BT_ASSERT(trimmer_it->trimmer_comp);
701
702 if (trimmer_it->trimmer_comp->begin.is_set &&
703 trimmer_it->trimmer_comp->end.is_set) {
704 /*
705 * Both trimming time range's bounds are set, so skip
706 * the
707 * `TRIMMER_ITERATOR_STATE_SET_BOUNDS_NS_FROM_ORIGIN`
708 * phase.
709 */
710 trimmer_it->state = TRIMMER_ITERATOR_STATE_SEEK_INITIALLY;
711 }
712
713 trimmer_it->begin = trimmer_it->trimmer_comp->begin;
714 trimmer_it->end = trimmer_it->trimmer_comp->end;
715 msg_iter_status =
716 bt_message_iterator_create_from_message_iterator(
717 self_msg_iter,
718 bt_self_component_filter_borrow_input_port_by_name(
719 trimmer_it->trimmer_comp->self_comp_filter, in_port_name),
720 &trimmer_it->upstream_iter);
721 if (msg_iter_status != BT_MESSAGE_ITERATOR_CREATE_FROM_MESSAGE_ITERATOR_STATUS_OK) {
722 status = (int) msg_iter_status;
723 goto error;
724 }
725
726 trimmer_it->output_messages = g_queue_new();
727 if (!trimmer_it->output_messages) {
728 status = BT_MESSAGE_ITERATOR_CLASS_INITIALIZE_METHOD_STATUS_MEMORY_ERROR;
729 goto error;
730 }
731
732 trimmer_it->stream_states = g_hash_table_new_full(g_direct_hash,
733 g_direct_equal, NULL,
734 (GDestroyNotify) destroy_trimmer_iterator_stream_state);
735 if (!trimmer_it->stream_states) {
736 status = BT_MESSAGE_ITERATOR_CLASS_INITIALIZE_METHOD_STATUS_MEMORY_ERROR;
737 goto error;
738 }
739
740 /*
741 * The trimmer requires upstream messages to have times, so it can
742 * always seek forward.
743 */
744 bt_self_message_iterator_configuration_set_can_seek_forward(
745 config, BT_TRUE);
746
747 trimmer_it->self_msg_iter = self_msg_iter;
748 bt_self_message_iterator_set_data(self_msg_iter, trimmer_it);
749
750 status = BT_MESSAGE_ITERATOR_CLASS_INITIALIZE_METHOD_STATUS_OK;
751 goto end;
752
753 error:
754 destroy_trimmer_iterator(trimmer_it);
755
756 end:
757 return status;
758 }
759
760 static inline
761 int get_msg_ns_from_origin(const bt_message *msg, int64_t *ns_from_origin,
762 bool *has_clock_snapshot)
763 {
764 const bt_clock_class *clock_class = NULL;
765 const bt_clock_snapshot *clock_snapshot = NULL;
766 int ret = 0;
767
768 BT_ASSERT_DBG(msg);
769 BT_ASSERT_DBG(ns_from_origin);
770 BT_ASSERT_DBG(has_clock_snapshot);
771
772 switch (bt_message_get_type(msg)) {
773 case BT_MESSAGE_TYPE_EVENT:
774 clock_class =
775 bt_message_event_borrow_stream_class_default_clock_class_const(
776 msg);
777 if (G_UNLIKELY(!clock_class)) {
778 goto error;
779 }
780
781 clock_snapshot = bt_message_event_borrow_default_clock_snapshot_const(
782 msg);
783 break;
784 case BT_MESSAGE_TYPE_PACKET_BEGINNING:
785 {
786 const bt_packet *packet = bt_message_packet_beginning_borrow_packet_const(msg);
787 const bt_stream *stream = bt_packet_borrow_stream_const(packet);
788 const bt_stream_class *stream_class = bt_stream_borrow_class_const(stream);
789
790 if (!bt_stream_class_packets_have_beginning_default_clock_snapshot(stream_class)) {
791 goto no_clock_snapshot;
792 }
793
794 clock_snapshot = bt_message_packet_beginning_borrow_default_clock_snapshot_const(
795 msg);
796 break;
797 }
798 case BT_MESSAGE_TYPE_PACKET_END:
799 {
800 const bt_packet *packet = bt_message_packet_end_borrow_packet_const(msg);
801 const bt_stream *stream = bt_packet_borrow_stream_const(packet);
802 const bt_stream_class *stream_class = bt_stream_borrow_class_const(stream);
803
804 if (!bt_stream_class_packets_have_end_default_clock_snapshot(stream_class)) {
805 goto no_clock_snapshot;
806 }
807
808 clock_snapshot = bt_message_packet_end_borrow_default_clock_snapshot_const(
809 msg);
810 break;
811 }
812 case BT_MESSAGE_TYPE_STREAM_BEGINNING:
813 {
814 enum bt_message_stream_clock_snapshot_state cs_state;
815
816 clock_class =
817 bt_message_stream_beginning_borrow_stream_class_default_clock_class_const(msg);
818 if (G_UNLIKELY(!clock_class)) {
819 goto error;
820 }
821
822 cs_state = bt_message_stream_beginning_borrow_default_clock_snapshot_const(msg, &clock_snapshot);
823 if (cs_state != BT_MESSAGE_STREAM_CLOCK_SNAPSHOT_STATE_KNOWN) {
824 goto no_clock_snapshot;
825 }
826
827 break;
828 }
829 case BT_MESSAGE_TYPE_STREAM_END:
830 {
831 enum bt_message_stream_clock_snapshot_state cs_state;
832
833 clock_class =
834 bt_message_stream_end_borrow_stream_class_default_clock_class_const(msg);
835 if (G_UNLIKELY(!clock_class)) {
836 goto error;
837 }
838
839 cs_state = bt_message_stream_end_borrow_default_clock_snapshot_const(msg, &clock_snapshot);
840 if (cs_state != BT_MESSAGE_STREAM_CLOCK_SNAPSHOT_STATE_KNOWN) {
841 goto no_clock_snapshot;
842 }
843
844 break;
845 }
846 case BT_MESSAGE_TYPE_DISCARDED_EVENTS:
847 {
848 const bt_stream *stream = bt_message_discarded_events_borrow_stream_const(msg);
849 const bt_stream_class *stream_class = bt_stream_borrow_class_const(stream);
850
851 if (!bt_stream_class_discarded_events_have_default_clock_snapshots(stream_class)) {
852 goto no_clock_snapshot;
853 }
854
855 clock_snapshot = bt_message_discarded_events_borrow_beginning_default_clock_snapshot_const(
856 msg);
857 break;
858 }
859 case BT_MESSAGE_TYPE_DISCARDED_PACKETS:
860 {
861 const bt_stream *stream = bt_message_discarded_packets_borrow_stream_const(msg);
862 const bt_stream_class *stream_class = bt_stream_borrow_class_const(stream);
863
864 if (!bt_stream_class_discarded_packets_have_default_clock_snapshots(stream_class)) {
865 goto no_clock_snapshot;
866 }
867
868 clock_snapshot = bt_message_discarded_packets_borrow_beginning_default_clock_snapshot_const(
869 msg);
870 break;
871 }
872 case BT_MESSAGE_TYPE_MESSAGE_ITERATOR_INACTIVITY:
873 clock_snapshot =
874 bt_message_message_iterator_inactivity_borrow_clock_snapshot_const(
875 msg);
876 break;
877 default:
878 goto no_clock_snapshot;
879 }
880
881 ret = bt_clock_snapshot_get_ns_from_origin(clock_snapshot,
882 ns_from_origin);
883 if (G_UNLIKELY(ret)) {
884 goto error;
885 }
886
887 *has_clock_snapshot = true;
888 goto end;
889
890 no_clock_snapshot:
891 *has_clock_snapshot = false;
892 goto end;
893
894 error:
895 ret = -1;
896
897 end:
898 return ret;
899 }
900
901 static inline
902 void put_messages(bt_message_array_const msgs, uint64_t count)
903 {
904 uint64_t i;
905
906 for (i = 0; i < count; i++) {
907 BT_MESSAGE_PUT_REF_AND_RESET(msgs[i]);
908 }
909 }
910
911 static inline
912 int set_trimmer_iterator_bound(struct trimmer_iterator *trimmer_it,
913 struct trimmer_bound *bound, int64_t ns_from_origin,
914 bool is_gmt)
915 {
916 struct trimmer_comp *trimmer_comp = trimmer_it->trimmer_comp;
917 struct tm tm;
918 struct tm *res;
919 time_t time_seconds = (time_t) (ns_from_origin / NS_PER_S);
920 int ret = 0;
921
922 BT_ASSERT(!bound->is_set);
923 errno = 0;
924
925 /* We only need to extract the date from this time */
926 if (is_gmt) {
927 res = bt_gmtime_r(&time_seconds, &tm);
928 } else {
929 res = bt_localtime_r(&time_seconds, &tm);
930 }
931
932 if (!res) {
933 BT_COMP_LOGE_APPEND_CAUSE_ERRNO(trimmer_comp->self_comp,
934 "Cannot convert timestamp to date and time",
935 ": ts=%" PRId64, (int64_t) time_seconds);
936 ret = -1;
937 goto end;
938 }
939
940 ret = set_bound_ns_from_origin(bound, tm.tm_year + 1900, tm.tm_mon + 1,
941 tm.tm_mday, bound->time.hour, bound->time.minute,
942 bound->time.second, bound->time.ns, is_gmt);
943
944 end:
945 return ret;
946 }
947
948 static
949 bt_message_iterator_class_next_method_status
950 state_set_trimmer_iterator_bounds(
951 struct trimmer_iterator *trimmer_it)
952 {
953 bt_message_iterator_next_status upstream_iter_status =
954 BT_MESSAGE_ITERATOR_NEXT_STATUS_OK;
955 struct trimmer_comp *trimmer_comp = trimmer_it->trimmer_comp;
956 bt_message_array_const msgs;
957 uint64_t count = 0;
958 int64_t ns_from_origin = INT64_MIN;
959 uint64_t i;
960 int ret;
961
962 BT_ASSERT(!trimmer_it->begin.is_set ||
963 !trimmer_it->end.is_set);
964
965 while (true) {
966 upstream_iter_status =
967 bt_message_iterator_next(
968 trimmer_it->upstream_iter, &msgs, &count);
969 if (upstream_iter_status != BT_MESSAGE_ITERATOR_NEXT_STATUS_OK) {
970 goto end;
971 }
972
973 for (i = 0; i < count; i++) {
974 const bt_message *msg = msgs[i];
975 bool has_ns_from_origin;
976 ret = get_msg_ns_from_origin(msg, &ns_from_origin,
977 &has_ns_from_origin);
978 if (ret) {
979 goto error;
980 }
981
982 if (!has_ns_from_origin) {
983 continue;
984 }
985
986 BT_ASSERT_DBG(ns_from_origin != INT64_MIN &&
987 ns_from_origin != INT64_MAX);
988 put_messages(msgs, count);
989 goto found;
990 }
991
992 put_messages(msgs, count);
993 }
994
995 found:
996 if (!trimmer_it->begin.is_set) {
997 BT_ASSERT(!trimmer_it->begin.is_infinite);
998 ret = set_trimmer_iterator_bound(trimmer_it, &trimmer_it->begin,
999 ns_from_origin, trimmer_comp->is_gmt);
1000 if (ret) {
1001 goto error;
1002 }
1003 }
1004
1005 if (!trimmer_it->end.is_set) {
1006 BT_ASSERT(!trimmer_it->end.is_infinite);
1007 ret = set_trimmer_iterator_bound(trimmer_it, &trimmer_it->end,
1008 ns_from_origin, trimmer_comp->is_gmt);
1009 if (ret) {
1010 goto error;
1011 }
1012 }
1013
1014 ret = validate_trimmer_bounds(trimmer_it->trimmer_comp,
1015 &trimmer_it->begin, &trimmer_it->end);
1016 if (ret) {
1017 goto error;
1018 }
1019
1020 goto end;
1021
1022 error:
1023 put_messages(msgs, count);
1024 upstream_iter_status = BT_MESSAGE_ITERATOR_NEXT_STATUS_ERROR;
1025
1026 end:
1027 return (int) upstream_iter_status;
1028 }
1029
1030 static
1031 bt_message_iterator_class_next_method_status state_seek_initially(
1032 struct trimmer_iterator *trimmer_it)
1033 {
1034 struct trimmer_comp *trimmer_comp = trimmer_it->trimmer_comp;
1035 bt_message_iterator_class_next_method_status status;
1036
1037 BT_ASSERT(trimmer_it->begin.is_set);
1038
1039 if (trimmer_it->begin.is_infinite) {
1040 bt_bool can_seek;
1041
1042 status = (int) bt_message_iterator_can_seek_beginning(
1043 trimmer_it->upstream_iter, &can_seek);
1044 if (status != BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_OK) {
1045 if (status < 0) {
1046 BT_COMP_LOGE_APPEND_CAUSE(trimmer_comp->self_comp,
1047 "Cannot make upstream message iterator initially seek its beginning.");
1048 }
1049
1050 goto end;
1051 }
1052
1053 if (!can_seek) {
1054 BT_COMP_LOGE_APPEND_CAUSE(trimmer_comp->self_comp,
1055 "Cannot make upstream message iterator initially seek its beginning.");
1056 status = BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_ERROR;
1057 goto end;
1058 }
1059
1060 status = (int) bt_message_iterator_seek_beginning(
1061 trimmer_it->upstream_iter);
1062 } else {
1063 bt_bool can_seek;
1064
1065 status = (int) bt_message_iterator_can_seek_ns_from_origin(
1066 trimmer_it->upstream_iter, trimmer_it->begin.ns_from_origin,
1067 &can_seek);
1068
1069 if (status != BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_OK) {
1070 if (status < 0) {
1071 BT_COMP_LOGE_APPEND_CAUSE(trimmer_comp->self_comp,
1072 "Cannot make upstream message iterator initially seek: seek-ns-from-origin=%" PRId64,
1073 trimmer_it->begin.ns_from_origin);
1074 }
1075
1076 goto end;
1077 }
1078
1079 if (!can_seek) {
1080 BT_COMP_LOGE_APPEND_CAUSE(trimmer_comp->self_comp,
1081 "Cannot make upstream message iterator initially seek: seek-ns-from-origin=%" PRId64,
1082 trimmer_it->begin.ns_from_origin);
1083 status = BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_ERROR;
1084 goto end;
1085 }
1086
1087 status = (int) bt_message_iterator_seek_ns_from_origin(
1088 trimmer_it->upstream_iter, trimmer_it->begin.ns_from_origin);
1089 }
1090
1091 if (status == BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_OK) {
1092 trimmer_it->state = TRIMMER_ITERATOR_STATE_TRIM;
1093 }
1094
1095 end:
1096 return status;
1097 }
1098
1099 static inline
1100 void push_message(struct trimmer_iterator *trimmer_it, const bt_message *msg)
1101 {
1102 g_queue_push_head(trimmer_it->output_messages, (void *) msg);
1103 }
1104
1105 static inline
1106 const bt_message *pop_message(struct trimmer_iterator *trimmer_it)
1107 {
1108 return g_queue_pop_tail(trimmer_it->output_messages);
1109 }
1110
1111 static inline
1112 int clock_raw_value_from_ns_from_origin(const bt_clock_class *clock_class,
1113 int64_t ns_from_origin, uint64_t *raw_value)
1114 {
1115
1116 int64_t cc_offset_s;
1117 uint64_t cc_offset_cycles;
1118 uint64_t cc_freq;
1119
1120 bt_clock_class_get_offset(clock_class, &cc_offset_s, &cc_offset_cycles);
1121 cc_freq = bt_clock_class_get_frequency(clock_class);
1122 return bt_common_clock_value_from_ns_from_origin(cc_offset_s,
1123 cc_offset_cycles, cc_freq, ns_from_origin, raw_value);
1124 }
1125
1126 static inline
1127 bt_message_iterator_class_next_method_status
1128 end_stream(struct trimmer_iterator *trimmer_it,
1129 struct trimmer_iterator_stream_state *sstate)
1130 {
1131 bt_message_iterator_class_next_method_status status =
1132 BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_OK;
1133 /* Initialize to silence maybe-uninitialized warning. */
1134 uint64_t raw_value = 0;
1135 bt_message *msg = NULL;
1136
1137 BT_ASSERT(!trimmer_it->end.is_infinite);
1138 BT_ASSERT(sstate->stream);
1139
1140 /*
1141 * If we haven't seen a message with a clock snapshot, we don't know if the trimmer's end bound is within
1142 * the clock's range, so it wouldn't be safe to try to convert ns_from_origin to a clock value.
1143 *
1144 * Also, it would be a bit of a lie to generate a stream end message with the end bound as its
1145 * clock snapshot, because we don't really know if the stream existed at that time. If we have
1146 * seen a message with a clock snapshot and the stream is cut short by another message with a
1147 * clock snapshot, then we are sure that the the end bound time is not below the clock range,
1148 * and we know the stream was active at that time (and that we cut it short).
1149 */
1150 if (sstate->seen_clock_snapshot) {
1151 const bt_clock_class *clock_class;
1152 int ret;
1153
1154 clock_class = bt_stream_class_borrow_default_clock_class_const(
1155 bt_stream_borrow_class_const(sstate->stream));
1156 BT_ASSERT(clock_class);
1157 ret = clock_raw_value_from_ns_from_origin(clock_class,
1158 trimmer_it->end.ns_from_origin, &raw_value);
1159 if (ret) {
1160 status = BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_ERROR;
1161 goto end;
1162 }
1163 }
1164
1165 if (sstate->cur_packet) {
1166 /*
1167 * Create and push a packet end message, making its time
1168 * the trimming range's end time.
1169 *
1170 * We know that we must have seen a clock snapshot, the one in
1171 * the packet beginning message, since trimmer currently
1172 * requires packet messages to have clock snapshots (see comment
1173 * in create_stream_state_entry).
1174 */
1175 BT_ASSERT(sstate->seen_clock_snapshot);
1176
1177 msg = bt_message_packet_end_create_with_default_clock_snapshot(
1178 trimmer_it->self_msg_iter, sstate->cur_packet,
1179 raw_value);
1180 if (!msg) {
1181 status = BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_MEMORY_ERROR;
1182 goto end;
1183 }
1184
1185 push_message(trimmer_it, msg);
1186 msg = NULL;
1187 BT_PACKET_PUT_REF_AND_RESET(sstate->cur_packet);
1188 }
1189
1190 /* Create and push a stream end message. */
1191 msg = bt_message_stream_end_create(trimmer_it->self_msg_iter,
1192 sstate->stream);
1193 if (!msg) {
1194 status = BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_MEMORY_ERROR;
1195 goto end;
1196 }
1197
1198 if (sstate->seen_clock_snapshot) {
1199 bt_message_stream_end_set_default_clock_snapshot(msg, raw_value);
1200 }
1201
1202 push_message(trimmer_it, msg);
1203 msg = NULL;
1204
1205 /*
1206 * Just to make sure that we don't use this stream state again
1207 * in the future without an obvious error.
1208 */
1209 sstate->stream = NULL;
1210
1211 end:
1212 bt_message_put_ref(msg);
1213 return status;
1214 }
1215
1216 static inline
1217 bt_message_iterator_class_next_method_status end_iterator_streams(
1218 struct trimmer_iterator *trimmer_it)
1219 {
1220 bt_message_iterator_class_next_method_status status =
1221 BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_OK;
1222 GHashTableIter iter;
1223 gpointer key, sstate;
1224
1225 if (trimmer_it->end.is_infinite) {
1226 /*
1227 * An infinite trimming range's end time guarantees that
1228 * we received (and pushed) all the appropriate end
1229 * messages.
1230 */
1231 goto remove_all;
1232 }
1233
1234 /*
1235 * End each stream and then remove them from the hash table of
1236 * stream states to release unneeded references.
1237 */
1238 g_hash_table_iter_init(&iter, trimmer_it->stream_states);
1239
1240 while (g_hash_table_iter_next(&iter, &key, &sstate)) {
1241 status = end_stream(trimmer_it, sstate);
1242 if (status) {
1243 goto end;
1244 }
1245 }
1246
1247 remove_all:
1248 g_hash_table_remove_all(trimmer_it->stream_states);
1249
1250 end:
1251 return status;
1252 }
1253
1254 static
1255 bt_message_iterator_class_next_method_status
1256 create_stream_state_entry(
1257 struct trimmer_iterator *trimmer_it,
1258 const struct bt_stream *stream,
1259 struct trimmer_iterator_stream_state **stream_state)
1260 {
1261 struct trimmer_comp *trimmer_comp = trimmer_it->trimmer_comp;
1262 bt_message_iterator_class_next_method_status status;
1263 struct trimmer_iterator_stream_state *sstate;
1264 const bt_stream_class *sc;
1265
1266 BT_ASSERT(!bt_g_hash_table_contains(trimmer_it->stream_states, stream));
1267
1268 /*
1269 * Validate right now that the stream's class
1270 * has a registered default clock class so that
1271 * an existing stream state guarantees existing
1272 * default clock snapshots for its associated
1273 * messages.
1274 *
1275 * Also check that clock snapshots are always
1276 * known.
1277 */
1278 sc = bt_stream_borrow_class_const(stream);
1279 if (!bt_stream_class_borrow_default_clock_class_const(sc)) {
1280 BT_COMP_LOGE_APPEND_CAUSE(trimmer_comp->self_comp,
1281 "Unsupported stream: stream class does "
1282 "not have a default clock class: "
1283 "stream-addr=%p, "
1284 "stream-id=%" PRIu64 ", "
1285 "stream-name=\"%s\"",
1286 stream, bt_stream_get_id(stream),
1287 bt_stream_get_name(stream));
1288 status = BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_ERROR;
1289 goto end;
1290 }
1291
1292 /*
1293 * Temporary: make sure packet beginning, packet
1294 * end, discarded events, and discarded packets
1295 * messages have default clock snapshots until
1296 * the support for not having them is
1297 * implemented.
1298 */
1299 if (bt_stream_class_supports_packets(sc)) {
1300 if (!bt_stream_class_packets_have_beginning_default_clock_snapshot(
1301 sc)) {
1302 BT_COMP_LOGE_APPEND_CAUSE(trimmer_comp->self_comp,
1303 "Unsupported stream: packets have no beginning clock snapshot: "
1304 "stream-addr=%p, "
1305 "stream-id=%" PRIu64 ", "
1306 "stream-name=\"%s\"",
1307 stream, bt_stream_get_id(stream),
1308 bt_stream_get_name(stream));
1309 status = BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_ERROR;
1310 goto end;
1311 }
1312
1313 if (!bt_stream_class_packets_have_end_default_clock_snapshot(
1314 sc)) {
1315 BT_COMP_LOGE_APPEND_CAUSE(trimmer_comp->self_comp,
1316 "Unsupported stream: packets have no end clock snapshot: "
1317 "stream-addr=%p, "
1318 "stream-id=%" PRIu64 ", "
1319 "stream-name=\"%s\"",
1320 stream, bt_stream_get_id(stream),
1321 bt_stream_get_name(stream));
1322 status = BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_ERROR;
1323 goto end;
1324 }
1325
1326 if (bt_stream_class_supports_discarded_packets(sc) &&
1327 !bt_stream_class_discarded_packets_have_default_clock_snapshots(sc)) {
1328 BT_COMP_LOGE_APPEND_CAUSE(trimmer_comp->self_comp,
1329 "Unsupported stream: discarded packets "
1330 "have no clock snapshots: "
1331 "stream-addr=%p, "
1332 "stream-id=%" PRIu64 ", "
1333 "stream-name=\"%s\"",
1334 stream, bt_stream_get_id(stream),
1335 bt_stream_get_name(stream));
1336 status = BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_ERROR;
1337 goto end;
1338 }
1339 }
1340
1341 if (bt_stream_class_supports_discarded_events(sc) &&
1342 !bt_stream_class_discarded_events_have_default_clock_snapshots(sc)) {
1343 BT_COMP_LOGE_APPEND_CAUSE(trimmer_comp->self_comp,
1344 "Unsupported stream: discarded events have no clock snapshots: "
1345 "stream-addr=%p, "
1346 "stream-id=%" PRIu64 ", "
1347 "stream-name=\"%s\"",
1348 stream, bt_stream_get_id(stream),
1349 bt_stream_get_name(stream));
1350 status = BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_ERROR;
1351 goto end;
1352 }
1353
1354 sstate = g_new0(struct trimmer_iterator_stream_state, 1);
1355 if (!sstate) {
1356 status = BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_MEMORY_ERROR;
1357 goto end;
1358 }
1359
1360 sstate->stream = stream;
1361
1362 g_hash_table_insert(trimmer_it->stream_states, (void *) stream, sstate);
1363
1364 *stream_state = sstate;
1365
1366 status = BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_OK;
1367
1368 end:
1369 return status;
1370 }
1371
1372 static
1373 struct trimmer_iterator_stream_state *get_stream_state_entry(
1374 struct trimmer_iterator *trimmer_it,
1375 const struct bt_stream *stream)
1376 {
1377 struct trimmer_iterator_stream_state *sstate;
1378
1379 BT_ASSERT_DBG(stream);
1380 sstate = g_hash_table_lookup(trimmer_it->stream_states, stream);
1381 BT_ASSERT_DBG(sstate);
1382
1383 return sstate;
1384 }
1385
1386 /*
1387 * Handles a message which is associated to a given stream state. This
1388 * _could_ make the iterator's output message queue grow; this could
1389 * also consume the message without pushing anything to this queue, only
1390 * modifying the stream state.
1391 *
1392 * This function consumes the `msg` reference, _whatever the outcome_.
1393 *
1394 * If non-NULL, `ns_from_origin` is the message's time, as given by
1395 * get_msg_ns_from_origin(). If NULL, the message doesn't have a time.
1396 *
1397 * This function sets `reached_end` if handling this message made the
1398 * iterator reach the end of the trimming range. Note that the output
1399 * message queue could contain messages even if this function sets
1400 * `reached_end`.
1401 */
1402 static
1403 bt_message_iterator_class_next_method_status
1404 handle_message_with_stream(
1405 struct trimmer_iterator *trimmer_it, const bt_message *msg,
1406 const struct bt_stream *stream, const int64_t *ns_from_origin,
1407 bool *reached_end)
1408 {
1409 bt_message_iterator_class_next_method_status status =
1410 BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_OK;
1411 bt_message_type msg_type = bt_message_get_type(msg);
1412 int ret;
1413 struct trimmer_iterator_stream_state *sstate = NULL;
1414
1415 /*
1416 * Retrieve the stream's state - except if the message is stream
1417 * beginning, in which case we don't know about about this stream yet.
1418 */
1419 if (msg_type != BT_MESSAGE_TYPE_STREAM_BEGINNING) {
1420 sstate = get_stream_state_entry(trimmer_it, stream);
1421 }
1422
1423 switch (msg_type) {
1424 case BT_MESSAGE_TYPE_EVENT:
1425 /*
1426 * Event messages always have a clock snapshot if the stream
1427 * class has a clock class. And we know it has, otherwise we
1428 * couldn't be using the trimmer component.
1429 */
1430 BT_ASSERT_DBG(ns_from_origin);
1431
1432 if (G_UNLIKELY(!trimmer_it->end.is_infinite &&
1433 *ns_from_origin > trimmer_it->end.ns_from_origin)) {
1434 status = end_iterator_streams(trimmer_it);
1435 *reached_end = true;
1436 break;
1437 }
1438
1439 sstate->seen_clock_snapshot = true;
1440
1441 push_message(trimmer_it, msg);
1442 msg = NULL;
1443 break;
1444
1445 case BT_MESSAGE_TYPE_PACKET_BEGINNING:
1446 /*
1447 * Packet beginning messages won't have a clock snapshot if
1448 * stream_class->packets_have_beginning_default_clock_snapshot
1449 * is false. But for now, assume they always do.
1450 */
1451 BT_ASSERT(ns_from_origin);
1452 BT_ASSERT(!sstate->cur_packet);
1453
1454 if (G_UNLIKELY(!trimmer_it->end.is_infinite &&
1455 *ns_from_origin > trimmer_it->end.ns_from_origin)) {
1456 status = end_iterator_streams(trimmer_it);
1457 *reached_end = true;
1458 break;
1459 }
1460
1461 sstate->cur_packet =
1462 bt_message_packet_beginning_borrow_packet_const(msg);
1463 bt_packet_get_ref(sstate->cur_packet);
1464
1465 sstate->seen_clock_snapshot = true;
1466
1467 push_message(trimmer_it, msg);
1468 msg = NULL;
1469 break;
1470
1471 case BT_MESSAGE_TYPE_PACKET_END:
1472 /*
1473 * Packet end messages won't have a clock snapshot if
1474 * stream_class->packets_have_end_default_clock_snapshot
1475 * is false. But for now, assume they always do.
1476 */
1477 BT_ASSERT(ns_from_origin);
1478 BT_ASSERT(sstate->cur_packet);
1479
1480 if (G_UNLIKELY(!trimmer_it->end.is_infinite &&
1481 *ns_from_origin > trimmer_it->end.ns_from_origin)) {
1482 status = end_iterator_streams(trimmer_it);
1483 *reached_end = true;
1484 break;
1485 }
1486
1487 BT_PACKET_PUT_REF_AND_RESET(sstate->cur_packet);
1488
1489 sstate->seen_clock_snapshot = true;
1490
1491 push_message(trimmer_it, msg);
1492 msg = NULL;
1493 break;
1494
1495 case BT_MESSAGE_TYPE_DISCARDED_EVENTS:
1496 case BT_MESSAGE_TYPE_DISCARDED_PACKETS:
1497 {
1498 /*
1499 * `ns_from_origin` is the message's time range's
1500 * beginning time here.
1501 */
1502 int64_t end_ns_from_origin;
1503 const bt_clock_snapshot *end_cs;
1504
1505 BT_ASSERT(ns_from_origin);
1506
1507 sstate->seen_clock_snapshot = true;
1508
1509 if (bt_message_get_type(msg) ==
1510 BT_MESSAGE_TYPE_DISCARDED_EVENTS) {
1511 /*
1512 * Safe to ignore the return value because we
1513 * know there's a default clock and it's always
1514 * known.
1515 */
1516 end_cs = bt_message_discarded_events_borrow_end_default_clock_snapshot_const(
1517 msg);
1518 } else {
1519 /*
1520 * Safe to ignore the return value because we
1521 * know there's a default clock and it's always
1522 * known.
1523 */
1524 end_cs = bt_message_discarded_packets_borrow_end_default_clock_snapshot_const(
1525 msg);
1526 }
1527
1528 if (bt_clock_snapshot_get_ns_from_origin(end_cs,
1529 &end_ns_from_origin)) {
1530 status = BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_ERROR;
1531 goto end;
1532 }
1533
1534 if (!trimmer_it->end.is_infinite &&
1535 *ns_from_origin > trimmer_it->end.ns_from_origin) {
1536 status = end_iterator_streams(trimmer_it);
1537 *reached_end = true;
1538 break;
1539 }
1540
1541 if (!trimmer_it->end.is_infinite &&
1542 end_ns_from_origin > trimmer_it->end.ns_from_origin) {
1543 /*
1544 * This message's end time is outside the
1545 * trimming time range: replace it with a new
1546 * message having an end time equal to the
1547 * trimming time range's end and without a
1548 * count.
1549 */
1550 const bt_clock_class *clock_class =
1551 bt_clock_snapshot_borrow_clock_class_const(
1552 end_cs);
1553 const bt_clock_snapshot *begin_cs;
1554 bt_message *new_msg;
1555 uint64_t end_raw_value;
1556
1557 ret = clock_raw_value_from_ns_from_origin(clock_class,
1558 trimmer_it->end.ns_from_origin, &end_raw_value);
1559 if (ret) {
1560 status = BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_ERROR;
1561 goto end;
1562 }
1563
1564 if (msg_type == BT_MESSAGE_TYPE_DISCARDED_EVENTS) {
1565 begin_cs = bt_message_discarded_events_borrow_beginning_default_clock_snapshot_const(
1566 msg);
1567 new_msg = bt_message_discarded_events_create_with_default_clock_snapshots(
1568 trimmer_it->self_msg_iter,
1569 sstate->stream,
1570 bt_clock_snapshot_get_value(begin_cs),
1571 end_raw_value);
1572 } else {
1573 begin_cs = bt_message_discarded_packets_borrow_beginning_default_clock_snapshot_const(
1574 msg);
1575 new_msg = bt_message_discarded_packets_create_with_default_clock_snapshots(
1576 trimmer_it->self_msg_iter,
1577 sstate->stream,
1578 bt_clock_snapshot_get_value(begin_cs),
1579 end_raw_value);
1580 }
1581
1582 if (!new_msg) {
1583 status = BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_MEMORY_ERROR;
1584 goto end;
1585 }
1586
1587 /* Replace the original message */
1588 BT_MESSAGE_MOVE_REF(msg, new_msg);
1589 }
1590
1591 push_message(trimmer_it, msg);
1592 msg = NULL;
1593 break;
1594 }
1595
1596 case BT_MESSAGE_TYPE_STREAM_BEGINNING:
1597 /*
1598 * If this message has a time and this time is greater than the
1599 * trimmer's end bound, it triggers the end of the trim window.
1600 */
1601 if (G_UNLIKELY(ns_from_origin && !trimmer_it->end.is_infinite &&
1602 *ns_from_origin > trimmer_it->end.ns_from_origin)) {
1603 status = end_iterator_streams(trimmer_it);
1604 *reached_end = true;
1605 break;
1606 }
1607
1608 /* Learn about this stream. */
1609 status = create_stream_state_entry(trimmer_it, stream, &sstate);
1610 if (status != BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_OK) {
1611 goto end;
1612 }
1613
1614 if (ns_from_origin) {
1615 sstate->seen_clock_snapshot = true;
1616 }
1617
1618 push_message(trimmer_it, msg);
1619 msg = NULL;
1620 break;
1621 case BT_MESSAGE_TYPE_STREAM_END:
1622 {
1623 gboolean removed;
1624
1625 /*
1626 * If this message has a time and this time is greater than the
1627 * trimmer's end bound, it triggers the end of the trim window.
1628 */
1629 if (G_UNLIKELY(ns_from_origin && !trimmer_it->end.is_infinite &&
1630 *ns_from_origin > trimmer_it->end.ns_from_origin)) {
1631 status = end_iterator_streams(trimmer_it);
1632 *reached_end = true;
1633 break;
1634 }
1635
1636 /*
1637 * Either the stream end message's time is within the trimmer's
1638 * bounds, or it doesn't have a time. In both cases, pass
1639 * the message unmodified.
1640 */
1641 push_message(trimmer_it, msg);
1642 msg = NULL;
1643
1644 /* Forget about this stream. */
1645 removed = g_hash_table_remove(trimmer_it->stream_states, sstate->stream);
1646 BT_ASSERT(removed);
1647 break;
1648 }
1649 default:
1650 break;
1651 }
1652
1653 end:
1654 /* We release the message's reference whatever the outcome */
1655 bt_message_put_ref(msg);
1656 return status;
1657 }
1658
1659 /*
1660 * Handles an input message. This _could_ make the iterator's output
1661 * message queue grow; this could also consume the message without
1662 * pushing anything to this queue, only modifying the stream state.
1663 *
1664 * This function consumes the `msg` reference, _whatever the outcome_.
1665 *
1666 * This function sets `reached_end` if handling this message made the
1667 * iterator reach the end of the trimming range. Note that the output
1668 * message queue could contain messages even if this function sets
1669 * `reached_end`.
1670 */
1671 static inline
1672 bt_message_iterator_class_next_method_status handle_message(
1673 struct trimmer_iterator *trimmer_it, const bt_message *msg,
1674 bool *reached_end)
1675 {
1676 bt_message_iterator_class_next_method_status status;
1677 const bt_stream *stream = NULL;
1678 int64_t ns_from_origin = INT64_MIN;
1679 bool has_ns_from_origin = false;
1680 int ret;
1681
1682 /* Find message's associated stream */
1683 switch (bt_message_get_type(msg)) {
1684 case BT_MESSAGE_TYPE_EVENT:
1685 stream = bt_event_borrow_stream_const(
1686 bt_message_event_borrow_event_const(msg));
1687 break;
1688 case BT_MESSAGE_TYPE_PACKET_BEGINNING:
1689 stream = bt_packet_borrow_stream_const(
1690 bt_message_packet_beginning_borrow_packet_const(msg));
1691 break;
1692 case BT_MESSAGE_TYPE_PACKET_END:
1693 stream = bt_packet_borrow_stream_const(
1694 bt_message_packet_end_borrow_packet_const(msg));
1695 break;
1696 case BT_MESSAGE_TYPE_DISCARDED_EVENTS:
1697 stream = bt_message_discarded_events_borrow_stream_const(msg);
1698 break;
1699 case BT_MESSAGE_TYPE_DISCARDED_PACKETS:
1700 stream = bt_message_discarded_packets_borrow_stream_const(msg);
1701 break;
1702 case BT_MESSAGE_TYPE_STREAM_BEGINNING:
1703 stream = bt_message_stream_beginning_borrow_stream_const(msg);
1704 break;
1705 case BT_MESSAGE_TYPE_STREAM_END:
1706 stream = bt_message_stream_end_borrow_stream_const(msg);
1707 break;
1708 default:
1709 break;
1710 }
1711
1712 /* Retrieve the message's time */
1713 ret = get_msg_ns_from_origin(msg, &ns_from_origin, &has_ns_from_origin);
1714 if (G_UNLIKELY(ret)) {
1715 status = BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_ERROR;
1716 goto end;
1717 }
1718
1719 if (G_LIKELY(stream)) {
1720 /* Message associated to a stream */
1721 status = handle_message_with_stream(trimmer_it, msg,
1722 stream, has_ns_from_origin ? &ns_from_origin : NULL, reached_end);
1723
1724 /*
1725 * handle_message_with_stream_state() unconditionally
1726 * consumes `msg`.
1727 */
1728 msg = NULL;
1729 } else {
1730 /*
1731 * Message not associated to a stream (message iterator
1732 * inactivity).
1733 */
1734 if (G_UNLIKELY(ns_from_origin > trimmer_it->end.ns_from_origin)) {
1735 BT_MESSAGE_PUT_REF_AND_RESET(msg);
1736 status = end_iterator_streams(trimmer_it);
1737 *reached_end = true;
1738 } else {
1739 push_message(trimmer_it, msg);
1740 status = BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_OK;
1741 msg = NULL;
1742 }
1743 }
1744
1745 end:
1746 /* We release the message's reference whatever the outcome */
1747 bt_message_put_ref(msg);
1748 return status;
1749 }
1750
1751 static inline
1752 void fill_message_array_from_output_messages(
1753 struct trimmer_iterator *trimmer_it,
1754 bt_message_array_const msgs, uint64_t capacity, uint64_t *count)
1755 {
1756 *count = 0;
1757
1758 /*
1759 * Move auto-seek messages to the output array (which is this
1760 * iterator's base message array).
1761 */
1762 while (capacity > 0 && !g_queue_is_empty(trimmer_it->output_messages)) {
1763 msgs[*count] = pop_message(trimmer_it);
1764 capacity--;
1765 (*count)++;
1766 }
1767
1768 BT_ASSERT_DBG(*count > 0);
1769 }
1770
1771 static inline
1772 bt_message_iterator_class_next_method_status state_ending(
1773 struct trimmer_iterator *trimmer_it,
1774 bt_message_array_const msgs, uint64_t capacity,
1775 uint64_t *count)
1776 {
1777 bt_message_iterator_class_next_method_status status =
1778 BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_OK;
1779
1780 if (g_queue_is_empty(trimmer_it->output_messages)) {
1781 trimmer_it->state = TRIMMER_ITERATOR_STATE_ENDED;
1782 status = BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_END;
1783 goto end;
1784 }
1785
1786 fill_message_array_from_output_messages(trimmer_it, msgs,
1787 capacity, count);
1788
1789 end:
1790 return status;
1791 }
1792
1793 static inline
1794 bt_message_iterator_class_next_method_status
1795 state_trim(struct trimmer_iterator *trimmer_it,
1796 bt_message_array_const msgs, uint64_t capacity,
1797 uint64_t *count)
1798 {
1799 bt_message_iterator_class_next_method_status status =
1800 BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_OK;
1801 bt_message_array_const my_msgs;
1802 uint64_t my_count;
1803 uint64_t i;
1804 bool reached_end = false;
1805
1806 while (g_queue_is_empty(trimmer_it->output_messages)) {
1807 status = (int) bt_message_iterator_next(
1808 trimmer_it->upstream_iter, &my_msgs, &my_count);
1809 if (G_UNLIKELY(status != BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_OK)) {
1810 if (status == BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_END) {
1811 status = end_iterator_streams(trimmer_it);
1812 if (status != BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_OK) {
1813 goto end;
1814 }
1815
1816 trimmer_it->state =
1817 TRIMMER_ITERATOR_STATE_ENDING;
1818 status = state_ending(trimmer_it, msgs,
1819 capacity, count);
1820 }
1821
1822 goto end;
1823 }
1824
1825 BT_ASSERT_DBG(my_count > 0);
1826
1827 for (i = 0; i < my_count; i++) {
1828 status = handle_message(trimmer_it, my_msgs[i],
1829 &reached_end);
1830
1831 /*
1832 * handle_message() unconditionally consumes the
1833 * message reference.
1834 */
1835 my_msgs[i] = NULL;
1836
1837 if (G_UNLIKELY(status !=
1838 BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_OK)) {
1839 put_messages(my_msgs, my_count);
1840 goto end;
1841 }
1842
1843 if (G_UNLIKELY(reached_end)) {
1844 /*
1845 * This message's time was passed the
1846 * trimming time range's end time: we
1847 * are done. Their might still be
1848 * messages in the output message queue,
1849 * so move to the "ending" state and
1850 * apply it immediately since
1851 * state_trim() is called within the
1852 * "next" method.
1853 */
1854 put_messages(my_msgs, my_count);
1855 trimmer_it->state =
1856 TRIMMER_ITERATOR_STATE_ENDING;
1857 status = state_ending(trimmer_it, msgs,
1858 capacity, count);
1859 goto end;
1860 }
1861 }
1862 }
1863
1864 /*
1865 * There's at least one message in the output message queue:
1866 * move the messages to the output message array.
1867 */
1868 BT_ASSERT_DBG(!g_queue_is_empty(trimmer_it->output_messages));
1869 fill_message_array_from_output_messages(trimmer_it, msgs,
1870 capacity, count);
1871
1872 end:
1873 return status;
1874 }
1875
1876 bt_message_iterator_class_next_method_status trimmer_msg_iter_next(
1877 bt_self_message_iterator *self_msg_iter,
1878 bt_message_array_const msgs, uint64_t capacity,
1879 uint64_t *count)
1880 {
1881 struct trimmer_iterator *trimmer_it =
1882 bt_self_message_iterator_get_data(self_msg_iter);
1883 bt_message_iterator_class_next_method_status status =
1884 BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_OK;
1885
1886 BT_ASSERT_DBG(trimmer_it);
1887
1888 if (G_LIKELY(trimmer_it->state == TRIMMER_ITERATOR_STATE_TRIM)) {
1889 status = state_trim(trimmer_it, msgs, capacity, count);
1890 if (status != BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_OK) {
1891 goto end;
1892 }
1893 } else {
1894 switch (trimmer_it->state) {
1895 case TRIMMER_ITERATOR_STATE_SET_BOUNDS_NS_FROM_ORIGIN:
1896 status = state_set_trimmer_iterator_bounds(trimmer_it);
1897 if (status != BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_OK) {
1898 goto end;
1899 }
1900
1901 status = state_seek_initially(trimmer_it);
1902 if (status != BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_OK) {
1903 goto end;
1904 }
1905
1906 status = state_trim(trimmer_it, msgs, capacity, count);
1907 if (status != BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_OK) {
1908 goto end;
1909 }
1910
1911 break;
1912 case TRIMMER_ITERATOR_STATE_SEEK_INITIALLY:
1913 status = state_seek_initially(trimmer_it);
1914 if (status != BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_OK) {
1915 goto end;
1916 }
1917
1918 status = state_trim(trimmer_it, msgs, capacity, count);
1919 if (status != BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_OK) {
1920 goto end;
1921 }
1922
1923 break;
1924 case TRIMMER_ITERATOR_STATE_ENDING:
1925 status = state_ending(trimmer_it, msgs, capacity,
1926 count);
1927 if (status != BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_OK) {
1928 goto end;
1929 }
1930
1931 break;
1932 case TRIMMER_ITERATOR_STATE_ENDED:
1933 status = BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_END;
1934 break;
1935 default:
1936 bt_common_abort();
1937 }
1938 }
1939
1940 end:
1941 return status;
1942 }
1943
1944 void trimmer_msg_iter_finalize(bt_self_message_iterator *self_msg_iter)
1945 {
1946 struct trimmer_iterator *trimmer_it =
1947 bt_self_message_iterator_get_data(self_msg_iter);
1948
1949 BT_ASSERT(trimmer_it);
1950 destroy_trimmer_iterator(trimmer_it);
1951 }
This page took 0.104422 seconds and 4 git commands to generate.