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_TAG "PLUGIN-UTILS-TRIMMER-FLT"
27 #include "compat/utc.h"
28 #include "compat/time.h"
29 #include <babeltrace2/babeltrace.h>
30 #include "common/common.h"
31 #include "plugins/plugins-common.h"
32 #include "common/assert.h"
39 #define NS_PER_S INT64_C(1000000000)
41 static const char * const in_port_name
= "in";
44 unsigned int hour
, minute
, second
, ns
;
47 struct trimmer_bound
{
49 * Nanoseconds from origin, valid if `is_set` is set and
50 * `is_infinite` is false.
52 int64_t ns_from_origin
;
54 /* True if this bound's full time (`ns_from_origin`) is set */
58 * True if this bound represents the infinity (negative or
59 * positive depending on which bound it is). If this is true,
60 * then we don't care about `ns_from_origin` above.
65 * This bound's time without the date; this time is used to set
66 * `ns_from_origin` once we know the date.
68 struct trimmer_time time
;
72 struct trimmer_bound begin
, end
;
76 enum trimmer_iterator_state
{
78 * Find the first message's date and set the bounds's times
81 TRIMMER_ITERATOR_STATE_SET_BOUNDS_NS_FROM_ORIGIN
,
84 * Initially seek to the trimming range's beginning time.
86 TRIMMER_ITERATOR_STATE_SEEK_INITIALLY
,
89 * Fill the output message queue for as long as received input
90 * messages are within the trimming time range.
92 TRIMMER_ITERATOR_STATE_TRIM
,
94 /* Flush the remaining messages in the output message queue */
95 TRIMMER_ITERATOR_STATE_ENDING
,
97 /* Trimming operation and message iterator is ended */
98 TRIMMER_ITERATOR_STATE_ENDED
,
101 struct trimmer_iterator
{
103 struct trimmer_comp
*trimmer_comp
;
106 bt_self_message_iterator
*self_msg_iter
;
108 enum trimmer_iterator_state state
;
111 bt_self_component_port_input_message_iterator
*upstream_iter
;
112 struct trimmer_bound begin
, end
;
115 * Queue of `const bt_message *` (owned by the queue).
117 * This is where the trimming operation pushes the messages to
118 * output by this message iterator.
120 GQueue
*output_messages
;
123 * Hash table of `bt_stream *` (weak) to
124 * `struct trimmer_iterator_stream_state *` (owned by the HT).
126 GHashTable
*stream_states
;
129 struct trimmer_iterator_stream_state
{
131 * True if both stream beginning and initial stream activity
132 * beginning messages were pushed for this stream.
137 * True if the last pushed message for this stream was a stream
138 * activity end message.
140 bool last_msg_is_stream_activity_end
;
143 * Time to use for a generated stream end activity message when
146 int64_t stream_act_end_ns_from_origin
;
149 const bt_stream
*stream
;
151 /* Owned by this (`NULL` initially and between packets) */
152 const bt_packet
*cur_packet
;
155 const bt_message
*stream_beginning_msg
;
159 void destroy_trimmer_comp(struct trimmer_comp
*trimmer_comp
)
161 BT_ASSERT(trimmer_comp
);
162 g_free(trimmer_comp
);
166 struct trimmer_comp
*create_trimmer_comp(void)
168 return g_new0(struct trimmer_comp
, 1);
172 void trimmer_finalize(bt_self_component_filter
*self_comp
)
174 struct trimmer_comp
*trimmer_comp
=
175 bt_self_component_get_data(
176 bt_self_component_filter_as_self_component(self_comp
));
179 destroy_trimmer_comp(trimmer_comp
);
184 * Sets the time (in ns from origin) of a trimmer bound from date and
187 * Returns a negative value if anything goes wrong.
190 int set_bound_ns_from_origin(struct trimmer_bound
*bound
,
191 unsigned int year
, unsigned int month
, unsigned int day
,
192 unsigned int hour
, unsigned int minute
, unsigned int second
,
193 unsigned int ns
, bool is_gmt
)
203 .tm_year
= year
- 1900,
208 result
= bt_timegm(&tm
);
210 result
= mktime(&tm
);
219 bound
->ns_from_origin
= (int64_t) result
;
220 bound
->ns_from_origin
*= NS_PER_S
;
221 bound
->ns_from_origin
+= ns
;
222 bound
->is_set
= true;
229 * Parses a timestamp, figuring out its format.
231 * Returns a negative value if anything goes wrong.
235 * YYYY-MM-DD hh:mm[:ss[.ns]]
239 * TODO: Check overflows.
242 int set_bound_from_str(const char *str
, struct trimmer_bound
*bound
,
247 unsigned int year
, month
, day
, hour
, minute
, second
, ns
;
250 /* Try `YYYY-MM-DD hh:mm:ss.ns` format */
251 s_ret
= sscanf(str
, "%u-%u-%u %u:%u:%u.%u%c", &year
, &month
, &day
,
252 &hour
, &minute
, &second
, &ns
, &dummy
);
254 ret
= set_bound_ns_from_origin(bound
, year
, month
, day
,
255 hour
, minute
, second
, ns
, is_gmt
);
259 /* Try `YYYY-MM-DD hh:mm:ss` format */
260 s_ret
= sscanf(str
, "%u-%u-%u %u:%u:%u%c", &year
, &month
, &day
,
261 &hour
, &minute
, &second
, &dummy
);
263 ret
= set_bound_ns_from_origin(bound
, year
, month
, day
,
264 hour
, minute
, second
, 0, is_gmt
);
268 /* Try `YYYY-MM-DD hh:mm` format */
269 s_ret
= sscanf(str
, "%u-%u-%u %u:%u%c", &year
, &month
, &day
,
270 &hour
, &minute
, &dummy
);
272 ret
= set_bound_ns_from_origin(bound
, year
, month
, day
,
273 hour
, minute
, 0, 0, is_gmt
);
277 /* Try `YYYY-MM-DD` format */
278 s_ret
= sscanf(str
, "%u-%u-%u%c", &year
, &month
, &day
, &dummy
);
280 ret
= set_bound_ns_from_origin(bound
, year
, month
, day
,
285 /* Try `hh:mm:ss.ns` format */
286 s_ret
= sscanf(str
, "%u:%u:%u.%u%c", &hour
, &minute
, &second
, &ns
,
289 bound
->time
.hour
= hour
;
290 bound
->time
.minute
= minute
;
291 bound
->time
.second
= second
;
296 /* Try `hh:mm:ss` format */
297 s_ret
= sscanf(str
, "%u:%u:%u%c", &hour
, &minute
, &second
, &dummy
);
299 bound
->time
.hour
= hour
;
300 bound
->time
.minute
= minute
;
301 bound
->time
.second
= second
;
306 /* Try `-s.ns` format */
307 s_ret
= sscanf(str
, "-%u.%u%c", &second
, &ns
, &dummy
);
309 bound
->ns_from_origin
= -((int64_t) second
) * NS_PER_S
;
310 bound
->ns_from_origin
-= (int64_t) ns
;
311 bound
->is_set
= true;
315 /* Try `s.ns` format */
316 s_ret
= sscanf(str
, "%u.%u%c", &second
, &ns
, &dummy
);
318 bound
->ns_from_origin
= ((int64_t) second
) * NS_PER_S
;
319 bound
->ns_from_origin
+= (int64_t) ns
;
320 bound
->is_set
= true;
324 /* Try `-s` format */
325 s_ret
= sscanf(str
, "-%u%c", &second
, &dummy
);
327 bound
->ns_from_origin
= -((int64_t) second
) * NS_PER_S
;
328 bound
->is_set
= true;
333 s_ret
= sscanf(str
, "%u%c", &second
, &dummy
);
335 bound
->ns_from_origin
= (int64_t) second
* NS_PER_S
;
336 bound
->is_set
= true;
340 BT_LOGE("Invalid date/time format: param=\"%s\"", str
);
348 * Sets a trimmer bound's properties from a parameter string/integer
351 * Returns a negative value if anything goes wrong.
354 int set_bound_from_param(const char *param_name
, const bt_value
*param
,
355 struct trimmer_bound
*bound
, bool is_gmt
)
361 if (bt_value_is_signed_integer(param
)) {
362 int64_t value
= bt_value_signed_integer_get(param
);
365 * Just convert it to a temporary string to handle
366 * everything the same way.
368 sprintf(tmp_arg
, "%" PRId64
, value
);
370 } else if (bt_value_is_string(param
)) {
371 arg
= bt_value_string_get(param
);
373 BT_LOGE("`%s` parameter must be an integer or a string value.",
379 ret
= set_bound_from_str(arg
, bound
, is_gmt
);
386 int validate_trimmer_bounds(struct trimmer_bound
*begin
,
387 struct trimmer_bound
*end
)
391 BT_ASSERT(begin
->is_set
);
392 BT_ASSERT(end
->is_set
);
394 if (!begin
->is_infinite
&& !end
->is_infinite
&&
395 begin
->ns_from_origin
> end
->ns_from_origin
) {
396 BT_LOGE("Trimming time range's beginning time is greater than end time: "
397 "begin-ns-from-origin=%" PRId64
", "
398 "end-ns-from-origin=%" PRId64
,
399 begin
->ns_from_origin
,
400 end
->ns_from_origin
);
405 if (!begin
->is_infinite
&& begin
->ns_from_origin
== INT64_MIN
) {
406 BT_LOGE("Invalid trimming time range's beginning time: "
407 "ns-from-origin=%" PRId64
,
408 begin
->ns_from_origin
);
413 if (!end
->is_infinite
&& end
->ns_from_origin
== INT64_MIN
) {
414 BT_LOGE("Invalid trimming time range's end time: "
415 "ns-from-origin=%" PRId64
,
416 end
->ns_from_origin
);
426 int init_trimmer_comp_from_params(struct trimmer_comp
*trimmer_comp
,
427 const bt_value
*params
)
429 const bt_value
*value
;
433 value
= bt_value_map_borrow_entry_value_const(params
, "gmt");
435 trimmer_comp
->is_gmt
= (bool) bt_value_bool_get(value
);
438 value
= bt_value_map_borrow_entry_value_const(params
, "begin");
440 if (set_bound_from_param("begin", value
,
441 &trimmer_comp
->begin
, trimmer_comp
->is_gmt
)) {
442 /* set_bound_from_param() logs errors */
443 ret
= BT_SELF_COMPONENT_STATUS_ERROR
;
447 trimmer_comp
->begin
.is_infinite
= true;
448 trimmer_comp
->begin
.is_set
= true;
451 value
= bt_value_map_borrow_entry_value_const(params
, "end");
453 if (set_bound_from_param("end", value
,
454 &trimmer_comp
->end
, trimmer_comp
->is_gmt
)) {
455 /* set_bound_from_param() logs errors */
456 ret
= BT_SELF_COMPONENT_STATUS_ERROR
;
460 trimmer_comp
->end
.is_infinite
= true;
461 trimmer_comp
->end
.is_set
= true;
465 if (trimmer_comp
->begin
.is_set
&& trimmer_comp
->end
.is_set
) {
466 /* validate_trimmer_bounds() logs errors */
467 ret
= validate_trimmer_bounds(&trimmer_comp
->begin
,
474 bt_self_component_status
trimmer_init(bt_self_component_filter
*self_comp
,
475 const bt_value
*params
, void *init_data
)
478 bt_self_component_status status
;
479 struct trimmer_comp
*trimmer_comp
= create_trimmer_comp();
482 status
= BT_SELF_COMPONENT_STATUS_NOMEM
;
486 status
= bt_self_component_filter_add_input_port(
487 self_comp
, in_port_name
, NULL
, NULL
);
488 if (status
!= BT_SELF_COMPONENT_STATUS_OK
) {
492 status
= bt_self_component_filter_add_output_port(
493 self_comp
, "out", NULL
, NULL
);
494 if (status
!= BT_SELF_COMPONENT_STATUS_OK
) {
498 ret
= init_trimmer_comp_from_params(trimmer_comp
, params
);
500 status
= BT_SELF_COMPONENT_STATUS_ERROR
;
504 bt_self_component_set_data(
505 bt_self_component_filter_as_self_component(self_comp
),
510 if (status
== BT_SELF_COMPONENT_STATUS_OK
) {
511 status
= BT_SELF_COMPONENT_STATUS_ERROR
;
515 destroy_trimmer_comp(trimmer_comp
);
523 void destroy_trimmer_iterator(struct trimmer_iterator
*trimmer_it
)
525 BT_ASSERT(trimmer_it
);
526 bt_self_component_port_input_message_iterator_put_ref(
527 trimmer_it
->upstream_iter
);
529 if (trimmer_it
->output_messages
) {
530 g_queue_free(trimmer_it
->output_messages
);
533 if (trimmer_it
->stream_states
) {
534 g_hash_table_destroy(trimmer_it
->stream_states
);
541 void destroy_trimmer_iterator_stream_state(
542 struct trimmer_iterator_stream_state
*sstate
)
545 BT_PACKET_PUT_REF_AND_RESET(sstate
->cur_packet
);
546 BT_MESSAGE_PUT_REF_AND_RESET(sstate
->stream_beginning_msg
);
551 bt_self_message_iterator_status
trimmer_msg_iter_init(
552 bt_self_message_iterator
*self_msg_iter
,
553 bt_self_component_filter
*self_comp
,
554 bt_self_component_port_output
*port
)
556 bt_self_message_iterator_status status
=
557 BT_SELF_MESSAGE_ITERATOR_STATUS_OK
;
558 struct trimmer_iterator
*trimmer_it
;
560 trimmer_it
= g_new0(struct trimmer_iterator
, 1);
562 status
= BT_SELF_MESSAGE_ITERATOR_STATUS_NOMEM
;
566 trimmer_it
->trimmer_comp
= bt_self_component_get_data(
567 bt_self_component_filter_as_self_component(self_comp
));
568 BT_ASSERT(trimmer_it
->trimmer_comp
);
570 if (trimmer_it
->trimmer_comp
->begin
.is_set
&&
571 trimmer_it
->trimmer_comp
->end
.is_set
) {
573 * Both trimming time range's bounds are set, so skip
575 * `TRIMMER_ITERATOR_STATE_SET_BOUNDS_NS_FROM_ORIGIN`
578 trimmer_it
->state
= TRIMMER_ITERATOR_STATE_SEEK_INITIALLY
;
581 trimmer_it
->begin
= trimmer_it
->trimmer_comp
->begin
;
582 trimmer_it
->end
= trimmer_it
->trimmer_comp
->end
;
583 trimmer_it
->upstream_iter
=
584 bt_self_component_port_input_message_iterator_create(
585 bt_self_component_filter_borrow_input_port_by_name(
586 self_comp
, in_port_name
));
587 if (!trimmer_it
->upstream_iter
) {
588 status
= BT_SELF_MESSAGE_ITERATOR_STATUS_ERROR
;
592 trimmer_it
->output_messages
= g_queue_new();
593 if (!trimmer_it
->output_messages
) {
594 status
= BT_SELF_MESSAGE_ITERATOR_STATUS_NOMEM
;
598 trimmer_it
->stream_states
= g_hash_table_new_full(g_direct_hash
,
599 g_direct_equal
, NULL
,
600 (GDestroyNotify
) destroy_trimmer_iterator_stream_state
);
601 if (!trimmer_it
->stream_states
) {
602 status
= BT_SELF_MESSAGE_ITERATOR_STATUS_NOMEM
;
606 trimmer_it
->self_msg_iter
= self_msg_iter
;
607 bt_self_message_iterator_set_data(self_msg_iter
, trimmer_it
);
610 if (status
!= BT_SELF_MESSAGE_ITERATOR_STATUS_OK
&& trimmer_it
) {
611 destroy_trimmer_iterator(trimmer_it
);
618 int get_msg_ns_from_origin(const bt_message
*msg
, int64_t *ns_from_origin
,
621 const bt_clock_class
*clock_class
= NULL
;
622 const bt_clock_snapshot
*clock_snapshot
= NULL
;
623 bt_message_stream_activity_clock_snapshot_state sa_cs_state
;
627 BT_ASSERT(ns_from_origin
);
630 switch (bt_message_get_type(msg
)) {
631 case BT_MESSAGE_TYPE_EVENT
:
633 bt_message_event_borrow_stream_class_default_clock_class_const(
635 if (G_UNLIKELY(!clock_class
)) {
639 clock_snapshot
= bt_message_event_borrow_default_clock_snapshot_const(
642 case BT_MESSAGE_TYPE_PACKET_BEGINNING
:
644 bt_message_packet_beginning_borrow_stream_class_default_clock_class_const(
646 if (G_UNLIKELY(!clock_class
)) {
650 clock_snapshot
= bt_message_packet_beginning_borrow_default_clock_snapshot_const(
653 case BT_MESSAGE_TYPE_PACKET_END
:
655 bt_message_packet_end_borrow_stream_class_default_clock_class_const(
657 if (G_UNLIKELY(!clock_class
)) {
661 clock_snapshot
= bt_message_packet_end_borrow_default_clock_snapshot_const(
664 case BT_MESSAGE_TYPE_DISCARDED_EVENTS
:
666 bt_message_discarded_events_borrow_stream_class_default_clock_class_const(
668 if (G_UNLIKELY(!clock_class
)) {
672 clock_snapshot
= bt_message_discarded_events_borrow_beginning_default_clock_snapshot_const(
675 case BT_MESSAGE_TYPE_DISCARDED_PACKETS
:
677 bt_message_discarded_packets_borrow_stream_class_default_clock_class_const(
679 if (G_UNLIKELY(!clock_class
)) {
683 clock_snapshot
= bt_message_discarded_packets_borrow_beginning_default_clock_snapshot_const(
686 case BT_MESSAGE_TYPE_STREAM_ACTIVITY_BEGINNING
:
688 bt_message_stream_activity_beginning_borrow_stream_class_default_clock_class_const(
690 if (G_UNLIKELY(!clock_class
)) {
694 sa_cs_state
= bt_message_stream_activity_beginning_borrow_default_clock_snapshot_const(
695 msg
, &clock_snapshot
);
696 if (sa_cs_state
== BT_MESSAGE_STREAM_ACTIVITY_CLOCK_SNAPSHOT_STATE_UNKNOWN
||
697 sa_cs_state
== BT_MESSAGE_STREAM_ACTIVITY_CLOCK_SNAPSHOT_STATE_INFINITE
) {
698 /* Lowest possible time to always include them */
699 *ns_from_origin
= INT64_MIN
;
700 goto no_clock_snapshot
;
704 case BT_MESSAGE_TYPE_STREAM_ACTIVITY_END
:
706 bt_message_stream_activity_end_borrow_stream_class_default_clock_class_const(
708 if (G_UNLIKELY(!clock_class
)) {
712 sa_cs_state
= bt_message_stream_activity_end_borrow_default_clock_snapshot_const(
713 msg
, &clock_snapshot
);
714 if (sa_cs_state
== BT_MESSAGE_STREAM_ACTIVITY_CLOCK_SNAPSHOT_STATE_UNKNOWN
) {
715 /* Lowest time to always include it */
716 *ns_from_origin
= INT64_MIN
;
717 goto no_clock_snapshot
;
718 } else if (sa_cs_state
== BT_MESSAGE_STREAM_ACTIVITY_CLOCK_SNAPSHOT_STATE_INFINITE
) {
719 /* Greatest time to always exclude it */
720 *ns_from_origin
= INT64_MAX
;
721 goto no_clock_snapshot
;
725 case BT_MESSAGE_TYPE_MESSAGE_ITERATOR_INACTIVITY
:
727 bt_message_message_iterator_inactivity_borrow_default_clock_snapshot_const(
731 goto no_clock_snapshot
;
734 ret
= bt_clock_snapshot_get_ns_from_origin(clock_snapshot
,
736 if (G_UNLIKELY(ret
)) {
754 void put_messages(bt_message_array_const msgs
, uint64_t count
)
758 for (i
= 0; i
< count
; i
++) {
759 BT_MESSAGE_PUT_REF_AND_RESET(msgs
[i
]);
764 int set_trimmer_iterator_bound(struct trimmer_bound
*bound
,
765 int64_t ns_from_origin
, bool is_gmt
)
768 time_t time_seconds
= (time_t) (ns_from_origin
/ NS_PER_S
);
771 BT_ASSERT(!bound
->is_set
);
774 /* We only need to extract the date from this time */
776 bt_gmtime_r(&time_seconds
, &tm
);
778 bt_localtime_r(&time_seconds
, &tm
);
782 BT_LOGE_ERRNO("Cannot convert timestamp to date and time",
783 "ts=%" PRId64
, (int64_t) time_seconds
);
788 ret
= set_bound_ns_from_origin(bound
, tm
.tm_year
+ 1900, tm
.tm_mon
+ 1,
789 tm
.tm_mday
, bound
->time
.hour
, bound
->time
.minute
,
790 bound
->time
.second
, bound
->time
.ns
, is_gmt
);
797 bt_self_message_iterator_status
state_set_trimmer_iterator_bounds(
798 struct trimmer_iterator
*trimmer_it
)
800 bt_message_iterator_status upstream_iter_status
=
801 BT_MESSAGE_ITERATOR_STATUS_OK
;
802 struct trimmer_comp
*trimmer_comp
= trimmer_it
->trimmer_comp
;
803 bt_message_array_const msgs
;
805 int64_t ns_from_origin
= INT64_MIN
;
809 BT_ASSERT(!trimmer_it
->begin
.is_set
||
810 !trimmer_it
->end
.is_set
);
813 upstream_iter_status
=
814 bt_self_component_port_input_message_iterator_next(
815 trimmer_it
->upstream_iter
, &msgs
, &count
);
816 if (upstream_iter_status
!= BT_MESSAGE_ITERATOR_STATUS_OK
) {
820 for (i
= 0; i
< count
; i
++) {
821 const bt_message
*msg
= msgs
[i
];
825 ret
= get_msg_ns_from_origin(msg
, &ns_from_origin
,
835 BT_ASSERT(ns_from_origin
!= INT64_MIN
&&
836 ns_from_origin
!= INT64_MAX
);
837 put_messages(msgs
, count
);
841 put_messages(msgs
, count
);
845 if (!trimmer_it
->begin
.is_set
) {
846 BT_ASSERT(!trimmer_it
->begin
.is_infinite
);
847 ret
= set_trimmer_iterator_bound(&trimmer_it
->begin
,
848 ns_from_origin
, trimmer_comp
->is_gmt
);
854 if (!trimmer_it
->end
.is_set
) {
855 BT_ASSERT(!trimmer_it
->end
.is_infinite
);
856 ret
= set_trimmer_iterator_bound(&trimmer_it
->end
,
857 ns_from_origin
, trimmer_comp
->is_gmt
);
863 ret
= validate_trimmer_bounds(&trimmer_it
->begin
,
872 put_messages(msgs
, count
);
873 upstream_iter_status
= BT_MESSAGE_ITERATOR_STATUS_ERROR
;
876 return (int) upstream_iter_status
;
880 bt_self_message_iterator_status
state_seek_initially(
881 struct trimmer_iterator
*trimmer_it
)
883 bt_self_message_iterator_status status
=
884 BT_SELF_MESSAGE_ITERATOR_STATUS_OK
;
886 BT_ASSERT(trimmer_it
->begin
.is_set
);
888 if (trimmer_it
->begin
.is_infinite
) {
889 if (!bt_self_component_port_input_message_iterator_can_seek_beginning(
890 trimmer_it
->upstream_iter
)) {
891 BT_LOGE_STR("Cannot make upstream message iterator initially seek its beginning.");
892 status
= BT_SELF_MESSAGE_ITERATOR_STATUS_ERROR
;
896 status
= (int) bt_self_component_port_input_message_iterator_seek_beginning(
897 trimmer_it
->upstream_iter
);
899 if (!bt_self_component_port_input_message_iterator_can_seek_ns_from_origin(
900 trimmer_it
->upstream_iter
,
901 trimmer_it
->begin
.ns_from_origin
)) {
902 BT_LOGE("Cannot make upstream message iterator initially seek: "
903 "seek-ns-from-origin=%" PRId64
,
904 trimmer_it
->begin
.ns_from_origin
);
905 status
= BT_SELF_MESSAGE_ITERATOR_STATUS_ERROR
;
909 status
= (int) bt_self_component_port_input_message_iterator_seek_ns_from_origin(
910 trimmer_it
->upstream_iter
, trimmer_it
->begin
.ns_from_origin
);
913 if (status
== BT_SELF_MESSAGE_ITERATOR_STATUS_OK
) {
914 trimmer_it
->state
= TRIMMER_ITERATOR_STATE_TRIM
;
922 void push_message(struct trimmer_iterator
*trimmer_it
, const bt_message
*msg
)
924 g_queue_push_head(trimmer_it
->output_messages
, (void *) msg
);
928 const bt_message
*pop_message(struct trimmer_iterator
*trimmer_it
)
930 return g_queue_pop_tail(trimmer_it
->output_messages
);
934 int clock_raw_value_from_ns_from_origin(const bt_clock_class
*clock_class
,
935 int64_t ns_from_origin
, uint64_t *raw_value
)
939 uint64_t cc_offset_cycles
;
942 bt_clock_class_get_offset(clock_class
, &cc_offset_s
, &cc_offset_cycles
);
943 cc_freq
= bt_clock_class_get_frequency(clock_class
);
944 return bt_common_clock_value_from_ns_from_origin(cc_offset_s
,
945 cc_offset_cycles
, cc_freq
, ns_from_origin
, raw_value
);
949 bt_self_message_iterator_status
end_stream(struct trimmer_iterator
*trimmer_it
,
950 struct trimmer_iterator_stream_state
*sstate
)
952 bt_self_message_iterator_status status
=
953 BT_SELF_MESSAGE_ITERATOR_STATUS_OK
;
955 const bt_clock_class
*clock_class
;
957 bt_message
*msg
= NULL
;
959 BT_ASSERT(!trimmer_it
->end
.is_infinite
);
961 if (!sstate
->stream
) {
965 if (sstate
->cur_packet
) {
967 * The last message could not have been a stream
968 * activity end message if we have a current packet.
970 BT_ASSERT(!sstate
->last_msg_is_stream_activity_end
);
973 * Create and push a packet end message, making its time
974 * the trimming range's end time.
976 clock_class
= bt_stream_class_borrow_default_clock_class_const(
977 bt_stream_borrow_class_const(sstate
->stream
));
978 BT_ASSERT(clock_class
);
979 ret
= clock_raw_value_from_ns_from_origin(clock_class
,
980 trimmer_it
->end
.ns_from_origin
, &raw_value
);
982 status
= BT_SELF_MESSAGE_ITERATOR_STATUS_ERROR
;
986 msg
= bt_message_packet_end_create_with_default_clock_snapshot(
987 trimmer_it
->self_msg_iter
, sstate
->cur_packet
,
990 status
= BT_SELF_MESSAGE_ITERATOR_STATUS_NOMEM
;
994 push_message(trimmer_it
, msg
);
996 BT_PACKET_PUT_REF_AND_RESET(sstate
->cur_packet
);
999 * Because we generated a packet end message, set the
1000 * stream activity end message's time to use to the
1001 * trimming range's end time (this packet end message's
1004 sstate
->stream_act_end_ns_from_origin
=
1005 trimmer_it
->end
.ns_from_origin
;
1008 if (!sstate
->last_msg_is_stream_activity_end
) {
1009 /* Create and push a stream activity end message */
1010 msg
= bt_message_stream_activity_end_create(
1011 trimmer_it
->self_msg_iter
, sstate
->stream
);
1013 status
= BT_SELF_MESSAGE_ITERATOR_STATUS_NOMEM
;
1017 clock_class
= bt_stream_class_borrow_default_clock_class_const(
1018 bt_stream_borrow_class_const(sstate
->stream
));
1019 BT_ASSERT(clock_class
);
1021 if (sstate
->stream_act_end_ns_from_origin
== INT64_MIN
) {
1023 * We received at least what is necessary to
1024 * have a stream state (stream beginning and
1025 * stream activity beginning messages), but
1026 * nothing else: use the trimmer range's end
1029 sstate
->stream_act_end_ns_from_origin
=
1030 trimmer_it
->end
.ns_from_origin
;
1033 ret
= clock_raw_value_from_ns_from_origin(clock_class
,
1034 sstate
->stream_act_end_ns_from_origin
, &raw_value
);
1036 status
= BT_SELF_MESSAGE_ITERATOR_STATUS_ERROR
;
1040 bt_message_stream_activity_end_set_default_clock_snapshot(
1042 push_message(trimmer_it
, msg
);
1046 /* Create and push a stream end message */
1047 msg
= bt_message_stream_end_create(trimmer_it
->self_msg_iter
,
1050 status
= BT_SELF_MESSAGE_ITERATOR_STATUS_NOMEM
;
1054 push_message(trimmer_it
, msg
);
1058 * Just to make sure that we don't use this stream state again
1059 * in the future without an obvious error.
1061 sstate
->stream
= NULL
;
1064 bt_message_put_ref(msg
);
1069 bt_self_message_iterator_status
end_iterator_streams(
1070 struct trimmer_iterator
*trimmer_it
)
1072 bt_self_message_iterator_status status
=
1073 BT_SELF_MESSAGE_ITERATOR_STATUS_OK
;
1074 GHashTableIter iter
;
1075 gpointer key
, sstate
;
1077 if (trimmer_it
->end
.is_infinite
) {
1079 * An infinite trimming range's end time guarantees that
1080 * we received (and pushed) all the appropriate end
1087 * End each stream and then remove them from the hash table of
1088 * stream states to release unneeded references.
1090 g_hash_table_iter_init(&iter
, trimmer_it
->stream_states
);
1092 while (g_hash_table_iter_next(&iter
, &key
, &sstate
)) {
1093 status
= end_stream(trimmer_it
, sstate
);
1100 g_hash_table_remove_all(trimmer_it
->stream_states
);
1107 bt_self_message_iterator_status
create_stream_beginning_activity_message(
1108 struct trimmer_iterator
*trimmer_it
,
1109 const bt_stream
*stream
,
1110 const bt_clock_class
*clock_class
, bt_message
**msg
)
1112 bt_message
*local_msg
;
1113 bt_self_message_iterator_status status
=
1114 BT_SELF_MESSAGE_ITERATOR_STATUS_OK
;
1117 BT_ASSERT(!trimmer_it
->begin
.is_infinite
);
1119 local_msg
= bt_message_stream_activity_beginning_create(
1120 trimmer_it
->self_msg_iter
, stream
);
1122 status
= BT_SELF_MESSAGE_ITERATOR_STATUS_NOMEM
;
1130 ret
= clock_raw_value_from_ns_from_origin(clock_class
,
1131 trimmer_it
->begin
.ns_from_origin
, &raw_value
);
1133 status
= BT_SELF_MESSAGE_ITERATOR_STATUS_ERROR
;
1134 bt_message_put_ref(local_msg
);
1138 bt_message_stream_activity_beginning_set_default_clock_snapshot(
1139 local_msg
, raw_value
);
1142 BT_MESSAGE_MOVE_REF(*msg
, local_msg
);
1149 * Makes sure to initialize a stream state, pushing the appropriate
1152 * `stream_act_beginning_msg` is an initial stream activity beginning
1153 * message to potentially use, depending on its clock snapshot state.
1154 * This function consumes `stream_act_beginning_msg` unconditionally.
1157 bt_self_message_iterator_status
ensure_stream_state_is_inited(
1158 struct trimmer_iterator
*trimmer_it
,
1159 struct trimmer_iterator_stream_state
*sstate
,
1160 const bt_message
*stream_act_beginning_msg
)
1162 bt_self_message_iterator_status status
=
1163 BT_SELF_MESSAGE_ITERATOR_STATUS_OK
;
1164 bt_message
*new_msg
= NULL
;
1165 const bt_clock_class
*clock_class
=
1166 bt_stream_class_borrow_default_clock_class_const(
1167 bt_stream_borrow_class_const(sstate
->stream
));
1169 BT_ASSERT(!sstate
->inited
);
1171 if (!sstate
->stream_beginning_msg
) {
1172 /* No initial stream beginning message: create one */
1173 sstate
->stream_beginning_msg
=
1174 bt_message_stream_beginning_create(
1175 trimmer_it
->self_msg_iter
, sstate
->stream
);
1176 if (!sstate
->stream_beginning_msg
) {
1177 status
= BT_SELF_MESSAGE_ITERATOR_STATUS_NOMEM
;
1182 /* Push initial stream beginning message */
1183 BT_ASSERT(sstate
->stream_beginning_msg
);
1184 push_message(trimmer_it
, sstate
->stream_beginning_msg
);
1185 sstate
->stream_beginning_msg
= NULL
;
1187 if (stream_act_beginning_msg
) {
1189 * Initial stream activity beginning message exists: if
1190 * its time is -inf, then create and push a new one
1191 * having the trimming range's beginning time. Otherwise
1192 * push it as is (known and unknown).
1194 const bt_clock_snapshot
*cs
;
1195 bt_message_stream_activity_clock_snapshot_state sa_cs_state
;
1197 sa_cs_state
= bt_message_stream_activity_beginning_borrow_default_clock_snapshot_const(
1198 stream_act_beginning_msg
, &cs
);
1199 if (sa_cs_state
== BT_MESSAGE_STREAM_ACTIVITY_CLOCK_SNAPSHOT_STATE_INFINITE
&&
1200 !trimmer_it
->begin
.is_infinite
) {
1202 * -inf time: use trimming range's beginning
1203 * time (which is not -inf).
1205 status
= create_stream_beginning_activity_message(
1206 trimmer_it
, sstate
->stream
, clock_class
,
1208 if (status
!= BT_SELF_MESSAGE_ITERATOR_STATUS_OK
) {
1212 push_message(trimmer_it
, new_msg
);
1215 /* Known/unknown: push as is */
1216 push_message(trimmer_it
, stream_act_beginning_msg
);
1217 stream_act_beginning_msg
= NULL
;
1220 BT_ASSERT(!trimmer_it
->begin
.is_infinite
);
1223 * No stream beginning activity message: create and push
1226 status
= create_stream_beginning_activity_message(
1227 trimmer_it
, sstate
->stream
, clock_class
, &new_msg
);
1228 if (status
!= BT_SELF_MESSAGE_ITERATOR_STATUS_OK
) {
1232 push_message(trimmer_it
, new_msg
);
1236 sstate
->inited
= true;
1239 bt_message_put_ref(new_msg
);
1240 bt_message_put_ref(stream_act_beginning_msg
);
1245 bt_self_message_iterator_status
ensure_cur_packet_exists(
1246 struct trimmer_iterator
*trimmer_it
,
1247 struct trimmer_iterator_stream_state
*sstate
,
1248 const bt_packet
*packet
)
1250 bt_self_message_iterator_status status
=
1251 BT_SELF_MESSAGE_ITERATOR_STATUS_OK
;
1253 const bt_clock_class
*clock_class
=
1254 bt_stream_class_borrow_default_clock_class_const(
1255 bt_stream_borrow_class_const(sstate
->stream
));
1256 bt_message
*msg
= NULL
;
1259 BT_ASSERT(!trimmer_it
->begin
.is_infinite
);
1260 BT_ASSERT(!sstate
->cur_packet
);
1263 * Create and push an initial packet beginning message,
1264 * making its time the trimming range's beginning time.
1266 ret
= clock_raw_value_from_ns_from_origin(clock_class
,
1267 trimmer_it
->begin
.ns_from_origin
, &raw_value
);
1269 status
= BT_SELF_MESSAGE_ITERATOR_STATUS_ERROR
;
1273 msg
= bt_message_packet_beginning_create_with_default_clock_snapshot(
1274 trimmer_it
->self_msg_iter
, packet
, raw_value
);
1276 status
= BT_SELF_MESSAGE_ITERATOR_STATUS_NOMEM
;
1280 push_message(trimmer_it
, msg
);
1283 /* Set packet as this stream's current packet */
1284 sstate
->cur_packet
= packet
;
1285 bt_packet_get_ref(sstate
->cur_packet
);
1288 bt_message_put_ref(msg
);
1293 * Handles a message which is associated to a given stream state. This
1294 * _could_ make the iterator's output message queue grow; this could
1295 * also consume the message without pushing anything to this queue, only
1296 * modifying the stream state.
1298 * This function consumes the `msg` reference, _whatever the outcome_.
1300 * `ns_from_origin` is the message's time, as given by
1301 * get_msg_ns_from_origin().
1303 * This function sets `reached_end` if handling this message made the
1304 * iterator reach the end of the trimming range. Note that the output
1305 * message queue could contain messages even if this function sets
1309 bt_self_message_iterator_status
handle_message_with_stream_state(
1310 struct trimmer_iterator
*trimmer_it
, const bt_message
*msg
,
1311 struct trimmer_iterator_stream_state
*sstate
,
1312 int64_t ns_from_origin
, bool *reached_end
)
1314 bt_self_message_iterator_status status
=
1315 BT_SELF_MESSAGE_ITERATOR_STATUS_OK
;
1316 bt_message_type msg_type
= bt_message_get_type(msg
);
1320 case BT_MESSAGE_TYPE_EVENT
:
1321 if (G_UNLIKELY(!trimmer_it
->end
.is_infinite
&&
1322 ns_from_origin
> trimmer_it
->end
.ns_from_origin
)) {
1323 status
= end_iterator_streams(trimmer_it
);
1324 *reached_end
= true;
1328 if (G_UNLIKELY(!sstate
->inited
)) {
1329 status
= ensure_stream_state_is_inited(trimmer_it
,
1331 if (status
!= BT_SELF_MESSAGE_ITERATOR_STATUS_OK
) {
1336 if (G_UNLIKELY(!sstate
->cur_packet
)) {
1337 const bt_event
*event
=
1338 bt_message_event_borrow_event_const(msg
);
1339 const bt_packet
*packet
= bt_event_borrow_packet_const(
1342 status
= ensure_cur_packet_exists(trimmer_it
, sstate
,
1344 if (status
!= BT_SELF_MESSAGE_ITERATOR_STATUS_OK
) {
1349 BT_ASSERT(sstate
->cur_packet
);
1350 push_message(trimmer_it
, msg
);
1353 case BT_MESSAGE_TYPE_PACKET_BEGINNING
:
1354 if (G_UNLIKELY(!trimmer_it
->end
.is_infinite
&&
1355 ns_from_origin
> trimmer_it
->end
.ns_from_origin
)) {
1356 status
= end_iterator_streams(trimmer_it
);
1357 *reached_end
= true;
1361 if (G_UNLIKELY(!sstate
->inited
)) {
1362 status
= ensure_stream_state_is_inited(trimmer_it
,
1364 if (status
!= BT_SELF_MESSAGE_ITERATOR_STATUS_OK
) {
1369 BT_ASSERT(!sstate
->cur_packet
);
1370 sstate
->cur_packet
=
1371 bt_message_packet_beginning_borrow_packet_const(msg
);
1372 bt_packet_get_ref(sstate
->cur_packet
);
1373 push_message(trimmer_it
, msg
);
1376 case BT_MESSAGE_TYPE_PACKET_END
:
1377 sstate
->stream_act_end_ns_from_origin
= ns_from_origin
;
1379 if (G_UNLIKELY(!trimmer_it
->end
.is_infinite
&&
1380 ns_from_origin
> trimmer_it
->end
.ns_from_origin
)) {
1381 status
= end_iterator_streams(trimmer_it
);
1382 *reached_end
= true;
1386 if (G_UNLIKELY(!sstate
->inited
)) {
1387 status
= ensure_stream_state_is_inited(trimmer_it
,
1389 if (status
!= BT_SELF_MESSAGE_ITERATOR_STATUS_OK
) {
1394 if (G_UNLIKELY(!sstate
->cur_packet
)) {
1395 const bt_packet
*packet
=
1396 bt_message_packet_end_borrow_packet_const(msg
);
1398 status
= ensure_cur_packet_exists(trimmer_it
, sstate
,
1400 if (status
!= BT_SELF_MESSAGE_ITERATOR_STATUS_OK
) {
1405 BT_ASSERT(sstate
->cur_packet
);
1406 BT_PACKET_PUT_REF_AND_RESET(sstate
->cur_packet
);
1407 push_message(trimmer_it
, msg
);
1410 case BT_MESSAGE_TYPE_DISCARDED_EVENTS
:
1411 case BT_MESSAGE_TYPE_DISCARDED_PACKETS
:
1414 * `ns_from_origin` is the message's time range's
1415 * beginning time here.
1417 int64_t end_ns_from_origin
;
1418 const bt_clock_snapshot
*end_cs
;
1420 if (bt_message_get_type(msg
) ==
1421 BT_MESSAGE_TYPE_DISCARDED_EVENTS
) {
1423 * Safe to ignore the return value because we
1424 * know there's a default clock and it's always
1427 end_cs
= bt_message_discarded_events_borrow_end_default_clock_snapshot_const(
1431 * Safe to ignore the return value because we
1432 * know there's a default clock and it's always
1435 end_cs
= bt_message_discarded_packets_borrow_end_default_clock_snapshot_const(
1439 if (bt_clock_snapshot_get_ns_from_origin(end_cs
,
1440 &end_ns_from_origin
)) {
1441 status
= BT_SELF_MESSAGE_ITERATOR_STATUS_ERROR
;
1445 sstate
->stream_act_end_ns_from_origin
= end_ns_from_origin
;
1447 if (!trimmer_it
->end
.is_infinite
&&
1448 ns_from_origin
> trimmer_it
->end
.ns_from_origin
) {
1449 status
= end_iterator_streams(trimmer_it
);
1450 *reached_end
= true;
1454 if (!trimmer_it
->end
.is_infinite
&&
1455 end_ns_from_origin
> trimmer_it
->end
.ns_from_origin
) {
1457 * This message's end time is outside the
1458 * trimming time range: replace it with a new
1459 * message having an end time equal to the
1460 * trimming time range's end and without a
1463 const bt_clock_class
*clock_class
=
1464 bt_clock_snapshot_borrow_clock_class_const(
1466 const bt_clock_snapshot
*begin_cs
;
1467 bt_message
*new_msg
;
1468 uint64_t end_raw_value
;
1470 ret
= clock_raw_value_from_ns_from_origin(clock_class
,
1471 trimmer_it
->end
.ns_from_origin
, &end_raw_value
);
1473 status
= BT_SELF_MESSAGE_ITERATOR_STATUS_ERROR
;
1477 if (msg_type
== BT_MESSAGE_TYPE_DISCARDED_EVENTS
) {
1478 begin_cs
= bt_message_discarded_events_borrow_beginning_default_clock_snapshot_const(
1480 new_msg
= bt_message_discarded_events_create_with_default_clock_snapshots(
1481 trimmer_it
->self_msg_iter
,
1483 bt_clock_snapshot_get_value(begin_cs
),
1486 begin_cs
= bt_message_discarded_packets_borrow_beginning_default_clock_snapshot_const(
1488 new_msg
= bt_message_discarded_packets_create_with_default_clock_snapshots(
1489 trimmer_it
->self_msg_iter
,
1491 bt_clock_snapshot_get_value(begin_cs
),
1496 status
= BT_SELF_MESSAGE_ITERATOR_STATUS_NOMEM
;
1500 /* Replace the original message */
1501 BT_MESSAGE_MOVE_REF(msg
, new_msg
);
1504 if (G_UNLIKELY(!sstate
->inited
)) {
1505 status
= ensure_stream_state_is_inited(trimmer_it
,
1507 if (status
!= BT_SELF_MESSAGE_ITERATOR_STATUS_OK
) {
1512 push_message(trimmer_it
, msg
);
1516 case BT_MESSAGE_TYPE_STREAM_ACTIVITY_BEGINNING
:
1517 if (!trimmer_it
->end
.is_infinite
&&
1518 ns_from_origin
> trimmer_it
->end
.ns_from_origin
) {
1520 * This only happens when the message's time is
1521 * known and is greater than the trimming
1522 * range's end time. Unknown and -inf times are
1524 * `trimmer_it->end.ns_from_origin`.
1526 status
= end_iterator_streams(trimmer_it
);
1527 *reached_end
= true;
1531 if (!sstate
->inited
) {
1532 status
= ensure_stream_state_is_inited(trimmer_it
,
1535 if (status
!= BT_SELF_MESSAGE_ITERATOR_STATUS_OK
) {
1539 push_message(trimmer_it
, msg
);
1544 case BT_MESSAGE_TYPE_STREAM_ACTIVITY_END
:
1545 if (trimmer_it
->end
.is_infinite
) {
1546 push_message(trimmer_it
, msg
);
1551 if (ns_from_origin
== INT64_MIN
) {
1552 /* Unknown: push as is if stream state is inited */
1553 if (sstate
->inited
) {
1554 push_message(trimmer_it
, msg
);
1556 sstate
->last_msg_is_stream_activity_end
= true;
1558 } else if (ns_from_origin
== INT64_MAX
) {
1559 /* Infinite: use trimming range's end time */
1560 sstate
->stream_act_end_ns_from_origin
=
1561 trimmer_it
->end
.ns_from_origin
;
1563 /* Known: check if outside of trimming range */
1564 if (ns_from_origin
> trimmer_it
->end
.ns_from_origin
) {
1565 sstate
->stream_act_end_ns_from_origin
=
1566 trimmer_it
->end
.ns_from_origin
;
1567 status
= end_iterator_streams(trimmer_it
);
1568 *reached_end
= true;
1572 if (!sstate
->inited
) {
1574 * First message for this stream is a
1575 * stream activity end: we can't deduce
1576 * anything about the stream activity
1577 * beginning's time, and using this
1578 * message's time would make a useless
1579 * pair of stream activity beginning/end
1580 * with the same time. Just skip this
1581 * message and wait for something
1587 push_message(trimmer_it
, msg
);
1589 sstate
->last_msg_is_stream_activity_end
= true;
1590 sstate
->stream_act_end_ns_from_origin
= ns_from_origin
;
1594 case BT_MESSAGE_TYPE_STREAM_BEGINNING
:
1596 * We don't know what follows at this point, so just
1597 * keep this message until we know what to do with it
1598 * (it will be used in ensure_stream_state_is_inited()).
1600 BT_ASSERT(!sstate
->inited
);
1601 BT_MESSAGE_MOVE_REF(sstate
->stream_beginning_msg
, msg
);
1603 case BT_MESSAGE_TYPE_STREAM_END
:
1604 if (sstate
->inited
) {
1606 * This is the end of an inited stream: end this
1607 * stream if its stream activity end message
1608 * time is not the trimming range's end time
1609 * (which means the final stream activity end
1610 * message had an infinite time). end_stream()
1611 * will generate its own stream end message.
1613 if (trimmer_it
->end
.is_infinite
) {
1614 push_message(trimmer_it
, msg
);
1616 g_hash_table_remove(trimmer_it
->stream_states
,
1618 } else if (sstate
->stream_act_end_ns_from_origin
<
1619 trimmer_it
->end
.ns_from_origin
) {
1620 status
= end_stream(trimmer_it
, sstate
);
1621 if (status
!= BT_SELF_MESSAGE_ITERATOR_STATUS_OK
) {
1625 /* We won't need this stream state again */
1626 g_hash_table_remove(trimmer_it
->stream_states
,
1630 /* We dont't need this stream state anymore */
1631 g_hash_table_remove(trimmer_it
->stream_states
, sstate
->stream
);
1640 /* We release the message's reference whatever the outcome */
1641 bt_message_put_ref(msg
);
1642 return BT_SELF_MESSAGE_ITERATOR_STATUS_OK
;
1646 * Handles an input message. This _could_ make the iterator's output
1647 * message queue grow; this could also consume the message without
1648 * pushing anything to this queue, only modifying the stream state.
1650 * This function consumes the `msg` reference, _whatever the outcome_.
1652 * This function sets `reached_end` if handling this message made the
1653 * iterator reach the end of the trimming range. Note that the output
1654 * message queue could contain messages even if this function sets
1658 bt_self_message_iterator_status
handle_message(
1659 struct trimmer_iterator
*trimmer_it
, const bt_message
*msg
,
1662 bt_self_message_iterator_status status
;
1663 const bt_stream
*stream
= NULL
;
1664 int64_t ns_from_origin
= INT64_MIN
;
1667 struct trimmer_iterator_stream_state
*sstate
= NULL
;
1669 /* Find message's associated stream */
1670 switch (bt_message_get_type(msg
)) {
1671 case BT_MESSAGE_TYPE_EVENT
:
1672 stream
= bt_event_borrow_stream_const(
1673 bt_message_event_borrow_event_const(msg
));
1675 case BT_MESSAGE_TYPE_PACKET_BEGINNING
:
1676 stream
= bt_packet_borrow_stream_const(
1677 bt_message_packet_beginning_borrow_packet_const(msg
));
1679 case BT_MESSAGE_TYPE_PACKET_END
:
1680 stream
= bt_packet_borrow_stream_const(
1681 bt_message_packet_end_borrow_packet_const(msg
));
1683 case BT_MESSAGE_TYPE_DISCARDED_EVENTS
:
1684 stream
= bt_message_discarded_events_borrow_stream_const(msg
);
1686 case BT_MESSAGE_TYPE_DISCARDED_PACKETS
:
1687 stream
= bt_message_discarded_packets_borrow_stream_const(msg
);
1689 case BT_MESSAGE_TYPE_STREAM_ACTIVITY_BEGINNING
:
1690 stream
= bt_message_stream_activity_beginning_borrow_stream_const(msg
);
1692 case BT_MESSAGE_TYPE_STREAM_ACTIVITY_END
:
1693 stream
= bt_message_stream_activity_end_borrow_stream_const(msg
);
1695 case BT_MESSAGE_TYPE_STREAM_BEGINNING
:
1696 stream
= bt_message_stream_beginning_borrow_stream_const(msg
);
1698 case BT_MESSAGE_TYPE_STREAM_END
:
1699 stream
= bt_message_stream_end_borrow_stream_const(msg
);
1705 if (G_LIKELY(stream
)) {
1706 /* Find stream state */
1707 sstate
= g_hash_table_lookup(trimmer_it
->stream_states
,
1709 if (G_UNLIKELY(!sstate
)) {
1710 /* No stream state yet: create one now */
1711 const bt_stream_class
*sc
;
1714 * Validate right now that the stream's class
1715 * has a registered default clock class so that
1716 * an existing stream state guarantees existing
1717 * default clock snapshots for its associated
1720 * Also check that clock snapshots are always
1723 sc
= bt_stream_borrow_class_const(stream
);
1724 if (!bt_stream_class_borrow_default_clock_class_const(sc
)) {
1725 BT_LOGE("Unsupported stream: stream class does "
1726 "not have a default clock class: "
1728 "stream-id=%" PRIu64
", "
1729 "stream-name=\"%s\"",
1730 stream
, bt_stream_get_id(stream
),
1731 bt_stream_get_name(stream
));
1732 status
= BT_SELF_MESSAGE_ITERATOR_STATUS_ERROR
;
1737 * Temporary: make sure packet beginning, packet
1738 * end, discarded events, and discarded packets
1739 * messages have default clock snapshots until
1740 * the support for not having them is
1743 if (!bt_stream_class_packets_have_beginning_default_clock_snapshot(
1745 BT_LOGE("Unsupported stream: packets have "
1746 "no beginning clock snapshot: "
1748 "stream-id=%" PRIu64
", "
1749 "stream-name=\"%s\"",
1750 stream
, bt_stream_get_id(stream
),
1751 bt_stream_get_name(stream
));
1752 status
= BT_SELF_MESSAGE_ITERATOR_STATUS_ERROR
;
1756 if (!bt_stream_class_packets_have_end_default_clock_snapshot(
1758 BT_LOGE("Unsupported stream: packets have "
1759 "no end clock snapshot: "
1761 "stream-id=%" PRIu64
", "
1762 "stream-name=\"%s\"",
1763 stream
, bt_stream_get_id(stream
),
1764 bt_stream_get_name(stream
));
1765 status
= BT_SELF_MESSAGE_ITERATOR_STATUS_ERROR
;
1769 if (bt_stream_class_supports_discarded_events(sc
) &&
1770 !bt_stream_class_discarded_events_have_default_clock_snapshots(sc
)) {
1771 BT_LOGE("Unsupported stream: discarded events "
1772 "have no clock snapshots: "
1774 "stream-id=%" PRIu64
", "
1775 "stream-name=\"%s\"",
1776 stream
, bt_stream_get_id(stream
),
1777 bt_stream_get_name(stream
));
1778 status
= BT_SELF_MESSAGE_ITERATOR_STATUS_ERROR
;
1782 if (bt_stream_class_supports_discarded_packets(sc
) &&
1783 !bt_stream_class_discarded_packets_have_default_clock_snapshots(sc
)) {
1784 BT_LOGE("Unsupported stream: discarded packets "
1785 "have no clock snapshots: "
1787 "stream-id=%" PRIu64
", "
1788 "stream-name=\"%s\"",
1789 stream
, bt_stream_get_id(stream
),
1790 bt_stream_get_name(stream
));
1791 status
= BT_SELF_MESSAGE_ITERATOR_STATUS_ERROR
;
1795 sstate
= g_new0(struct trimmer_iterator_stream_state
,
1798 status
= BT_SELF_MESSAGE_ITERATOR_STATUS_NOMEM
;
1802 sstate
->stream
= stream
;
1803 sstate
->stream_act_end_ns_from_origin
= INT64_MIN
;
1804 g_hash_table_insert(trimmer_it
->stream_states
,
1805 (void *) stream
, sstate
);
1809 /* Retrieve the message's time */
1810 ret
= get_msg_ns_from_origin(msg
, &ns_from_origin
, &skip
);
1811 if (G_UNLIKELY(ret
)) {
1812 status
= BT_SELF_MESSAGE_ITERATOR_STATUS_ERROR
;
1816 if (G_LIKELY(sstate
)) {
1817 /* Message associated to a stream */
1818 status
= handle_message_with_stream_state(trimmer_it
, msg
,
1819 sstate
, ns_from_origin
, reached_end
);
1822 * handle_message_with_stream_state() unconditionally
1828 * Message not associated to a stream (message iterator
1831 if (G_UNLIKELY(ns_from_origin
> trimmer_it
->end
.ns_from_origin
)) {
1832 BT_MESSAGE_PUT_REF_AND_RESET(msg
);
1833 status
= end_iterator_streams(trimmer_it
);
1834 *reached_end
= true;
1836 push_message(trimmer_it
, msg
);
1837 status
= BT_SELF_MESSAGE_ITERATOR_STATUS_OK
;
1843 /* We release the message's reference whatever the outcome */
1844 bt_message_put_ref(msg
);
1849 void fill_message_array_from_output_messages(
1850 struct trimmer_iterator
*trimmer_it
,
1851 bt_message_array_const msgs
, uint64_t capacity
, uint64_t *count
)
1856 * Move auto-seek messages to the output array (which is this
1857 * iterator's base message array).
1859 while (capacity
> 0 && !g_queue_is_empty(trimmer_it
->output_messages
)) {
1860 msgs
[*count
] = pop_message(trimmer_it
);
1865 BT_ASSERT(*count
> 0);
1869 bt_self_message_iterator_status
state_ending(
1870 struct trimmer_iterator
*trimmer_it
,
1871 bt_message_array_const msgs
, uint64_t capacity
,
1874 bt_self_message_iterator_status status
=
1875 BT_SELF_MESSAGE_ITERATOR_STATUS_OK
;
1877 if (g_queue_is_empty(trimmer_it
->output_messages
)) {
1878 trimmer_it
->state
= TRIMMER_ITERATOR_STATE_ENDED
;
1879 status
= BT_SELF_MESSAGE_ITERATOR_STATUS_END
;
1883 fill_message_array_from_output_messages(trimmer_it
, msgs
,
1891 bt_self_message_iterator_status
state_trim(struct trimmer_iterator
*trimmer_it
,
1892 bt_message_array_const msgs
, uint64_t capacity
,
1895 bt_self_message_iterator_status status
=
1896 BT_SELF_MESSAGE_ITERATOR_STATUS_OK
;
1897 bt_message_array_const my_msgs
;
1900 bool reached_end
= false;
1902 while (g_queue_is_empty(trimmer_it
->output_messages
)) {
1903 status
= (int) bt_self_component_port_input_message_iterator_next(
1904 trimmer_it
->upstream_iter
, &my_msgs
, &my_count
);
1905 if (G_UNLIKELY(status
!= BT_SELF_MESSAGE_ITERATOR_STATUS_OK
)) {
1906 if (status
== BT_SELF_MESSAGE_ITERATOR_STATUS_END
) {
1907 status
= end_iterator_streams(trimmer_it
);
1908 if (status
!= BT_SELF_MESSAGE_ITERATOR_STATUS_OK
) {
1913 TRIMMER_ITERATOR_STATE_ENDING
;
1914 status
= state_ending(trimmer_it
, msgs
,
1921 BT_ASSERT(my_count
> 0);
1923 for (i
= 0; i
< my_count
; i
++) {
1924 status
= handle_message(trimmer_it
, my_msgs
[i
],
1928 * handle_message() unconditionally consumes the
1929 * message reference.
1933 if (G_UNLIKELY(status
!=
1934 BT_SELF_MESSAGE_ITERATOR_STATUS_OK
)) {
1935 put_messages(my_msgs
, my_count
);
1939 if (G_UNLIKELY(reached_end
)) {
1941 * This message's time was passed the
1942 * trimming time range's end time: we
1943 * are done. Their might still be
1944 * messages in the output message queue,
1945 * so move to the "ending" state and
1946 * apply it immediately since
1947 * state_trim() is called within the
1950 put_messages(my_msgs
, my_count
);
1952 TRIMMER_ITERATOR_STATE_ENDING
;
1953 status
= state_ending(trimmer_it
, msgs
,
1961 * There's at least one message in the output message queue:
1962 * move the messages to the output message array.
1964 BT_ASSERT(!g_queue_is_empty(trimmer_it
->output_messages
));
1965 fill_message_array_from_output_messages(trimmer_it
, msgs
,
1973 bt_self_message_iterator_status
trimmer_msg_iter_next(
1974 bt_self_message_iterator
*self_msg_iter
,
1975 bt_message_array_const msgs
, uint64_t capacity
,
1978 struct trimmer_iterator
*trimmer_it
=
1979 bt_self_message_iterator_get_data(self_msg_iter
);
1980 bt_self_message_iterator_status status
=
1981 BT_SELF_MESSAGE_ITERATOR_STATUS_OK
;
1983 BT_ASSERT(trimmer_it
);
1985 if (G_LIKELY(trimmer_it
->state
== TRIMMER_ITERATOR_STATE_TRIM
)) {
1986 status
= state_trim(trimmer_it
, msgs
, capacity
, count
);
1987 if (status
!= BT_SELF_MESSAGE_ITERATOR_STATUS_OK
) {
1991 switch (trimmer_it
->state
) {
1992 case TRIMMER_ITERATOR_STATE_SET_BOUNDS_NS_FROM_ORIGIN
:
1993 status
= state_set_trimmer_iterator_bounds(trimmer_it
);
1994 if (status
!= BT_SELF_MESSAGE_ITERATOR_STATUS_OK
) {
1998 status
= state_seek_initially(trimmer_it
);
1999 if (status
!= BT_SELF_MESSAGE_ITERATOR_STATUS_OK
) {
2003 status
= state_trim(trimmer_it
, msgs
, capacity
, count
);
2004 if (status
!= BT_SELF_MESSAGE_ITERATOR_STATUS_OK
) {
2009 case TRIMMER_ITERATOR_STATE_SEEK_INITIALLY
:
2010 status
= state_seek_initially(trimmer_it
);
2011 if (status
!= BT_SELF_MESSAGE_ITERATOR_STATUS_OK
) {
2015 status
= state_trim(trimmer_it
, msgs
, capacity
, count
);
2016 if (status
!= BT_SELF_MESSAGE_ITERATOR_STATUS_OK
) {
2021 case TRIMMER_ITERATOR_STATE_ENDING
:
2022 status
= state_ending(trimmer_it
, msgs
, capacity
,
2024 if (status
!= BT_SELF_MESSAGE_ITERATOR_STATUS_OK
) {
2029 case TRIMMER_ITERATOR_STATE_ENDED
:
2030 status
= BT_SELF_MESSAGE_ITERATOR_STATUS_END
;
2042 void trimmer_msg_iter_finalize(bt_self_message_iterator
*self_msg_iter
)
2044 struct trimmer_iterator
*trimmer_it
=
2045 bt_self_message_iterator_get_data(self_msg_iter
);
2047 BT_ASSERT(trimmer_it
);
2048 destroy_trimmer_iterator(trimmer_it
);