2 * Copyright 2016 Jérémie Galarneau <jeremie.galarneau@efficios.com>
3 * Copyright 2019 Philippe Proulx <pproulx@efficios.com>
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:
12 * The above copyright notice and this permission notice shall be included in
13 * all copies or substantial portions of the Software.
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
24 #define BT_COMP_LOG_SELF_COMP (trimmer_comp->self_comp)
25 #define BT_LOG_OUTPUT_LEVEL (trimmer_comp->log_level)
26 #define BT_LOG_TAG "PLUGIN/FLT.UTILS.TRIMMER"
27 #include "logging/comp-logging.h"
29 #include "compat/utc.h"
30 #include "compat/time.h"
31 #include <babeltrace2/babeltrace.h>
32 #include "common/common.h"
33 #include "common/assert.h"
37 #include "compat/glib.h"
41 #define NS_PER_S INT64_C(1000000000)
43 static const char * const in_port_name
= "in";
46 unsigned int hour
, minute
, second
, ns
;
49 struct trimmer_bound
{
51 * Nanoseconds from origin, valid if `is_set` is set and
52 * `is_infinite` is false.
54 int64_t ns_from_origin
;
56 /* True if this bound's full time (`ns_from_origin`) is set */
60 * True if this bound represents the infinity (negative or
61 * positive depending on which bound it is). If this is true,
62 * then we don't care about `ns_from_origin` above.
67 * This bound's time without the date; this time is used to set
68 * `ns_from_origin` once we know the date.
70 struct trimmer_time time
;
74 struct trimmer_bound begin
, end
;
76 bt_logging_level log_level
;
77 bt_self_component
*self_comp
;
80 enum trimmer_iterator_state
{
82 * Find the first message's date and set the bounds's times
85 TRIMMER_ITERATOR_STATE_SET_BOUNDS_NS_FROM_ORIGIN
,
88 * Initially seek to the trimming range's beginning time.
90 TRIMMER_ITERATOR_STATE_SEEK_INITIALLY
,
93 * Fill the output message queue for as long as received input
94 * messages are within the trimming time range.
96 TRIMMER_ITERATOR_STATE_TRIM
,
98 /* Flush the remaining messages in the output message queue */
99 TRIMMER_ITERATOR_STATE_ENDING
,
101 /* Trimming operation and message iterator is ended */
102 TRIMMER_ITERATOR_STATE_ENDED
,
105 struct trimmer_iterator
{
107 struct trimmer_comp
*trimmer_comp
;
110 bt_self_message_iterator
*self_msg_iter
;
112 enum trimmer_iterator_state state
;
115 bt_self_component_port_input_message_iterator
*upstream_iter
;
116 struct trimmer_bound begin
, end
;
119 * Queue of `const bt_message *` (owned by the queue).
121 * This is where the trimming operation pushes the messages to
122 * output by this message iterator.
124 GQueue
*output_messages
;
127 * Hash table of `bt_stream *` (weak) to
128 * `struct trimmer_iterator_stream_state *` (owned by the HT).
130 GHashTable
*stream_states
;
133 struct trimmer_iterator_stream_state
{
135 const bt_stream
*stream
;
137 /* Have we seen a message with clock_snapshot going through this stream? */
138 bool seen_clock_snapshot
;
140 /* Owned by this (`NULL` initially and between packets) */
141 const bt_packet
*cur_packet
;
145 void destroy_trimmer_comp(struct trimmer_comp
*trimmer_comp
)
147 BT_ASSERT(trimmer_comp
);
148 g_free(trimmer_comp
);
152 struct trimmer_comp
*create_trimmer_comp(void)
154 return g_new0(struct trimmer_comp
, 1);
158 void trimmer_finalize(bt_self_component_filter
*self_comp
)
160 struct trimmer_comp
*trimmer_comp
=
161 bt_self_component_get_data(
162 bt_self_component_filter_as_self_component(self_comp
));
165 destroy_trimmer_comp(trimmer_comp
);
170 * Compile regex in `pattern`, and try to match `string`. If there's a match,
171 * return true and set `*match_info` to the list of matches. The list of
172 * matches must be freed by the caller. If there's no match, return false and
173 * set `*match_info` to NULL;
176 bool compile_and_match(const char *pattern
, const char *string
, GMatchInfo
**match_info
) {
177 bool matches
= false;
178 GError
*regex_error
= NULL
;
181 regex
= g_regex_new(pattern
, 0, 0, ®ex_error
);
186 matches
= g_regex_match(regex
, string
, 0, match_info
);
189 * g_regex_match allocates `*match_info` even if it returns
190 * FALSE. If there's no match, we have no use for it, so free
191 * it immediatly and don't return it to the caller.
193 g_match_info_free(*match_info
);
197 g_regex_unref(regex
);
202 g_error_free(regex_error
);
209 * Convert the captured text in match number `match_num` in `match_info`
210 * to an unsigned integer.
213 guint64
match_to_uint(const GMatchInfo
*match_info
, gint match_num
) {
214 gchar
*text
, *endptr
;
217 text
= g_match_info_fetch(match_info
, match_num
);
221 * Because the input is carefully sanitized with regexes by the caller,
222 * we assume that g_ascii_strtoull cannot fail.
225 result
= g_ascii_strtoull(text
, &endptr
, 10);
226 BT_ASSERT(endptr
> text
);
227 BT_ASSERT(errno
== 0);
235 * When parsing the nanoseconds part, .512 means .512000000, not .000000512.
236 * This function is like match_to_uint, but multiplies the parsed number to get
237 * the expected result.
240 guint64
match_to_uint_ns(const GMatchInfo
*match_info
, gint match_num
) {
243 gint start_pos
, end_pos
, power
;
244 static int pow10
[] = {
245 1, 10, 100, 1000, 10000, 100000, 1000000, 10000000, 100000000,
248 nanoseconds
= match_to_uint(match_info
, match_num
);
250 /* Multiply by 10 as many times as there are omitted digits. */
251 ret
= g_match_info_fetch_pos(match_info
, match_num
, &start_pos
, &end_pos
);
254 power
= 9 - (end_pos
- start_pos
);
255 BT_ASSERT(power
>= 0 && power
<= 8);
257 nanoseconds
*= pow10
[power
];
263 * Sets the time (in ns from origin) of a trimmer bound from date and
266 * Returns a negative value if anything goes wrong.
269 int set_bound_ns_from_origin(struct trimmer_bound
*bound
,
270 unsigned int year
, unsigned int month
, unsigned int day
,
271 unsigned int hour
, unsigned int minute
, unsigned int second
,
272 unsigned int ns
, bool is_gmt
)
282 .tm_year
= year
- 1900,
287 result
= bt_timegm(&tm
);
289 result
= mktime(&tm
);
298 bound
->ns_from_origin
= (int64_t) result
;
299 bound
->ns_from_origin
*= NS_PER_S
;
300 bound
->ns_from_origin
+= ns
;
301 bound
->is_set
= true;
308 * Parses a timestamp, figuring out its format.
310 * Returns a negative value if anything goes wrong.
314 * YYYY-MM-DD hh:mm[:ss[.ns]]
318 * TODO: Check overflows.
321 int set_bound_from_str(struct trimmer_comp
*trimmer_comp
,
322 const char *str
, struct trimmer_bound
*bound
, bool is_gmt
)
324 /* Matches YYYY-MM-DD */
325 #define DATE_RE "([0-9]{4})-([0-9]{2})-([0-9]{2})"
327 /* Matches HH:MM[:SS[.NS]] */
328 #define TIME_RE "([0-9]{2}):([0-9]{2})(?::([0-9]{2})(?:\\.([0-9]{1,9}))?)?"
330 /* Matches [-]SS[.NS] */
331 #define S_NS_RE "^(-?)([0-9]+)(?:\\.([0-9]{1,9}))?$"
333 GMatchInfo
*match_info
;
336 /* Try `YYYY-MM-DD hh:mm[:ss[.ns]]` format */
337 if (compile_and_match("^" DATE_RE
" " TIME_RE
"$", str
, &match_info
)) {
338 unsigned int year
= 0, month
= 0, day
= 0, hours
= 0, minutes
= 0, seconds
= 0, nanoseconds
= 0;
339 gint match_count
= g_match_info_get_match_count(match_info
);
341 BT_ASSERT(match_count
>= 6 && match_count
<= 8);
343 year
= match_to_uint(match_info
, 1);
344 month
= match_to_uint(match_info
, 2);
345 day
= match_to_uint(match_info
, 3);
346 hours
= match_to_uint(match_info
, 4);
347 minutes
= match_to_uint(match_info
, 5);
349 if (match_count
>= 7) {
350 seconds
= match_to_uint(match_info
, 6);
353 if (match_count
>= 8) {
354 nanoseconds
= match_to_uint_ns(match_info
, 7);
357 set_bound_ns_from_origin(bound
, year
, month
, day
, hours
, minutes
, seconds
, nanoseconds
, is_gmt
);
362 if (compile_and_match("^" DATE_RE
"$", str
, &match_info
)) {
363 unsigned int year
= 0, month
= 0, day
= 0;
365 BT_ASSERT(g_match_info_get_match_count(match_info
) == 4);
367 year
= match_to_uint(match_info
, 1);
368 month
= match_to_uint(match_info
, 2);
369 day
= match_to_uint(match_info
, 3);
371 set_bound_ns_from_origin(bound
, year
, month
, day
, 0, 0, 0, 0, is_gmt
);
376 /* Try `hh:mm[:ss[.ns]]` format */
377 if (compile_and_match("^" TIME_RE
"$", str
, &match_info
)) {
378 gint match_count
= g_match_info_get_match_count(match_info
);
379 BT_ASSERT(match_count
>= 3 && match_count
<= 5);
380 bound
->time
.hour
= match_to_uint(match_info
, 1);
381 bound
->time
.minute
= match_to_uint(match_info
, 2);
383 if (match_count
>= 4) {
384 bound
->time
.second
= match_to_uint(match_info
, 3);
387 if (match_count
>= 5) {
388 bound
->time
.ns
= match_to_uint_ns(match_info
, 4);
394 /* Try `[-]s[.ns]` format */
395 if (compile_and_match("^" S_NS_RE
"$", str
, &match_info
)) {
396 gboolean is_neg
, fetch_pos_ret
;
397 gint start_pos
, end_pos
, match_count
;
398 guint64 seconds
, nanoseconds
= 0;
400 match_count
= g_match_info_get_match_count(match_info
);
401 BT_ASSERT(match_count
>= 3 && match_count
<= 4);
403 /* Check for presence of negation sign. */
404 fetch_pos_ret
= g_match_info_fetch_pos(match_info
, 1, &start_pos
, &end_pos
);
405 BT_ASSERT(fetch_pos_ret
);
406 is_neg
= (end_pos
- start_pos
) > 0;
408 seconds
= match_to_uint(match_info
, 2);
410 if (match_count
>= 4) {
411 nanoseconds
= match_to_uint_ns(match_info
, 3);
414 bound
->ns_from_origin
= seconds
* NS_PER_S
+ nanoseconds
;
417 bound
->ns_from_origin
= -bound
->ns_from_origin
;
420 bound
->is_set
= true;
425 BT_COMP_LOGE("Invalid date/time format: param=\"%s\"", str
);
433 * Sets a trimmer bound's properties from a parameter string/integer
436 * Returns a negative value if anything goes wrong.
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
)
447 if (bt_value_is_signed_integer(param
)) {
448 int64_t value
= bt_value_integer_signed_get(param
);
451 * Just convert it to a temporary string to handle
452 * everything the same way.
454 sprintf(tmp_arg
, "%" PRId64
, value
);
456 } else if (bt_value_is_string(param
)) {
457 arg
= bt_value_string_get(param
);
459 BT_COMP_LOGE("`%s` parameter must be an integer or a string value.",
465 ret
= set_bound_from_str(trimmer_comp
, arg
, bound
, is_gmt
);
472 int validate_trimmer_bounds(struct trimmer_comp
*trimmer_comp
,
473 struct trimmer_bound
*begin
, struct trimmer_bound
*end
)
477 BT_ASSERT(begin
->is_set
);
478 BT_ASSERT(end
->is_set
);
480 if (!begin
->is_infinite
&& !end
->is_infinite
&&
481 begin
->ns_from_origin
> end
->ns_from_origin
) {
482 BT_COMP_LOGE("Trimming time range's beginning time is greater than end time: "
483 "begin-ns-from-origin=%" PRId64
", "
484 "end-ns-from-origin=%" PRId64
,
485 begin
->ns_from_origin
,
486 end
->ns_from_origin
);
491 if (!begin
->is_infinite
&& begin
->ns_from_origin
== INT64_MIN
) {
492 BT_COMP_LOGE("Invalid trimming time range's beginning time: "
493 "ns-from-origin=%" PRId64
,
494 begin
->ns_from_origin
);
499 if (!end
->is_infinite
&& end
->ns_from_origin
== INT64_MIN
) {
500 BT_COMP_LOGE("Invalid trimming time range's end time: "
501 "ns-from-origin=%" PRId64
,
502 end
->ns_from_origin
);
512 int init_trimmer_comp_from_params(struct trimmer_comp
*trimmer_comp
,
513 const bt_value
*params
)
515 const bt_value
*value
;
519 value
= bt_value_map_borrow_entry_value_const(params
, "gmt");
521 trimmer_comp
->is_gmt
= (bool) bt_value_bool_get(value
);
524 value
= bt_value_map_borrow_entry_value_const(params
, "begin");
526 if (set_bound_from_param(trimmer_comp
, "begin", value
,
527 &trimmer_comp
->begin
, trimmer_comp
->is_gmt
)) {
528 /* set_bound_from_param() logs errors */
533 trimmer_comp
->begin
.is_infinite
= true;
534 trimmer_comp
->begin
.is_set
= true;
537 value
= bt_value_map_borrow_entry_value_const(params
, "end");
539 if (set_bound_from_param(trimmer_comp
, "end", value
,
540 &trimmer_comp
->end
, trimmer_comp
->is_gmt
)) {
541 /* set_bound_from_param() logs errors */
546 trimmer_comp
->end
.is_infinite
= true;
547 trimmer_comp
->end
.is_set
= true;
551 if (trimmer_comp
->begin
.is_set
&& trimmer_comp
->end
.is_set
) {
552 /* validate_trimmer_bounds() logs errors */
553 ret
= validate_trimmer_bounds(trimmer_comp
,
554 &trimmer_comp
->begin
, &trimmer_comp
->end
);
560 bt_component_class_init_method_status
trimmer_init(
561 bt_self_component_filter
*self_comp_flt
,
562 const bt_value
*params
, void *init_data
)
565 bt_component_class_init_method_status status
=
566 BT_COMPONENT_CLASS_INIT_METHOD_STATUS_OK
;
567 bt_self_component_add_port_status add_port_status
;
568 struct trimmer_comp
*trimmer_comp
= create_trimmer_comp();
569 bt_self_component
*self_comp
=
570 bt_self_component_filter_as_self_component(self_comp_flt
);
572 status
= BT_COMPONENT_CLASS_INIT_METHOD_STATUS_MEMORY_ERROR
;
576 trimmer_comp
->log_level
= bt_component_get_logging_level(
577 bt_self_component_as_component(self_comp
));
578 trimmer_comp
->self_comp
= self_comp
;
579 add_port_status
= bt_self_component_filter_add_input_port(
580 self_comp_flt
, in_port_name
, NULL
, NULL
);
581 switch (add_port_status
) {
582 case BT_SELF_COMPONENT_ADD_PORT_STATUS_ERROR
:
583 status
= BT_COMPONENT_CLASS_INIT_METHOD_STATUS_ERROR
;
585 case BT_SELF_COMPONENT_ADD_PORT_STATUS_MEMORY_ERROR
:
586 status
= BT_COMPONENT_CLASS_INIT_METHOD_STATUS_MEMORY_ERROR
;
592 add_port_status
= bt_self_component_filter_add_output_port(
593 self_comp_flt
, "out", NULL
, NULL
);
594 switch (add_port_status
) {
595 case BT_SELF_COMPONENT_ADD_PORT_STATUS_ERROR
:
596 status
= BT_COMPONENT_CLASS_INIT_METHOD_STATUS_ERROR
;
598 case BT_SELF_COMPONENT_ADD_PORT_STATUS_MEMORY_ERROR
:
599 status
= BT_COMPONENT_CLASS_INIT_METHOD_STATUS_MEMORY_ERROR
;
605 ret
= init_trimmer_comp_from_params(trimmer_comp
, params
);
607 status
= BT_COMPONENT_CLASS_INIT_METHOD_STATUS_ERROR
;
611 bt_self_component_set_data(self_comp
, trimmer_comp
);
615 if (status
== BT_COMPONENT_CLASS_INIT_METHOD_STATUS_OK
) {
616 status
= BT_COMPONENT_CLASS_INIT_METHOD_STATUS_ERROR
;
620 destroy_trimmer_comp(trimmer_comp
);
628 void destroy_trimmer_iterator(struct trimmer_iterator
*trimmer_it
)
630 BT_ASSERT(trimmer_it
);
631 bt_self_component_port_input_message_iterator_put_ref(
632 trimmer_it
->upstream_iter
);
634 if (trimmer_it
->output_messages
) {
635 g_queue_free(trimmer_it
->output_messages
);
638 if (trimmer_it
->stream_states
) {
639 g_hash_table_destroy(trimmer_it
->stream_states
);
646 void destroy_trimmer_iterator_stream_state(
647 struct trimmer_iterator_stream_state
*sstate
)
650 BT_PACKET_PUT_REF_AND_RESET(sstate
->cur_packet
);
655 bt_component_class_message_iterator_init_method_status
trimmer_msg_iter_init(
656 bt_self_message_iterator
*self_msg_iter
,
657 bt_self_component_filter
*self_comp
,
658 bt_self_component_port_output
*port
)
660 bt_component_class_message_iterator_init_method_status status
=
661 BT_COMPONENT_CLASS_MESSAGE_ITERATOR_INIT_METHOD_STATUS_OK
;
662 struct trimmer_iterator
*trimmer_it
;
664 trimmer_it
= g_new0(struct trimmer_iterator
, 1);
666 status
= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_INIT_METHOD_STATUS_MEMORY_ERROR
;
670 trimmer_it
->trimmer_comp
= bt_self_component_get_data(
671 bt_self_component_filter_as_self_component(self_comp
));
672 BT_ASSERT(trimmer_it
->trimmer_comp
);
674 if (trimmer_it
->trimmer_comp
->begin
.is_set
&&
675 trimmer_it
->trimmer_comp
->end
.is_set
) {
677 * Both trimming time range's bounds are set, so skip
679 * `TRIMMER_ITERATOR_STATE_SET_BOUNDS_NS_FROM_ORIGIN`
682 trimmer_it
->state
= TRIMMER_ITERATOR_STATE_SEEK_INITIALLY
;
685 trimmer_it
->begin
= trimmer_it
->trimmer_comp
->begin
;
686 trimmer_it
->end
= trimmer_it
->trimmer_comp
->end
;
687 trimmer_it
->upstream_iter
=
688 bt_self_component_port_input_message_iterator_create_from_message_iterator(
690 bt_self_component_filter_borrow_input_port_by_name(
691 self_comp
, in_port_name
));
692 if (!trimmer_it
->upstream_iter
) {
693 status
= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_INIT_METHOD_STATUS_ERROR
;
697 trimmer_it
->output_messages
= g_queue_new();
698 if (!trimmer_it
->output_messages
) {
699 status
= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_INIT_METHOD_STATUS_MEMORY_ERROR
;
703 trimmer_it
->stream_states
= g_hash_table_new_full(g_direct_hash
,
704 g_direct_equal
, NULL
,
705 (GDestroyNotify
) destroy_trimmer_iterator_stream_state
);
706 if (!trimmer_it
->stream_states
) {
707 status
= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_INIT_METHOD_STATUS_MEMORY_ERROR
;
711 trimmer_it
->self_msg_iter
= self_msg_iter
;
712 bt_self_message_iterator_set_data(self_msg_iter
, trimmer_it
);
715 if (status
!= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_INIT_METHOD_STATUS_OK
&& trimmer_it
) {
716 destroy_trimmer_iterator(trimmer_it
);
723 int get_msg_ns_from_origin(const bt_message
*msg
, int64_t *ns_from_origin
,
724 bool *has_clock_snapshot
)
726 const bt_clock_class
*clock_class
= NULL
;
727 const bt_clock_snapshot
*clock_snapshot
= NULL
;
731 BT_ASSERT(ns_from_origin
);
732 BT_ASSERT(has_clock_snapshot
);
734 switch (bt_message_get_type(msg
)) {
735 case BT_MESSAGE_TYPE_EVENT
:
737 bt_message_event_borrow_stream_class_default_clock_class_const(
739 if (G_UNLIKELY(!clock_class
)) {
743 clock_snapshot
= bt_message_event_borrow_default_clock_snapshot_const(
746 case BT_MESSAGE_TYPE_PACKET_BEGINNING
:
748 bt_message_packet_beginning_borrow_stream_class_default_clock_class_const(
750 if (G_UNLIKELY(!clock_class
)) {
754 clock_snapshot
= bt_message_packet_beginning_borrow_default_clock_snapshot_const(
757 case BT_MESSAGE_TYPE_PACKET_END
:
759 bt_message_packet_end_borrow_stream_class_default_clock_class_const(
761 if (G_UNLIKELY(!clock_class
)) {
765 clock_snapshot
= bt_message_packet_end_borrow_default_clock_snapshot_const(
768 case BT_MESSAGE_TYPE_STREAM_BEGINNING
:
770 enum bt_message_stream_clock_snapshot_state cs_state
;
773 bt_message_stream_beginning_borrow_stream_class_default_clock_class_const(msg
);
774 if (G_UNLIKELY(!clock_class
)) {
778 cs_state
= bt_message_stream_beginning_borrow_default_clock_snapshot_const(msg
, &clock_snapshot
);
779 if (cs_state
!= BT_MESSAGE_STREAM_CLOCK_SNAPSHOT_STATE_KNOWN
) {
780 goto no_clock_snapshot
;
785 case BT_MESSAGE_TYPE_STREAM_END
:
787 enum bt_message_stream_clock_snapshot_state cs_state
;
790 bt_message_stream_end_borrow_stream_class_default_clock_class_const(msg
);
791 if (G_UNLIKELY(!clock_class
)) {
795 cs_state
= bt_message_stream_end_borrow_default_clock_snapshot_const(msg
, &clock_snapshot
);
796 if (cs_state
!= BT_MESSAGE_STREAM_CLOCK_SNAPSHOT_STATE_KNOWN
) {
797 goto no_clock_snapshot
;
802 case BT_MESSAGE_TYPE_DISCARDED_EVENTS
:
804 bt_message_discarded_events_borrow_stream_class_default_clock_class_const(
806 if (G_UNLIKELY(!clock_class
)) {
810 clock_snapshot
= bt_message_discarded_events_borrow_beginning_default_clock_snapshot_const(
813 case BT_MESSAGE_TYPE_DISCARDED_PACKETS
:
815 bt_message_discarded_packets_borrow_stream_class_default_clock_class_const(
817 if (G_UNLIKELY(!clock_class
)) {
821 clock_snapshot
= bt_message_discarded_packets_borrow_beginning_default_clock_snapshot_const(
824 case BT_MESSAGE_TYPE_MESSAGE_ITERATOR_INACTIVITY
:
826 bt_message_message_iterator_inactivity_borrow_default_clock_snapshot_const(
830 goto no_clock_snapshot
;
833 ret
= bt_clock_snapshot_get_ns_from_origin(clock_snapshot
,
835 if (G_UNLIKELY(ret
)) {
839 *has_clock_snapshot
= true;
843 *has_clock_snapshot
= false;
854 void put_messages(bt_message_array_const msgs
, uint64_t count
)
858 for (i
= 0; i
< count
; i
++) {
859 BT_MESSAGE_PUT_REF_AND_RESET(msgs
[i
]);
864 int set_trimmer_iterator_bound(struct trimmer_iterator
*trimmer_it
,
865 struct trimmer_bound
*bound
, int64_t ns_from_origin
,
868 struct trimmer_comp
*trimmer_comp
= trimmer_it
->trimmer_comp
;
871 time_t time_seconds
= (time_t) (ns_from_origin
/ NS_PER_S
);
874 BT_ASSERT(!bound
->is_set
);
877 /* We only need to extract the date from this time */
879 res
= bt_gmtime_r(&time_seconds
, &tm
);
881 res
= bt_localtime_r(&time_seconds
, &tm
);
885 BT_COMP_LOGE_ERRNO("Cannot convert timestamp to date and time",
886 ": ts=%" PRId64
, (int64_t) time_seconds
);
891 ret
= set_bound_ns_from_origin(bound
, tm
.tm_year
+ 1900, tm
.tm_mon
+ 1,
892 tm
.tm_mday
, bound
->time
.hour
, bound
->time
.minute
,
893 bound
->time
.second
, bound
->time
.ns
, is_gmt
);
900 bt_component_class_message_iterator_next_method_status
901 state_set_trimmer_iterator_bounds(
902 struct trimmer_iterator
*trimmer_it
)
904 bt_message_iterator_next_status upstream_iter_status
=
905 BT_MESSAGE_ITERATOR_NEXT_STATUS_OK
;
906 struct trimmer_comp
*trimmer_comp
= trimmer_it
->trimmer_comp
;
907 bt_message_array_const msgs
;
909 int64_t ns_from_origin
= INT64_MIN
;
913 BT_ASSERT(!trimmer_it
->begin
.is_set
||
914 !trimmer_it
->end
.is_set
);
917 upstream_iter_status
=
918 bt_self_component_port_input_message_iterator_next(
919 trimmer_it
->upstream_iter
, &msgs
, &count
);
920 if (upstream_iter_status
!= BT_MESSAGE_ITERATOR_NEXT_STATUS_OK
) {
924 for (i
= 0; i
< count
; i
++) {
925 const bt_message
*msg
= msgs
[i
];
926 bool has_ns_from_origin
;
929 ret
= get_msg_ns_from_origin(msg
, &ns_from_origin
,
930 &has_ns_from_origin
);
935 if (!has_ns_from_origin
) {
939 BT_ASSERT(ns_from_origin
!= INT64_MIN
&&
940 ns_from_origin
!= INT64_MAX
);
941 put_messages(msgs
, count
);
945 put_messages(msgs
, count
);
949 if (!trimmer_it
->begin
.is_set
) {
950 BT_ASSERT(!trimmer_it
->begin
.is_infinite
);
951 ret
= set_trimmer_iterator_bound(trimmer_it
, &trimmer_it
->begin
,
952 ns_from_origin
, trimmer_comp
->is_gmt
);
958 if (!trimmer_it
->end
.is_set
) {
959 BT_ASSERT(!trimmer_it
->end
.is_infinite
);
960 ret
= set_trimmer_iterator_bound(trimmer_it
, &trimmer_it
->end
,
961 ns_from_origin
, trimmer_comp
->is_gmt
);
967 ret
= validate_trimmer_bounds(trimmer_it
->trimmer_comp
,
968 &trimmer_it
->begin
, &trimmer_it
->end
);
976 put_messages(msgs
, count
);
977 upstream_iter_status
= BT_MESSAGE_ITERATOR_NEXT_STATUS_ERROR
;
980 return (int) upstream_iter_status
;
984 bt_component_class_message_iterator_next_method_status
state_seek_initially(
985 struct trimmer_iterator
*trimmer_it
)
987 struct trimmer_comp
*trimmer_comp
= trimmer_it
->trimmer_comp
;
988 bt_component_class_message_iterator_next_method_status status
=
989 BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK
;
991 BT_ASSERT(trimmer_it
->begin
.is_set
);
993 if (trimmer_it
->begin
.is_infinite
) {
994 if (!bt_self_component_port_input_message_iterator_can_seek_beginning(
995 trimmer_it
->upstream_iter
)) {
996 BT_COMP_LOGE_STR("Cannot make upstream message iterator initially seek its beginning.");
997 status
= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_ERROR
;
1001 status
= (int) bt_self_component_port_input_message_iterator_seek_beginning(
1002 trimmer_it
->upstream_iter
);
1004 if (!bt_self_component_port_input_message_iterator_can_seek_ns_from_origin(
1005 trimmer_it
->upstream_iter
,
1006 trimmer_it
->begin
.ns_from_origin
)) {
1007 BT_COMP_LOGE("Cannot make upstream message iterator initially seek: "
1008 "seek-ns-from-origin=%" PRId64
,
1009 trimmer_it
->begin
.ns_from_origin
);
1010 status
= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_ERROR
;
1014 status
= (int) bt_self_component_port_input_message_iterator_seek_ns_from_origin(
1015 trimmer_it
->upstream_iter
, trimmer_it
->begin
.ns_from_origin
);
1018 if (status
== BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK
) {
1019 trimmer_it
->state
= TRIMMER_ITERATOR_STATE_TRIM
;
1027 void push_message(struct trimmer_iterator
*trimmer_it
, const bt_message
*msg
)
1029 g_queue_push_head(trimmer_it
->output_messages
, (void *) msg
);
1033 const bt_message
*pop_message(struct trimmer_iterator
*trimmer_it
)
1035 return g_queue_pop_tail(trimmer_it
->output_messages
);
1039 int clock_raw_value_from_ns_from_origin(const bt_clock_class
*clock_class
,
1040 int64_t ns_from_origin
, uint64_t *raw_value
)
1043 int64_t cc_offset_s
;
1044 uint64_t cc_offset_cycles
;
1047 bt_clock_class_get_offset(clock_class
, &cc_offset_s
, &cc_offset_cycles
);
1048 cc_freq
= bt_clock_class_get_frequency(clock_class
);
1049 return bt_common_clock_value_from_ns_from_origin(cc_offset_s
,
1050 cc_offset_cycles
, cc_freq
, ns_from_origin
, raw_value
);
1054 bt_component_class_message_iterator_next_method_status
1055 end_stream(struct trimmer_iterator
*trimmer_it
,
1056 struct trimmer_iterator_stream_state
*sstate
)
1058 bt_component_class_message_iterator_next_method_status status
=
1059 BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK
;
1060 /* Initialize to silence maybe-uninitialized warning. */
1061 uint64_t raw_value
= 0;
1062 bt_message
*msg
= NULL
;
1064 BT_ASSERT(!trimmer_it
->end
.is_infinite
);
1065 BT_ASSERT(sstate
->stream
);
1068 * If we haven't seen a message with a clock snapshot, we don't know if the trimmer's end bound is within
1069 * the clock's range, so it wouldn't be safe to try to convert ns_from_origin to a clock value.
1071 * Also, it would be a bit of a lie to generate a stream end message with the end bound as its
1072 * clock snapshot, because we don't really know if the stream existed at that time. If we have
1073 * seen a message with a clock snapshot and the stream is cut short by another message with a
1074 * clock snapshot, then we are sure that the the end bound time is not below the clock range,
1075 * and we know the stream was active at that time (and that we cut it short).
1077 if (sstate
->seen_clock_snapshot
) {
1078 const bt_clock_class
*clock_class
;
1081 clock_class
= bt_stream_class_borrow_default_clock_class_const(
1082 bt_stream_borrow_class_const(sstate
->stream
));
1083 BT_ASSERT(clock_class
);
1084 ret
= clock_raw_value_from_ns_from_origin(clock_class
,
1085 trimmer_it
->end
.ns_from_origin
, &raw_value
);
1087 status
= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_ERROR
;
1092 if (sstate
->cur_packet
) {
1094 * Create and push a packet end message, making its time
1095 * the trimming range's end time.
1097 * We know that we must have seen a clock snapshot, the one in
1098 * the packet beginning message, since trimmer currently
1099 * requires packet messages to have clock snapshots (see comment
1100 * in create_stream_state_entry).
1102 BT_ASSERT(sstate
->seen_clock_snapshot
);
1104 msg
= bt_message_packet_end_create_with_default_clock_snapshot(
1105 trimmer_it
->self_msg_iter
, sstate
->cur_packet
,
1108 status
= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_MEMORY_ERROR
;
1112 push_message(trimmer_it
, msg
);
1114 BT_PACKET_PUT_REF_AND_RESET(sstate
->cur_packet
);
1117 /* Create and push a stream end message. */
1118 msg
= bt_message_stream_end_create(trimmer_it
->self_msg_iter
,
1121 status
= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_MEMORY_ERROR
;
1125 if (sstate
->seen_clock_snapshot
) {
1126 bt_message_stream_end_set_default_clock_snapshot(msg
, raw_value
);
1129 push_message(trimmer_it
, msg
);
1133 * Just to make sure that we don't use this stream state again
1134 * in the future without an obvious error.
1136 sstate
->stream
= NULL
;
1139 bt_message_put_ref(msg
);
1144 bt_component_class_message_iterator_next_method_status
end_iterator_streams(
1145 struct trimmer_iterator
*trimmer_it
)
1147 bt_component_class_message_iterator_next_method_status status
=
1148 BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK
;
1149 GHashTableIter iter
;
1150 gpointer key
, sstate
;
1152 if (trimmer_it
->end
.is_infinite
) {
1154 * An infinite trimming range's end time guarantees that
1155 * we received (and pushed) all the appropriate end
1162 * End each stream and then remove them from the hash table of
1163 * stream states to release unneeded references.
1165 g_hash_table_iter_init(&iter
, trimmer_it
->stream_states
);
1167 while (g_hash_table_iter_next(&iter
, &key
, &sstate
)) {
1168 status
= end_stream(trimmer_it
, sstate
);
1175 g_hash_table_remove_all(trimmer_it
->stream_states
);
1182 bt_component_class_message_iterator_next_method_status
1183 create_stream_state_entry(
1184 struct trimmer_iterator
*trimmer_it
,
1185 const struct bt_stream
*stream
,
1186 struct trimmer_iterator_stream_state
**stream_state
)
1188 struct trimmer_comp
*trimmer_comp
= trimmer_it
->trimmer_comp
;
1189 bt_component_class_message_iterator_next_method_status status
;
1190 struct trimmer_iterator_stream_state
*sstate
;
1191 const bt_stream_class
*sc
;
1193 BT_ASSERT(!bt_g_hash_table_contains(trimmer_it
->stream_states
, stream
));
1196 * Validate right now that the stream's class
1197 * has a registered default clock class so that
1198 * an existing stream state guarantees existing
1199 * default clock snapshots for its associated
1202 * Also check that clock snapshots are always
1205 sc
= bt_stream_borrow_class_const(stream
);
1206 if (!bt_stream_class_borrow_default_clock_class_const(sc
)) {
1207 BT_COMP_LOGE("Unsupported stream: stream class does "
1208 "not have a default clock class: "
1210 "stream-id=%" PRIu64
", "
1211 "stream-name=\"%s\"",
1212 stream
, bt_stream_get_id(stream
),
1213 bt_stream_get_name(stream
));
1214 status
= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_ERROR
;
1219 * Temporary: make sure packet beginning, packet
1220 * end, discarded events, and discarded packets
1221 * messages have default clock snapshots until
1222 * the support for not having them is
1225 if (!bt_stream_class_packets_have_beginning_default_clock_snapshot(
1227 BT_COMP_LOGE("Unsupported stream: packets have "
1228 "no beginning clock snapshot: "
1230 "stream-id=%" PRIu64
", "
1231 "stream-name=\"%s\"",
1232 stream
, bt_stream_get_id(stream
),
1233 bt_stream_get_name(stream
));
1234 status
= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_ERROR
;
1238 if (!bt_stream_class_packets_have_end_default_clock_snapshot(
1240 BT_COMP_LOGE("Unsupported stream: packets have "
1241 "no end clock snapshot: "
1243 "stream-id=%" PRIu64
", "
1244 "stream-name=\"%s\"",
1245 stream
, bt_stream_get_id(stream
),
1246 bt_stream_get_name(stream
));
1247 status
= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_ERROR
;
1251 if (bt_stream_class_supports_discarded_events(sc
) &&
1252 !bt_stream_class_discarded_events_have_default_clock_snapshots(sc
)) {
1253 BT_COMP_LOGE("Unsupported stream: discarded events "
1254 "have no clock snapshots: "
1256 "stream-id=%" PRIu64
", "
1257 "stream-name=\"%s\"",
1258 stream
, bt_stream_get_id(stream
),
1259 bt_stream_get_name(stream
));
1260 status
= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_ERROR
;
1264 if (bt_stream_class_supports_discarded_packets(sc
) &&
1265 !bt_stream_class_discarded_packets_have_default_clock_snapshots(sc
)) {
1266 BT_COMP_LOGE("Unsupported stream: discarded packets "
1267 "have no clock snapshots: "
1269 "stream-id=%" PRIu64
", "
1270 "stream-name=\"%s\"",
1271 stream
, bt_stream_get_id(stream
),
1272 bt_stream_get_name(stream
));
1273 status
= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_ERROR
;
1277 sstate
= g_new0(struct trimmer_iterator_stream_state
, 1);
1279 status
= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_MEMORY_ERROR
;
1283 sstate
->stream
= stream
;
1285 g_hash_table_insert(trimmer_it
->stream_states
, (void *) stream
, sstate
);
1287 *stream_state
= sstate
;
1289 status
= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK
;
1296 struct trimmer_iterator_stream_state
*get_stream_state_entry(
1297 struct trimmer_iterator
*trimmer_it
,
1298 const struct bt_stream
*stream
)
1300 struct trimmer_iterator_stream_state
*sstate
;
1303 sstate
= g_hash_table_lookup(trimmer_it
->stream_states
, stream
);
1310 * Handles a message which is associated to a given stream state. This
1311 * _could_ make the iterator's output message queue grow; this could
1312 * also consume the message without pushing anything to this queue, only
1313 * modifying the stream state.
1315 * This function consumes the `msg` reference, _whatever the outcome_.
1317 * If non-NULL, `ns_from_origin` is the message's time, as given by
1318 * get_msg_ns_from_origin(). If NULL, the message doesn't have a time.
1320 * This function sets `reached_end` if handling this message made the
1321 * iterator reach the end of the trimming range. Note that the output
1322 * message queue could contain messages even if this function sets
1326 bt_component_class_message_iterator_next_method_status
1327 handle_message_with_stream(
1328 struct trimmer_iterator
*trimmer_it
, const bt_message
*msg
,
1329 const struct bt_stream
*stream
, const int64_t *ns_from_origin
,
1332 bt_component_class_message_iterator_next_method_status status
=
1333 BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK
;
1334 bt_message_type msg_type
= bt_message_get_type(msg
);
1336 struct trimmer_iterator_stream_state
*sstate
= NULL
;
1339 * Retrieve the stream's state - except if the message is stream
1340 * beginning, in which case we don't know about about this stream yet.
1342 if (msg_type
!= BT_MESSAGE_TYPE_STREAM_BEGINNING
) {
1343 sstate
= get_stream_state_entry(trimmer_it
, stream
);
1347 case BT_MESSAGE_TYPE_EVENT
:
1349 * Event messages always have a clock snapshot if the stream
1350 * class has a clock class. And we know it has, otherwise we
1351 * couldn't be using the trimmer component.
1353 BT_ASSERT(ns_from_origin
);
1355 if (G_UNLIKELY(!trimmer_it
->end
.is_infinite
&&
1356 *ns_from_origin
> trimmer_it
->end
.ns_from_origin
)) {
1357 status
= end_iterator_streams(trimmer_it
);
1358 *reached_end
= true;
1362 sstate
->seen_clock_snapshot
= true;
1364 push_message(trimmer_it
, msg
);
1368 case BT_MESSAGE_TYPE_PACKET_BEGINNING
:
1370 * Packet beginning messages won't have a clock snapshot if
1371 * stream_class->packets_have_beginning_default_clock_snapshot
1372 * is false. But for now, assume they always do.
1374 BT_ASSERT(ns_from_origin
);
1375 BT_ASSERT(!sstate
->cur_packet
);
1377 if (G_UNLIKELY(!trimmer_it
->end
.is_infinite
&&
1378 *ns_from_origin
> trimmer_it
->end
.ns_from_origin
)) {
1379 status
= end_iterator_streams(trimmer_it
);
1380 *reached_end
= true;
1384 sstate
->cur_packet
=
1385 bt_message_packet_beginning_borrow_packet_const(msg
);
1386 bt_packet_get_ref(sstate
->cur_packet
);
1388 sstate
->seen_clock_snapshot
= true;
1390 push_message(trimmer_it
, msg
);
1394 case BT_MESSAGE_TYPE_PACKET_END
:
1396 * Packet end messages won't have a clock snapshot if
1397 * stream_class->packets_have_end_default_clock_snapshot
1398 * is false. But for now, assume they always do.
1400 BT_ASSERT(ns_from_origin
);
1401 BT_ASSERT(sstate
->cur_packet
);
1403 if (G_UNLIKELY(!trimmer_it
->end
.is_infinite
&&
1404 *ns_from_origin
> trimmer_it
->end
.ns_from_origin
)) {
1405 status
= end_iterator_streams(trimmer_it
);
1406 *reached_end
= true;
1410 BT_PACKET_PUT_REF_AND_RESET(sstate
->cur_packet
);
1412 sstate
->seen_clock_snapshot
= true;
1414 push_message(trimmer_it
, msg
);
1418 case BT_MESSAGE_TYPE_DISCARDED_EVENTS
:
1419 case BT_MESSAGE_TYPE_DISCARDED_PACKETS
:
1422 * `ns_from_origin` is the message's time range's
1423 * beginning time here.
1425 int64_t end_ns_from_origin
;
1426 const bt_clock_snapshot
*end_cs
;
1428 BT_ASSERT(ns_from_origin
);
1430 sstate
->seen_clock_snapshot
= true;
1432 if (bt_message_get_type(msg
) ==
1433 BT_MESSAGE_TYPE_DISCARDED_EVENTS
) {
1435 * Safe to ignore the return value because we
1436 * know there's a default clock and it's always
1439 end_cs
= bt_message_discarded_events_borrow_end_default_clock_snapshot_const(
1443 * Safe to ignore the return value because we
1444 * know there's a default clock and it's always
1447 end_cs
= bt_message_discarded_packets_borrow_end_default_clock_snapshot_const(
1451 if (bt_clock_snapshot_get_ns_from_origin(end_cs
,
1452 &end_ns_from_origin
)) {
1453 status
= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_ERROR
;
1457 if (!trimmer_it
->end
.is_infinite
&&
1458 *ns_from_origin
> trimmer_it
->end
.ns_from_origin
) {
1459 status
= end_iterator_streams(trimmer_it
);
1460 *reached_end
= true;
1464 if (!trimmer_it
->end
.is_infinite
&&
1465 end_ns_from_origin
> trimmer_it
->end
.ns_from_origin
) {
1467 * This message's end time is outside the
1468 * trimming time range: replace it with a new
1469 * message having an end time equal to the
1470 * trimming time range's end and without a
1473 const bt_clock_class
*clock_class
=
1474 bt_clock_snapshot_borrow_clock_class_const(
1476 const bt_clock_snapshot
*begin_cs
;
1477 bt_message
*new_msg
;
1478 uint64_t end_raw_value
;
1480 ret
= clock_raw_value_from_ns_from_origin(clock_class
,
1481 trimmer_it
->end
.ns_from_origin
, &end_raw_value
);
1483 status
= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_ERROR
;
1487 if (msg_type
== BT_MESSAGE_TYPE_DISCARDED_EVENTS
) {
1488 begin_cs
= bt_message_discarded_events_borrow_beginning_default_clock_snapshot_const(
1490 new_msg
= bt_message_discarded_events_create_with_default_clock_snapshots(
1491 trimmer_it
->self_msg_iter
,
1493 bt_clock_snapshot_get_value(begin_cs
),
1496 begin_cs
= bt_message_discarded_packets_borrow_beginning_default_clock_snapshot_const(
1498 new_msg
= bt_message_discarded_packets_create_with_default_clock_snapshots(
1499 trimmer_it
->self_msg_iter
,
1501 bt_clock_snapshot_get_value(begin_cs
),
1506 status
= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_MEMORY_ERROR
;
1510 /* Replace the original message */
1511 BT_MESSAGE_MOVE_REF(msg
, new_msg
);
1514 push_message(trimmer_it
, msg
);
1519 case BT_MESSAGE_TYPE_STREAM_BEGINNING
:
1521 * If this message has a time and this time is greater than the
1522 * trimmer's end bound, it triggers the end of the trim window.
1524 if (G_UNLIKELY(ns_from_origin
&& !trimmer_it
->end
.is_infinite
&&
1525 *ns_from_origin
> trimmer_it
->end
.ns_from_origin
)) {
1526 status
= end_iterator_streams(trimmer_it
);
1527 *reached_end
= true;
1531 /* Learn about this stream. */
1532 status
= create_stream_state_entry(trimmer_it
, stream
, &sstate
);
1533 if (status
!= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK
) {
1537 if (ns_from_origin
) {
1538 sstate
->seen_clock_snapshot
= true;
1541 push_message(trimmer_it
, msg
);
1544 case BT_MESSAGE_TYPE_STREAM_END
:
1549 * If this message has a time and this time is greater than the
1550 * trimmer's end bound, it triggers the end of the trim window.
1552 if (G_UNLIKELY(ns_from_origin
&& !trimmer_it
->end
.is_infinite
&&
1553 *ns_from_origin
> trimmer_it
->end
.ns_from_origin
)) {
1554 status
= end_iterator_streams(trimmer_it
);
1555 *reached_end
= true;
1560 * Either the stream end message's time is within the trimmer's
1561 * bounds, or it doesn't have a time. In both cases, pass
1562 * the message unmodified.
1564 push_message(trimmer_it
, msg
);
1567 /* Forget about this stream. */
1568 removed
= g_hash_table_remove(trimmer_it
->stream_states
, sstate
->stream
);
1577 /* We release the message's reference whatever the outcome */
1578 bt_message_put_ref(msg
);
1583 * Handles an input message. This _could_ make the iterator's output
1584 * message queue grow; this could also consume the message without
1585 * pushing anything to this queue, only modifying the stream state.
1587 * This function consumes the `msg` reference, _whatever the outcome_.
1589 * This function sets `reached_end` if handling this message made the
1590 * iterator reach the end of the trimming range. Note that the output
1591 * message queue could contain messages even if this function sets
1595 bt_component_class_message_iterator_next_method_status
handle_message(
1596 struct trimmer_iterator
*trimmer_it
, const bt_message
*msg
,
1599 bt_component_class_message_iterator_next_method_status status
;
1600 const bt_stream
*stream
= NULL
;
1601 int64_t ns_from_origin
= INT64_MIN
;
1602 bool has_ns_from_origin
= false;
1605 /* Find message's associated stream */
1606 switch (bt_message_get_type(msg
)) {
1607 case BT_MESSAGE_TYPE_EVENT
:
1608 stream
= bt_event_borrow_stream_const(
1609 bt_message_event_borrow_event_const(msg
));
1611 case BT_MESSAGE_TYPE_PACKET_BEGINNING
:
1612 stream
= bt_packet_borrow_stream_const(
1613 bt_message_packet_beginning_borrow_packet_const(msg
));
1615 case BT_MESSAGE_TYPE_PACKET_END
:
1616 stream
= bt_packet_borrow_stream_const(
1617 bt_message_packet_end_borrow_packet_const(msg
));
1619 case BT_MESSAGE_TYPE_DISCARDED_EVENTS
:
1620 stream
= bt_message_discarded_events_borrow_stream_const(msg
);
1622 case BT_MESSAGE_TYPE_DISCARDED_PACKETS
:
1623 stream
= bt_message_discarded_packets_borrow_stream_const(msg
);
1625 case BT_MESSAGE_TYPE_STREAM_BEGINNING
:
1626 stream
= bt_message_stream_beginning_borrow_stream_const(msg
);
1628 case BT_MESSAGE_TYPE_STREAM_END
:
1629 stream
= bt_message_stream_end_borrow_stream_const(msg
);
1635 /* Retrieve the message's time */
1636 ret
= get_msg_ns_from_origin(msg
, &ns_from_origin
, &has_ns_from_origin
);
1637 if (G_UNLIKELY(ret
)) {
1638 status
= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_ERROR
;
1642 if (G_LIKELY(stream
)) {
1643 /* Message associated to a stream */
1644 status
= handle_message_with_stream(trimmer_it
, msg
,
1645 stream
, has_ns_from_origin
? &ns_from_origin
: NULL
, reached_end
);
1648 * handle_message_with_stream_state() unconditionally
1654 * Message not associated to a stream (message iterator
1657 if (G_UNLIKELY(ns_from_origin
> trimmer_it
->end
.ns_from_origin
)) {
1658 BT_MESSAGE_PUT_REF_AND_RESET(msg
);
1659 status
= end_iterator_streams(trimmer_it
);
1660 *reached_end
= true;
1662 push_message(trimmer_it
, msg
);
1663 status
= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK
;
1669 /* We release the message's reference whatever the outcome */
1670 bt_message_put_ref(msg
);
1675 void fill_message_array_from_output_messages(
1676 struct trimmer_iterator
*trimmer_it
,
1677 bt_message_array_const msgs
, uint64_t capacity
, uint64_t *count
)
1682 * Move auto-seek messages to the output array (which is this
1683 * iterator's base message array).
1685 while (capacity
> 0 && !g_queue_is_empty(trimmer_it
->output_messages
)) {
1686 msgs
[*count
] = pop_message(trimmer_it
);
1691 BT_ASSERT(*count
> 0);
1695 bt_component_class_message_iterator_next_method_status
state_ending(
1696 struct trimmer_iterator
*trimmer_it
,
1697 bt_message_array_const msgs
, uint64_t capacity
,
1700 bt_component_class_message_iterator_next_method_status status
=
1701 BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK
;
1703 if (g_queue_is_empty(trimmer_it
->output_messages
)) {
1704 trimmer_it
->state
= TRIMMER_ITERATOR_STATE_ENDED
;
1705 status
= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_END
;
1709 fill_message_array_from_output_messages(trimmer_it
, msgs
,
1717 bt_component_class_message_iterator_next_method_status
1718 state_trim(struct trimmer_iterator
*trimmer_it
,
1719 bt_message_array_const msgs
, uint64_t capacity
,
1722 bt_component_class_message_iterator_next_method_status status
=
1723 BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK
;
1724 bt_message_array_const my_msgs
;
1727 bool reached_end
= false;
1729 while (g_queue_is_empty(trimmer_it
->output_messages
)) {
1730 status
= (int) bt_self_component_port_input_message_iterator_next(
1731 trimmer_it
->upstream_iter
, &my_msgs
, &my_count
);
1732 if (G_UNLIKELY(status
!= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK
)) {
1733 if (status
== BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_END
) {
1734 status
= end_iterator_streams(trimmer_it
);
1735 if (status
!= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK
) {
1740 TRIMMER_ITERATOR_STATE_ENDING
;
1741 status
= state_ending(trimmer_it
, msgs
,
1748 BT_ASSERT(my_count
> 0);
1750 for (i
= 0; i
< my_count
; i
++) {
1751 status
= handle_message(trimmer_it
, my_msgs
[i
],
1755 * handle_message() unconditionally consumes the
1756 * message reference.
1760 if (G_UNLIKELY(status
!=
1761 BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK
)) {
1762 put_messages(my_msgs
, my_count
);
1766 if (G_UNLIKELY(reached_end
)) {
1768 * This message's time was passed the
1769 * trimming time range's end time: we
1770 * are done. Their might still be
1771 * messages in the output message queue,
1772 * so move to the "ending" state and
1773 * apply it immediately since
1774 * state_trim() is called within the
1777 put_messages(my_msgs
, my_count
);
1779 TRIMMER_ITERATOR_STATE_ENDING
;
1780 status
= state_ending(trimmer_it
, msgs
,
1788 * There's at least one message in the output message queue:
1789 * move the messages to the output message array.
1791 BT_ASSERT(!g_queue_is_empty(trimmer_it
->output_messages
));
1792 fill_message_array_from_output_messages(trimmer_it
, msgs
,
1800 bt_component_class_message_iterator_next_method_status
trimmer_msg_iter_next(
1801 bt_self_message_iterator
*self_msg_iter
,
1802 bt_message_array_const msgs
, uint64_t capacity
,
1805 struct trimmer_iterator
*trimmer_it
=
1806 bt_self_message_iterator_get_data(self_msg_iter
);
1807 bt_component_class_message_iterator_next_method_status status
=
1808 BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK
;
1810 BT_ASSERT(trimmer_it
);
1812 if (G_LIKELY(trimmer_it
->state
== TRIMMER_ITERATOR_STATE_TRIM
)) {
1813 status
= state_trim(trimmer_it
, msgs
, capacity
, count
);
1814 if (status
!= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK
) {
1818 switch (trimmer_it
->state
) {
1819 case TRIMMER_ITERATOR_STATE_SET_BOUNDS_NS_FROM_ORIGIN
:
1820 status
= state_set_trimmer_iterator_bounds(trimmer_it
);
1821 if (status
!= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK
) {
1825 status
= state_seek_initially(trimmer_it
);
1826 if (status
!= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK
) {
1830 status
= state_trim(trimmer_it
, msgs
, capacity
, count
);
1831 if (status
!= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK
) {
1836 case TRIMMER_ITERATOR_STATE_SEEK_INITIALLY
:
1837 status
= state_seek_initially(trimmer_it
);
1838 if (status
!= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK
) {
1842 status
= state_trim(trimmer_it
, msgs
, capacity
, count
);
1843 if (status
!= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK
) {
1848 case TRIMMER_ITERATOR_STATE_ENDING
:
1849 status
= state_ending(trimmer_it
, msgs
, capacity
,
1851 if (status
!= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK
) {
1856 case TRIMMER_ITERATOR_STATE_ENDED
:
1857 status
= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_END
;
1869 void trimmer_msg_iter_finalize(bt_self_message_iterator
*self_msg_iter
)
1871 struct trimmer_iterator
*trimmer_it
=
1872 bt_self_message_iterator_get_data(self_msg_iter
);
1874 BT_ASSERT(trimmer_it
);
1875 destroy_trimmer_iterator(trimmer_it
);