2 * SPDX-License-Identifier: MIT
4 * Copyright 2016 Jérémie Galarneau <jeremie.galarneau@efficios.com>
5 * Copyright 2019 Philippe Proulx <pproulx@efficios.com>
8 #define BT_LOG_OUTPUT_LEVEL (trimmer_comp->log_level)
9 #define BT_LOG_TAG "PLUGIN/FLT.UTILS.TRIMMER"
10 #include "logging/comp-logging.h"
12 #include "compat/utc.h"
13 #include "compat/time.h"
14 #include <babeltrace2/babeltrace.h>
15 #include "common/common.h"
16 #include "common/assert.h"
21 #include "compat/glib.h"
22 #include "plugins/common/param-validation/param-validation.h"
26 #define NS_PER_S INT64_C(1000000000)
28 static const char * const in_port_name
= "in";
31 unsigned int hour
, minute
, second
, ns
;
34 struct trimmer_bound
{
36 * Nanoseconds from origin, valid if `is_set` is set and
37 * `is_infinite` is false.
39 int64_t ns_from_origin
;
41 /* True if this bound's full time (`ns_from_origin`) is set */
45 * True if this bound represents the infinity (negative or
46 * positive depending on which bound it is). If this is true,
47 * then we don't care about `ns_from_origin` above.
52 * This bound's time without the date; this time is used to set
53 * `ns_from_origin` once we know the date.
55 struct trimmer_time time
;
59 struct trimmer_bound begin
, end
;
61 bt_logging_level log_level
;
62 bt_self_component
*self_comp
;
63 bt_self_component_filter
*self_comp_filter
;
66 enum trimmer_iterator_state
{
68 * Find the first message's date and set the bounds's times
71 TRIMMER_ITERATOR_STATE_SET_BOUNDS_NS_FROM_ORIGIN
,
74 * Initially seek to the trimming range's beginning time.
76 TRIMMER_ITERATOR_STATE_SEEK_INITIALLY
,
79 * Fill the output message queue for as long as received input
80 * messages are within the trimming time range.
82 TRIMMER_ITERATOR_STATE_TRIM
,
84 /* Flush the remaining messages in the output message queue */
85 TRIMMER_ITERATOR_STATE_ENDING
,
87 /* Trimming operation and message iterator is ended */
88 TRIMMER_ITERATOR_STATE_ENDED
,
91 struct trimmer_iterator
{
93 struct trimmer_comp
*trimmer_comp
;
96 bt_self_message_iterator
*self_msg_iter
;
98 enum trimmer_iterator_state state
;
101 bt_message_iterator
*upstream_iter
;
102 struct trimmer_bound begin
, end
;
105 * Queue of `const bt_message *` (owned by the queue).
107 * This is where the trimming operation pushes the messages to
108 * output by this message iterator.
110 GQueue
*output_messages
;
113 * Hash table of `bt_stream *` (weak) to
114 * `struct trimmer_iterator_stream_state *` (owned by the HT).
116 GHashTable
*stream_states
;
119 struct trimmer_iterator_stream_state
{
121 const bt_stream
*stream
;
123 /* Have we seen a message with clock_snapshot going through this stream? */
124 bool seen_clock_snapshot
;
126 /* Owned by this (`NULL` initially and between packets) */
127 const bt_packet
*cur_packet
;
131 void destroy_trimmer_comp(struct trimmer_comp
*trimmer_comp
)
133 BT_ASSERT(trimmer_comp
);
134 g_free(trimmer_comp
);
138 struct trimmer_comp
*create_trimmer_comp(void)
140 return g_new0(struct trimmer_comp
, 1);
144 void trimmer_finalize(bt_self_component_filter
*self_comp
)
146 struct trimmer_comp
*trimmer_comp
=
147 bt_self_component_get_data(
148 bt_self_component_filter_as_self_component(self_comp
));
151 destroy_trimmer_comp(trimmer_comp
);
156 * Compile regex in `pattern`, and try to match `string`. If there's a match,
157 * return true and set `*match_info` to the list of matches. The list of
158 * matches must be freed by the caller. If there's no match, return false and
159 * set `*match_info` to NULL;
162 bool compile_and_match(const char *pattern
, const char *string
, GMatchInfo
**match_info
) {
163 bool matches
= false;
164 GError
*regex_error
= NULL
;
167 regex
= g_regex_new(pattern
, 0, 0, ®ex_error
);
172 matches
= g_regex_match(regex
, string
, 0, match_info
);
175 * g_regex_match allocates `*match_info` even if it returns
176 * FALSE. If there's no match, we have no use for it, so free
177 * it immediatly and don't return it to the caller.
179 g_match_info_free(*match_info
);
183 g_regex_unref(regex
);
188 g_error_free(regex_error
);
195 * Convert the captured text in match number `match_num` in `match_info`
196 * to an unsigned integer.
199 guint64
match_to_uint(const GMatchInfo
*match_info
, gint match_num
) {
200 gchar
*text
, *endptr
;
203 text
= g_match_info_fetch(match_info
, match_num
);
207 * Because the input is carefully sanitized with regexes by the caller,
208 * we assume that g_ascii_strtoull cannot fail.
211 result
= g_ascii_strtoull(text
, &endptr
, 10);
212 BT_ASSERT(endptr
> text
);
213 BT_ASSERT(errno
== 0);
221 * When parsing the nanoseconds part, .512 means .512000000, not .000000512.
222 * This function is like match_to_uint, but multiplies the parsed number to get
223 * the expected result.
226 guint64
match_to_uint_ns(const GMatchInfo
*match_info
, gint match_num
) {
229 gint start_pos
, end_pos
, power
;
230 static int pow10
[] = {
231 1, 10, 100, 1000, 10000, 100000, 1000000, 10000000, 100000000,
234 nanoseconds
= match_to_uint(match_info
, match_num
);
236 /* Multiply by 10 as many times as there are omitted digits. */
237 ret
= g_match_info_fetch_pos(match_info
, match_num
, &start_pos
, &end_pos
);
240 power
= 9 - (end_pos
- start_pos
);
241 BT_ASSERT(power
>= 0 && power
<= 8);
243 nanoseconds
*= pow10
[power
];
249 * Sets the time (in ns from origin) of a trimmer bound from date and
252 * Returns a negative value if anything goes wrong.
255 int set_bound_ns_from_origin(struct trimmer_bound
*bound
,
256 unsigned int year
, unsigned int month
, unsigned int day
,
257 unsigned int hour
, unsigned int minute
, unsigned int second
,
258 unsigned int ns
, bool is_gmt
)
268 .tm_year
= year
- 1900,
273 result
= bt_timegm(&tm
);
275 result
= mktime(&tm
);
284 bound
->ns_from_origin
= (int64_t) result
;
285 bound
->ns_from_origin
*= NS_PER_S
;
286 bound
->ns_from_origin
+= ns
;
287 bound
->is_set
= true;
294 * Parses a timestamp, figuring out its format.
296 * Returns a negative value if anything goes wrong.
300 * YYYY-MM-DD hh:mm[:ss[.ns]]
304 * TODO: Check overflows.
307 int set_bound_from_str(struct trimmer_comp
*trimmer_comp
,
308 const char *str
, struct trimmer_bound
*bound
, bool is_gmt
)
310 /* Matches YYYY-MM-DD */
311 #define DATE_RE "([0-9]{4})-([0-9]{2})-([0-9]{2})"
313 /* Matches HH:MM[:SS[.NS]] */
314 #define TIME_RE "([0-9]{2}):([0-9]{2})(?::([0-9]{2})(?:\\.([0-9]{1,9}))?)?"
316 /* Matches [-]SS[.NS] */
317 #define S_NS_RE "^(-?)([0-9]+)(?:\\.([0-9]{1,9}))?$"
319 GMatchInfo
*match_info
;
322 /* Try `YYYY-MM-DD hh:mm[:ss[.ns]]` format */
323 if (compile_and_match("^" DATE_RE
" " TIME_RE
"$", str
, &match_info
)) {
324 unsigned int year
= 0, month
= 0, day
= 0, hours
= 0, minutes
= 0, seconds
= 0, nanoseconds
= 0;
325 gint match_count
= g_match_info_get_match_count(match_info
);
327 BT_ASSERT(match_count
>= 6 && match_count
<= 8);
329 year
= match_to_uint(match_info
, 1);
330 month
= match_to_uint(match_info
, 2);
331 day
= match_to_uint(match_info
, 3);
332 hours
= match_to_uint(match_info
, 4);
333 minutes
= match_to_uint(match_info
, 5);
335 if (match_count
>= 7) {
336 seconds
= match_to_uint(match_info
, 6);
339 if (match_count
>= 8) {
340 nanoseconds
= match_to_uint_ns(match_info
, 7);
343 set_bound_ns_from_origin(bound
, year
, month
, day
, hours
, minutes
, seconds
, nanoseconds
, is_gmt
);
348 if (compile_and_match("^" DATE_RE
"$", str
, &match_info
)) {
349 unsigned int year
= 0, month
= 0, day
= 0;
351 BT_ASSERT(g_match_info_get_match_count(match_info
) == 4);
353 year
= match_to_uint(match_info
, 1);
354 month
= match_to_uint(match_info
, 2);
355 day
= match_to_uint(match_info
, 3);
357 set_bound_ns_from_origin(bound
, year
, month
, day
, 0, 0, 0, 0, is_gmt
);
362 /* Try `hh:mm[:ss[.ns]]` format */
363 if (compile_and_match("^" TIME_RE
"$", str
, &match_info
)) {
364 gint match_count
= g_match_info_get_match_count(match_info
);
365 BT_ASSERT(match_count
>= 3 && match_count
<= 5);
366 bound
->time
.hour
= match_to_uint(match_info
, 1);
367 bound
->time
.minute
= match_to_uint(match_info
, 2);
369 if (match_count
>= 4) {
370 bound
->time
.second
= match_to_uint(match_info
, 3);
373 if (match_count
>= 5) {
374 bound
->time
.ns
= match_to_uint_ns(match_info
, 4);
380 /* Try `[-]s[.ns]` format */
381 if (compile_and_match("^" S_NS_RE
"$", str
, &match_info
)) {
382 gboolean is_neg
, fetch_pos_ret
;
383 gint start_pos
, end_pos
, match_count
;
384 guint64 seconds
, nanoseconds
= 0;
386 match_count
= g_match_info_get_match_count(match_info
);
387 BT_ASSERT(match_count
>= 3 && match_count
<= 4);
389 /* Check for presence of negation sign. */
390 fetch_pos_ret
= g_match_info_fetch_pos(match_info
, 1, &start_pos
, &end_pos
);
391 BT_ASSERT(fetch_pos_ret
);
392 is_neg
= (end_pos
- start_pos
) > 0;
394 seconds
= match_to_uint(match_info
, 2);
396 if (match_count
>= 4) {
397 nanoseconds
= match_to_uint_ns(match_info
, 3);
400 bound
->ns_from_origin
= seconds
* NS_PER_S
+ nanoseconds
;
403 bound
->ns_from_origin
= -bound
->ns_from_origin
;
406 bound
->is_set
= true;
411 BT_COMP_LOGE_APPEND_CAUSE(trimmer_comp
->self_comp
,
412 "Invalid date/time format: param=\"%s\"", str
);
416 g_match_info_free(match_info
);
422 * Sets a trimmer bound's properties from a parameter string/integer
425 * Returns a negative value if anything goes wrong.
428 int set_bound_from_param(struct trimmer_comp
*trimmer_comp
,
429 const char *param_name
, const bt_value
*param
,
430 struct trimmer_bound
*bound
, bool is_gmt
)
436 if (bt_value_is_signed_integer(param
)) {
437 int64_t value
= bt_value_integer_signed_get(param
);
440 * Just convert it to a temporary string to handle
441 * everything the same way.
443 sprintf(tmp_arg
, "%" PRId64
, value
);
446 BT_ASSERT(bt_value_is_string(param
));
447 arg
= bt_value_string_get(param
);
450 ret
= set_bound_from_str(trimmer_comp
, arg
, bound
, is_gmt
);
456 int validate_trimmer_bounds(struct trimmer_comp
*trimmer_comp
,
457 struct trimmer_bound
*begin
, struct trimmer_bound
*end
)
461 BT_ASSERT(begin
->is_set
);
462 BT_ASSERT(end
->is_set
);
464 if (!begin
->is_infinite
&& !end
->is_infinite
&&
465 begin
->ns_from_origin
> end
->ns_from_origin
) {
466 BT_COMP_LOGE_APPEND_CAUSE(trimmer_comp
->self_comp
,
467 "Trimming time range's beginning time is greater than end time: "
468 "begin-ns-from-origin=%" PRId64
", "
469 "end-ns-from-origin=%" PRId64
,
470 begin
->ns_from_origin
,
471 end
->ns_from_origin
);
476 if (!begin
->is_infinite
&& begin
->ns_from_origin
== INT64_MIN
) {
477 BT_COMP_LOGE_APPEND_CAUSE(trimmer_comp
->self_comp
,
478 "Invalid trimming time range's beginning time: "
479 "ns-from-origin=%" PRId64
,
480 begin
->ns_from_origin
);
485 if (!end
->is_infinite
&& end
->ns_from_origin
== INT64_MIN
) {
486 BT_COMP_LOGE_APPEND_CAUSE(trimmer_comp
->self_comp
,
487 "Invalid trimming time range's end time: "
488 "ns-from-origin=%" PRId64
,
489 end
->ns_from_origin
);
499 enum bt_param_validation_status
validate_bound_type(
500 const bt_value
*value
,
501 struct bt_param_validation_context
*context
)
503 enum bt_param_validation_status status
= BT_PARAM_VALIDATION_STATUS_OK
;
505 if (!bt_value_is_signed_integer(value
) &&
506 !bt_value_is_string(value
)) {
507 status
= bt_param_validation_error(context
,
508 "unexpected type: expected-types=[%s, %s], actual-type=%s",
509 bt_common_value_type_string(BT_VALUE_TYPE_SIGNED_INTEGER
),
510 bt_common_value_type_string(BT_VALUE_TYPE_STRING
),
511 bt_common_value_type_string(bt_value_get_type(value
)));
518 struct bt_param_validation_map_value_entry_descr trimmer_params
[] = {
519 { "gmt", BT_PARAM_VALIDATION_MAP_VALUE_ENTRY_OPTIONAL
, { .type
= BT_VALUE_TYPE_BOOL
} },
520 { "begin", BT_PARAM_VALIDATION_MAP_VALUE_ENTRY_OPTIONAL
, { .validation_func
= validate_bound_type
} },
521 { "end", BT_PARAM_VALIDATION_MAP_VALUE_ENTRY_OPTIONAL
, { .validation_func
= validate_bound_type
} },
522 BT_PARAM_VALIDATION_MAP_VALUE_ENTRY_END
526 bt_component_class_initialize_method_status
init_trimmer_comp_from_params(
527 struct trimmer_comp
*trimmer_comp
,
528 const bt_value
*params
)
530 const bt_value
*value
;
531 bt_component_class_initialize_method_status status
;
532 enum bt_param_validation_status validation_status
;
533 gchar
*validate_error
= NULL
;
535 validation_status
= bt_param_validation_validate(params
,
536 trimmer_params
, &validate_error
);
537 if (validation_status
== BT_PARAM_VALIDATION_STATUS_MEMORY_ERROR
) {
538 status
= BT_COMPONENT_CLASS_INITIALIZE_METHOD_STATUS_MEMORY_ERROR
;
540 } else if (validation_status
== BT_PARAM_VALIDATION_STATUS_VALIDATION_ERROR
) {
541 status
= BT_COMPONENT_CLASS_INITIALIZE_METHOD_STATUS_ERROR
;
542 BT_COMP_LOGE_APPEND_CAUSE(trimmer_comp
->self_comp
, "%s",
548 value
= bt_value_map_borrow_entry_value_const(params
, "gmt");
550 trimmer_comp
->is_gmt
= (bool) bt_value_bool_get(value
);
553 value
= bt_value_map_borrow_entry_value_const(params
, "begin");
555 if (set_bound_from_param(trimmer_comp
, "begin", value
,
556 &trimmer_comp
->begin
, trimmer_comp
->is_gmt
)) {
557 /* set_bound_from_param() logs errors */
558 status
= BT_COMPONENT_CLASS_INITIALIZE_METHOD_STATUS_ERROR
;
562 trimmer_comp
->begin
.is_infinite
= true;
563 trimmer_comp
->begin
.is_set
= true;
566 value
= bt_value_map_borrow_entry_value_const(params
, "end");
568 if (set_bound_from_param(trimmer_comp
, "end", value
,
569 &trimmer_comp
->end
, trimmer_comp
->is_gmt
)) {
570 /* set_bound_from_param() logs errors */
571 status
= BT_COMPONENT_CLASS_INITIALIZE_METHOD_STATUS_ERROR
;
575 trimmer_comp
->end
.is_infinite
= true;
576 trimmer_comp
->end
.is_set
= true;
579 if (trimmer_comp
->begin
.is_set
&& trimmer_comp
->end
.is_set
) {
580 /* validate_trimmer_bounds() logs errors */
581 if (validate_trimmer_bounds(trimmer_comp
,
582 &trimmer_comp
->begin
, &trimmer_comp
->end
)) {
583 status
= BT_COMPONENT_CLASS_INITIALIZE_METHOD_STATUS_ERROR
;
588 status
= BT_COMPONENT_CLASS_INITIALIZE_METHOD_STATUS_OK
;
591 g_free(validate_error
);
596 bt_component_class_initialize_method_status
trimmer_init(
597 bt_self_component_filter
*self_comp_flt
,
598 bt_self_component_filter_configuration
*config
,
599 const bt_value
*params
, void *init_data
)
601 bt_component_class_initialize_method_status status
;
602 bt_self_component_add_port_status add_port_status
;
603 struct trimmer_comp
*trimmer_comp
= create_trimmer_comp();
604 bt_self_component
*self_comp
=
605 bt_self_component_filter_as_self_component(self_comp_flt
);
608 status
= BT_COMPONENT_CLASS_INITIALIZE_METHOD_STATUS_MEMORY_ERROR
;
612 trimmer_comp
->log_level
= bt_component_get_logging_level(
613 bt_self_component_as_component(self_comp
));
614 trimmer_comp
->self_comp
= self_comp
;
615 trimmer_comp
->self_comp_filter
= self_comp_flt
;
617 add_port_status
= bt_self_component_filter_add_input_port(
618 self_comp_flt
, in_port_name
, NULL
, NULL
);
619 if (add_port_status
!= BT_SELF_COMPONENT_ADD_PORT_STATUS_OK
) {
620 status
= (int) add_port_status
;
624 add_port_status
= bt_self_component_filter_add_output_port(
625 self_comp_flt
, "out", NULL
, NULL
);
626 if (add_port_status
!= BT_SELF_COMPONENT_ADD_PORT_STATUS_OK
) {
627 status
= (int) add_port_status
;
631 status
= init_trimmer_comp_from_params(trimmer_comp
, params
);
632 if (status
!= BT_COMPONENT_CLASS_INITIALIZE_METHOD_STATUS_OK
) {
636 bt_self_component_set_data(self_comp
, trimmer_comp
);
638 status
= BT_COMPONENT_CLASS_INITIALIZE_METHOD_STATUS_OK
;
643 destroy_trimmer_comp(trimmer_comp
);
651 void destroy_trimmer_iterator(struct trimmer_iterator
*trimmer_it
)
657 bt_message_iterator_put_ref(
658 trimmer_it
->upstream_iter
);
660 if (trimmer_it
->output_messages
) {
661 g_queue_free(trimmer_it
->output_messages
);
664 if (trimmer_it
->stream_states
) {
665 g_hash_table_destroy(trimmer_it
->stream_states
);
674 void destroy_trimmer_iterator_stream_state(
675 struct trimmer_iterator_stream_state
*sstate
)
678 BT_PACKET_PUT_REF_AND_RESET(sstate
->cur_packet
);
683 bt_message_iterator_class_initialize_method_status
trimmer_msg_iter_init(
684 bt_self_message_iterator
*self_msg_iter
,
685 bt_self_message_iterator_configuration
*config
,
686 bt_self_component_port_output
*port
)
688 bt_message_iterator_class_initialize_method_status status
;
689 bt_message_iterator_create_from_message_iterator_status
691 struct trimmer_iterator
*trimmer_it
;
692 bt_self_component
*self_comp
=
693 bt_self_message_iterator_borrow_component(self_msg_iter
);
695 trimmer_it
= g_new0(struct trimmer_iterator
, 1);
697 status
= BT_MESSAGE_ITERATOR_CLASS_INITIALIZE_METHOD_STATUS_MEMORY_ERROR
;
701 trimmer_it
->trimmer_comp
= bt_self_component_get_data(self_comp
);
702 BT_ASSERT(trimmer_it
->trimmer_comp
);
704 if (trimmer_it
->trimmer_comp
->begin
.is_set
&&
705 trimmer_it
->trimmer_comp
->end
.is_set
) {
707 * Both trimming time range's bounds are set, so skip
709 * `TRIMMER_ITERATOR_STATE_SET_BOUNDS_NS_FROM_ORIGIN`
712 trimmer_it
->state
= TRIMMER_ITERATOR_STATE_SEEK_INITIALLY
;
715 trimmer_it
->begin
= trimmer_it
->trimmer_comp
->begin
;
716 trimmer_it
->end
= trimmer_it
->trimmer_comp
->end
;
718 bt_message_iterator_create_from_message_iterator(
720 bt_self_component_filter_borrow_input_port_by_name(
721 trimmer_it
->trimmer_comp
->self_comp_filter
, in_port_name
),
722 &trimmer_it
->upstream_iter
);
723 if (msg_iter_status
!= BT_MESSAGE_ITERATOR_CREATE_FROM_MESSAGE_ITERATOR_STATUS_OK
) {
724 status
= (int) msg_iter_status
;
728 trimmer_it
->output_messages
= g_queue_new();
729 if (!trimmer_it
->output_messages
) {
730 status
= BT_MESSAGE_ITERATOR_CLASS_INITIALIZE_METHOD_STATUS_MEMORY_ERROR
;
734 trimmer_it
->stream_states
= g_hash_table_new_full(g_direct_hash
,
735 g_direct_equal
, NULL
,
736 (GDestroyNotify
) destroy_trimmer_iterator_stream_state
);
737 if (!trimmer_it
->stream_states
) {
738 status
= BT_MESSAGE_ITERATOR_CLASS_INITIALIZE_METHOD_STATUS_MEMORY_ERROR
;
743 * The trimmer requires upstream messages to have times, so it can
744 * always seek forward.
746 bt_self_message_iterator_configuration_set_can_seek_forward(
749 trimmer_it
->self_msg_iter
= self_msg_iter
;
750 bt_self_message_iterator_set_data(self_msg_iter
, trimmer_it
);
752 status
= BT_MESSAGE_ITERATOR_CLASS_INITIALIZE_METHOD_STATUS_OK
;
756 destroy_trimmer_iterator(trimmer_it
);
763 int get_msg_ns_from_origin(const bt_message
*msg
, int64_t *ns_from_origin
,
764 bool *has_clock_snapshot
)
766 const bt_clock_class
*clock_class
= NULL
;
767 const bt_clock_snapshot
*clock_snapshot
= NULL
;
771 BT_ASSERT_DBG(ns_from_origin
);
772 BT_ASSERT_DBG(has_clock_snapshot
);
774 switch (bt_message_get_type(msg
)) {
775 case BT_MESSAGE_TYPE_EVENT
:
777 bt_message_event_borrow_stream_class_default_clock_class_const(
779 if (G_UNLIKELY(!clock_class
)) {
783 clock_snapshot
= bt_message_event_borrow_default_clock_snapshot_const(
786 case BT_MESSAGE_TYPE_PACKET_BEGINNING
:
788 bt_message_packet_beginning_borrow_stream_class_default_clock_class_const(
790 if (G_UNLIKELY(!clock_class
)) {
794 clock_snapshot
= bt_message_packet_beginning_borrow_default_clock_snapshot_const(
797 case BT_MESSAGE_TYPE_PACKET_END
:
799 bt_message_packet_end_borrow_stream_class_default_clock_class_const(
801 if (G_UNLIKELY(!clock_class
)) {
805 clock_snapshot
= bt_message_packet_end_borrow_default_clock_snapshot_const(
808 case BT_MESSAGE_TYPE_STREAM_BEGINNING
:
810 enum bt_message_stream_clock_snapshot_state cs_state
;
813 bt_message_stream_beginning_borrow_stream_class_default_clock_class_const(msg
);
814 if (G_UNLIKELY(!clock_class
)) {
818 cs_state
= bt_message_stream_beginning_borrow_default_clock_snapshot_const(msg
, &clock_snapshot
);
819 if (cs_state
!= BT_MESSAGE_STREAM_CLOCK_SNAPSHOT_STATE_KNOWN
) {
820 goto no_clock_snapshot
;
825 case BT_MESSAGE_TYPE_STREAM_END
:
827 enum bt_message_stream_clock_snapshot_state cs_state
;
830 bt_message_stream_end_borrow_stream_class_default_clock_class_const(msg
);
831 if (G_UNLIKELY(!clock_class
)) {
835 cs_state
= bt_message_stream_end_borrow_default_clock_snapshot_const(msg
, &clock_snapshot
);
836 if (cs_state
!= BT_MESSAGE_STREAM_CLOCK_SNAPSHOT_STATE_KNOWN
) {
837 goto no_clock_snapshot
;
842 case BT_MESSAGE_TYPE_DISCARDED_EVENTS
:
844 bt_message_discarded_events_borrow_stream_class_default_clock_class_const(
846 if (G_UNLIKELY(!clock_class
)) {
850 clock_snapshot
= bt_message_discarded_events_borrow_beginning_default_clock_snapshot_const(
853 case BT_MESSAGE_TYPE_DISCARDED_PACKETS
:
855 bt_message_discarded_packets_borrow_stream_class_default_clock_class_const(
857 if (G_UNLIKELY(!clock_class
)) {
861 clock_snapshot
= bt_message_discarded_packets_borrow_beginning_default_clock_snapshot_const(
864 case BT_MESSAGE_TYPE_MESSAGE_ITERATOR_INACTIVITY
:
866 bt_message_message_iterator_inactivity_borrow_clock_snapshot_const(
870 goto no_clock_snapshot
;
873 ret
= bt_clock_snapshot_get_ns_from_origin(clock_snapshot
,
875 if (G_UNLIKELY(ret
)) {
879 *has_clock_snapshot
= true;
883 *has_clock_snapshot
= false;
894 void put_messages(bt_message_array_const msgs
, uint64_t count
)
898 for (i
= 0; i
< count
; i
++) {
899 BT_MESSAGE_PUT_REF_AND_RESET(msgs
[i
]);
904 int set_trimmer_iterator_bound(struct trimmer_iterator
*trimmer_it
,
905 struct trimmer_bound
*bound
, int64_t ns_from_origin
,
908 struct trimmer_comp
*trimmer_comp
= trimmer_it
->trimmer_comp
;
911 time_t time_seconds
= (time_t) (ns_from_origin
/ NS_PER_S
);
914 BT_ASSERT(!bound
->is_set
);
917 /* We only need to extract the date from this time */
919 res
= bt_gmtime_r(&time_seconds
, &tm
);
921 res
= bt_localtime_r(&time_seconds
, &tm
);
925 BT_COMP_LOGE_APPEND_CAUSE_ERRNO(trimmer_comp
->self_comp
,
926 "Cannot convert timestamp to date and time",
927 ": ts=%" PRId64
, (int64_t) time_seconds
);
932 ret
= set_bound_ns_from_origin(bound
, tm
.tm_year
+ 1900, tm
.tm_mon
+ 1,
933 tm
.tm_mday
, bound
->time
.hour
, bound
->time
.minute
,
934 bound
->time
.second
, bound
->time
.ns
, is_gmt
);
941 bt_message_iterator_class_next_method_status
942 state_set_trimmer_iterator_bounds(
943 struct trimmer_iterator
*trimmer_it
)
945 bt_message_iterator_next_status upstream_iter_status
=
946 BT_MESSAGE_ITERATOR_NEXT_STATUS_OK
;
947 struct trimmer_comp
*trimmer_comp
= trimmer_it
->trimmer_comp
;
948 bt_message_array_const msgs
;
950 int64_t ns_from_origin
= INT64_MIN
;
954 BT_ASSERT(!trimmer_it
->begin
.is_set
||
955 !trimmer_it
->end
.is_set
);
958 upstream_iter_status
=
959 bt_message_iterator_next(
960 trimmer_it
->upstream_iter
, &msgs
, &count
);
961 if (upstream_iter_status
!= BT_MESSAGE_ITERATOR_NEXT_STATUS_OK
) {
965 for (i
= 0; i
< count
; i
++) {
966 const bt_message
*msg
= msgs
[i
];
967 bool has_ns_from_origin
;
968 ret
= get_msg_ns_from_origin(msg
, &ns_from_origin
,
969 &has_ns_from_origin
);
974 if (!has_ns_from_origin
) {
978 BT_ASSERT_DBG(ns_from_origin
!= INT64_MIN
&&
979 ns_from_origin
!= INT64_MAX
);
980 put_messages(msgs
, count
);
984 put_messages(msgs
, count
);
988 if (!trimmer_it
->begin
.is_set
) {
989 BT_ASSERT(!trimmer_it
->begin
.is_infinite
);
990 ret
= set_trimmer_iterator_bound(trimmer_it
, &trimmer_it
->begin
,
991 ns_from_origin
, trimmer_comp
->is_gmt
);
997 if (!trimmer_it
->end
.is_set
) {
998 BT_ASSERT(!trimmer_it
->end
.is_infinite
);
999 ret
= set_trimmer_iterator_bound(trimmer_it
, &trimmer_it
->end
,
1000 ns_from_origin
, trimmer_comp
->is_gmt
);
1006 ret
= validate_trimmer_bounds(trimmer_it
->trimmer_comp
,
1007 &trimmer_it
->begin
, &trimmer_it
->end
);
1015 put_messages(msgs
, count
);
1016 upstream_iter_status
= BT_MESSAGE_ITERATOR_NEXT_STATUS_ERROR
;
1019 return (int) upstream_iter_status
;
1023 bt_message_iterator_class_next_method_status
state_seek_initially(
1024 struct trimmer_iterator
*trimmer_it
)
1026 struct trimmer_comp
*trimmer_comp
= trimmer_it
->trimmer_comp
;
1027 bt_message_iterator_class_next_method_status status
;
1029 BT_ASSERT(trimmer_it
->begin
.is_set
);
1031 if (trimmer_it
->begin
.is_infinite
) {
1034 status
= (int) bt_message_iterator_can_seek_beginning(
1035 trimmer_it
->upstream_iter
, &can_seek
);
1036 if (status
!= BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_OK
) {
1038 BT_COMP_LOGE_APPEND_CAUSE(trimmer_comp
->self_comp
,
1039 "Cannot make upstream message iterator initially seek its beginning.");
1046 BT_COMP_LOGE_APPEND_CAUSE(trimmer_comp
->self_comp
,
1047 "Cannot make upstream message iterator initially seek its beginning.");
1048 status
= BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_ERROR
;
1052 status
= (int) bt_message_iterator_seek_beginning(
1053 trimmer_it
->upstream_iter
);
1057 status
= (int) bt_message_iterator_can_seek_ns_from_origin(
1058 trimmer_it
->upstream_iter
, trimmer_it
->begin
.ns_from_origin
,
1061 if (status
!= BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_OK
) {
1063 BT_COMP_LOGE_APPEND_CAUSE(trimmer_comp
->self_comp
,
1064 "Cannot make upstream message iterator initially seek: seek-ns-from-origin=%" PRId64
,
1065 trimmer_it
->begin
.ns_from_origin
);
1072 BT_COMP_LOGE_APPEND_CAUSE(trimmer_comp
->self_comp
,
1073 "Cannot make upstream message iterator initially seek: seek-ns-from-origin=%" PRId64
,
1074 trimmer_it
->begin
.ns_from_origin
);
1075 status
= BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_ERROR
;
1079 status
= (int) bt_message_iterator_seek_ns_from_origin(
1080 trimmer_it
->upstream_iter
, trimmer_it
->begin
.ns_from_origin
);
1083 if (status
== BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_OK
) {
1084 trimmer_it
->state
= TRIMMER_ITERATOR_STATE_TRIM
;
1092 void push_message(struct trimmer_iterator
*trimmer_it
, const bt_message
*msg
)
1094 g_queue_push_head(trimmer_it
->output_messages
, (void *) msg
);
1098 const bt_message
*pop_message(struct trimmer_iterator
*trimmer_it
)
1100 return g_queue_pop_tail(trimmer_it
->output_messages
);
1104 int clock_raw_value_from_ns_from_origin(const bt_clock_class
*clock_class
,
1105 int64_t ns_from_origin
, uint64_t *raw_value
)
1108 int64_t cc_offset_s
;
1109 uint64_t cc_offset_cycles
;
1112 bt_clock_class_get_offset(clock_class
, &cc_offset_s
, &cc_offset_cycles
);
1113 cc_freq
= bt_clock_class_get_frequency(clock_class
);
1114 return bt_common_clock_value_from_ns_from_origin(cc_offset_s
,
1115 cc_offset_cycles
, cc_freq
, ns_from_origin
, raw_value
);
1119 bt_message_iterator_class_next_method_status
1120 end_stream(struct trimmer_iterator
*trimmer_it
,
1121 struct trimmer_iterator_stream_state
*sstate
)
1123 bt_message_iterator_class_next_method_status status
=
1124 BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_OK
;
1125 /* Initialize to silence maybe-uninitialized warning. */
1126 uint64_t raw_value
= 0;
1127 bt_message
*msg
= NULL
;
1129 BT_ASSERT(!trimmer_it
->end
.is_infinite
);
1130 BT_ASSERT(sstate
->stream
);
1133 * If we haven't seen a message with a clock snapshot, we don't know if the trimmer's end bound is within
1134 * the clock's range, so it wouldn't be safe to try to convert ns_from_origin to a clock value.
1136 * Also, it would be a bit of a lie to generate a stream end message with the end bound as its
1137 * clock snapshot, because we don't really know if the stream existed at that time. If we have
1138 * seen a message with a clock snapshot and the stream is cut short by another message with a
1139 * clock snapshot, then we are sure that the the end bound time is not below the clock range,
1140 * and we know the stream was active at that time (and that we cut it short).
1142 if (sstate
->seen_clock_snapshot
) {
1143 const bt_clock_class
*clock_class
;
1146 clock_class
= bt_stream_class_borrow_default_clock_class_const(
1147 bt_stream_borrow_class_const(sstate
->stream
));
1148 BT_ASSERT(clock_class
);
1149 ret
= clock_raw_value_from_ns_from_origin(clock_class
,
1150 trimmer_it
->end
.ns_from_origin
, &raw_value
);
1152 status
= BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_ERROR
;
1157 if (sstate
->cur_packet
) {
1159 * Create and push a packet end message, making its time
1160 * the trimming range's end time.
1162 * We know that we must have seen a clock snapshot, the one in
1163 * the packet beginning message, since trimmer currently
1164 * requires packet messages to have clock snapshots (see comment
1165 * in create_stream_state_entry).
1167 BT_ASSERT(sstate
->seen_clock_snapshot
);
1169 msg
= bt_message_packet_end_create_with_default_clock_snapshot(
1170 trimmer_it
->self_msg_iter
, sstate
->cur_packet
,
1173 status
= BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_MEMORY_ERROR
;
1177 push_message(trimmer_it
, msg
);
1179 BT_PACKET_PUT_REF_AND_RESET(sstate
->cur_packet
);
1182 /* Create and push a stream end message. */
1183 msg
= bt_message_stream_end_create(trimmer_it
->self_msg_iter
,
1186 status
= BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_MEMORY_ERROR
;
1190 if (sstate
->seen_clock_snapshot
) {
1191 bt_message_stream_end_set_default_clock_snapshot(msg
, raw_value
);
1194 push_message(trimmer_it
, msg
);
1198 * Just to make sure that we don't use this stream state again
1199 * in the future without an obvious error.
1201 sstate
->stream
= NULL
;
1204 bt_message_put_ref(msg
);
1209 bt_message_iterator_class_next_method_status
end_iterator_streams(
1210 struct trimmer_iterator
*trimmer_it
)
1212 bt_message_iterator_class_next_method_status status
=
1213 BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_OK
;
1214 GHashTableIter iter
;
1215 gpointer key
, sstate
;
1217 if (trimmer_it
->end
.is_infinite
) {
1219 * An infinite trimming range's end time guarantees that
1220 * we received (and pushed) all the appropriate end
1227 * End each stream and then remove them from the hash table of
1228 * stream states to release unneeded references.
1230 g_hash_table_iter_init(&iter
, trimmer_it
->stream_states
);
1232 while (g_hash_table_iter_next(&iter
, &key
, &sstate
)) {
1233 status
= end_stream(trimmer_it
, sstate
);
1240 g_hash_table_remove_all(trimmer_it
->stream_states
);
1247 bt_message_iterator_class_next_method_status
1248 create_stream_state_entry(
1249 struct trimmer_iterator
*trimmer_it
,
1250 const struct bt_stream
*stream
,
1251 struct trimmer_iterator_stream_state
**stream_state
)
1253 struct trimmer_comp
*trimmer_comp
= trimmer_it
->trimmer_comp
;
1254 bt_message_iterator_class_next_method_status status
;
1255 struct trimmer_iterator_stream_state
*sstate
;
1256 const bt_stream_class
*sc
;
1258 BT_ASSERT(!bt_g_hash_table_contains(trimmer_it
->stream_states
, stream
));
1261 * Validate right now that the stream's class
1262 * has a registered default clock class so that
1263 * an existing stream state guarantees existing
1264 * default clock snapshots for its associated
1267 * Also check that clock snapshots are always
1270 sc
= bt_stream_borrow_class_const(stream
);
1271 if (!bt_stream_class_borrow_default_clock_class_const(sc
)) {
1272 BT_COMP_LOGE_APPEND_CAUSE(trimmer_comp
->self_comp
,
1273 "Unsupported stream: stream class does "
1274 "not have a default clock class: "
1276 "stream-id=%" PRIu64
", "
1277 "stream-name=\"%s\"",
1278 stream
, bt_stream_get_id(stream
),
1279 bt_stream_get_name(stream
));
1280 status
= BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_ERROR
;
1285 * Temporary: make sure packet beginning, packet
1286 * end, discarded events, and discarded packets
1287 * messages have default clock snapshots until
1288 * the support for not having them is
1291 if (bt_stream_class_supports_packets(sc
)) {
1292 if (!bt_stream_class_packets_have_beginning_default_clock_snapshot(
1294 BT_COMP_LOGE_APPEND_CAUSE(trimmer_comp
->self_comp
,
1295 "Unsupported stream: packets have no beginning clock snapshot: "
1297 "stream-id=%" PRIu64
", "
1298 "stream-name=\"%s\"",
1299 stream
, bt_stream_get_id(stream
),
1300 bt_stream_get_name(stream
));
1301 status
= BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_ERROR
;
1305 if (!bt_stream_class_packets_have_end_default_clock_snapshot(
1307 BT_COMP_LOGE_APPEND_CAUSE(trimmer_comp
->self_comp
,
1308 "Unsupported stream: packets have no end clock snapshot: "
1310 "stream-id=%" PRIu64
", "
1311 "stream-name=\"%s\"",
1312 stream
, bt_stream_get_id(stream
),
1313 bt_stream_get_name(stream
));
1314 status
= BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_ERROR
;
1318 if (bt_stream_class_supports_discarded_packets(sc
) &&
1319 !bt_stream_class_discarded_packets_have_default_clock_snapshots(sc
)) {
1320 BT_COMP_LOGE_APPEND_CAUSE(trimmer_comp
->self_comp
,
1321 "Unsupported stream: discarded packets "
1322 "have no clock snapshots: "
1324 "stream-id=%" PRIu64
", "
1325 "stream-name=\"%s\"",
1326 stream
, bt_stream_get_id(stream
),
1327 bt_stream_get_name(stream
));
1328 status
= BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_ERROR
;
1333 if (bt_stream_class_supports_discarded_events(sc
) &&
1334 !bt_stream_class_discarded_events_have_default_clock_snapshots(sc
)) {
1335 BT_COMP_LOGE_APPEND_CAUSE(trimmer_comp
->self_comp
,
1336 "Unsupported stream: discarded events have no clock snapshots: "
1338 "stream-id=%" PRIu64
", "
1339 "stream-name=\"%s\"",
1340 stream
, bt_stream_get_id(stream
),
1341 bt_stream_get_name(stream
));
1342 status
= BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_ERROR
;
1346 sstate
= g_new0(struct trimmer_iterator_stream_state
, 1);
1348 status
= BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_MEMORY_ERROR
;
1352 sstate
->stream
= stream
;
1354 g_hash_table_insert(trimmer_it
->stream_states
, (void *) stream
, sstate
);
1356 *stream_state
= sstate
;
1358 status
= BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_OK
;
1365 struct trimmer_iterator_stream_state
*get_stream_state_entry(
1366 struct trimmer_iterator
*trimmer_it
,
1367 const struct bt_stream
*stream
)
1369 struct trimmer_iterator_stream_state
*sstate
;
1371 BT_ASSERT_DBG(stream
);
1372 sstate
= g_hash_table_lookup(trimmer_it
->stream_states
, stream
);
1373 BT_ASSERT_DBG(sstate
);
1379 * Handles a message which is associated to a given stream state. This
1380 * _could_ make the iterator's output message queue grow; this could
1381 * also consume the message without pushing anything to this queue, only
1382 * modifying the stream state.
1384 * This function consumes the `msg` reference, _whatever the outcome_.
1386 * If non-NULL, `ns_from_origin` is the message's time, as given by
1387 * get_msg_ns_from_origin(). If NULL, the message doesn't have a time.
1389 * This function sets `reached_end` if handling this message made the
1390 * iterator reach the end of the trimming range. Note that the output
1391 * message queue could contain messages even if this function sets
1395 bt_message_iterator_class_next_method_status
1396 handle_message_with_stream(
1397 struct trimmer_iterator
*trimmer_it
, const bt_message
*msg
,
1398 const struct bt_stream
*stream
, const int64_t *ns_from_origin
,
1401 bt_message_iterator_class_next_method_status status
=
1402 BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_OK
;
1403 bt_message_type msg_type
= bt_message_get_type(msg
);
1405 struct trimmer_iterator_stream_state
*sstate
= NULL
;
1408 * Retrieve the stream's state - except if the message is stream
1409 * beginning, in which case we don't know about about this stream yet.
1411 if (msg_type
!= BT_MESSAGE_TYPE_STREAM_BEGINNING
) {
1412 sstate
= get_stream_state_entry(trimmer_it
, stream
);
1416 case BT_MESSAGE_TYPE_EVENT
:
1418 * Event messages always have a clock snapshot if the stream
1419 * class has a clock class. And we know it has, otherwise we
1420 * couldn't be using the trimmer component.
1422 BT_ASSERT_DBG(ns_from_origin
);
1424 if (G_UNLIKELY(!trimmer_it
->end
.is_infinite
&&
1425 *ns_from_origin
> trimmer_it
->end
.ns_from_origin
)) {
1426 status
= end_iterator_streams(trimmer_it
);
1427 *reached_end
= true;
1431 sstate
->seen_clock_snapshot
= true;
1433 push_message(trimmer_it
, msg
);
1437 case BT_MESSAGE_TYPE_PACKET_BEGINNING
:
1439 * Packet beginning messages won't have a clock snapshot if
1440 * stream_class->packets_have_beginning_default_clock_snapshot
1441 * is false. But for now, assume they always do.
1443 BT_ASSERT(ns_from_origin
);
1444 BT_ASSERT(!sstate
->cur_packet
);
1446 if (G_UNLIKELY(!trimmer_it
->end
.is_infinite
&&
1447 *ns_from_origin
> trimmer_it
->end
.ns_from_origin
)) {
1448 status
= end_iterator_streams(trimmer_it
);
1449 *reached_end
= true;
1453 sstate
->cur_packet
=
1454 bt_message_packet_beginning_borrow_packet_const(msg
);
1455 bt_packet_get_ref(sstate
->cur_packet
);
1457 sstate
->seen_clock_snapshot
= true;
1459 push_message(trimmer_it
, msg
);
1463 case BT_MESSAGE_TYPE_PACKET_END
:
1465 * Packet end messages won't have a clock snapshot if
1466 * stream_class->packets_have_end_default_clock_snapshot
1467 * is false. But for now, assume they always do.
1469 BT_ASSERT(ns_from_origin
);
1470 BT_ASSERT(sstate
->cur_packet
);
1472 if (G_UNLIKELY(!trimmer_it
->end
.is_infinite
&&
1473 *ns_from_origin
> trimmer_it
->end
.ns_from_origin
)) {
1474 status
= end_iterator_streams(trimmer_it
);
1475 *reached_end
= true;
1479 BT_PACKET_PUT_REF_AND_RESET(sstate
->cur_packet
);
1481 sstate
->seen_clock_snapshot
= true;
1483 push_message(trimmer_it
, msg
);
1487 case BT_MESSAGE_TYPE_DISCARDED_EVENTS
:
1488 case BT_MESSAGE_TYPE_DISCARDED_PACKETS
:
1491 * `ns_from_origin` is the message's time range's
1492 * beginning time here.
1494 int64_t end_ns_from_origin
;
1495 const bt_clock_snapshot
*end_cs
;
1497 BT_ASSERT(ns_from_origin
);
1499 sstate
->seen_clock_snapshot
= true;
1501 if (bt_message_get_type(msg
) ==
1502 BT_MESSAGE_TYPE_DISCARDED_EVENTS
) {
1504 * Safe to ignore the return value because we
1505 * know there's a default clock and it's always
1508 end_cs
= bt_message_discarded_events_borrow_end_default_clock_snapshot_const(
1512 * Safe to ignore the return value because we
1513 * know there's a default clock and it's always
1516 end_cs
= bt_message_discarded_packets_borrow_end_default_clock_snapshot_const(
1520 if (bt_clock_snapshot_get_ns_from_origin(end_cs
,
1521 &end_ns_from_origin
)) {
1522 status
= BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_ERROR
;
1526 if (!trimmer_it
->end
.is_infinite
&&
1527 *ns_from_origin
> trimmer_it
->end
.ns_from_origin
) {
1528 status
= end_iterator_streams(trimmer_it
);
1529 *reached_end
= true;
1533 if (!trimmer_it
->end
.is_infinite
&&
1534 end_ns_from_origin
> trimmer_it
->end
.ns_from_origin
) {
1536 * This message's end time is outside the
1537 * trimming time range: replace it with a new
1538 * message having an end time equal to the
1539 * trimming time range's end and without a
1542 const bt_clock_class
*clock_class
=
1543 bt_clock_snapshot_borrow_clock_class_const(
1545 const bt_clock_snapshot
*begin_cs
;
1546 bt_message
*new_msg
;
1547 uint64_t end_raw_value
;
1549 ret
= clock_raw_value_from_ns_from_origin(clock_class
,
1550 trimmer_it
->end
.ns_from_origin
, &end_raw_value
);
1552 status
= BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_ERROR
;
1556 if (msg_type
== BT_MESSAGE_TYPE_DISCARDED_EVENTS
) {
1557 begin_cs
= bt_message_discarded_events_borrow_beginning_default_clock_snapshot_const(
1559 new_msg
= bt_message_discarded_events_create_with_default_clock_snapshots(
1560 trimmer_it
->self_msg_iter
,
1562 bt_clock_snapshot_get_value(begin_cs
),
1565 begin_cs
= bt_message_discarded_packets_borrow_beginning_default_clock_snapshot_const(
1567 new_msg
= bt_message_discarded_packets_create_with_default_clock_snapshots(
1568 trimmer_it
->self_msg_iter
,
1570 bt_clock_snapshot_get_value(begin_cs
),
1575 status
= BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_MEMORY_ERROR
;
1579 /* Replace the original message */
1580 BT_MESSAGE_MOVE_REF(msg
, new_msg
);
1583 push_message(trimmer_it
, msg
);
1588 case BT_MESSAGE_TYPE_STREAM_BEGINNING
:
1590 * If this message has a time and this time is greater than the
1591 * trimmer's end bound, it triggers the end of the trim window.
1593 if (G_UNLIKELY(ns_from_origin
&& !trimmer_it
->end
.is_infinite
&&
1594 *ns_from_origin
> trimmer_it
->end
.ns_from_origin
)) {
1595 status
= end_iterator_streams(trimmer_it
);
1596 *reached_end
= true;
1600 /* Learn about this stream. */
1601 status
= create_stream_state_entry(trimmer_it
, stream
, &sstate
);
1602 if (status
!= BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_OK
) {
1606 if (ns_from_origin
) {
1607 sstate
->seen_clock_snapshot
= true;
1610 push_message(trimmer_it
, msg
);
1613 case BT_MESSAGE_TYPE_STREAM_END
:
1618 * If this message has a time and this time is greater than the
1619 * trimmer's end bound, it triggers the end of the trim window.
1621 if (G_UNLIKELY(ns_from_origin
&& !trimmer_it
->end
.is_infinite
&&
1622 *ns_from_origin
> trimmer_it
->end
.ns_from_origin
)) {
1623 status
= end_iterator_streams(trimmer_it
);
1624 *reached_end
= true;
1629 * Either the stream end message's time is within the trimmer's
1630 * bounds, or it doesn't have a time. In both cases, pass
1631 * the message unmodified.
1633 push_message(trimmer_it
, msg
);
1636 /* Forget about this stream. */
1637 removed
= g_hash_table_remove(trimmer_it
->stream_states
, sstate
->stream
);
1646 /* We release the message's reference whatever the outcome */
1647 bt_message_put_ref(msg
);
1652 * Handles an input message. This _could_ make the iterator's output
1653 * message queue grow; this could also consume the message without
1654 * pushing anything to this queue, only modifying the stream state.
1656 * This function consumes the `msg` reference, _whatever the outcome_.
1658 * This function sets `reached_end` if handling this message made the
1659 * iterator reach the end of the trimming range. Note that the output
1660 * message queue could contain messages even if this function sets
1664 bt_message_iterator_class_next_method_status
handle_message(
1665 struct trimmer_iterator
*trimmer_it
, const bt_message
*msg
,
1668 bt_message_iterator_class_next_method_status status
;
1669 const bt_stream
*stream
= NULL
;
1670 int64_t ns_from_origin
= INT64_MIN
;
1671 bool has_ns_from_origin
= false;
1674 /* Find message's associated stream */
1675 switch (bt_message_get_type(msg
)) {
1676 case BT_MESSAGE_TYPE_EVENT
:
1677 stream
= bt_event_borrow_stream_const(
1678 bt_message_event_borrow_event_const(msg
));
1680 case BT_MESSAGE_TYPE_PACKET_BEGINNING
:
1681 stream
= bt_packet_borrow_stream_const(
1682 bt_message_packet_beginning_borrow_packet_const(msg
));
1684 case BT_MESSAGE_TYPE_PACKET_END
:
1685 stream
= bt_packet_borrow_stream_const(
1686 bt_message_packet_end_borrow_packet_const(msg
));
1688 case BT_MESSAGE_TYPE_DISCARDED_EVENTS
:
1689 stream
= bt_message_discarded_events_borrow_stream_const(msg
);
1691 case BT_MESSAGE_TYPE_DISCARDED_PACKETS
:
1692 stream
= bt_message_discarded_packets_borrow_stream_const(msg
);
1694 case BT_MESSAGE_TYPE_STREAM_BEGINNING
:
1695 stream
= bt_message_stream_beginning_borrow_stream_const(msg
);
1697 case BT_MESSAGE_TYPE_STREAM_END
:
1698 stream
= bt_message_stream_end_borrow_stream_const(msg
);
1704 /* Retrieve the message's time */
1705 ret
= get_msg_ns_from_origin(msg
, &ns_from_origin
, &has_ns_from_origin
);
1706 if (G_UNLIKELY(ret
)) {
1707 status
= BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_ERROR
;
1711 if (G_LIKELY(stream
)) {
1712 /* Message associated to a stream */
1713 status
= handle_message_with_stream(trimmer_it
, msg
,
1714 stream
, has_ns_from_origin
? &ns_from_origin
: NULL
, reached_end
);
1717 * handle_message_with_stream_state() unconditionally
1723 * Message not associated to a stream (message iterator
1726 if (G_UNLIKELY(ns_from_origin
> trimmer_it
->end
.ns_from_origin
)) {
1727 BT_MESSAGE_PUT_REF_AND_RESET(msg
);
1728 status
= end_iterator_streams(trimmer_it
);
1729 *reached_end
= true;
1731 push_message(trimmer_it
, msg
);
1732 status
= BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_OK
;
1738 /* We release the message's reference whatever the outcome */
1739 bt_message_put_ref(msg
);
1744 void fill_message_array_from_output_messages(
1745 struct trimmer_iterator
*trimmer_it
,
1746 bt_message_array_const msgs
, uint64_t capacity
, uint64_t *count
)
1751 * Move auto-seek messages to the output array (which is this
1752 * iterator's base message array).
1754 while (capacity
> 0 && !g_queue_is_empty(trimmer_it
->output_messages
)) {
1755 msgs
[*count
] = pop_message(trimmer_it
);
1760 BT_ASSERT_DBG(*count
> 0);
1764 bt_message_iterator_class_next_method_status
state_ending(
1765 struct trimmer_iterator
*trimmer_it
,
1766 bt_message_array_const msgs
, uint64_t capacity
,
1769 bt_message_iterator_class_next_method_status status
=
1770 BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_OK
;
1772 if (g_queue_is_empty(trimmer_it
->output_messages
)) {
1773 trimmer_it
->state
= TRIMMER_ITERATOR_STATE_ENDED
;
1774 status
= BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_END
;
1778 fill_message_array_from_output_messages(trimmer_it
, msgs
,
1786 bt_message_iterator_class_next_method_status
1787 state_trim(struct trimmer_iterator
*trimmer_it
,
1788 bt_message_array_const msgs
, uint64_t capacity
,
1791 bt_message_iterator_class_next_method_status status
=
1792 BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_OK
;
1793 bt_message_array_const my_msgs
;
1796 bool reached_end
= false;
1798 while (g_queue_is_empty(trimmer_it
->output_messages
)) {
1799 status
= (int) bt_message_iterator_next(
1800 trimmer_it
->upstream_iter
, &my_msgs
, &my_count
);
1801 if (G_UNLIKELY(status
!= BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_OK
)) {
1802 if (status
== BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_END
) {
1803 status
= end_iterator_streams(trimmer_it
);
1804 if (status
!= BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_OK
) {
1809 TRIMMER_ITERATOR_STATE_ENDING
;
1810 status
= state_ending(trimmer_it
, msgs
,
1817 BT_ASSERT_DBG(my_count
> 0);
1819 for (i
= 0; i
< my_count
; i
++) {
1820 status
= handle_message(trimmer_it
, my_msgs
[i
],
1824 * handle_message() unconditionally consumes the
1825 * message reference.
1829 if (G_UNLIKELY(status
!=
1830 BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_OK
)) {
1831 put_messages(my_msgs
, my_count
);
1835 if (G_UNLIKELY(reached_end
)) {
1837 * This message's time was passed the
1838 * trimming time range's end time: we
1839 * are done. Their might still be
1840 * messages in the output message queue,
1841 * so move to the "ending" state and
1842 * apply it immediately since
1843 * state_trim() is called within the
1846 put_messages(my_msgs
, my_count
);
1848 TRIMMER_ITERATOR_STATE_ENDING
;
1849 status
= state_ending(trimmer_it
, msgs
,
1857 * There's at least one message in the output message queue:
1858 * move the messages to the output message array.
1860 BT_ASSERT_DBG(!g_queue_is_empty(trimmer_it
->output_messages
));
1861 fill_message_array_from_output_messages(trimmer_it
, msgs
,
1869 bt_message_iterator_class_next_method_status
trimmer_msg_iter_next(
1870 bt_self_message_iterator
*self_msg_iter
,
1871 bt_message_array_const msgs
, uint64_t capacity
,
1874 struct trimmer_iterator
*trimmer_it
=
1875 bt_self_message_iterator_get_data(self_msg_iter
);
1876 bt_message_iterator_class_next_method_status status
=
1877 BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_OK
;
1879 BT_ASSERT_DBG(trimmer_it
);
1881 if (G_LIKELY(trimmer_it
->state
== TRIMMER_ITERATOR_STATE_TRIM
)) {
1882 status
= state_trim(trimmer_it
, msgs
, capacity
, count
);
1883 if (status
!= BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_OK
) {
1887 switch (trimmer_it
->state
) {
1888 case TRIMMER_ITERATOR_STATE_SET_BOUNDS_NS_FROM_ORIGIN
:
1889 status
= state_set_trimmer_iterator_bounds(trimmer_it
);
1890 if (status
!= BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_OK
) {
1894 status
= state_seek_initially(trimmer_it
);
1895 if (status
!= BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_OK
) {
1899 status
= state_trim(trimmer_it
, msgs
, capacity
, count
);
1900 if (status
!= BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_OK
) {
1905 case TRIMMER_ITERATOR_STATE_SEEK_INITIALLY
:
1906 status
= state_seek_initially(trimmer_it
);
1907 if (status
!= BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_OK
) {
1911 status
= state_trim(trimmer_it
, msgs
, capacity
, count
);
1912 if (status
!= BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_OK
) {
1917 case TRIMMER_ITERATOR_STATE_ENDING
:
1918 status
= state_ending(trimmer_it
, msgs
, capacity
,
1920 if (status
!= BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_OK
) {
1925 case TRIMMER_ITERATOR_STATE_ENDED
:
1926 status
= BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_END
;
1938 void trimmer_msg_iter_finalize(bt_self_message_iterator
*self_msg_iter
)
1940 struct trimmer_iterator
*trimmer_it
=
1941 bt_self_message_iterator_get_data(self_msg_iter
);
1943 BT_ASSERT(trimmer_it
);
1944 destroy_trimmer_iterator(trimmer_it
);