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