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