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