projects / babeltrace.git / blob
? search:


7cf741cf1ae0f0563b9f56f279a5e13b5156d83a
[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_COMP_LOG_SELF_COMP (trimmer_comp->self_comp)
25 #define BT_LOG_OUTPUT_LEVEL (trimmer_comp->log_level)
26 #define BT_LOG_TAG "PLUGIN/FLT.UTILS.TRIMMER"
27 #include "plugins/comp-logging.h"
28
29 #include "compat/utc.h"
30 #include "compat/time.h"
31 #include <babeltrace2/babeltrace.h>
32 #include "common/common.h"
33 #include "common/assert.h"
34 #include <stdint.h>
35 #include <inttypes.h>
36 #include <glib.h>
37 #include "compat/glib.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("Invalid date/time format: param=\"%s\"", str);
426 ret = -1;
427
428 end:
429 return ret;
430 }
431
432 /*
433 * Sets a trimmer bound's properties from a parameter string/integer
434 * value.
435 *
436 * Returns a negative value if anything goes wrong.
437 */
438 static
439 int set_bound_from_param(struct trimmer_comp *trimmer_comp,
440 const char *param_name, const bt_value *param,
441 struct trimmer_bound *bound, bool is_gmt)
442 {
443 int ret;
444 const char *arg;
445 char tmp_arg[64];
446
447 if (bt_value_is_signed_integer(param)) {
448 int64_t value = bt_value_integer_signed_get(param);
449
450 /*
451 * Just convert it to a temporary string to handle
452 * everything the same way.
453 */
454 sprintf(tmp_arg, "%" PRId64, value);
455 arg = tmp_arg;
456 } else if (bt_value_is_string(param)) {
457 arg = bt_value_string_get(param);
458 } else {
459 BT_COMP_LOGE("`%s` parameter must be an integer or a string value.",
460 param_name);
461 ret = -1;
462 goto end;
463 }
464
465 ret = set_bound_from_str(trimmer_comp, arg, bound, is_gmt);
466
467 end:
468 return ret;
469 }
470
471 static
472 int validate_trimmer_bounds(struct trimmer_comp *trimmer_comp,
473 struct trimmer_bound *begin, struct trimmer_bound *end)
474 {
475 int ret = 0;
476
477 BT_ASSERT(begin->is_set);
478 BT_ASSERT(end->is_set);
479
480 if (!begin->is_infinite && !end->is_infinite &&
481 begin->ns_from_origin > end->ns_from_origin) {
482 BT_COMP_LOGE("Trimming time range's beginning time is greater than end time: "
483 "begin-ns-from-origin=%" PRId64 ", "
484 "end-ns-from-origin=%" PRId64,
485 begin->ns_from_origin,
486 end->ns_from_origin);
487 ret = -1;
488 goto end;
489 }
490
491 if (!begin->is_infinite && begin->ns_from_origin == INT64_MIN) {
492 BT_COMP_LOGE("Invalid trimming time range's beginning time: "
493 "ns-from-origin=%" PRId64,
494 begin->ns_from_origin);
495 ret = -1;
496 goto end;
497 }
498
499 if (!end->is_infinite && end->ns_from_origin == INT64_MIN) {
500 BT_COMP_LOGE("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 int init_trimmer_comp_from_params(struct trimmer_comp *trimmer_comp,
513 const bt_value *params)
514 {
515 const bt_value *value;
516 int ret = 0;
517
518 BT_ASSERT(params);
519 value = bt_value_map_borrow_entry_value_const(params, "gmt");
520 if (value) {
521 trimmer_comp->is_gmt = (bool) bt_value_bool_get(value);
522 }
523
524 value = bt_value_map_borrow_entry_value_const(params, "begin");
525 if (value) {
526 if (set_bound_from_param(trimmer_comp, "begin", value,
527 &trimmer_comp->begin, trimmer_comp->is_gmt)) {
528 /* set_bound_from_param() logs errors */
529 ret = -1;
530 goto end;
531 }
532 } else {
533 trimmer_comp->begin.is_infinite = true;
534 trimmer_comp->begin.is_set = true;
535 }
536
537 value = bt_value_map_borrow_entry_value_const(params, "end");
538 if (value) {
539 if (set_bound_from_param(trimmer_comp, "end", value,
540 &trimmer_comp->end, trimmer_comp->is_gmt)) {
541 /* set_bound_from_param() logs errors */
542 ret = -1;
543 goto end;
544 }
545 } else {
546 trimmer_comp->end.is_infinite = true;
547 trimmer_comp->end.is_set = true;
548 }
549
550 end:
551 if (trimmer_comp->begin.is_set && trimmer_comp->end.is_set) {
552 /* validate_trimmer_bounds() logs errors */
553 ret = validate_trimmer_bounds(trimmer_comp,
554 &trimmer_comp->begin, &trimmer_comp->end);
555 }
556
557 return ret;
558 }
559
560 bt_component_class_init_method_status trimmer_init(
561 bt_self_component_filter *self_comp_flt,
562 const bt_value *params, void *init_data)
563 {
564 int ret;
565 bt_component_class_init_method_status status =
566 BT_COMPONENT_CLASS_INIT_METHOD_STATUS_OK;
567 bt_self_component_add_port_status add_port_status;
568 struct trimmer_comp *trimmer_comp = create_trimmer_comp();
569 bt_self_component *self_comp =
570 bt_self_component_filter_as_self_component(self_comp_flt);
571 if (!trimmer_comp) {
572 status = BT_COMPONENT_CLASS_INIT_METHOD_STATUS_MEMORY_ERROR;
573 goto error;
574 }
575
576 trimmer_comp->log_level = bt_component_get_logging_level(
577 bt_self_component_as_component(self_comp));
578 trimmer_comp->self_comp = self_comp;
579 add_port_status = bt_self_component_filter_add_input_port(
580 self_comp_flt, in_port_name, NULL, NULL);
581 switch (add_port_status) {
582 case BT_SELF_COMPONENT_ADD_PORT_STATUS_ERROR:
583 status = BT_COMPONENT_CLASS_INIT_METHOD_STATUS_ERROR;
584 goto error;
585 case BT_SELF_COMPONENT_ADD_PORT_STATUS_MEMORY_ERROR:
586 status = BT_COMPONENT_CLASS_INIT_METHOD_STATUS_MEMORY_ERROR;
587 goto error;
588 default:
589 break;
590 }
591
592 add_port_status = bt_self_component_filter_add_output_port(
593 self_comp_flt, "out", NULL, NULL);
594 switch (add_port_status) {
595 case BT_SELF_COMPONENT_ADD_PORT_STATUS_ERROR:
596 status = BT_COMPONENT_CLASS_INIT_METHOD_STATUS_ERROR;
597 goto error;
598 case BT_SELF_COMPONENT_ADD_PORT_STATUS_MEMORY_ERROR:
599 status = BT_COMPONENT_CLASS_INIT_METHOD_STATUS_MEMORY_ERROR;
600 goto error;
601 default:
602 break;
603 }
604
605 ret = init_trimmer_comp_from_params(trimmer_comp, params);
606 if (ret) {
607 status = BT_COMPONENT_CLASS_INIT_METHOD_STATUS_ERROR;
608 goto error;
609 }
610
611 bt_self_component_set_data(self_comp, trimmer_comp);
612 goto end;
613
614 error:
615 if (status == BT_COMPONENT_CLASS_INIT_METHOD_STATUS_OK) {
616 status = BT_COMPONENT_CLASS_INIT_METHOD_STATUS_ERROR;
617 }
618
619 if (trimmer_comp) {
620 destroy_trimmer_comp(trimmer_comp);
621 }
622
623 end:
624 return status;
625 }
626
627 static
628 void destroy_trimmer_iterator(struct trimmer_iterator *trimmer_it)
629 {
630 BT_ASSERT(trimmer_it);
631 bt_self_component_port_input_message_iterator_put_ref(
632 trimmer_it->upstream_iter);
633
634 if (trimmer_it->output_messages) {
635 g_queue_free(trimmer_it->output_messages);
636 }
637
638 if (trimmer_it->stream_states) {
639 g_hash_table_destroy(trimmer_it->stream_states);
640 }
641
642 g_free(trimmer_it);
643 }
644
645 static
646 void destroy_trimmer_iterator_stream_state(
647 struct trimmer_iterator_stream_state *sstate)
648 {
649 BT_ASSERT(sstate);
650 BT_PACKET_PUT_REF_AND_RESET(sstate->cur_packet);
651 g_free(sstate);
652 }
653
654 BT_HIDDEN
655 bt_component_class_message_iterator_init_method_status trimmer_msg_iter_init(
656 bt_self_message_iterator *self_msg_iter,
657 bt_self_component_filter *self_comp,
658 bt_self_component_port_output *port)
659 {
660 bt_component_class_message_iterator_init_method_status status =
661 BT_COMPONENT_CLASS_MESSAGE_ITERATOR_INIT_METHOD_STATUS_OK;
662 struct trimmer_iterator *trimmer_it;
663
664 trimmer_it = g_new0(struct trimmer_iterator, 1);
665 if (!trimmer_it) {
666 status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_INIT_METHOD_STATUS_MEMORY_ERROR;
667 goto end;
668 }
669
670 trimmer_it->trimmer_comp = bt_self_component_get_data(
671 bt_self_component_filter_as_self_component(self_comp));
672 BT_ASSERT(trimmer_it->trimmer_comp);
673
674 if (trimmer_it->trimmer_comp->begin.is_set &&
675 trimmer_it->trimmer_comp->end.is_set) {
676 /*
677 * Both trimming time range's bounds are set, so skip
678 * the
679 * `TRIMMER_ITERATOR_STATE_SET_BOUNDS_NS_FROM_ORIGIN`
680 * phase.
681 */
682 trimmer_it->state = TRIMMER_ITERATOR_STATE_SEEK_INITIALLY;
683 }
684
685 trimmer_it->begin = trimmer_it->trimmer_comp->begin;
686 trimmer_it->end = trimmer_it->trimmer_comp->end;
687 trimmer_it->upstream_iter =
688 bt_self_component_port_input_message_iterator_create(
689 bt_self_component_filter_borrow_input_port_by_name(
690 self_comp, in_port_name));
691 if (!trimmer_it->upstream_iter) {
692 status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_INIT_METHOD_STATUS_ERROR;
693 goto end;
694 }
695
696 trimmer_it->output_messages = g_queue_new();
697 if (!trimmer_it->output_messages) {
698 status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_INIT_METHOD_STATUS_MEMORY_ERROR;
699 goto end;
700 }
701
702 trimmer_it->stream_states = g_hash_table_new_full(g_direct_hash,
703 g_direct_equal, NULL,
704 (GDestroyNotify) destroy_trimmer_iterator_stream_state);
705 if (!trimmer_it->stream_states) {
706 status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_INIT_METHOD_STATUS_MEMORY_ERROR;
707 goto end;
708 }
709
710 trimmer_it->self_msg_iter = self_msg_iter;
711 bt_self_message_iterator_set_data(self_msg_iter, trimmer_it);
712
713 end:
714 if (status != BT_COMPONENT_CLASS_MESSAGE_ITERATOR_INIT_METHOD_STATUS_OK && trimmer_it) {
715 destroy_trimmer_iterator(trimmer_it);
716 }
717
718 return status;
719 }
720
721 static inline
722 int get_msg_ns_from_origin(const bt_message *msg, int64_t *ns_from_origin,
723 bool *has_clock_snapshot)
724 {
725 const bt_clock_class *clock_class = NULL;
726 const bt_clock_snapshot *clock_snapshot = NULL;
727 int ret = 0;
728
729 BT_ASSERT(msg);
730 BT_ASSERT(ns_from_origin);
731 BT_ASSERT(has_clock_snapshot);
732
733 switch (bt_message_get_type(msg)) {
734 case BT_MESSAGE_TYPE_EVENT:
735 clock_class =
736 bt_message_event_borrow_stream_class_default_clock_class_const(
737 msg);
738 if (G_UNLIKELY(!clock_class)) {
739 goto error;
740 }
741
742 clock_snapshot = bt_message_event_borrow_default_clock_snapshot_const(
743 msg);
744 break;
745 case BT_MESSAGE_TYPE_PACKET_BEGINNING:
746 clock_class =
747 bt_message_packet_beginning_borrow_stream_class_default_clock_class_const(
748 msg);
749 if (G_UNLIKELY(!clock_class)) {
750 goto error;
751 }
752
753 clock_snapshot = bt_message_packet_beginning_borrow_default_clock_snapshot_const(
754 msg);
755 break;
756 case BT_MESSAGE_TYPE_PACKET_END:
757 clock_class =
758 bt_message_packet_end_borrow_stream_class_default_clock_class_const(
759 msg);
760 if (G_UNLIKELY(!clock_class)) {
761 goto error;
762 }
763
764 clock_snapshot = bt_message_packet_end_borrow_default_clock_snapshot_const(
765 msg);
766 break;
767 case BT_MESSAGE_TYPE_STREAM_BEGINNING:
768 {
769 enum bt_message_stream_clock_snapshot_state cs_state;
770
771 clock_class =
772 bt_message_stream_beginning_borrow_stream_class_default_clock_class_const(msg);
773 if (G_UNLIKELY(!clock_class)) {
774 goto error;
775 }
776
777 cs_state = bt_message_stream_beginning_borrow_default_clock_snapshot_const(msg, &clock_snapshot);
778 if (cs_state != BT_MESSAGE_STREAM_CLOCK_SNAPSHOT_STATE_KNOWN) {
779 goto no_clock_snapshot;
780 }
781
782 break;
783 }
784 case BT_MESSAGE_TYPE_STREAM_END:
785 {
786 enum bt_message_stream_clock_snapshot_state cs_state;
787
788 clock_class =
789 bt_message_stream_end_borrow_stream_class_default_clock_class_const(msg);
790 if (G_UNLIKELY(!clock_class)) {
791 goto error;
792 }
793
794 cs_state = bt_message_stream_end_borrow_default_clock_snapshot_const(msg, &clock_snapshot);
795 if (cs_state != BT_MESSAGE_STREAM_CLOCK_SNAPSHOT_STATE_KNOWN) {
796 goto no_clock_snapshot;
797 }
798
799 break;
800 }
801 case BT_MESSAGE_TYPE_DISCARDED_EVENTS:
802 clock_class =
803 bt_message_discarded_events_borrow_stream_class_default_clock_class_const(
804 msg);
805 if (G_UNLIKELY(!clock_class)) {
806 goto error;
807 }
808
809 clock_snapshot = bt_message_discarded_events_borrow_beginning_default_clock_snapshot_const(
810 msg);
811 break;
812 case BT_MESSAGE_TYPE_DISCARDED_PACKETS:
813 clock_class =
814 bt_message_discarded_packets_borrow_stream_class_default_clock_class_const(
815 msg);
816 if (G_UNLIKELY(!clock_class)) {
817 goto error;
818 }
819
820 clock_snapshot = bt_message_discarded_packets_borrow_beginning_default_clock_snapshot_const(
821 msg);
822 break;
823 case BT_MESSAGE_TYPE_MESSAGE_ITERATOR_INACTIVITY:
824 clock_snapshot =
825 bt_message_message_iterator_inactivity_borrow_default_clock_snapshot_const(
826 msg);
827 break;
828 default:
829 goto no_clock_snapshot;
830 }
831
832 ret = bt_clock_snapshot_get_ns_from_origin(clock_snapshot,
833 ns_from_origin);
834 if (G_UNLIKELY(ret)) {
835 goto error;
836 }
837
838 *has_clock_snapshot = true;
839 goto end;
840
841 no_clock_snapshot:
842 *has_clock_snapshot = false;
843 goto end;
844
845 error:
846 ret = -1;
847
848 end:
849 return ret;
850 }
851
852 static inline
853 void put_messages(bt_message_array_const msgs, uint64_t count)
854 {
855 uint64_t i;
856
857 for (i = 0; i < count; i++) {
858 BT_MESSAGE_PUT_REF_AND_RESET(msgs[i]);
859 }
860 }
861
862 static inline
863 int set_trimmer_iterator_bound(struct trimmer_iterator *trimmer_it,
864 struct trimmer_bound *bound, int64_t ns_from_origin,
865 bool is_gmt)
866 {
867 struct trimmer_comp *trimmer_comp = trimmer_it->trimmer_comp;
868 struct tm tm;
869 struct tm *res;
870 time_t time_seconds = (time_t) (ns_from_origin / NS_PER_S);
871 int ret = 0;
872
873 BT_ASSERT(!bound->is_set);
874 errno = 0;
875
876 /* We only need to extract the date from this time */
877 if (is_gmt) {
878 res = bt_gmtime_r(&time_seconds, &tm);
879 } else {
880 res = bt_localtime_r(&time_seconds, &tm);
881 }
882
883 if (!res) {
884 BT_COMP_LOGE_ERRNO("Cannot convert timestamp to date and time",
885 ": ts=%" PRId64, (int64_t) time_seconds);
886 ret = -1;
887 goto end;
888 }
889
890 ret = set_bound_ns_from_origin(bound, tm.tm_year + 1900, tm.tm_mon + 1,
891 tm.tm_mday, bound->time.hour, bound->time.minute,
892 bound->time.second, bound->time.ns, is_gmt);
893
894 end:
895 return ret;
896 }
897
898 static
899 bt_component_class_message_iterator_next_method_status
900 state_set_trimmer_iterator_bounds(
901 struct trimmer_iterator *trimmer_it)
902 {
903 bt_message_iterator_next_status upstream_iter_status =
904 BT_MESSAGE_ITERATOR_NEXT_STATUS_OK;
905 struct trimmer_comp *trimmer_comp = trimmer_it->trimmer_comp;
906 bt_message_array_const msgs;
907 uint64_t count = 0;
908 int64_t ns_from_origin = INT64_MIN;
909 uint64_t i;
910 int ret;
911
912 BT_ASSERT(!trimmer_it->begin.is_set ||
913 !trimmer_it->end.is_set);
914
915 while (true) {
916 upstream_iter_status =
917 bt_self_component_port_input_message_iterator_next(
918 trimmer_it->upstream_iter, &msgs, &count);
919 if (upstream_iter_status != BT_MESSAGE_ITERATOR_NEXT_STATUS_OK) {
920 goto end;
921 }
922
923 for (i = 0; i < count; i++) {
924 const bt_message *msg = msgs[i];
925 bool has_ns_from_origin;
926 int ret;
927
928 ret = get_msg_ns_from_origin(msg, &ns_from_origin,
929 &has_ns_from_origin);
930 if (ret) {
931 goto error;
932 }
933
934 if (!has_ns_from_origin) {
935 continue;
936 }
937
938 BT_ASSERT(ns_from_origin != INT64_MIN &&
939 ns_from_origin != INT64_MAX);
940 put_messages(msgs, count);
941 goto found;
942 }
943
944 put_messages(msgs, count);
945 }
946
947 found:
948 if (!trimmer_it->begin.is_set) {
949 BT_ASSERT(!trimmer_it->begin.is_infinite);
950 ret = set_trimmer_iterator_bound(trimmer_it, &trimmer_it->begin,
951 ns_from_origin, trimmer_comp->is_gmt);
952 if (ret) {
953 goto error;
954 }
955 }
956
957 if (!trimmer_it->end.is_set) {
958 BT_ASSERT(!trimmer_it->end.is_infinite);
959 ret = set_trimmer_iterator_bound(trimmer_it, &trimmer_it->end,
960 ns_from_origin, trimmer_comp->is_gmt);
961 if (ret) {
962 goto error;
963 }
964 }
965
966 ret = validate_trimmer_bounds(trimmer_it->trimmer_comp,
967 &trimmer_it->begin, &trimmer_it->end);
968 if (ret) {
969 goto error;
970 }
971
972 goto end;
973
974 error:
975 put_messages(msgs, count);
976 upstream_iter_status = BT_MESSAGE_ITERATOR_NEXT_STATUS_ERROR;
977
978 end:
979 return (int) upstream_iter_status;
980 }
981
982 static
983 bt_component_class_message_iterator_next_method_status state_seek_initially(
984 struct trimmer_iterator *trimmer_it)
985 {
986 struct trimmer_comp *trimmer_comp = trimmer_it->trimmer_comp;
987 bt_component_class_message_iterator_next_method_status status =
988 BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK;
989
990 BT_ASSERT(trimmer_it->begin.is_set);
991
992 if (trimmer_it->begin.is_infinite) {
993 if (!bt_self_component_port_input_message_iterator_can_seek_beginning(
994 trimmer_it->upstream_iter)) {
995 BT_COMP_LOGE_STR("Cannot make upstream message iterator initially seek its beginning.");
996 status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_ERROR;
997 goto end;
998 }
999
1000 status = (int) bt_self_component_port_input_message_iterator_seek_beginning(
1001 trimmer_it->upstream_iter);
1002 } else {
1003 if (!bt_self_component_port_input_message_iterator_can_seek_ns_from_origin(
1004 trimmer_it->upstream_iter,
1005 trimmer_it->begin.ns_from_origin)) {
1006 BT_COMP_LOGE("Cannot make upstream message iterator initially seek: "
1007 "seek-ns-from-origin=%" PRId64,
1008 trimmer_it->begin.ns_from_origin);
1009 status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_ERROR;
1010 goto end;
1011 }
1012
1013 status = (int) bt_self_component_port_input_message_iterator_seek_ns_from_origin(
1014 trimmer_it->upstream_iter, trimmer_it->begin.ns_from_origin);
1015 }
1016
1017 if (status == BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK) {
1018 trimmer_it->state = TRIMMER_ITERATOR_STATE_TRIM;
1019 }
1020
1021 end:
1022 return status;
1023 }
1024
1025 static inline
1026 void push_message(struct trimmer_iterator *trimmer_it, const bt_message *msg)
1027 {
1028 g_queue_push_head(trimmer_it->output_messages, (void *) msg);
1029 }
1030
1031 static inline
1032 const bt_message *pop_message(struct trimmer_iterator *trimmer_it)
1033 {
1034 return g_queue_pop_tail(trimmer_it->output_messages);
1035 }
1036
1037 static inline
1038 int clock_raw_value_from_ns_from_origin(const bt_clock_class *clock_class,
1039 int64_t ns_from_origin, uint64_t *raw_value)
1040 {
1041
1042 int64_t cc_offset_s;
1043 uint64_t cc_offset_cycles;
1044 uint64_t cc_freq;
1045
1046 bt_clock_class_get_offset(clock_class, &cc_offset_s, &cc_offset_cycles);
1047 cc_freq = bt_clock_class_get_frequency(clock_class);
1048 return bt_common_clock_value_from_ns_from_origin(cc_offset_s,
1049 cc_offset_cycles, cc_freq, ns_from_origin, raw_value);
1050 }
1051
1052 static inline
1053 bt_component_class_message_iterator_next_method_status
1054 end_stream(struct trimmer_iterator *trimmer_it,
1055 struct trimmer_iterator_stream_state *sstate)
1056 {
1057 bt_component_class_message_iterator_next_method_status status =
1058 BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK;
1059 /* Initialize to silence maybe-uninitialized warning. */
1060 uint64_t raw_value = 0;
1061 bt_message *msg = NULL;
1062
1063 BT_ASSERT(!trimmer_it->end.is_infinite);
1064 BT_ASSERT(sstate->stream);
1065
1066 /*
1067 * If we haven't seen a message with a clock snapshot, we don't know if the trimmer's end bound is within
1068 * the clock's range, so it wouldn't be safe to try to convert ns_from_origin to a clock value.
1069 *
1070 * Also, it would be a bit of a lie to generate a stream end message with the end bound as its
1071 * clock snapshot, because we don't really know if the stream existed at that time. If we have
1072 * seen a message with a clock snapshot and the stream is cut short by another message with a
1073 * clock snapshot, then we are sure that the the end bound time is not below the clock range,
1074 * and we know the stream was active at that time (and that we cut it short).
1075 */
1076 if (sstate->seen_clock_snapshot) {
1077 const bt_clock_class *clock_class;
1078 int ret;
1079
1080 clock_class = bt_stream_class_borrow_default_clock_class_const(
1081 bt_stream_borrow_class_const(sstate->stream));
1082 BT_ASSERT(clock_class);
1083 ret = clock_raw_value_from_ns_from_origin(clock_class,
1084 trimmer_it->end.ns_from_origin, &raw_value);
1085 if (ret) {
1086 status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_ERROR;
1087 goto end;
1088 }
1089 }
1090
1091 if (sstate->cur_packet) {
1092 /*
1093 * Create and push a packet end message, making its time
1094 * the trimming range's end time.
1095 *
1096 * We know that we must have seen a clock snapshot, the one in
1097 * the packet beginning message, since trimmer currently
1098 * requires packet messages to have clock snapshots (see comment
1099 * in create_stream_state_entry).
1100 */
1101 BT_ASSERT(sstate->seen_clock_snapshot);
1102
1103 msg = bt_message_packet_end_create_with_default_clock_snapshot(
1104 trimmer_it->self_msg_iter, sstate->cur_packet,
1105 raw_value);
1106 if (!msg) {
1107 status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_MEMORY_ERROR;
1108 goto end;
1109 }
1110
1111 push_message(trimmer_it, msg);
1112 msg = NULL;
1113 BT_PACKET_PUT_REF_AND_RESET(sstate->cur_packet);
1114 }
1115
1116 /* Create and push a stream end message. */
1117 msg = bt_message_stream_end_create(trimmer_it->self_msg_iter,
1118 sstate->stream);
1119 if (!msg) {
1120 status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_MEMORY_ERROR;
1121 goto end;
1122 }
1123
1124 if (sstate->seen_clock_snapshot) {
1125 bt_message_stream_end_set_default_clock_snapshot(msg, raw_value);
1126 }
1127
1128 push_message(trimmer_it, msg);
1129 msg = NULL;
1130
1131 /*
1132 * Just to make sure that we don't use this stream state again
1133 * in the future without an obvious error.
1134 */
1135 sstate->stream = NULL;
1136
1137 end:
1138 bt_message_put_ref(msg);
1139 return status;
1140 }
1141
1142 static inline
1143 bt_component_class_message_iterator_next_method_status end_iterator_streams(
1144 struct trimmer_iterator *trimmer_it)
1145 {
1146 bt_component_class_message_iterator_next_method_status status =
1147 BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK;
1148 GHashTableIter iter;
1149 gpointer key, sstate;
1150
1151 if (trimmer_it->end.is_infinite) {
1152 /*
1153 * An infinite trimming range's end time guarantees that
1154 * we received (and pushed) all the appropriate end
1155 * messages.
1156 */
1157 goto remove_all;
1158 }
1159
1160 /*
1161 * End each stream and then remove them from the hash table of
1162 * stream states to release unneeded references.
1163 */
1164 g_hash_table_iter_init(&iter, trimmer_it->stream_states);
1165
1166 while (g_hash_table_iter_next(&iter, &key, &sstate)) {
1167 status = end_stream(trimmer_it, sstate);
1168 if (status) {
1169 goto end;
1170 }
1171 }
1172
1173 remove_all:
1174 g_hash_table_remove_all(trimmer_it->stream_states);
1175
1176 end:
1177 return status;
1178 }
1179
1180 static
1181 bt_component_class_message_iterator_next_method_status
1182 create_stream_state_entry(
1183 struct trimmer_iterator *trimmer_it,
1184 const struct bt_stream *stream,
1185 struct trimmer_iterator_stream_state **stream_state)
1186 {
1187 struct trimmer_comp *trimmer_comp = trimmer_it->trimmer_comp;
1188 bt_component_class_message_iterator_next_method_status status;
1189 struct trimmer_iterator_stream_state *sstate;
1190 const bt_stream_class *sc;
1191
1192 BT_ASSERT(!bt_g_hash_table_contains(trimmer_it->stream_states, stream));
1193
1194 /*
1195 * Validate right now that the stream's class
1196 * has a registered default clock class so that
1197 * an existing stream state guarantees existing
1198 * default clock snapshots for its associated
1199 * messages.
1200 *
1201 * Also check that clock snapshots are always
1202 * known.
1203 */
1204 sc = bt_stream_borrow_class_const(stream);
1205 if (!bt_stream_class_borrow_default_clock_class_const(sc)) {
1206 BT_COMP_LOGE("Unsupported stream: stream class does "
1207 "not have a default clock class: "
1208 "stream-addr=%p, "
1209 "stream-id=%" PRIu64 ", "
1210 "stream-name=\"%s\"",
1211 stream, bt_stream_get_id(stream),
1212 bt_stream_get_name(stream));
1213 status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_ERROR;
1214 goto end;
1215 }
1216
1217 /*
1218 * Temporary: make sure packet beginning, packet
1219 * end, discarded events, and discarded packets
1220 * messages have default clock snapshots until
1221 * the support for not having them is
1222 * implemented.
1223 */
1224 if (!bt_stream_class_packets_have_beginning_default_clock_snapshot(
1225 sc)) {
1226 BT_COMP_LOGE("Unsupported stream: packets have "
1227 "no beginning clock snapshot: "
1228 "stream-addr=%p, "
1229 "stream-id=%" PRIu64 ", "
1230 "stream-name=\"%s\"",
1231 stream, bt_stream_get_id(stream),
1232 bt_stream_get_name(stream));
1233 status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_ERROR;
1234 goto end;
1235 }
1236
1237 if (!bt_stream_class_packets_have_end_default_clock_snapshot(
1238 sc)) {
1239 BT_COMP_LOGE("Unsupported stream: packets have "
1240 "no end clock snapshot: "
1241 "stream-addr=%p, "
1242 "stream-id=%" PRIu64 ", "
1243 "stream-name=\"%s\"",
1244 stream, bt_stream_get_id(stream),
1245 bt_stream_get_name(stream));
1246 status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_ERROR;
1247 goto end;
1248 }
1249
1250 if (bt_stream_class_supports_discarded_events(sc) &&
1251 !bt_stream_class_discarded_events_have_default_clock_snapshots(sc)) {
1252 BT_COMP_LOGE("Unsupported stream: discarded events "
1253 "have no clock snapshots: "
1254 "stream-addr=%p, "
1255 "stream-id=%" PRIu64 ", "
1256 "stream-name=\"%s\"",
1257 stream, bt_stream_get_id(stream),
1258 bt_stream_get_name(stream));
1259 status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_ERROR;
1260 goto end;
1261 }
1262
1263 if (bt_stream_class_supports_discarded_packets(sc) &&
1264 !bt_stream_class_discarded_packets_have_default_clock_snapshots(sc)) {
1265 BT_COMP_LOGE("Unsupported stream: discarded packets "
1266 "have no clock snapshots: "
1267 "stream-addr=%p, "
1268 "stream-id=%" PRIu64 ", "
1269 "stream-name=\"%s\"",
1270 stream, bt_stream_get_id(stream),
1271 bt_stream_get_name(stream));
1272 status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_ERROR;
1273 goto end;
1274 }
1275
1276 sstate = g_new0(struct trimmer_iterator_stream_state, 1);
1277 if (!sstate) {
1278 status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_MEMORY_ERROR;
1279 goto end;
1280 }
1281
1282 sstate->stream = stream;
1283
1284 g_hash_table_insert(trimmer_it->stream_states, (void *) stream, sstate);
1285
1286 *stream_state = sstate;
1287
1288 status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK;
1289
1290 end:
1291 return status;
1292 }
1293
1294 static
1295 struct trimmer_iterator_stream_state *get_stream_state_entry(
1296 struct trimmer_iterator *trimmer_it,
1297 const struct bt_stream *stream)
1298 {
1299 struct trimmer_iterator_stream_state *sstate;
1300
1301 BT_ASSERT(stream);
1302 sstate = g_hash_table_lookup(trimmer_it->stream_states, stream);
1303 BT_ASSERT(sstate);
1304
1305 return sstate;
1306 }
1307
1308 /*
1309 * Handles a message which is associated to a given stream state. This
1310 * _could_ make the iterator's output message queue grow; this could
1311 * also consume the message without pushing anything to this queue, only
1312 * modifying the stream state.
1313 *
1314 * This function consumes the `msg` reference, _whatever the outcome_.
1315 *
1316 * If non-NULL, `ns_from_origin` is the message's time, as given by
1317 * get_msg_ns_from_origin(). If NULL, the message doesn't have a time.
1318 *
1319 * This function sets `reached_end` if handling this message made the
1320 * iterator reach the end of the trimming range. Note that the output
1321 * message queue could contain messages even if this function sets
1322 * `reached_end`.
1323 */
1324 static
1325 bt_component_class_message_iterator_next_method_status
1326 handle_message_with_stream(
1327 struct trimmer_iterator *trimmer_it, const bt_message *msg,
1328 const struct bt_stream *stream, const int64_t *ns_from_origin,
1329 bool *reached_end)
1330 {
1331 bt_component_class_message_iterator_next_method_status status =
1332 BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK;
1333 bt_message_type msg_type = bt_message_get_type(msg);
1334 int ret;
1335 struct trimmer_iterator_stream_state *sstate = NULL;
1336
1337 /*
1338 * Retrieve the stream's state - except if the message is stream
1339 * beginning, in which case we don't know about about this stream yet.
1340 */
1341 if (msg_type != BT_MESSAGE_TYPE_STREAM_BEGINNING) {
1342 sstate = get_stream_state_entry(trimmer_it, stream);
1343 }
1344
1345 switch (msg_type) {
1346 case BT_MESSAGE_TYPE_EVENT:
1347 /*
1348 * Event messages always have a clock snapshot if the stream
1349 * class has a clock class. And we know it has, otherwise we
1350 * couldn't be using the trimmer component.
1351 */
1352 BT_ASSERT(ns_from_origin);
1353
1354 if (G_UNLIKELY(!trimmer_it->end.is_infinite &&
1355 *ns_from_origin > trimmer_it->end.ns_from_origin)) {
1356 status = end_iterator_streams(trimmer_it);
1357 *reached_end = true;
1358 break;
1359 }
1360
1361 sstate->seen_clock_snapshot = true;
1362
1363 push_message(trimmer_it, msg);
1364 msg = NULL;
1365 break;
1366
1367 case BT_MESSAGE_TYPE_PACKET_BEGINNING:
1368 /*
1369 * Packet beginning messages won't have a clock snapshot if
1370 * stream_class->packets_have_beginning_default_clock_snapshot
1371 * is false. But for now, assume they always do.
1372 */
1373 BT_ASSERT(ns_from_origin);
1374 BT_ASSERT(!sstate->cur_packet);
1375
1376 if (G_UNLIKELY(!trimmer_it->end.is_infinite &&
1377 *ns_from_origin > trimmer_it->end.ns_from_origin)) {
1378 status = end_iterator_streams(trimmer_it);
1379 *reached_end = true;
1380 break;
1381 }
1382
1383 sstate->cur_packet =
1384 bt_message_packet_beginning_borrow_packet_const(msg);
1385 bt_packet_get_ref(sstate->cur_packet);
1386
1387 sstate->seen_clock_snapshot = true;
1388
1389 push_message(trimmer_it, msg);
1390 msg = NULL;
1391 break;
1392
1393 case BT_MESSAGE_TYPE_PACKET_END:
1394 /*
1395 * Packet end messages won't have a clock snapshot if
1396 * stream_class->packets_have_end_default_clock_snapshot
1397 * is false. But for now, assume they always do.
1398 */
1399 BT_ASSERT(ns_from_origin);
1400 BT_ASSERT(sstate->cur_packet);
1401
1402 if (G_UNLIKELY(!trimmer_it->end.is_infinite &&
1403 *ns_from_origin > trimmer_it->end.ns_from_origin)) {
1404 status = end_iterator_streams(trimmer_it);
1405 *reached_end = true;
1406 break;
1407 }
1408
1409 BT_PACKET_PUT_REF_AND_RESET(sstate->cur_packet);
1410
1411 sstate->seen_clock_snapshot = true;
1412
1413 push_message(trimmer_it, msg);
1414 msg = NULL;
1415 break;
1416
1417 case BT_MESSAGE_TYPE_DISCARDED_EVENTS:
1418 case BT_MESSAGE_TYPE_DISCARDED_PACKETS:
1419 {
1420 /*
1421 * `ns_from_origin` is the message's time range's
1422 * beginning time here.
1423 */
1424 int64_t end_ns_from_origin;
1425 const bt_clock_snapshot *end_cs;
1426
1427 BT_ASSERT(ns_from_origin);
1428
1429 sstate->seen_clock_snapshot = true;
1430
1431 if (bt_message_get_type(msg) ==
1432 BT_MESSAGE_TYPE_DISCARDED_EVENTS) {
1433 /*
1434 * Safe to ignore the return value because we
1435 * know there's a default clock and it's always
1436 * known.
1437 */
1438 end_cs = bt_message_discarded_events_borrow_end_default_clock_snapshot_const(
1439 msg);
1440 } else {
1441 /*
1442 * Safe to ignore the return value because we
1443 * know there's a default clock and it's always
1444 * known.
1445 */
1446 end_cs = bt_message_discarded_packets_borrow_end_default_clock_snapshot_const(
1447 msg);
1448 }
1449
1450 if (bt_clock_snapshot_get_ns_from_origin(end_cs,
1451 &end_ns_from_origin)) {
1452 status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_ERROR;
1453 goto end;
1454 }
1455
1456 if (!trimmer_it->end.is_infinite &&
1457 *ns_from_origin > trimmer_it->end.ns_from_origin) {
1458 status = end_iterator_streams(trimmer_it);
1459 *reached_end = true;
1460 break;
1461 }
1462
1463 if (!trimmer_it->end.is_infinite &&
1464 end_ns_from_origin > trimmer_it->end.ns_from_origin) {
1465 /*
1466 * This message's end time is outside the
1467 * trimming time range: replace it with a new
1468 * message having an end time equal to the
1469 * trimming time range's end and without a
1470 * count.
1471 */
1472 const bt_clock_class *clock_class =
1473 bt_clock_snapshot_borrow_clock_class_const(
1474 end_cs);
1475 const bt_clock_snapshot *begin_cs;
1476 bt_message *new_msg;
1477 uint64_t end_raw_value;
1478
1479 ret = clock_raw_value_from_ns_from_origin(clock_class,
1480 trimmer_it->end.ns_from_origin, &end_raw_value);
1481 if (ret) {
1482 status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_ERROR;
1483 goto end;
1484 }
1485
1486 if (msg_type == BT_MESSAGE_TYPE_DISCARDED_EVENTS) {
1487 begin_cs = bt_message_discarded_events_borrow_beginning_default_clock_snapshot_const(
1488 msg);
1489 new_msg = bt_message_discarded_events_create_with_default_clock_snapshots(
1490 trimmer_it->self_msg_iter,
1491 sstate->stream,
1492 bt_clock_snapshot_get_value(begin_cs),
1493 end_raw_value);
1494 } else {
1495 begin_cs = bt_message_discarded_packets_borrow_beginning_default_clock_snapshot_const(
1496 msg);
1497 new_msg = bt_message_discarded_packets_create_with_default_clock_snapshots(
1498 trimmer_it->self_msg_iter,
1499 sstate->stream,
1500 bt_clock_snapshot_get_value(begin_cs),
1501 end_raw_value);
1502 }
1503
1504 if (!new_msg) {
1505 status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_MEMORY_ERROR;
1506 goto end;
1507 }
1508
1509 /* Replace the original message */
1510 BT_MESSAGE_MOVE_REF(msg, new_msg);
1511 }
1512
1513 push_message(trimmer_it, msg);
1514 msg = NULL;
1515 break;
1516 }
1517
1518 case BT_MESSAGE_TYPE_STREAM_BEGINNING:
1519 /*
1520 * If this message has a time and this time is greater than the
1521 * trimmer's end bound, it triggers the end of the trim window.
1522 */
1523 if (G_UNLIKELY(ns_from_origin && !trimmer_it->end.is_infinite &&
1524 *ns_from_origin > trimmer_it->end.ns_from_origin)) {
1525 status = end_iterator_streams(trimmer_it);
1526 *reached_end = true;
1527 break;
1528 }
1529
1530 /* Learn about this stream. */
1531 status = create_stream_state_entry(trimmer_it, stream, &sstate);
1532 if (status != BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK) {
1533 goto end;
1534 }
1535
1536 if (ns_from_origin) {
1537 sstate->seen_clock_snapshot = true;
1538 }
1539
1540 push_message(trimmer_it, msg);
1541 msg = NULL;
1542 break;
1543 case BT_MESSAGE_TYPE_STREAM_END:
1544 {
1545 gboolean removed;
1546
1547 /*
1548 * If this message has a time and this time is greater than the
1549 * trimmer's end bound, it triggers the end of the trim window.
1550 */
1551 if (G_UNLIKELY(ns_from_origin && !trimmer_it->end.is_infinite &&
1552 *ns_from_origin > trimmer_it->end.ns_from_origin)) {
1553 status = end_iterator_streams(trimmer_it);
1554 *reached_end = true;
1555 break;
1556 }
1557
1558 /*
1559 * Either the stream end message's time is within the trimmer's
1560 * bounds, or it doesn't have a time. In both cases, pass
1561 * the message unmodified.
1562 */
1563 push_message(trimmer_it, msg);
1564 msg = NULL;
1565
1566 /* Forget about this stream. */
1567 removed = g_hash_table_remove(trimmer_it->stream_states, sstate->stream);
1568 BT_ASSERT(removed);
1569 break;
1570 }
1571 default:
1572 break;
1573 }
1574
1575 end:
1576 /* We release the message's reference whatever the outcome */
1577 bt_message_put_ref(msg);
1578 return status;
1579 }
1580
1581 /*
1582 * Handles an input message. This _could_ make the iterator's output
1583 * message queue grow; this could also consume the message without
1584 * pushing anything to this queue, only modifying the stream state.
1585 *
1586 * This function consumes the `msg` reference, _whatever the outcome_.
1587 *
1588 * This function sets `reached_end` if handling this message made the
1589 * iterator reach the end of the trimming range. Note that the output
1590 * message queue could contain messages even if this function sets
1591 * `reached_end`.
1592 */
1593 static inline
1594 bt_component_class_message_iterator_next_method_status handle_message(
1595 struct trimmer_iterator *trimmer_it, const bt_message *msg,
1596 bool *reached_end)
1597 {
1598 bt_component_class_message_iterator_next_method_status status;
1599 const bt_stream *stream = NULL;
1600 int64_t ns_from_origin = INT64_MIN;
1601 bool has_ns_from_origin = false;
1602 int ret;
1603
1604 /* Find message's associated stream */
1605 switch (bt_message_get_type(msg)) {
1606 case BT_MESSAGE_TYPE_EVENT:
1607 stream = bt_event_borrow_stream_const(
1608 bt_message_event_borrow_event_const(msg));
1609 break;
1610 case BT_MESSAGE_TYPE_PACKET_BEGINNING:
1611 stream = bt_packet_borrow_stream_const(
1612 bt_message_packet_beginning_borrow_packet_const(msg));
1613 break;
1614 case BT_MESSAGE_TYPE_PACKET_END:
1615 stream = bt_packet_borrow_stream_const(
1616 bt_message_packet_end_borrow_packet_const(msg));
1617 break;
1618 case BT_MESSAGE_TYPE_DISCARDED_EVENTS:
1619 stream = bt_message_discarded_events_borrow_stream_const(msg);
1620 break;
1621 case BT_MESSAGE_TYPE_DISCARDED_PACKETS:
1622 stream = bt_message_discarded_packets_borrow_stream_const(msg);
1623 break;
1624 case BT_MESSAGE_TYPE_STREAM_BEGINNING:
1625 stream = bt_message_stream_beginning_borrow_stream_const(msg);
1626 break;
1627 case BT_MESSAGE_TYPE_STREAM_END:
1628 stream = bt_message_stream_end_borrow_stream_const(msg);
1629 break;
1630 default:
1631 break;
1632 }
1633
1634 /* Retrieve the message's time */
1635 ret = get_msg_ns_from_origin(msg, &ns_from_origin, &has_ns_from_origin);
1636 if (G_UNLIKELY(ret)) {
1637 status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_ERROR;
1638 goto end;
1639 }
1640
1641 if (G_LIKELY(stream)) {
1642 /* Message associated to a stream */
1643 status = handle_message_with_stream(trimmer_it, msg,
1644 stream, has_ns_from_origin ? &ns_from_origin : NULL, reached_end);
1645
1646 /*
1647 * handle_message_with_stream_state() unconditionally
1648 * consumes `msg`.
1649 */
1650 msg = NULL;
1651 } else {
1652 /*
1653 * Message not associated to a stream (message iterator
1654 * inactivity).
1655 */
1656 if (G_UNLIKELY(ns_from_origin > trimmer_it->end.ns_from_origin)) {
1657 BT_MESSAGE_PUT_REF_AND_RESET(msg);
1658 status = end_iterator_streams(trimmer_it);
1659 *reached_end = true;
1660 } else {
1661 push_message(trimmer_it, msg);
1662 status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK;
1663 msg = NULL;
1664 }
1665 }
1666
1667 end:
1668 /* We release the message's reference whatever the outcome */
1669 bt_message_put_ref(msg);
1670 return status;
1671 }
1672
1673 static inline
1674 void fill_message_array_from_output_messages(
1675 struct trimmer_iterator *trimmer_it,
1676 bt_message_array_const msgs, uint64_t capacity, uint64_t *count)
1677 {
1678 *count = 0;
1679
1680 /*
1681 * Move auto-seek messages to the output array (which is this
1682 * iterator's base message array).
1683 */
1684 while (capacity > 0 && !g_queue_is_empty(trimmer_it->output_messages)) {
1685 msgs[*count] = pop_message(trimmer_it);
1686 capacity--;
1687 (*count)++;
1688 }
1689
1690 BT_ASSERT(*count > 0);
1691 }
1692
1693 static inline
1694 bt_component_class_message_iterator_next_method_status state_ending(
1695 struct trimmer_iterator *trimmer_it,
1696 bt_message_array_const msgs, uint64_t capacity,
1697 uint64_t *count)
1698 {
1699 bt_component_class_message_iterator_next_method_status status =
1700 BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK;
1701
1702 if (g_queue_is_empty(trimmer_it->output_messages)) {
1703 trimmer_it->state = TRIMMER_ITERATOR_STATE_ENDED;
1704 status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_END;
1705 goto end;
1706 }
1707
1708 fill_message_array_from_output_messages(trimmer_it, msgs,
1709 capacity, count);
1710
1711 end:
1712 return status;
1713 }
1714
1715 static inline
1716 bt_component_class_message_iterator_next_method_status
1717 state_trim(struct trimmer_iterator *trimmer_it,
1718 bt_message_array_const msgs, uint64_t capacity,
1719 uint64_t *count)
1720 {
1721 bt_component_class_message_iterator_next_method_status status =
1722 BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK;
1723 bt_message_array_const my_msgs;
1724 uint64_t my_count;
1725 uint64_t i;
1726 bool reached_end = false;
1727
1728 while (g_queue_is_empty(trimmer_it->output_messages)) {
1729 status = (int) bt_self_component_port_input_message_iterator_next(
1730 trimmer_it->upstream_iter, &my_msgs, &my_count);
1731 if (G_UNLIKELY(status != BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK)) {
1732 if (status == BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_END) {
1733 status = end_iterator_streams(trimmer_it);
1734 if (status != BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK) {
1735 goto end;
1736 }
1737
1738 trimmer_it->state =
1739 TRIMMER_ITERATOR_STATE_ENDING;
1740 status = state_ending(trimmer_it, msgs,
1741 capacity, count);
1742 }
1743
1744 goto end;
1745 }
1746
1747 BT_ASSERT(my_count > 0);
1748
1749 for (i = 0; i < my_count; i++) {
1750 status = handle_message(trimmer_it, my_msgs[i],
1751 &reached_end);
1752
1753 /*
1754 * handle_message() unconditionally consumes the
1755 * message reference.
1756 */
1757 my_msgs[i] = NULL;
1758
1759 if (G_UNLIKELY(status !=
1760 BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK)) {
1761 put_messages(my_msgs, my_count);
1762 goto end;
1763 }
1764
1765 if (G_UNLIKELY(reached_end)) {
1766 /*
1767 * This message's time was passed the
1768 * trimming time range's end time: we
1769 * are done. Their might still be
1770 * messages in the output message queue,
1771 * so move to the "ending" state and
1772 * apply it immediately since
1773 * state_trim() is called within the
1774 * "next" method.
1775 */
1776 put_messages(my_msgs, my_count);
1777 trimmer_it->state =
1778 TRIMMER_ITERATOR_STATE_ENDING;
1779 status = state_ending(trimmer_it, msgs,
1780 capacity, count);
1781 goto end;
1782 }
1783 }
1784 }
1785
1786 /*
1787 * There's at least one message in the output message queue:
1788 * move the messages to the output message array.
1789 */
1790 BT_ASSERT(!g_queue_is_empty(trimmer_it->output_messages));
1791 fill_message_array_from_output_messages(trimmer_it, msgs,
1792 capacity, count);
1793
1794 end:
1795 return status;
1796 }
1797
1798 BT_HIDDEN
1799 bt_component_class_message_iterator_next_method_status trimmer_msg_iter_next(
1800 bt_self_message_iterator *self_msg_iter,
1801 bt_message_array_const msgs, uint64_t capacity,
1802 uint64_t *count)
1803 {
1804 struct trimmer_iterator *trimmer_it =
1805 bt_self_message_iterator_get_data(self_msg_iter);
1806 bt_component_class_message_iterator_next_method_status status =
1807 BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK;
1808
1809 BT_ASSERT(trimmer_it);
1810
1811 if (G_LIKELY(trimmer_it->state == TRIMMER_ITERATOR_STATE_TRIM)) {
1812 status = state_trim(trimmer_it, msgs, capacity, count);
1813 if (status != BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK) {
1814 goto end;
1815 }
1816 } else {
1817 switch (trimmer_it->state) {
1818 case TRIMMER_ITERATOR_STATE_SET_BOUNDS_NS_FROM_ORIGIN:
1819 status = state_set_trimmer_iterator_bounds(trimmer_it);
1820 if (status != BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK) {
1821 goto end;
1822 }
1823
1824 status = state_seek_initially(trimmer_it);
1825 if (status != BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK) {
1826 goto end;
1827 }
1828
1829 status = state_trim(trimmer_it, msgs, capacity, count);
1830 if (status != BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK) {
1831 goto end;
1832 }
1833
1834 break;
1835 case TRIMMER_ITERATOR_STATE_SEEK_INITIALLY:
1836 status = state_seek_initially(trimmer_it);
1837 if (status != BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK) {
1838 goto end;
1839 }
1840
1841 status = state_trim(trimmer_it, msgs, capacity, count);
1842 if (status != BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK) {
1843 goto end;
1844 }
1845
1846 break;
1847 case TRIMMER_ITERATOR_STATE_ENDING:
1848 status = state_ending(trimmer_it, msgs, capacity,
1849 count);
1850 if (status != BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK) {
1851 goto end;
1852 }
1853
1854 break;
1855 case TRIMMER_ITERATOR_STATE_ENDED:
1856 status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_END;
1857 break;
1858 default:
1859 abort();
1860 }
1861 }
1862
1863 end:
1864 return status;
1865 }
1866
1867 BT_HIDDEN
1868 void trimmer_msg_iter_finalize(bt_self_message_iterator *self_msg_iter)
1869 {
1870 struct trimmer_iterator *trimmer_it =
1871 bt_self_message_iterator_get_data(self_msg_iter);
1872
1873 BT_ASSERT(trimmer_it);
1874 destroy_trimmer_iterator(trimmer_it);
1875 }
BabelTrace - Trace Format Babel Tower
This page took 0.070529 seconds and 3 git commands to generate.