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_LOG_OUTPUT_LEVEL (trimmer_comp->log_level)
25 #define BT_LOG_TAG "PLUGIN/FLT.UTILS.TRIMMER"
26 #include "logging/comp-logging.h"
28 #include "compat/utc.h"
29 #include "compat/time.h"
30 #include <babeltrace2/babeltrace.h>
31 #include "common/common.h"
32 #include "common/assert.h"
36 #include "compat/glib.h"
40 #define NS_PER_S INT64_C(1000000000)
42 static const char * const in_port_name
= "in";
45 unsigned int hour
, minute
, second
, ns
;
48 struct trimmer_bound
{
50 * Nanoseconds from origin, valid if `is_set` is set and
51 * `is_infinite` is false.
53 int64_t ns_from_origin
;
55 /* True if this bound's full time (`ns_from_origin`) is set */
59 * True if this bound represents the infinity (negative or
60 * positive depending on which bound it is). If this is true,
61 * then we don't care about `ns_from_origin` above.
66 * This bound's time without the date; this time is used to set
67 * `ns_from_origin` once we know the date.
69 struct trimmer_time time
;
73 struct trimmer_bound begin
, end
;
75 bt_logging_level log_level
;
76 bt_self_component
*self_comp
;
79 enum trimmer_iterator_state
{
81 * Find the first message's date and set the bounds's times
84 TRIMMER_ITERATOR_STATE_SET_BOUNDS_NS_FROM_ORIGIN
,
87 * Initially seek to the trimming range's beginning time.
89 TRIMMER_ITERATOR_STATE_SEEK_INITIALLY
,
92 * Fill the output message queue for as long as received input
93 * messages are within the trimming time range.
95 TRIMMER_ITERATOR_STATE_TRIM
,
97 /* Flush the remaining messages in the output message queue */
98 TRIMMER_ITERATOR_STATE_ENDING
,
100 /* Trimming operation and message iterator is ended */
101 TRIMMER_ITERATOR_STATE_ENDED
,
104 struct trimmer_iterator
{
106 struct trimmer_comp
*trimmer_comp
;
109 bt_self_message_iterator
*self_msg_iter
;
111 enum trimmer_iterator_state state
;
114 bt_self_component_port_input_message_iterator
*upstream_iter
;
115 struct trimmer_bound begin
, end
;
118 * Queue of `const bt_message *` (owned by the queue).
120 * This is where the trimming operation pushes the messages to
121 * output by this message iterator.
123 GQueue
*output_messages
;
126 * Hash table of `bt_stream *` (weak) to
127 * `struct trimmer_iterator_stream_state *` (owned by the HT).
129 GHashTable
*stream_states
;
132 struct trimmer_iterator_stream_state
{
134 const bt_stream
*stream
;
136 /* Have we seen a message with clock_snapshot going through this stream? */
137 bool seen_clock_snapshot
;
139 /* Owned by this (`NULL` initially and between packets) */
140 const bt_packet
*cur_packet
;
144 void destroy_trimmer_comp(struct trimmer_comp
*trimmer_comp
)
146 BT_ASSERT(trimmer_comp
);
147 g_free(trimmer_comp
);
151 struct trimmer_comp
*create_trimmer_comp(void)
153 return g_new0(struct trimmer_comp
, 1);
157 void trimmer_finalize(bt_self_component_filter
*self_comp
)
159 struct trimmer_comp
*trimmer_comp
=
160 bt_self_component_get_data(
161 bt_self_component_filter_as_self_component(self_comp
));
164 destroy_trimmer_comp(trimmer_comp
);
169 * Compile regex in `pattern`, and try to match `string`. If there's a match,
170 * return true and set `*match_info` to the list of matches. The list of
171 * matches must be freed by the caller. If there's no match, return false and
172 * set `*match_info` to NULL;
175 bool compile_and_match(const char *pattern
, const char *string
, GMatchInfo
**match_info
) {
176 bool matches
= false;
177 GError
*regex_error
= NULL
;
180 regex
= g_regex_new(pattern
, 0, 0, ®ex_error
);
185 matches
= g_regex_match(regex
, string
, 0, match_info
);
188 * g_regex_match allocates `*match_info` even if it returns
189 * FALSE. If there's no match, we have no use for it, so free
190 * it immediatly and don't return it to the caller.
192 g_match_info_free(*match_info
);
196 g_regex_unref(regex
);
201 g_error_free(regex_error
);
208 * Convert the captured text in match number `match_num` in `match_info`
209 * to an unsigned integer.
212 guint64
match_to_uint(const GMatchInfo
*match_info
, gint match_num
) {
213 gchar
*text
, *endptr
;
216 text
= g_match_info_fetch(match_info
, match_num
);
220 * Because the input is carefully sanitized with regexes by the caller,
221 * we assume that g_ascii_strtoull cannot fail.
224 result
= g_ascii_strtoull(text
, &endptr
, 10);
225 BT_ASSERT(endptr
> text
);
226 BT_ASSERT(errno
== 0);
234 * When parsing the nanoseconds part, .512 means .512000000, not .000000512.
235 * This function is like match_to_uint, but multiplies the parsed number to get
236 * the expected result.
239 guint64
match_to_uint_ns(const GMatchInfo
*match_info
, gint match_num
) {
242 gint start_pos
, end_pos
, power
;
243 static int pow10
[] = {
244 1, 10, 100, 1000, 10000, 100000, 1000000, 10000000, 100000000,
247 nanoseconds
= match_to_uint(match_info
, match_num
);
249 /* Multiply by 10 as many times as there are omitted digits. */
250 ret
= g_match_info_fetch_pos(match_info
, match_num
, &start_pos
, &end_pos
);
253 power
= 9 - (end_pos
- start_pos
);
254 BT_ASSERT(power
>= 0 && power
<= 8);
256 nanoseconds
*= pow10
[power
];
262 * Sets the time (in ns from origin) of a trimmer bound from date and
265 * Returns a negative value if anything goes wrong.
268 int set_bound_ns_from_origin(struct trimmer_bound
*bound
,
269 unsigned int year
, unsigned int month
, unsigned int day
,
270 unsigned int hour
, unsigned int minute
, unsigned int second
,
271 unsigned int ns
, bool is_gmt
)
281 .tm_year
= year
- 1900,
286 result
= bt_timegm(&tm
);
288 result
= mktime(&tm
);
297 bound
->ns_from_origin
= (int64_t) result
;
298 bound
->ns_from_origin
*= NS_PER_S
;
299 bound
->ns_from_origin
+= ns
;
300 bound
->is_set
= true;
307 * Parses a timestamp, figuring out its format.
309 * Returns a negative value if anything goes wrong.
313 * YYYY-MM-DD hh:mm[:ss[.ns]]
317 * TODO: Check overflows.
320 int set_bound_from_str(struct trimmer_comp
*trimmer_comp
,
321 const char *str
, struct trimmer_bound
*bound
, bool is_gmt
)
323 /* Matches YYYY-MM-DD */
324 #define DATE_RE "([0-9]{4})-([0-9]{2})-([0-9]{2})"
326 /* Matches HH:MM[:SS[.NS]] */
327 #define TIME_RE "([0-9]{2}):([0-9]{2})(?::([0-9]{2})(?:\\.([0-9]{1,9}))?)?"
329 /* Matches [-]SS[.NS] */
330 #define S_NS_RE "^(-?)([0-9]+)(?:\\.([0-9]{1,9}))?$"
332 GMatchInfo
*match_info
;
335 /* Try `YYYY-MM-DD hh:mm[:ss[.ns]]` format */
336 if (compile_and_match("^" DATE_RE
" " TIME_RE
"$", str
, &match_info
)) {
337 unsigned int year
= 0, month
= 0, day
= 0, hours
= 0, minutes
= 0, seconds
= 0, nanoseconds
= 0;
338 gint match_count
= g_match_info_get_match_count(match_info
);
340 BT_ASSERT(match_count
>= 6 && match_count
<= 8);
342 year
= match_to_uint(match_info
, 1);
343 month
= match_to_uint(match_info
, 2);
344 day
= match_to_uint(match_info
, 3);
345 hours
= match_to_uint(match_info
, 4);
346 minutes
= match_to_uint(match_info
, 5);
348 if (match_count
>= 7) {
349 seconds
= match_to_uint(match_info
, 6);
352 if (match_count
>= 8) {
353 nanoseconds
= match_to_uint_ns(match_info
, 7);
356 set_bound_ns_from_origin(bound
, year
, month
, day
, hours
, minutes
, seconds
, nanoseconds
, is_gmt
);
361 if (compile_and_match("^" DATE_RE
"$", str
, &match_info
)) {
362 unsigned int year
= 0, month
= 0, day
= 0;
364 BT_ASSERT(g_match_info_get_match_count(match_info
) == 4);
366 year
= match_to_uint(match_info
, 1);
367 month
= match_to_uint(match_info
, 2);
368 day
= match_to_uint(match_info
, 3);
370 set_bound_ns_from_origin(bound
, year
, month
, day
, 0, 0, 0, 0, is_gmt
);
375 /* Try `hh:mm[:ss[.ns]]` format */
376 if (compile_and_match("^" TIME_RE
"$", str
, &match_info
)) {
377 gint match_count
= g_match_info_get_match_count(match_info
);
378 BT_ASSERT(match_count
>= 3 && match_count
<= 5);
379 bound
->time
.hour
= match_to_uint(match_info
, 1);
380 bound
->time
.minute
= match_to_uint(match_info
, 2);
382 if (match_count
>= 4) {
383 bound
->time
.second
= match_to_uint(match_info
, 3);
386 if (match_count
>= 5) {
387 bound
->time
.ns
= match_to_uint_ns(match_info
, 4);
393 /* Try `[-]s[.ns]` format */
394 if (compile_and_match("^" S_NS_RE
"$", str
, &match_info
)) {
395 gboolean is_neg
, fetch_pos_ret
;
396 gint start_pos
, end_pos
, match_count
;
397 guint64 seconds
, nanoseconds
= 0;
399 match_count
= g_match_info_get_match_count(match_info
);
400 BT_ASSERT(match_count
>= 3 && match_count
<= 4);
402 /* Check for presence of negation sign. */
403 fetch_pos_ret
= g_match_info_fetch_pos(match_info
, 1, &start_pos
, &end_pos
);
404 BT_ASSERT(fetch_pos_ret
);
405 is_neg
= (end_pos
- start_pos
) > 0;
407 seconds
= match_to_uint(match_info
, 2);
409 if (match_count
>= 4) {
410 nanoseconds
= match_to_uint_ns(match_info
, 3);
413 bound
->ns_from_origin
= seconds
* NS_PER_S
+ nanoseconds
;
416 bound
->ns_from_origin
= -bound
->ns_from_origin
;
419 bound
->is_set
= true;
424 BT_COMP_LOGE_APPEND_CAUSE(trimmer_comp
->self_comp
,
425 "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_APPEND_CAUSE(trimmer_comp
->self_comp
,
460 "`%s` parameter must be an integer or a string value.",
466 ret
= set_bound_from_str(trimmer_comp
, arg
, bound
, is_gmt
);
473 int validate_trimmer_bounds(struct trimmer_comp
*trimmer_comp
,
474 struct trimmer_bound
*begin
, struct trimmer_bound
*end
)
478 BT_ASSERT(begin
->is_set
);
479 BT_ASSERT(end
->is_set
);
481 if (!begin
->is_infinite
&& !end
->is_infinite
&&
482 begin
->ns_from_origin
> end
->ns_from_origin
) {
483 BT_COMP_LOGE_APPEND_CAUSE(trimmer_comp
->self_comp
,
484 "Trimming time range's beginning time is greater than end time: "
485 "begin-ns-from-origin=%" PRId64
", "
486 "end-ns-from-origin=%" PRId64
,
487 begin
->ns_from_origin
,
488 end
->ns_from_origin
);
493 if (!begin
->is_infinite
&& begin
->ns_from_origin
== INT64_MIN
) {
494 BT_COMP_LOGE_APPEND_CAUSE(trimmer_comp
->self_comp
,
495 "Invalid trimming time range's beginning time: "
496 "ns-from-origin=%" PRId64
,
497 begin
->ns_from_origin
);
502 if (!end
->is_infinite
&& end
->ns_from_origin
== INT64_MIN
) {
503 BT_COMP_LOGE_APPEND_CAUSE(trimmer_comp
->self_comp
,
504 "Invalid trimming time range's end time: "
505 "ns-from-origin=%" PRId64
,
506 end
->ns_from_origin
);
516 int init_trimmer_comp_from_params(struct trimmer_comp
*trimmer_comp
,
517 const bt_value
*params
)
519 const bt_value
*value
;
523 value
= bt_value_map_borrow_entry_value_const(params
, "gmt");
525 trimmer_comp
->is_gmt
= (bool) bt_value_bool_get(value
);
528 value
= bt_value_map_borrow_entry_value_const(params
, "begin");
530 if (set_bound_from_param(trimmer_comp
, "begin", value
,
531 &trimmer_comp
->begin
, trimmer_comp
->is_gmt
)) {
532 /* set_bound_from_param() logs errors */
537 trimmer_comp
->begin
.is_infinite
= true;
538 trimmer_comp
->begin
.is_set
= true;
541 value
= bt_value_map_borrow_entry_value_const(params
, "end");
543 if (set_bound_from_param(trimmer_comp
, "end", value
,
544 &trimmer_comp
->end
, trimmer_comp
->is_gmt
)) {
545 /* set_bound_from_param() logs errors */
550 trimmer_comp
->end
.is_infinite
= true;
551 trimmer_comp
->end
.is_set
= true;
555 if (trimmer_comp
->begin
.is_set
&& trimmer_comp
->end
.is_set
) {
556 /* validate_trimmer_bounds() logs errors */
557 ret
= validate_trimmer_bounds(trimmer_comp
,
558 &trimmer_comp
->begin
, &trimmer_comp
->end
);
564 bt_component_class_initialize_method_status
trimmer_init(
565 bt_self_component_filter
*self_comp_flt
,
566 bt_self_component_filter_configuration
*config
,
567 const bt_value
*params
, void *init_data
)
570 bt_component_class_initialize_method_status status
=
571 BT_COMPONENT_CLASS_INITIALIZE_METHOD_STATUS_OK
;
572 bt_self_component_add_port_status add_port_status
;
573 struct trimmer_comp
*trimmer_comp
= create_trimmer_comp();
574 bt_self_component
*self_comp
=
575 bt_self_component_filter_as_self_component(self_comp_flt
);
577 status
= BT_COMPONENT_CLASS_INITIALIZE_METHOD_STATUS_MEMORY_ERROR
;
581 trimmer_comp
->log_level
= bt_component_get_logging_level(
582 bt_self_component_as_component(self_comp
));
583 trimmer_comp
->self_comp
= self_comp
;
584 add_port_status
= bt_self_component_filter_add_input_port(
585 self_comp_flt
, in_port_name
, NULL
, NULL
);
586 switch (add_port_status
) {
587 case BT_SELF_COMPONENT_ADD_PORT_STATUS_ERROR
:
588 status
= BT_COMPONENT_CLASS_INITIALIZE_METHOD_STATUS_ERROR
;
590 case BT_SELF_COMPONENT_ADD_PORT_STATUS_MEMORY_ERROR
:
591 status
= BT_COMPONENT_CLASS_INITIALIZE_METHOD_STATUS_MEMORY_ERROR
;
597 add_port_status
= bt_self_component_filter_add_output_port(
598 self_comp_flt
, "out", NULL
, NULL
);
599 switch (add_port_status
) {
600 case BT_SELF_COMPONENT_ADD_PORT_STATUS_ERROR
:
601 status
= BT_COMPONENT_CLASS_INITIALIZE_METHOD_STATUS_ERROR
;
603 case BT_SELF_COMPONENT_ADD_PORT_STATUS_MEMORY_ERROR
:
604 status
= BT_COMPONENT_CLASS_INITIALIZE_METHOD_STATUS_MEMORY_ERROR
;
610 ret
= init_trimmer_comp_from_params(trimmer_comp
, params
);
612 status
= BT_COMPONENT_CLASS_INITIALIZE_METHOD_STATUS_ERROR
;
616 bt_self_component_set_data(self_comp
, trimmer_comp
);
620 if (status
== BT_COMPONENT_CLASS_INITIALIZE_METHOD_STATUS_OK
) {
621 status
= BT_COMPONENT_CLASS_INITIALIZE_METHOD_STATUS_ERROR
;
625 destroy_trimmer_comp(trimmer_comp
);
633 void destroy_trimmer_iterator(struct trimmer_iterator
*trimmer_it
)
639 bt_self_component_port_input_message_iterator_put_ref(
640 trimmer_it
->upstream_iter
);
642 if (trimmer_it
->output_messages
) {
643 g_queue_free(trimmer_it
->output_messages
);
646 if (trimmer_it
->stream_states
) {
647 g_hash_table_destroy(trimmer_it
->stream_states
);
656 void destroy_trimmer_iterator_stream_state(
657 struct trimmer_iterator_stream_state
*sstate
)
660 BT_PACKET_PUT_REF_AND_RESET(sstate
->cur_packet
);
665 bt_component_class_message_iterator_initialize_method_status
trimmer_msg_iter_init(
666 bt_self_message_iterator
*self_msg_iter
,
667 bt_self_message_iterator_configuration
*config
,
668 bt_self_component_filter
*self_comp
,
669 bt_self_component_port_output
*port
)
671 bt_component_class_message_iterator_initialize_method_status status
;
672 bt_self_component_port_input_message_iterator_create_from_message_iterator_status
674 struct trimmer_iterator
*trimmer_it
;
676 trimmer_it
= g_new0(struct trimmer_iterator
, 1);
678 status
= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_INITIALIZE_METHOD_STATUS_MEMORY_ERROR
;
682 trimmer_it
->trimmer_comp
= bt_self_component_get_data(
683 bt_self_component_filter_as_self_component(self_comp
));
684 BT_ASSERT(trimmer_it
->trimmer_comp
);
686 if (trimmer_it
->trimmer_comp
->begin
.is_set
&&
687 trimmer_it
->trimmer_comp
->end
.is_set
) {
689 * Both trimming time range's bounds are set, so skip
691 * `TRIMMER_ITERATOR_STATE_SET_BOUNDS_NS_FROM_ORIGIN`
694 trimmer_it
->state
= TRIMMER_ITERATOR_STATE_SEEK_INITIALLY
;
697 trimmer_it
->begin
= trimmer_it
->trimmer_comp
->begin
;
698 trimmer_it
->end
= trimmer_it
->trimmer_comp
->end
;
700 bt_self_component_port_input_message_iterator_create_from_message_iterator(
702 bt_self_component_filter_borrow_input_port_by_name(
703 self_comp
, in_port_name
), &trimmer_it
->upstream_iter
);
704 if (msg_iter_status
!= BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_CREATE_FROM_MESSAGE_ITERATOR_STATUS_OK
) {
705 status
= (int) msg_iter_status
;
709 trimmer_it
->output_messages
= g_queue_new();
710 if (!trimmer_it
->output_messages
) {
711 status
= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_INITIALIZE_METHOD_STATUS_MEMORY_ERROR
;
715 trimmer_it
->stream_states
= g_hash_table_new_full(g_direct_hash
,
716 g_direct_equal
, NULL
,
717 (GDestroyNotify
) destroy_trimmer_iterator_stream_state
);
718 if (!trimmer_it
->stream_states
) {
719 status
= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_INITIALIZE_METHOD_STATUS_MEMORY_ERROR
;
724 * The trimmer requires upstream messages to have times, so it can
725 * always seek forward.
727 bt_self_message_iterator_configuration_set_can_seek_forward(
730 trimmer_it
->self_msg_iter
= self_msg_iter
;
731 bt_self_message_iterator_set_data(self_msg_iter
, trimmer_it
);
733 status
= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_INITIALIZE_METHOD_STATUS_OK
;
737 destroy_trimmer_iterator(trimmer_it
);
744 int get_msg_ns_from_origin(const bt_message
*msg
, int64_t *ns_from_origin
,
745 bool *has_clock_snapshot
)
747 const bt_clock_class
*clock_class
= NULL
;
748 const bt_clock_snapshot
*clock_snapshot
= NULL
;
752 BT_ASSERT(ns_from_origin
);
753 BT_ASSERT(has_clock_snapshot
);
755 switch (bt_message_get_type(msg
)) {
756 case BT_MESSAGE_TYPE_EVENT
:
758 bt_message_event_borrow_stream_class_default_clock_class_const(
760 if (G_UNLIKELY(!clock_class
)) {
764 clock_snapshot
= bt_message_event_borrow_default_clock_snapshot_const(
767 case BT_MESSAGE_TYPE_PACKET_BEGINNING
:
769 bt_message_packet_beginning_borrow_stream_class_default_clock_class_const(
771 if (G_UNLIKELY(!clock_class
)) {
775 clock_snapshot
= bt_message_packet_beginning_borrow_default_clock_snapshot_const(
778 case BT_MESSAGE_TYPE_PACKET_END
:
780 bt_message_packet_end_borrow_stream_class_default_clock_class_const(
782 if (G_UNLIKELY(!clock_class
)) {
786 clock_snapshot
= bt_message_packet_end_borrow_default_clock_snapshot_const(
789 case BT_MESSAGE_TYPE_STREAM_BEGINNING
:
791 enum bt_message_stream_clock_snapshot_state cs_state
;
794 bt_message_stream_beginning_borrow_stream_class_default_clock_class_const(msg
);
795 if (G_UNLIKELY(!clock_class
)) {
799 cs_state
= bt_message_stream_beginning_borrow_default_clock_snapshot_const(msg
, &clock_snapshot
);
800 if (cs_state
!= BT_MESSAGE_STREAM_CLOCK_SNAPSHOT_STATE_KNOWN
) {
801 goto no_clock_snapshot
;
806 case BT_MESSAGE_TYPE_STREAM_END
:
808 enum bt_message_stream_clock_snapshot_state cs_state
;
811 bt_message_stream_end_borrow_stream_class_default_clock_class_const(msg
);
812 if (G_UNLIKELY(!clock_class
)) {
816 cs_state
= bt_message_stream_end_borrow_default_clock_snapshot_const(msg
, &clock_snapshot
);
817 if (cs_state
!= BT_MESSAGE_STREAM_CLOCK_SNAPSHOT_STATE_KNOWN
) {
818 goto no_clock_snapshot
;
823 case BT_MESSAGE_TYPE_DISCARDED_EVENTS
:
825 bt_message_discarded_events_borrow_stream_class_default_clock_class_const(
827 if (G_UNLIKELY(!clock_class
)) {
831 clock_snapshot
= bt_message_discarded_events_borrow_beginning_default_clock_snapshot_const(
834 case BT_MESSAGE_TYPE_DISCARDED_PACKETS
:
836 bt_message_discarded_packets_borrow_stream_class_default_clock_class_const(
838 if (G_UNLIKELY(!clock_class
)) {
842 clock_snapshot
= bt_message_discarded_packets_borrow_beginning_default_clock_snapshot_const(
845 case BT_MESSAGE_TYPE_MESSAGE_ITERATOR_INACTIVITY
:
847 bt_message_message_iterator_inactivity_borrow_default_clock_snapshot_const(
851 goto no_clock_snapshot
;
854 ret
= bt_clock_snapshot_get_ns_from_origin(clock_snapshot
,
856 if (G_UNLIKELY(ret
)) {
860 *has_clock_snapshot
= true;
864 *has_clock_snapshot
= false;
875 void put_messages(bt_message_array_const msgs
, uint64_t count
)
879 for (i
= 0; i
< count
; i
++) {
880 BT_MESSAGE_PUT_REF_AND_RESET(msgs
[i
]);
885 int set_trimmer_iterator_bound(struct trimmer_iterator
*trimmer_it
,
886 struct trimmer_bound
*bound
, int64_t ns_from_origin
,
889 struct trimmer_comp
*trimmer_comp
= trimmer_it
->trimmer_comp
;
892 time_t time_seconds
= (time_t) (ns_from_origin
/ NS_PER_S
);
895 BT_ASSERT(!bound
->is_set
);
898 /* We only need to extract the date from this time */
900 res
= bt_gmtime_r(&time_seconds
, &tm
);
902 res
= bt_localtime_r(&time_seconds
, &tm
);
906 BT_COMP_LOGE_APPEND_CAUSE_ERRNO(trimmer_comp
->self_comp
,
907 "Cannot convert timestamp to date and time",
908 ": ts=%" PRId64
, (int64_t) time_seconds
);
913 ret
= set_bound_ns_from_origin(bound
, tm
.tm_year
+ 1900, tm
.tm_mon
+ 1,
914 tm
.tm_mday
, bound
->time
.hour
, bound
->time
.minute
,
915 bound
->time
.second
, bound
->time
.ns
, is_gmt
);
922 bt_component_class_message_iterator_next_method_status
923 state_set_trimmer_iterator_bounds(
924 struct trimmer_iterator
*trimmer_it
)
926 bt_message_iterator_next_status upstream_iter_status
=
927 BT_MESSAGE_ITERATOR_NEXT_STATUS_OK
;
928 struct trimmer_comp
*trimmer_comp
= trimmer_it
->trimmer_comp
;
929 bt_message_array_const msgs
;
931 int64_t ns_from_origin
= INT64_MIN
;
935 BT_ASSERT(!trimmer_it
->begin
.is_set
||
936 !trimmer_it
->end
.is_set
);
939 upstream_iter_status
=
940 bt_self_component_port_input_message_iterator_next(
941 trimmer_it
->upstream_iter
, &msgs
, &count
);
942 if (upstream_iter_status
!= BT_MESSAGE_ITERATOR_NEXT_STATUS_OK
) {
946 for (i
= 0; i
< count
; i
++) {
947 const bt_message
*msg
= msgs
[i
];
948 bool has_ns_from_origin
;
951 ret
= get_msg_ns_from_origin(msg
, &ns_from_origin
,
952 &has_ns_from_origin
);
957 if (!has_ns_from_origin
) {
961 BT_ASSERT(ns_from_origin
!= INT64_MIN
&&
962 ns_from_origin
!= INT64_MAX
);
963 put_messages(msgs
, count
);
967 put_messages(msgs
, count
);
971 if (!trimmer_it
->begin
.is_set
) {
972 BT_ASSERT(!trimmer_it
->begin
.is_infinite
);
973 ret
= set_trimmer_iterator_bound(trimmer_it
, &trimmer_it
->begin
,
974 ns_from_origin
, trimmer_comp
->is_gmt
);
980 if (!trimmer_it
->end
.is_set
) {
981 BT_ASSERT(!trimmer_it
->end
.is_infinite
);
982 ret
= set_trimmer_iterator_bound(trimmer_it
, &trimmer_it
->end
,
983 ns_from_origin
, trimmer_comp
->is_gmt
);
989 ret
= validate_trimmer_bounds(trimmer_it
->trimmer_comp
,
990 &trimmer_it
->begin
, &trimmer_it
->end
);
998 put_messages(msgs
, count
);
999 upstream_iter_status
= BT_MESSAGE_ITERATOR_NEXT_STATUS_ERROR
;
1002 return (int) upstream_iter_status
;
1006 bt_component_class_message_iterator_next_method_status
state_seek_initially(
1007 struct trimmer_iterator
*trimmer_it
)
1009 struct trimmer_comp
*trimmer_comp
= trimmer_it
->trimmer_comp
;
1010 bt_component_class_message_iterator_next_method_status status
;
1012 BT_ASSERT(trimmer_it
->begin
.is_set
);
1014 if (trimmer_it
->begin
.is_infinite
) {
1017 status
= (int) bt_self_component_port_input_message_iterator_can_seek_beginning(
1018 trimmer_it
->upstream_iter
, &can_seek
);
1019 if (status
!= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK
) {
1021 BT_COMP_LOGE_APPEND_CAUSE(trimmer_comp
->self_comp
,
1022 "Cannot make upstream message iterator initially seek its beginning.");
1029 BT_COMP_LOGE_APPEND_CAUSE(trimmer_comp
->self_comp
,
1030 "Cannot make upstream message iterator initially seek its beginning.");
1031 status
= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_ERROR
;
1035 status
= (int) bt_self_component_port_input_message_iterator_seek_beginning(
1036 trimmer_it
->upstream_iter
);
1040 status
= (int) bt_self_component_port_input_message_iterator_can_seek_ns_from_origin(
1041 trimmer_it
->upstream_iter
, trimmer_it
->begin
.ns_from_origin
,
1044 if (status
!= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK
) {
1046 BT_COMP_LOGE_APPEND_CAUSE(trimmer_comp
->self_comp
,
1047 "Cannot make upstream message iterator initially seek: seek-ns-from-origin=%" PRId64
,
1048 trimmer_it
->begin
.ns_from_origin
);
1055 BT_COMP_LOGE_APPEND_CAUSE(trimmer_comp
->self_comp
,
1056 "Cannot make upstream message iterator initially seek: seek-ns-from-origin=%" PRId64
,
1057 trimmer_it
->begin
.ns_from_origin
);
1058 status
= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_ERROR
;
1062 status
= (int) bt_self_component_port_input_message_iterator_seek_ns_from_origin(
1063 trimmer_it
->upstream_iter
, trimmer_it
->begin
.ns_from_origin
);
1066 if (status
== BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK
) {
1067 trimmer_it
->state
= TRIMMER_ITERATOR_STATE_TRIM
;
1075 void push_message(struct trimmer_iterator
*trimmer_it
, const bt_message
*msg
)
1077 g_queue_push_head(trimmer_it
->output_messages
, (void *) msg
);
1081 const bt_message
*pop_message(struct trimmer_iterator
*trimmer_it
)
1083 return g_queue_pop_tail(trimmer_it
->output_messages
);
1087 int clock_raw_value_from_ns_from_origin(const bt_clock_class
*clock_class
,
1088 int64_t ns_from_origin
, uint64_t *raw_value
)
1091 int64_t cc_offset_s
;
1092 uint64_t cc_offset_cycles
;
1095 bt_clock_class_get_offset(clock_class
, &cc_offset_s
, &cc_offset_cycles
);
1096 cc_freq
= bt_clock_class_get_frequency(clock_class
);
1097 return bt_common_clock_value_from_ns_from_origin(cc_offset_s
,
1098 cc_offset_cycles
, cc_freq
, ns_from_origin
, raw_value
);
1102 bt_component_class_message_iterator_next_method_status
1103 end_stream(struct trimmer_iterator
*trimmer_it
,
1104 struct trimmer_iterator_stream_state
*sstate
)
1106 bt_component_class_message_iterator_next_method_status status
=
1107 BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK
;
1108 /* Initialize to silence maybe-uninitialized warning. */
1109 uint64_t raw_value
= 0;
1110 bt_message
*msg
= NULL
;
1112 BT_ASSERT(!trimmer_it
->end
.is_infinite
);
1113 BT_ASSERT(sstate
->stream
);
1116 * If we haven't seen a message with a clock snapshot, we don't know if the trimmer's end bound is within
1117 * the clock's range, so it wouldn't be safe to try to convert ns_from_origin to a clock value.
1119 * Also, it would be a bit of a lie to generate a stream end message with the end bound as its
1120 * clock snapshot, because we don't really know if the stream existed at that time. If we have
1121 * seen a message with a clock snapshot and the stream is cut short by another message with a
1122 * clock snapshot, then we are sure that the the end bound time is not below the clock range,
1123 * and we know the stream was active at that time (and that we cut it short).
1125 if (sstate
->seen_clock_snapshot
) {
1126 const bt_clock_class
*clock_class
;
1129 clock_class
= bt_stream_class_borrow_default_clock_class_const(
1130 bt_stream_borrow_class_const(sstate
->stream
));
1131 BT_ASSERT(clock_class
);
1132 ret
= clock_raw_value_from_ns_from_origin(clock_class
,
1133 trimmer_it
->end
.ns_from_origin
, &raw_value
);
1135 status
= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_ERROR
;
1140 if (sstate
->cur_packet
) {
1142 * Create and push a packet end message, making its time
1143 * the trimming range's end time.
1145 * We know that we must have seen a clock snapshot, the one in
1146 * the packet beginning message, since trimmer currently
1147 * requires packet messages to have clock snapshots (see comment
1148 * in create_stream_state_entry).
1150 BT_ASSERT(sstate
->seen_clock_snapshot
);
1152 msg
= bt_message_packet_end_create_with_default_clock_snapshot(
1153 trimmer_it
->self_msg_iter
, sstate
->cur_packet
,
1156 status
= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_MEMORY_ERROR
;
1160 push_message(trimmer_it
, msg
);
1162 BT_PACKET_PUT_REF_AND_RESET(sstate
->cur_packet
);
1165 /* Create and push a stream end message. */
1166 msg
= bt_message_stream_end_create(trimmer_it
->self_msg_iter
,
1169 status
= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_MEMORY_ERROR
;
1173 if (sstate
->seen_clock_snapshot
) {
1174 bt_message_stream_end_set_default_clock_snapshot(msg
, raw_value
);
1177 push_message(trimmer_it
, msg
);
1181 * Just to make sure that we don't use this stream state again
1182 * in the future without an obvious error.
1184 sstate
->stream
= NULL
;
1187 bt_message_put_ref(msg
);
1192 bt_component_class_message_iterator_next_method_status
end_iterator_streams(
1193 struct trimmer_iterator
*trimmer_it
)
1195 bt_component_class_message_iterator_next_method_status status
=
1196 BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK
;
1197 GHashTableIter iter
;
1198 gpointer key
, sstate
;
1200 if (trimmer_it
->end
.is_infinite
) {
1202 * An infinite trimming range's end time guarantees that
1203 * we received (and pushed) all the appropriate end
1210 * End each stream and then remove them from the hash table of
1211 * stream states to release unneeded references.
1213 g_hash_table_iter_init(&iter
, trimmer_it
->stream_states
);
1215 while (g_hash_table_iter_next(&iter
, &key
, &sstate
)) {
1216 status
= end_stream(trimmer_it
, sstate
);
1223 g_hash_table_remove_all(trimmer_it
->stream_states
);
1230 bt_component_class_message_iterator_next_method_status
1231 create_stream_state_entry(
1232 struct trimmer_iterator
*trimmer_it
,
1233 const struct bt_stream
*stream
,
1234 struct trimmer_iterator_stream_state
**stream_state
)
1236 struct trimmer_comp
*trimmer_comp
= trimmer_it
->trimmer_comp
;
1237 bt_component_class_message_iterator_next_method_status status
;
1238 struct trimmer_iterator_stream_state
*sstate
;
1239 const bt_stream_class
*sc
;
1241 BT_ASSERT(!bt_g_hash_table_contains(trimmer_it
->stream_states
, stream
));
1244 * Validate right now that the stream's class
1245 * has a registered default clock class so that
1246 * an existing stream state guarantees existing
1247 * default clock snapshots for its associated
1250 * Also check that clock snapshots are always
1253 sc
= bt_stream_borrow_class_const(stream
);
1254 if (!bt_stream_class_borrow_default_clock_class_const(sc
)) {
1255 BT_COMP_LOGE_APPEND_CAUSE(trimmer_comp
->self_comp
,
1256 "Unsupported stream: stream class does "
1257 "not have a default clock class: "
1259 "stream-id=%" PRIu64
", "
1260 "stream-name=\"%s\"",
1261 stream
, bt_stream_get_id(stream
),
1262 bt_stream_get_name(stream
));
1263 status
= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_ERROR
;
1268 * Temporary: make sure packet beginning, packet
1269 * end, discarded events, and discarded packets
1270 * messages have default clock snapshots until
1271 * the support for not having them is
1274 if (!bt_stream_class_packets_have_beginning_default_clock_snapshot(
1276 BT_COMP_LOGE_APPEND_CAUSE(trimmer_comp
->self_comp
,
1277 "Unsupported stream: packets have no beginning clock snapshot: "
1279 "stream-id=%" PRIu64
", "
1280 "stream-name=\"%s\"",
1281 stream
, bt_stream_get_id(stream
),
1282 bt_stream_get_name(stream
));
1283 status
= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_ERROR
;
1287 if (!bt_stream_class_packets_have_end_default_clock_snapshot(
1289 BT_COMP_LOGE_APPEND_CAUSE(trimmer_comp
->self_comp
,
1290 "Unsupported stream: packets have no end clock snapshot: "
1292 "stream-id=%" PRIu64
", "
1293 "stream-name=\"%s\"",
1294 stream
, bt_stream_get_id(stream
),
1295 bt_stream_get_name(stream
));
1296 status
= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_ERROR
;
1300 if (bt_stream_class_supports_discarded_events(sc
) &&
1301 !bt_stream_class_discarded_events_have_default_clock_snapshots(sc
)) {
1302 BT_COMP_LOGE_APPEND_CAUSE(trimmer_comp
->self_comp
,
1303 "Unsupported stream: discarded events have no clock snapshots: "
1305 "stream-id=%" PRIu64
", "
1306 "stream-name=\"%s\"",
1307 stream
, bt_stream_get_id(stream
),
1308 bt_stream_get_name(stream
));
1309 status
= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_ERROR
;
1313 if (bt_stream_class_supports_discarded_packets(sc
) &&
1314 !bt_stream_class_discarded_packets_have_default_clock_snapshots(sc
)) {
1315 BT_COMP_LOGE_APPEND_CAUSE(trimmer_comp
->self_comp
,
1316 "Unsupported stream: discarded packets "
1317 "have no clock snapshots: "
1319 "stream-id=%" PRIu64
", "
1320 "stream-name=\"%s\"",
1321 stream
, bt_stream_get_id(stream
),
1322 bt_stream_get_name(stream
));
1323 status
= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_ERROR
;
1327 sstate
= g_new0(struct trimmer_iterator_stream_state
, 1);
1329 status
= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_MEMORY_ERROR
;
1333 sstate
->stream
= stream
;
1335 g_hash_table_insert(trimmer_it
->stream_states
, (void *) stream
, sstate
);
1337 *stream_state
= sstate
;
1339 status
= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK
;
1346 struct trimmer_iterator_stream_state
*get_stream_state_entry(
1347 struct trimmer_iterator
*trimmer_it
,
1348 const struct bt_stream
*stream
)
1350 struct trimmer_iterator_stream_state
*sstate
;
1353 sstate
= g_hash_table_lookup(trimmer_it
->stream_states
, stream
);
1360 * Handles a message which is associated to a given stream state. This
1361 * _could_ make the iterator's output message queue grow; this could
1362 * also consume the message without pushing anything to this queue, only
1363 * modifying the stream state.
1365 * This function consumes the `msg` reference, _whatever the outcome_.
1367 * If non-NULL, `ns_from_origin` is the message's time, as given by
1368 * get_msg_ns_from_origin(). If NULL, the message doesn't have a time.
1370 * This function sets `reached_end` if handling this message made the
1371 * iterator reach the end of the trimming range. Note that the output
1372 * message queue could contain messages even if this function sets
1376 bt_component_class_message_iterator_next_method_status
1377 handle_message_with_stream(
1378 struct trimmer_iterator
*trimmer_it
, const bt_message
*msg
,
1379 const struct bt_stream
*stream
, const int64_t *ns_from_origin
,
1382 bt_component_class_message_iterator_next_method_status status
=
1383 BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK
;
1384 bt_message_type msg_type
= bt_message_get_type(msg
);
1386 struct trimmer_iterator_stream_state
*sstate
= NULL
;
1389 * Retrieve the stream's state - except if the message is stream
1390 * beginning, in which case we don't know about about this stream yet.
1392 if (msg_type
!= BT_MESSAGE_TYPE_STREAM_BEGINNING
) {
1393 sstate
= get_stream_state_entry(trimmer_it
, stream
);
1397 case BT_MESSAGE_TYPE_EVENT
:
1399 * Event messages always have a clock snapshot if the stream
1400 * class has a clock class. And we know it has, otherwise we
1401 * couldn't be using the trimmer component.
1403 BT_ASSERT(ns_from_origin
);
1405 if (G_UNLIKELY(!trimmer_it
->end
.is_infinite
&&
1406 *ns_from_origin
> trimmer_it
->end
.ns_from_origin
)) {
1407 status
= end_iterator_streams(trimmer_it
);
1408 *reached_end
= true;
1412 sstate
->seen_clock_snapshot
= true;
1414 push_message(trimmer_it
, msg
);
1418 case BT_MESSAGE_TYPE_PACKET_BEGINNING
:
1420 * Packet beginning messages won't have a clock snapshot if
1421 * stream_class->packets_have_beginning_default_clock_snapshot
1422 * is false. But for now, assume they always do.
1424 BT_ASSERT(ns_from_origin
);
1425 BT_ASSERT(!sstate
->cur_packet
);
1427 if (G_UNLIKELY(!trimmer_it
->end
.is_infinite
&&
1428 *ns_from_origin
> trimmer_it
->end
.ns_from_origin
)) {
1429 status
= end_iterator_streams(trimmer_it
);
1430 *reached_end
= true;
1434 sstate
->cur_packet
=
1435 bt_message_packet_beginning_borrow_packet_const(msg
);
1436 bt_packet_get_ref(sstate
->cur_packet
);
1438 sstate
->seen_clock_snapshot
= true;
1440 push_message(trimmer_it
, msg
);
1444 case BT_MESSAGE_TYPE_PACKET_END
:
1446 * Packet end messages won't have a clock snapshot if
1447 * stream_class->packets_have_end_default_clock_snapshot
1448 * is false. But for now, assume they always do.
1450 BT_ASSERT(ns_from_origin
);
1451 BT_ASSERT(sstate
->cur_packet
);
1453 if (G_UNLIKELY(!trimmer_it
->end
.is_infinite
&&
1454 *ns_from_origin
> trimmer_it
->end
.ns_from_origin
)) {
1455 status
= end_iterator_streams(trimmer_it
);
1456 *reached_end
= true;
1460 BT_PACKET_PUT_REF_AND_RESET(sstate
->cur_packet
);
1462 sstate
->seen_clock_snapshot
= true;
1464 push_message(trimmer_it
, msg
);
1468 case BT_MESSAGE_TYPE_DISCARDED_EVENTS
:
1469 case BT_MESSAGE_TYPE_DISCARDED_PACKETS
:
1472 * `ns_from_origin` is the message's time range's
1473 * beginning time here.
1475 int64_t end_ns_from_origin
;
1476 const bt_clock_snapshot
*end_cs
;
1478 BT_ASSERT(ns_from_origin
);
1480 sstate
->seen_clock_snapshot
= true;
1482 if (bt_message_get_type(msg
) ==
1483 BT_MESSAGE_TYPE_DISCARDED_EVENTS
) {
1485 * Safe to ignore the return value because we
1486 * know there's a default clock and it's always
1489 end_cs
= bt_message_discarded_events_borrow_end_default_clock_snapshot_const(
1493 * Safe to ignore the return value because we
1494 * know there's a default clock and it's always
1497 end_cs
= bt_message_discarded_packets_borrow_end_default_clock_snapshot_const(
1501 if (bt_clock_snapshot_get_ns_from_origin(end_cs
,
1502 &end_ns_from_origin
)) {
1503 status
= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_ERROR
;
1507 if (!trimmer_it
->end
.is_infinite
&&
1508 *ns_from_origin
> trimmer_it
->end
.ns_from_origin
) {
1509 status
= end_iterator_streams(trimmer_it
);
1510 *reached_end
= true;
1514 if (!trimmer_it
->end
.is_infinite
&&
1515 end_ns_from_origin
> trimmer_it
->end
.ns_from_origin
) {
1517 * This message's end time is outside the
1518 * trimming time range: replace it with a new
1519 * message having an end time equal to the
1520 * trimming time range's end and without a
1523 const bt_clock_class
*clock_class
=
1524 bt_clock_snapshot_borrow_clock_class_const(
1526 const bt_clock_snapshot
*begin_cs
;
1527 bt_message
*new_msg
;
1528 uint64_t end_raw_value
;
1530 ret
= clock_raw_value_from_ns_from_origin(clock_class
,
1531 trimmer_it
->end
.ns_from_origin
, &end_raw_value
);
1533 status
= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_ERROR
;
1537 if (msg_type
== BT_MESSAGE_TYPE_DISCARDED_EVENTS
) {
1538 begin_cs
= bt_message_discarded_events_borrow_beginning_default_clock_snapshot_const(
1540 new_msg
= bt_message_discarded_events_create_with_default_clock_snapshots(
1541 trimmer_it
->self_msg_iter
,
1543 bt_clock_snapshot_get_value(begin_cs
),
1546 begin_cs
= bt_message_discarded_packets_borrow_beginning_default_clock_snapshot_const(
1548 new_msg
= bt_message_discarded_packets_create_with_default_clock_snapshots(
1549 trimmer_it
->self_msg_iter
,
1551 bt_clock_snapshot_get_value(begin_cs
),
1556 status
= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_MEMORY_ERROR
;
1560 /* Replace the original message */
1561 BT_MESSAGE_MOVE_REF(msg
, new_msg
);
1564 push_message(trimmer_it
, msg
);
1569 case BT_MESSAGE_TYPE_STREAM_BEGINNING
:
1571 * If this message has a time and this time is greater than the
1572 * trimmer's end bound, it triggers the end of the trim window.
1574 if (G_UNLIKELY(ns_from_origin
&& !trimmer_it
->end
.is_infinite
&&
1575 *ns_from_origin
> trimmer_it
->end
.ns_from_origin
)) {
1576 status
= end_iterator_streams(trimmer_it
);
1577 *reached_end
= true;
1581 /* Learn about this stream. */
1582 status
= create_stream_state_entry(trimmer_it
, stream
, &sstate
);
1583 if (status
!= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK
) {
1587 if (ns_from_origin
) {
1588 sstate
->seen_clock_snapshot
= true;
1591 push_message(trimmer_it
, msg
);
1594 case BT_MESSAGE_TYPE_STREAM_END
:
1599 * If this message has a time and this time is greater than the
1600 * trimmer's end bound, it triggers the end of the trim window.
1602 if (G_UNLIKELY(ns_from_origin
&& !trimmer_it
->end
.is_infinite
&&
1603 *ns_from_origin
> trimmer_it
->end
.ns_from_origin
)) {
1604 status
= end_iterator_streams(trimmer_it
);
1605 *reached_end
= true;
1610 * Either the stream end message's time is within the trimmer's
1611 * bounds, or it doesn't have a time. In both cases, pass
1612 * the message unmodified.
1614 push_message(trimmer_it
, msg
);
1617 /* Forget about this stream. */
1618 removed
= g_hash_table_remove(trimmer_it
->stream_states
, sstate
->stream
);
1627 /* We release the message's reference whatever the outcome */
1628 bt_message_put_ref(msg
);
1633 * Handles an input message. This _could_ make the iterator's output
1634 * message queue grow; this could also consume the message without
1635 * pushing anything to this queue, only modifying the stream state.
1637 * This function consumes the `msg` reference, _whatever the outcome_.
1639 * This function sets `reached_end` if handling this message made the
1640 * iterator reach the end of the trimming range. Note that the output
1641 * message queue could contain messages even if this function sets
1645 bt_component_class_message_iterator_next_method_status
handle_message(
1646 struct trimmer_iterator
*trimmer_it
, const bt_message
*msg
,
1649 bt_component_class_message_iterator_next_method_status status
;
1650 const bt_stream
*stream
= NULL
;
1651 int64_t ns_from_origin
= INT64_MIN
;
1652 bool has_ns_from_origin
= false;
1655 /* Find message's associated stream */
1656 switch (bt_message_get_type(msg
)) {
1657 case BT_MESSAGE_TYPE_EVENT
:
1658 stream
= bt_event_borrow_stream_const(
1659 bt_message_event_borrow_event_const(msg
));
1661 case BT_MESSAGE_TYPE_PACKET_BEGINNING
:
1662 stream
= bt_packet_borrow_stream_const(
1663 bt_message_packet_beginning_borrow_packet_const(msg
));
1665 case BT_MESSAGE_TYPE_PACKET_END
:
1666 stream
= bt_packet_borrow_stream_const(
1667 bt_message_packet_end_borrow_packet_const(msg
));
1669 case BT_MESSAGE_TYPE_DISCARDED_EVENTS
:
1670 stream
= bt_message_discarded_events_borrow_stream_const(msg
);
1672 case BT_MESSAGE_TYPE_DISCARDED_PACKETS
:
1673 stream
= bt_message_discarded_packets_borrow_stream_const(msg
);
1675 case BT_MESSAGE_TYPE_STREAM_BEGINNING
:
1676 stream
= bt_message_stream_beginning_borrow_stream_const(msg
);
1678 case BT_MESSAGE_TYPE_STREAM_END
:
1679 stream
= bt_message_stream_end_borrow_stream_const(msg
);
1685 /* Retrieve the message's time */
1686 ret
= get_msg_ns_from_origin(msg
, &ns_from_origin
, &has_ns_from_origin
);
1687 if (G_UNLIKELY(ret
)) {
1688 status
= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_ERROR
;
1692 if (G_LIKELY(stream
)) {
1693 /* Message associated to a stream */
1694 status
= handle_message_with_stream(trimmer_it
, msg
,
1695 stream
, has_ns_from_origin
? &ns_from_origin
: NULL
, reached_end
);
1698 * handle_message_with_stream_state() unconditionally
1704 * Message not associated to a stream (message iterator
1707 if (G_UNLIKELY(ns_from_origin
> trimmer_it
->end
.ns_from_origin
)) {
1708 BT_MESSAGE_PUT_REF_AND_RESET(msg
);
1709 status
= end_iterator_streams(trimmer_it
);
1710 *reached_end
= true;
1712 push_message(trimmer_it
, msg
);
1713 status
= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK
;
1719 /* We release the message's reference whatever the outcome */
1720 bt_message_put_ref(msg
);
1725 void fill_message_array_from_output_messages(
1726 struct trimmer_iterator
*trimmer_it
,
1727 bt_message_array_const msgs
, uint64_t capacity
, uint64_t *count
)
1732 * Move auto-seek messages to the output array (which is this
1733 * iterator's base message array).
1735 while (capacity
> 0 && !g_queue_is_empty(trimmer_it
->output_messages
)) {
1736 msgs
[*count
] = pop_message(trimmer_it
);
1741 BT_ASSERT(*count
> 0);
1745 bt_component_class_message_iterator_next_method_status
state_ending(
1746 struct trimmer_iterator
*trimmer_it
,
1747 bt_message_array_const msgs
, uint64_t capacity
,
1750 bt_component_class_message_iterator_next_method_status status
=
1751 BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK
;
1753 if (g_queue_is_empty(trimmer_it
->output_messages
)) {
1754 trimmer_it
->state
= TRIMMER_ITERATOR_STATE_ENDED
;
1755 status
= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_END
;
1759 fill_message_array_from_output_messages(trimmer_it
, msgs
,
1767 bt_component_class_message_iterator_next_method_status
1768 state_trim(struct trimmer_iterator
*trimmer_it
,
1769 bt_message_array_const msgs
, uint64_t capacity
,
1772 bt_component_class_message_iterator_next_method_status status
=
1773 BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK
;
1774 bt_message_array_const my_msgs
;
1777 bool reached_end
= false;
1779 while (g_queue_is_empty(trimmer_it
->output_messages
)) {
1780 status
= (int) bt_self_component_port_input_message_iterator_next(
1781 trimmer_it
->upstream_iter
, &my_msgs
, &my_count
);
1782 if (G_UNLIKELY(status
!= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK
)) {
1783 if (status
== BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_END
) {
1784 status
= end_iterator_streams(trimmer_it
);
1785 if (status
!= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK
) {
1790 TRIMMER_ITERATOR_STATE_ENDING
;
1791 status
= state_ending(trimmer_it
, msgs
,
1798 BT_ASSERT(my_count
> 0);
1800 for (i
= 0; i
< my_count
; i
++) {
1801 status
= handle_message(trimmer_it
, my_msgs
[i
],
1805 * handle_message() unconditionally consumes the
1806 * message reference.
1810 if (G_UNLIKELY(status
!=
1811 BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK
)) {
1812 put_messages(my_msgs
, my_count
);
1816 if (G_UNLIKELY(reached_end
)) {
1818 * This message's time was passed the
1819 * trimming time range's end time: we
1820 * are done. Their might still be
1821 * messages in the output message queue,
1822 * so move to the "ending" state and
1823 * apply it immediately since
1824 * state_trim() is called within the
1827 put_messages(my_msgs
, my_count
);
1829 TRIMMER_ITERATOR_STATE_ENDING
;
1830 status
= state_ending(trimmer_it
, msgs
,
1838 * There's at least one message in the output message queue:
1839 * move the messages to the output message array.
1841 BT_ASSERT(!g_queue_is_empty(trimmer_it
->output_messages
));
1842 fill_message_array_from_output_messages(trimmer_it
, msgs
,
1850 bt_component_class_message_iterator_next_method_status
trimmer_msg_iter_next(
1851 bt_self_message_iterator
*self_msg_iter
,
1852 bt_message_array_const msgs
, uint64_t capacity
,
1855 struct trimmer_iterator
*trimmer_it
=
1856 bt_self_message_iterator_get_data(self_msg_iter
);
1857 bt_component_class_message_iterator_next_method_status status
=
1858 BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK
;
1860 BT_ASSERT(trimmer_it
);
1862 if (G_LIKELY(trimmer_it
->state
== TRIMMER_ITERATOR_STATE_TRIM
)) {
1863 status
= state_trim(trimmer_it
, msgs
, capacity
, count
);
1864 if (status
!= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK
) {
1868 switch (trimmer_it
->state
) {
1869 case TRIMMER_ITERATOR_STATE_SET_BOUNDS_NS_FROM_ORIGIN
:
1870 status
= state_set_trimmer_iterator_bounds(trimmer_it
);
1871 if (status
!= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK
) {
1875 status
= state_seek_initially(trimmer_it
);
1876 if (status
!= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK
) {
1880 status
= state_trim(trimmer_it
, msgs
, capacity
, count
);
1881 if (status
!= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK
) {
1886 case TRIMMER_ITERATOR_STATE_SEEK_INITIALLY
:
1887 status
= state_seek_initially(trimmer_it
);
1888 if (status
!= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK
) {
1892 status
= state_trim(trimmer_it
, msgs
, capacity
, count
);
1893 if (status
!= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK
) {
1898 case TRIMMER_ITERATOR_STATE_ENDING
:
1899 status
= state_ending(trimmer_it
, msgs
, capacity
,
1901 if (status
!= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK
) {
1906 case TRIMMER_ITERATOR_STATE_ENDED
:
1907 status
= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_END
;
1919 void trimmer_msg_iter_finalize(bt_self_message_iterator
*self_msg_iter
)
1921 struct trimmer_iterator
*trimmer_it
=
1922 bt_self_message_iterator_get_data(self_msg_iter
);
1924 BT_ASSERT(trimmer_it
);
1925 destroy_trimmer_iterator(trimmer_it
);