projects / babeltrace.git / blob
? search:


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