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 "common/assert.h"
38 #define NS_PER_S INT64_C(1000000000)
40 static const char * const in_port_name
= "in";
43 unsigned int hour
, minute
, second
, ns
;
46 struct trimmer_bound
{
48 * Nanoseconds from origin, valid if `is_set` is set and
49 * `is_infinite` is false.
51 int64_t ns_from_origin
;
53 /* True if this bound's full time (`ns_from_origin`) is set */
57 * True if this bound represents the infinity (negative or
58 * positive depending on which bound it is). If this is true,
59 * then we don't care about `ns_from_origin` above.
64 * This bound's time without the date; this time is used to set
65 * `ns_from_origin` once we know the date.
67 struct trimmer_time time
;
71 struct trimmer_bound begin
, end
;
75 enum trimmer_iterator_state
{
77 * Find the first message's date and set the bounds's times
80 TRIMMER_ITERATOR_STATE_SET_BOUNDS_NS_FROM_ORIGIN
,
83 * Initially seek to the trimming range's beginning time.
85 TRIMMER_ITERATOR_STATE_SEEK_INITIALLY
,
88 * Fill the output message queue for as long as received input
89 * messages are within the trimming time range.
91 TRIMMER_ITERATOR_STATE_TRIM
,
93 /* Flush the remaining messages in the output message queue */
94 TRIMMER_ITERATOR_STATE_ENDING
,
96 /* Trimming operation and message iterator is ended */
97 TRIMMER_ITERATOR_STATE_ENDED
,
100 struct trimmer_iterator
{
102 struct trimmer_comp
*trimmer_comp
;
105 bt_self_message_iterator
*self_msg_iter
;
107 enum trimmer_iterator_state state
;
110 bt_self_component_port_input_message_iterator
*upstream_iter
;
111 struct trimmer_bound begin
, end
;
114 * Queue of `const bt_message *` (owned by the queue).
116 * This is where the trimming operation pushes the messages to
117 * output by this message iterator.
119 GQueue
*output_messages
;
122 * Hash table of `bt_stream *` (weak) to
123 * `struct trimmer_iterator_stream_state *` (owned by the HT).
125 GHashTable
*stream_states
;
128 struct trimmer_iterator_stream_state
{
130 * True if both stream beginning and initial stream activity
131 * beginning messages were pushed for this stream.
136 * True if the last pushed message for this stream was a stream
137 * activity end message.
139 bool last_msg_is_stream_activity_end
;
142 * Time to use for a generated stream end activity message when
145 int64_t stream_act_end_ns_from_origin
;
148 const bt_stream
*stream
;
150 /* Owned by this (`NULL` initially and between packets) */
151 const bt_packet
*cur_packet
;
154 const bt_message
*stream_beginning_msg
;
158 void destroy_trimmer_comp(struct trimmer_comp
*trimmer_comp
)
160 BT_ASSERT(trimmer_comp
);
161 g_free(trimmer_comp
);
165 struct trimmer_comp
*create_trimmer_comp(void)
167 return g_new0(struct trimmer_comp
, 1);
171 void trimmer_finalize(bt_self_component_filter
*self_comp
)
173 struct trimmer_comp
*trimmer_comp
=
174 bt_self_component_get_data(
175 bt_self_component_filter_as_self_component(self_comp
));
178 destroy_trimmer_comp(trimmer_comp
);
183 * Sets the time (in ns from origin) of a trimmer bound from date and
186 * Returns a negative value if anything goes wrong.
189 int set_bound_ns_from_origin(struct trimmer_bound
*bound
,
190 unsigned int year
, unsigned int month
, unsigned int day
,
191 unsigned int hour
, unsigned int minute
, unsigned int second
,
192 unsigned int ns
, bool is_gmt
)
202 .tm_year
= year
- 1900,
207 result
= bt_timegm(&tm
);
209 result
= mktime(&tm
);
218 bound
->ns_from_origin
= (int64_t) result
;
219 bound
->ns_from_origin
*= NS_PER_S
;
220 bound
->ns_from_origin
+= ns
;
221 bound
->is_set
= true;
228 * Parses a timestamp, figuring out its format.
230 * Returns a negative value if anything goes wrong.
234 * YYYY-MM-DD hh:mm[:ss[.ns]]
238 * TODO: Check overflows.
241 int set_bound_from_str(const char *str
, struct trimmer_bound
*bound
,
246 unsigned int year
, month
, day
, hour
, minute
, second
, ns
;
249 /* Try `YYYY-MM-DD hh:mm:ss.ns` format */
250 s_ret
= sscanf(str
, "%u-%u-%u %u:%u:%u.%u%c", &year
, &month
, &day
,
251 &hour
, &minute
, &second
, &ns
, &dummy
);
253 ret
= set_bound_ns_from_origin(bound
, year
, month
, day
,
254 hour
, minute
, second
, ns
, is_gmt
);
258 /* Try `YYYY-MM-DD hh:mm:ss` format */
259 s_ret
= sscanf(str
, "%u-%u-%u %u:%u:%u%c", &year
, &month
, &day
,
260 &hour
, &minute
, &second
, &dummy
);
262 ret
= set_bound_ns_from_origin(bound
, year
, month
, day
,
263 hour
, minute
, second
, 0, is_gmt
);
267 /* Try `YYYY-MM-DD hh:mm` format */
268 s_ret
= sscanf(str
, "%u-%u-%u %u:%u%c", &year
, &month
, &day
,
269 &hour
, &minute
, &dummy
);
271 ret
= set_bound_ns_from_origin(bound
, year
, month
, day
,
272 hour
, minute
, 0, 0, is_gmt
);
276 /* Try `YYYY-MM-DD` format */
277 s_ret
= sscanf(str
, "%u-%u-%u%c", &year
, &month
, &day
, &dummy
);
279 ret
= set_bound_ns_from_origin(bound
, year
, month
, day
,
284 /* Try `hh:mm:ss.ns` format */
285 s_ret
= sscanf(str
, "%u:%u:%u.%u%c", &hour
, &minute
, &second
, &ns
,
288 bound
->time
.hour
= hour
;
289 bound
->time
.minute
= minute
;
290 bound
->time
.second
= second
;
295 /* Try `hh:mm:ss` format */
296 s_ret
= sscanf(str
, "%u:%u:%u%c", &hour
, &minute
, &second
, &dummy
);
298 bound
->time
.hour
= hour
;
299 bound
->time
.minute
= minute
;
300 bound
->time
.second
= second
;
305 /* Try `-s.ns` format */
306 s_ret
= sscanf(str
, "-%u.%u%c", &second
, &ns
, &dummy
);
308 bound
->ns_from_origin
= -((int64_t) second
) * NS_PER_S
;
309 bound
->ns_from_origin
-= (int64_t) ns
;
310 bound
->is_set
= true;
314 /* Try `s.ns` format */
315 s_ret
= sscanf(str
, "%u.%u%c", &second
, &ns
, &dummy
);
317 bound
->ns_from_origin
= ((int64_t) second
) * NS_PER_S
;
318 bound
->ns_from_origin
+= (int64_t) ns
;
319 bound
->is_set
= true;
323 /* Try `-s` format */
324 s_ret
= sscanf(str
, "-%u%c", &second
, &dummy
);
326 bound
->ns_from_origin
= -((int64_t) second
) * NS_PER_S
;
327 bound
->is_set
= true;
332 s_ret
= sscanf(str
, "%u%c", &second
, &dummy
);
334 bound
->ns_from_origin
= (int64_t) second
* NS_PER_S
;
335 bound
->is_set
= true;
339 BT_LOGE("Invalid date/time format: param=\"%s\"", str
);
347 * Sets a trimmer bound's properties from a parameter string/integer
350 * Returns a negative value if anything goes wrong.
353 int set_bound_from_param(const char *param_name
, const bt_value
*param
,
354 struct trimmer_bound
*bound
, bool is_gmt
)
360 if (bt_value_is_signed_integer(param
)) {
361 int64_t value
= bt_value_signed_integer_get(param
);
364 * Just convert it to a temporary string to handle
365 * everything the same way.
367 sprintf(tmp_arg
, "%" PRId64
, value
);
369 } else if (bt_value_is_string(param
)) {
370 arg
= bt_value_string_get(param
);
372 BT_LOGE("`%s` parameter must be an integer or a string value.",
378 ret
= set_bound_from_str(arg
, bound
, is_gmt
);
385 int validate_trimmer_bounds(struct trimmer_bound
*begin
,
386 struct trimmer_bound
*end
)
390 BT_ASSERT(begin
->is_set
);
391 BT_ASSERT(end
->is_set
);
393 if (!begin
->is_infinite
&& !end
->is_infinite
&&
394 begin
->ns_from_origin
> end
->ns_from_origin
) {
395 BT_LOGE("Trimming time range's beginning time is greater than end time: "
396 "begin-ns-from-origin=%" PRId64
", "
397 "end-ns-from-origin=%" PRId64
,
398 begin
->ns_from_origin
,
399 end
->ns_from_origin
);
404 if (!begin
->is_infinite
&& begin
->ns_from_origin
== INT64_MIN
) {
405 BT_LOGE("Invalid trimming time range's beginning time: "
406 "ns-from-origin=%" PRId64
,
407 begin
->ns_from_origin
);
412 if (!end
->is_infinite
&& end
->ns_from_origin
== INT64_MIN
) {
413 BT_LOGE("Invalid trimming time range's end time: "
414 "ns-from-origin=%" PRId64
,
415 end
->ns_from_origin
);
425 int init_trimmer_comp_from_params(struct trimmer_comp
*trimmer_comp
,
426 const bt_value
*params
)
428 const bt_value
*value
;
432 value
= bt_value_map_borrow_entry_value_const(params
, "gmt");
434 trimmer_comp
->is_gmt
= (bool) bt_value_bool_get(value
);
437 value
= bt_value_map_borrow_entry_value_const(params
, "begin");
439 if (set_bound_from_param("begin", value
,
440 &trimmer_comp
->begin
, trimmer_comp
->is_gmt
)) {
441 /* set_bound_from_param() logs errors */
442 ret
= BT_SELF_COMPONENT_STATUS_ERROR
;
446 trimmer_comp
->begin
.is_infinite
= true;
447 trimmer_comp
->begin
.is_set
= true;
450 value
= bt_value_map_borrow_entry_value_const(params
, "end");
452 if (set_bound_from_param("end", value
,
453 &trimmer_comp
->end
, trimmer_comp
->is_gmt
)) {
454 /* set_bound_from_param() logs errors */
455 ret
= BT_SELF_COMPONENT_STATUS_ERROR
;
459 trimmer_comp
->end
.is_infinite
= true;
460 trimmer_comp
->end
.is_set
= true;
464 if (trimmer_comp
->begin
.is_set
&& trimmer_comp
->end
.is_set
) {
465 /* validate_trimmer_bounds() logs errors */
466 ret
= validate_trimmer_bounds(&trimmer_comp
->begin
,
473 bt_self_component_status
trimmer_init(bt_self_component_filter
*self_comp
,
474 const bt_value
*params
, void *init_data
)
477 bt_self_component_status status
;
478 struct trimmer_comp
*trimmer_comp
= create_trimmer_comp();
481 status
= BT_SELF_COMPONENT_STATUS_NOMEM
;
485 status
= bt_self_component_filter_add_input_port(
486 self_comp
, in_port_name
, NULL
, NULL
);
487 if (status
!= BT_SELF_COMPONENT_STATUS_OK
) {
491 status
= bt_self_component_filter_add_output_port(
492 self_comp
, "out", NULL
, NULL
);
493 if (status
!= BT_SELF_COMPONENT_STATUS_OK
) {
497 ret
= init_trimmer_comp_from_params(trimmer_comp
, params
);
499 status
= BT_SELF_COMPONENT_STATUS_ERROR
;
503 bt_self_component_set_data(
504 bt_self_component_filter_as_self_component(self_comp
),
509 if (status
== BT_SELF_COMPONENT_STATUS_OK
) {
510 status
= BT_SELF_COMPONENT_STATUS_ERROR
;
514 destroy_trimmer_comp(trimmer_comp
);
522 void destroy_trimmer_iterator(struct trimmer_iterator
*trimmer_it
)
524 BT_ASSERT(trimmer_it
);
525 bt_self_component_port_input_message_iterator_put_ref(
526 trimmer_it
->upstream_iter
);
528 if (trimmer_it
->output_messages
) {
529 g_queue_free(trimmer_it
->output_messages
);
532 if (trimmer_it
->stream_states
) {
533 g_hash_table_destroy(trimmer_it
->stream_states
);
540 void destroy_trimmer_iterator_stream_state(
541 struct trimmer_iterator_stream_state
*sstate
)
544 BT_PACKET_PUT_REF_AND_RESET(sstate
->cur_packet
);
545 BT_MESSAGE_PUT_REF_AND_RESET(sstate
->stream_beginning_msg
);
550 bt_self_message_iterator_status
trimmer_msg_iter_init(
551 bt_self_message_iterator
*self_msg_iter
,
552 bt_self_component_filter
*self_comp
,
553 bt_self_component_port_output
*port
)
555 bt_self_message_iterator_status status
=
556 BT_SELF_MESSAGE_ITERATOR_STATUS_OK
;
557 struct trimmer_iterator
*trimmer_it
;
559 trimmer_it
= g_new0(struct trimmer_iterator
, 1);
561 status
= BT_SELF_MESSAGE_ITERATOR_STATUS_NOMEM
;
565 trimmer_it
->trimmer_comp
= bt_self_component_get_data(
566 bt_self_component_filter_as_self_component(self_comp
));
567 BT_ASSERT(trimmer_it
->trimmer_comp
);
569 if (trimmer_it
->trimmer_comp
->begin
.is_set
&&
570 trimmer_it
->trimmer_comp
->end
.is_set
) {
572 * Both trimming time range's bounds are set, so skip
574 * `TRIMMER_ITERATOR_STATE_SET_BOUNDS_NS_FROM_ORIGIN`
577 trimmer_it
->state
= TRIMMER_ITERATOR_STATE_SEEK_INITIALLY
;
580 trimmer_it
->begin
= trimmer_it
->trimmer_comp
->begin
;
581 trimmer_it
->end
= trimmer_it
->trimmer_comp
->end
;
582 trimmer_it
->upstream_iter
=
583 bt_self_component_port_input_message_iterator_create(
584 bt_self_component_filter_borrow_input_port_by_name(
585 self_comp
, in_port_name
));
586 if (!trimmer_it
->upstream_iter
) {
587 status
= BT_SELF_MESSAGE_ITERATOR_STATUS_ERROR
;
591 trimmer_it
->output_messages
= g_queue_new();
592 if (!trimmer_it
->output_messages
) {
593 status
= BT_SELF_MESSAGE_ITERATOR_STATUS_NOMEM
;
597 trimmer_it
->stream_states
= g_hash_table_new_full(g_direct_hash
,
598 g_direct_equal
, NULL
,
599 (GDestroyNotify
) destroy_trimmer_iterator_stream_state
);
600 if (!trimmer_it
->stream_states
) {
601 status
= BT_SELF_MESSAGE_ITERATOR_STATUS_NOMEM
;
605 trimmer_it
->self_msg_iter
= self_msg_iter
;
606 bt_self_message_iterator_set_data(self_msg_iter
, trimmer_it
);
609 if (status
!= BT_SELF_MESSAGE_ITERATOR_STATUS_OK
&& trimmer_it
) {
610 destroy_trimmer_iterator(trimmer_it
);
617 int get_msg_ns_from_origin(const bt_message
*msg
, int64_t *ns_from_origin
,
620 const bt_clock_class
*clock_class
= NULL
;
621 const bt_clock_snapshot
*clock_snapshot
= NULL
;
622 bt_message_stream_activity_clock_snapshot_state sa_cs_state
;
626 BT_ASSERT(ns_from_origin
);
629 switch (bt_message_get_type(msg
)) {
630 case BT_MESSAGE_TYPE_EVENT
:
632 bt_message_event_borrow_stream_class_default_clock_class_const(
634 if (G_UNLIKELY(!clock_class
)) {
638 clock_snapshot
= bt_message_event_borrow_default_clock_snapshot_const(
641 case BT_MESSAGE_TYPE_PACKET_BEGINNING
:
643 bt_message_packet_beginning_borrow_stream_class_default_clock_class_const(
645 if (G_UNLIKELY(!clock_class
)) {
649 clock_snapshot
= bt_message_packet_beginning_borrow_default_clock_snapshot_const(
652 case BT_MESSAGE_TYPE_PACKET_END
:
654 bt_message_packet_end_borrow_stream_class_default_clock_class_const(
656 if (G_UNLIKELY(!clock_class
)) {
660 clock_snapshot
= bt_message_packet_end_borrow_default_clock_snapshot_const(
663 case BT_MESSAGE_TYPE_DISCARDED_EVENTS
:
665 bt_message_discarded_events_borrow_stream_class_default_clock_class_const(
667 if (G_UNLIKELY(!clock_class
)) {
671 clock_snapshot
= bt_message_discarded_events_borrow_beginning_default_clock_snapshot_const(
674 case BT_MESSAGE_TYPE_DISCARDED_PACKETS
:
676 bt_message_discarded_packets_borrow_stream_class_default_clock_class_const(
678 if (G_UNLIKELY(!clock_class
)) {
682 clock_snapshot
= bt_message_discarded_packets_borrow_beginning_default_clock_snapshot_const(
685 case BT_MESSAGE_TYPE_STREAM_ACTIVITY_BEGINNING
:
687 bt_message_stream_activity_beginning_borrow_stream_class_default_clock_class_const(
689 if (G_UNLIKELY(!clock_class
)) {
693 sa_cs_state
= bt_message_stream_activity_beginning_borrow_default_clock_snapshot_const(
694 msg
, &clock_snapshot
);
695 if (sa_cs_state
== BT_MESSAGE_STREAM_ACTIVITY_CLOCK_SNAPSHOT_STATE_UNKNOWN
||
696 sa_cs_state
== BT_MESSAGE_STREAM_ACTIVITY_CLOCK_SNAPSHOT_STATE_INFINITE
) {
697 /* Lowest possible time to always include them */
698 *ns_from_origin
= INT64_MIN
;
699 goto no_clock_snapshot
;
703 case BT_MESSAGE_TYPE_STREAM_ACTIVITY_END
:
705 bt_message_stream_activity_end_borrow_stream_class_default_clock_class_const(
707 if (G_UNLIKELY(!clock_class
)) {
711 sa_cs_state
= bt_message_stream_activity_end_borrow_default_clock_snapshot_const(
712 msg
, &clock_snapshot
);
713 if (sa_cs_state
== BT_MESSAGE_STREAM_ACTIVITY_CLOCK_SNAPSHOT_STATE_UNKNOWN
) {
714 /* Lowest time to always include it */
715 *ns_from_origin
= INT64_MIN
;
716 goto no_clock_snapshot
;
717 } else if (sa_cs_state
== BT_MESSAGE_STREAM_ACTIVITY_CLOCK_SNAPSHOT_STATE_INFINITE
) {
718 /* Greatest time to always exclude it */
719 *ns_from_origin
= INT64_MAX
;
720 goto no_clock_snapshot
;
724 case BT_MESSAGE_TYPE_MESSAGE_ITERATOR_INACTIVITY
:
726 bt_message_message_iterator_inactivity_borrow_default_clock_snapshot_const(
730 goto no_clock_snapshot
;
733 ret
= bt_clock_snapshot_get_ns_from_origin(clock_snapshot
,
735 if (G_UNLIKELY(ret
)) {
753 void put_messages(bt_message_array_const msgs
, uint64_t count
)
757 for (i
= 0; i
< count
; i
++) {
758 BT_MESSAGE_PUT_REF_AND_RESET(msgs
[i
]);
763 int set_trimmer_iterator_bound(struct trimmer_bound
*bound
,
764 int64_t ns_from_origin
, bool is_gmt
)
767 time_t time_seconds
= (time_t) (ns_from_origin
/ NS_PER_S
);
770 BT_ASSERT(!bound
->is_set
);
773 /* We only need to extract the date from this time */
775 bt_gmtime_r(&time_seconds
, &tm
);
777 bt_localtime_r(&time_seconds
, &tm
);
781 BT_LOGE_ERRNO("Cannot convert timestamp to date and time",
782 "ts=%" PRId64
, (int64_t) time_seconds
);
787 ret
= set_bound_ns_from_origin(bound
, tm
.tm_year
+ 1900, tm
.tm_mon
+ 1,
788 tm
.tm_mday
, bound
->time
.hour
, bound
->time
.minute
,
789 bound
->time
.second
, bound
->time
.ns
, is_gmt
);
796 bt_self_message_iterator_status
state_set_trimmer_iterator_bounds(
797 struct trimmer_iterator
*trimmer_it
)
799 bt_message_iterator_status upstream_iter_status
=
800 BT_MESSAGE_ITERATOR_STATUS_OK
;
801 struct trimmer_comp
*trimmer_comp
= trimmer_it
->trimmer_comp
;
802 bt_message_array_const msgs
;
804 int64_t ns_from_origin
= INT64_MIN
;
808 BT_ASSERT(!trimmer_it
->begin
.is_set
||
809 !trimmer_it
->end
.is_set
);
812 upstream_iter_status
=
813 bt_self_component_port_input_message_iterator_next(
814 trimmer_it
->upstream_iter
, &msgs
, &count
);
815 if (upstream_iter_status
!= BT_MESSAGE_ITERATOR_STATUS_OK
) {
819 for (i
= 0; i
< count
; i
++) {
820 const bt_message
*msg
= msgs
[i
];
824 ret
= get_msg_ns_from_origin(msg
, &ns_from_origin
,
834 BT_ASSERT(ns_from_origin
!= INT64_MIN
&&
835 ns_from_origin
!= INT64_MAX
);
836 put_messages(msgs
, count
);
840 put_messages(msgs
, count
);
844 if (!trimmer_it
->begin
.is_set
) {
845 BT_ASSERT(!trimmer_it
->begin
.is_infinite
);
846 ret
= set_trimmer_iterator_bound(&trimmer_it
->begin
,
847 ns_from_origin
, trimmer_comp
->is_gmt
);
853 if (!trimmer_it
->end
.is_set
) {
854 BT_ASSERT(!trimmer_it
->end
.is_infinite
);
855 ret
= set_trimmer_iterator_bound(&trimmer_it
->end
,
856 ns_from_origin
, trimmer_comp
->is_gmt
);
862 ret
= validate_trimmer_bounds(&trimmer_it
->begin
,
871 put_messages(msgs
, count
);
872 upstream_iter_status
= BT_MESSAGE_ITERATOR_STATUS_ERROR
;
875 return (int) upstream_iter_status
;
879 bt_self_message_iterator_status
state_seek_initially(
880 struct trimmer_iterator
*trimmer_it
)
882 bt_self_message_iterator_status status
=
883 BT_SELF_MESSAGE_ITERATOR_STATUS_OK
;
885 BT_ASSERT(trimmer_it
->begin
.is_set
);
887 if (trimmer_it
->begin
.is_infinite
) {
888 if (!bt_self_component_port_input_message_iterator_can_seek_beginning(
889 trimmer_it
->upstream_iter
)) {
890 BT_LOGE_STR("Cannot make upstream message iterator initially seek its beginning.");
891 status
= BT_SELF_MESSAGE_ITERATOR_STATUS_ERROR
;
895 status
= (int) bt_self_component_port_input_message_iterator_seek_beginning(
896 trimmer_it
->upstream_iter
);
898 if (!bt_self_component_port_input_message_iterator_can_seek_ns_from_origin(
899 trimmer_it
->upstream_iter
,
900 trimmer_it
->begin
.ns_from_origin
)) {
901 BT_LOGE("Cannot make upstream message iterator initially seek: "
902 "seek-ns-from-origin=%" PRId64
,
903 trimmer_it
->begin
.ns_from_origin
);
904 status
= BT_SELF_MESSAGE_ITERATOR_STATUS_ERROR
;
908 status
= (int) bt_self_component_port_input_message_iterator_seek_ns_from_origin(
909 trimmer_it
->upstream_iter
, trimmer_it
->begin
.ns_from_origin
);
912 if (status
== BT_SELF_MESSAGE_ITERATOR_STATUS_OK
) {
913 trimmer_it
->state
= TRIMMER_ITERATOR_STATE_TRIM
;
921 void push_message(struct trimmer_iterator
*trimmer_it
, const bt_message
*msg
)
923 g_queue_push_head(trimmer_it
->output_messages
, (void *) msg
);
927 const bt_message
*pop_message(struct trimmer_iterator
*trimmer_it
)
929 return g_queue_pop_tail(trimmer_it
->output_messages
);
933 int clock_raw_value_from_ns_from_origin(const bt_clock_class
*clock_class
,
934 int64_t ns_from_origin
, uint64_t *raw_value
)
938 uint64_t cc_offset_cycles
;
941 bt_clock_class_get_offset(clock_class
, &cc_offset_s
, &cc_offset_cycles
);
942 cc_freq
= bt_clock_class_get_frequency(clock_class
);
943 return bt_common_clock_value_from_ns_from_origin(cc_offset_s
,
944 cc_offset_cycles
, cc_freq
, ns_from_origin
, raw_value
);
948 bt_self_message_iterator_status
end_stream(struct trimmer_iterator
*trimmer_it
,
949 struct trimmer_iterator_stream_state
*sstate
)
951 bt_self_message_iterator_status status
=
952 BT_SELF_MESSAGE_ITERATOR_STATUS_OK
;
954 const bt_clock_class
*clock_class
;
956 bt_message
*msg
= NULL
;
958 BT_ASSERT(!trimmer_it
->end
.is_infinite
);
960 if (!sstate
->stream
) {
964 if (sstate
->cur_packet
) {
966 * The last message could not have been a stream
967 * activity end message if we have a current packet.
969 BT_ASSERT(!sstate
->last_msg_is_stream_activity_end
);
972 * Create and push a packet end message, making its time
973 * the trimming range's end time.
975 clock_class
= bt_stream_class_borrow_default_clock_class_const(
976 bt_stream_borrow_class_const(sstate
->stream
));
977 BT_ASSERT(clock_class
);
978 ret
= clock_raw_value_from_ns_from_origin(clock_class
,
979 trimmer_it
->end
.ns_from_origin
, &raw_value
);
981 status
= BT_SELF_MESSAGE_ITERATOR_STATUS_ERROR
;
985 msg
= bt_message_packet_end_create_with_default_clock_snapshot(
986 trimmer_it
->self_msg_iter
, sstate
->cur_packet
,
989 status
= BT_SELF_MESSAGE_ITERATOR_STATUS_NOMEM
;
993 push_message(trimmer_it
, msg
);
995 BT_PACKET_PUT_REF_AND_RESET(sstate
->cur_packet
);
998 * Because we generated a packet end message, set the
999 * stream activity end message's time to use to the
1000 * trimming range's end time (this packet end message's
1003 sstate
->stream_act_end_ns_from_origin
=
1004 trimmer_it
->end
.ns_from_origin
;
1007 if (!sstate
->last_msg_is_stream_activity_end
) {
1008 /* Create and push a stream activity end message */
1009 msg
= bt_message_stream_activity_end_create(
1010 trimmer_it
->self_msg_iter
, sstate
->stream
);
1012 status
= BT_SELF_MESSAGE_ITERATOR_STATUS_NOMEM
;
1016 clock_class
= bt_stream_class_borrow_default_clock_class_const(
1017 bt_stream_borrow_class_const(sstate
->stream
));
1018 BT_ASSERT(clock_class
);
1020 if (sstate
->stream_act_end_ns_from_origin
== INT64_MIN
) {
1022 * We received at least what is necessary to
1023 * have a stream state (stream beginning and
1024 * stream activity beginning messages), but
1025 * nothing else: use the trimmer range's end
1028 sstate
->stream_act_end_ns_from_origin
=
1029 trimmer_it
->end
.ns_from_origin
;
1032 ret
= clock_raw_value_from_ns_from_origin(clock_class
,
1033 sstate
->stream_act_end_ns_from_origin
, &raw_value
);
1035 status
= BT_SELF_MESSAGE_ITERATOR_STATUS_ERROR
;
1039 bt_message_stream_activity_end_set_default_clock_snapshot(
1041 push_message(trimmer_it
, msg
);
1045 /* Create and push a stream end message */
1046 msg
= bt_message_stream_end_create(trimmer_it
->self_msg_iter
,
1049 status
= BT_SELF_MESSAGE_ITERATOR_STATUS_NOMEM
;
1053 push_message(trimmer_it
, msg
);
1057 * Just to make sure that we don't use this stream state again
1058 * in the future without an obvious error.
1060 sstate
->stream
= NULL
;
1063 bt_message_put_ref(msg
);
1068 bt_self_message_iterator_status
end_iterator_streams(
1069 struct trimmer_iterator
*trimmer_it
)
1071 bt_self_message_iterator_status status
=
1072 BT_SELF_MESSAGE_ITERATOR_STATUS_OK
;
1073 GHashTableIter iter
;
1074 gpointer key
, sstate
;
1076 if (trimmer_it
->end
.is_infinite
) {
1078 * An infinite trimming range's end time guarantees that
1079 * we received (and pushed) all the appropriate end
1086 * End each stream and then remove them from the hash table of
1087 * stream states to release unneeded references.
1089 g_hash_table_iter_init(&iter
, trimmer_it
->stream_states
);
1091 while (g_hash_table_iter_next(&iter
, &key
, &sstate
)) {
1092 status
= end_stream(trimmer_it
, sstate
);
1099 g_hash_table_remove_all(trimmer_it
->stream_states
);
1106 bt_self_message_iterator_status
create_stream_beginning_activity_message(
1107 struct trimmer_iterator
*trimmer_it
,
1108 const bt_stream
*stream
,
1109 const bt_clock_class
*clock_class
, bt_message
**msg
)
1111 bt_message
*local_msg
;
1112 bt_self_message_iterator_status status
=
1113 BT_SELF_MESSAGE_ITERATOR_STATUS_OK
;
1116 BT_ASSERT(!trimmer_it
->begin
.is_infinite
);
1118 local_msg
= bt_message_stream_activity_beginning_create(
1119 trimmer_it
->self_msg_iter
, stream
);
1121 status
= BT_SELF_MESSAGE_ITERATOR_STATUS_NOMEM
;
1129 ret
= clock_raw_value_from_ns_from_origin(clock_class
,
1130 trimmer_it
->begin
.ns_from_origin
, &raw_value
);
1132 status
= BT_SELF_MESSAGE_ITERATOR_STATUS_ERROR
;
1133 bt_message_put_ref(local_msg
);
1137 bt_message_stream_activity_beginning_set_default_clock_snapshot(
1138 local_msg
, raw_value
);
1141 BT_MESSAGE_MOVE_REF(*msg
, local_msg
);
1148 * Makes sure to initialize a stream state, pushing the appropriate
1151 * `stream_act_beginning_msg` is an initial stream activity beginning
1152 * message to potentially use, depending on its clock snapshot state.
1153 * This function consumes `stream_act_beginning_msg` unconditionally.
1156 bt_self_message_iterator_status
ensure_stream_state_is_inited(
1157 struct trimmer_iterator
*trimmer_it
,
1158 struct trimmer_iterator_stream_state
*sstate
,
1159 const bt_message
*stream_act_beginning_msg
)
1161 bt_self_message_iterator_status status
=
1162 BT_SELF_MESSAGE_ITERATOR_STATUS_OK
;
1163 bt_message
*new_msg
= NULL
;
1164 const bt_clock_class
*clock_class
=
1165 bt_stream_class_borrow_default_clock_class_const(
1166 bt_stream_borrow_class_const(sstate
->stream
));
1168 BT_ASSERT(!sstate
->inited
);
1170 if (!sstate
->stream_beginning_msg
) {
1171 /* No initial stream beginning message: create one */
1172 sstate
->stream_beginning_msg
=
1173 bt_message_stream_beginning_create(
1174 trimmer_it
->self_msg_iter
, sstate
->stream
);
1175 if (!sstate
->stream_beginning_msg
) {
1176 status
= BT_SELF_MESSAGE_ITERATOR_STATUS_NOMEM
;
1181 /* Push initial stream beginning message */
1182 BT_ASSERT(sstate
->stream_beginning_msg
);
1183 push_message(trimmer_it
, sstate
->stream_beginning_msg
);
1184 sstate
->stream_beginning_msg
= NULL
;
1186 if (stream_act_beginning_msg
) {
1188 * Initial stream activity beginning message exists: if
1189 * its time is -inf, then create and push a new one
1190 * having the trimming range's beginning time. Otherwise
1191 * push it as is (known and unknown).
1193 const bt_clock_snapshot
*cs
;
1194 bt_message_stream_activity_clock_snapshot_state sa_cs_state
;
1196 sa_cs_state
= bt_message_stream_activity_beginning_borrow_default_clock_snapshot_const(
1197 stream_act_beginning_msg
, &cs
);
1198 if (sa_cs_state
== BT_MESSAGE_STREAM_ACTIVITY_CLOCK_SNAPSHOT_STATE_INFINITE
&&
1199 !trimmer_it
->begin
.is_infinite
) {
1201 * -inf time: use trimming range's beginning
1202 * time (which is not -inf).
1204 status
= create_stream_beginning_activity_message(
1205 trimmer_it
, sstate
->stream
, clock_class
,
1207 if (status
!= BT_SELF_MESSAGE_ITERATOR_STATUS_OK
) {
1211 push_message(trimmer_it
, new_msg
);
1214 /* Known/unknown: push as is */
1215 push_message(trimmer_it
, stream_act_beginning_msg
);
1216 stream_act_beginning_msg
= NULL
;
1219 BT_ASSERT(!trimmer_it
->begin
.is_infinite
);
1222 * No stream beginning activity message: create and push
1225 status
= create_stream_beginning_activity_message(
1226 trimmer_it
, sstate
->stream
, clock_class
, &new_msg
);
1227 if (status
!= BT_SELF_MESSAGE_ITERATOR_STATUS_OK
) {
1231 push_message(trimmer_it
, new_msg
);
1235 sstate
->inited
= true;
1238 bt_message_put_ref(new_msg
);
1239 bt_message_put_ref(stream_act_beginning_msg
);
1244 bt_self_message_iterator_status
ensure_cur_packet_exists(
1245 struct trimmer_iterator
*trimmer_it
,
1246 struct trimmer_iterator_stream_state
*sstate
,
1247 const bt_packet
*packet
)
1249 bt_self_message_iterator_status status
=
1250 BT_SELF_MESSAGE_ITERATOR_STATUS_OK
;
1252 const bt_clock_class
*clock_class
=
1253 bt_stream_class_borrow_default_clock_class_const(
1254 bt_stream_borrow_class_const(sstate
->stream
));
1255 bt_message
*msg
= NULL
;
1258 BT_ASSERT(!trimmer_it
->begin
.is_infinite
);
1259 BT_ASSERT(!sstate
->cur_packet
);
1262 * Create and push an initial packet beginning message,
1263 * making its time the trimming range's beginning time.
1265 ret
= clock_raw_value_from_ns_from_origin(clock_class
,
1266 trimmer_it
->begin
.ns_from_origin
, &raw_value
);
1268 status
= BT_SELF_MESSAGE_ITERATOR_STATUS_ERROR
;
1272 msg
= bt_message_packet_beginning_create_with_default_clock_snapshot(
1273 trimmer_it
->self_msg_iter
, packet
, raw_value
);
1275 status
= BT_SELF_MESSAGE_ITERATOR_STATUS_NOMEM
;
1279 push_message(trimmer_it
, msg
);
1282 /* Set packet as this stream's current packet */
1283 sstate
->cur_packet
= packet
;
1284 bt_packet_get_ref(sstate
->cur_packet
);
1287 bt_message_put_ref(msg
);
1292 * Handles a message which is associated to a given stream state. This
1293 * _could_ make the iterator's output message queue grow; this could
1294 * also consume the message without pushing anything to this queue, only
1295 * modifying the stream state.
1297 * This function consumes the `msg` reference, _whatever the outcome_.
1299 * `ns_from_origin` is the message's time, as given by
1300 * get_msg_ns_from_origin().
1302 * This function sets `reached_end` if handling this message made the
1303 * iterator reach the end of the trimming range. Note that the output
1304 * message queue could contain messages even if this function sets
1308 bt_self_message_iterator_status
handle_message_with_stream_state(
1309 struct trimmer_iterator
*trimmer_it
, const bt_message
*msg
,
1310 struct trimmer_iterator_stream_state
*sstate
,
1311 int64_t ns_from_origin
, bool *reached_end
)
1313 bt_self_message_iterator_status status
=
1314 BT_SELF_MESSAGE_ITERATOR_STATUS_OK
;
1315 bt_message_type msg_type
= bt_message_get_type(msg
);
1319 case BT_MESSAGE_TYPE_EVENT
:
1320 if (G_UNLIKELY(!trimmer_it
->end
.is_infinite
&&
1321 ns_from_origin
> trimmer_it
->end
.ns_from_origin
)) {
1322 status
= end_iterator_streams(trimmer_it
);
1323 *reached_end
= true;
1327 if (G_UNLIKELY(!sstate
->inited
)) {
1328 status
= ensure_stream_state_is_inited(trimmer_it
,
1330 if (status
!= BT_SELF_MESSAGE_ITERATOR_STATUS_OK
) {
1335 if (G_UNLIKELY(!sstate
->cur_packet
)) {
1336 const bt_event
*event
=
1337 bt_message_event_borrow_event_const(msg
);
1338 const bt_packet
*packet
= bt_event_borrow_packet_const(
1341 status
= ensure_cur_packet_exists(trimmer_it
, sstate
,
1343 if (status
!= BT_SELF_MESSAGE_ITERATOR_STATUS_OK
) {
1348 BT_ASSERT(sstate
->cur_packet
);
1349 push_message(trimmer_it
, msg
);
1352 case BT_MESSAGE_TYPE_PACKET_BEGINNING
:
1353 if (G_UNLIKELY(!trimmer_it
->end
.is_infinite
&&
1354 ns_from_origin
> trimmer_it
->end
.ns_from_origin
)) {
1355 status
= end_iterator_streams(trimmer_it
);
1356 *reached_end
= true;
1360 if (G_UNLIKELY(!sstate
->inited
)) {
1361 status
= ensure_stream_state_is_inited(trimmer_it
,
1363 if (status
!= BT_SELF_MESSAGE_ITERATOR_STATUS_OK
) {
1368 BT_ASSERT(!sstate
->cur_packet
);
1369 sstate
->cur_packet
=
1370 bt_message_packet_beginning_borrow_packet_const(msg
);
1371 bt_packet_get_ref(sstate
->cur_packet
);
1372 push_message(trimmer_it
, msg
);
1375 case BT_MESSAGE_TYPE_PACKET_END
:
1376 sstate
->stream_act_end_ns_from_origin
= ns_from_origin
;
1378 if (G_UNLIKELY(!trimmer_it
->end
.is_infinite
&&
1379 ns_from_origin
> trimmer_it
->end
.ns_from_origin
)) {
1380 status
= end_iterator_streams(trimmer_it
);
1381 *reached_end
= true;
1385 if (G_UNLIKELY(!sstate
->inited
)) {
1386 status
= ensure_stream_state_is_inited(trimmer_it
,
1388 if (status
!= BT_SELF_MESSAGE_ITERATOR_STATUS_OK
) {
1393 if (G_UNLIKELY(!sstate
->cur_packet
)) {
1394 const bt_packet
*packet
=
1395 bt_message_packet_end_borrow_packet_const(msg
);
1397 status
= ensure_cur_packet_exists(trimmer_it
, sstate
,
1399 if (status
!= BT_SELF_MESSAGE_ITERATOR_STATUS_OK
) {
1404 BT_ASSERT(sstate
->cur_packet
);
1405 BT_PACKET_PUT_REF_AND_RESET(sstate
->cur_packet
);
1406 push_message(trimmer_it
, msg
);
1409 case BT_MESSAGE_TYPE_DISCARDED_EVENTS
:
1410 case BT_MESSAGE_TYPE_DISCARDED_PACKETS
:
1413 * `ns_from_origin` is the message's time range's
1414 * beginning time here.
1416 int64_t end_ns_from_origin
;
1417 const bt_clock_snapshot
*end_cs
;
1419 if (bt_message_get_type(msg
) ==
1420 BT_MESSAGE_TYPE_DISCARDED_EVENTS
) {
1422 * Safe to ignore the return value because we
1423 * know there's a default clock and it's always
1426 end_cs
= bt_message_discarded_events_borrow_end_default_clock_snapshot_const(
1430 * Safe to ignore the return value because we
1431 * know there's a default clock and it's always
1434 end_cs
= bt_message_discarded_packets_borrow_end_default_clock_snapshot_const(
1438 if (bt_clock_snapshot_get_ns_from_origin(end_cs
,
1439 &end_ns_from_origin
)) {
1440 status
= BT_SELF_MESSAGE_ITERATOR_STATUS_ERROR
;
1444 sstate
->stream_act_end_ns_from_origin
= end_ns_from_origin
;
1446 if (!trimmer_it
->end
.is_infinite
&&
1447 ns_from_origin
> trimmer_it
->end
.ns_from_origin
) {
1448 status
= end_iterator_streams(trimmer_it
);
1449 *reached_end
= true;
1453 if (!trimmer_it
->end
.is_infinite
&&
1454 end_ns_from_origin
> trimmer_it
->end
.ns_from_origin
) {
1456 * This message's end time is outside the
1457 * trimming time range: replace it with a new
1458 * message having an end time equal to the
1459 * trimming time range's end and without a
1462 const bt_clock_class
*clock_class
=
1463 bt_clock_snapshot_borrow_clock_class_const(
1465 const bt_clock_snapshot
*begin_cs
;
1466 bt_message
*new_msg
;
1467 uint64_t end_raw_value
;
1469 ret
= clock_raw_value_from_ns_from_origin(clock_class
,
1470 trimmer_it
->end
.ns_from_origin
, &end_raw_value
);
1472 status
= BT_SELF_MESSAGE_ITERATOR_STATUS_ERROR
;
1476 if (msg_type
== BT_MESSAGE_TYPE_DISCARDED_EVENTS
) {
1477 begin_cs
= bt_message_discarded_events_borrow_beginning_default_clock_snapshot_const(
1479 new_msg
= bt_message_discarded_events_create_with_default_clock_snapshots(
1480 trimmer_it
->self_msg_iter
,
1482 bt_clock_snapshot_get_value(begin_cs
),
1485 begin_cs
= bt_message_discarded_packets_borrow_beginning_default_clock_snapshot_const(
1487 new_msg
= bt_message_discarded_packets_create_with_default_clock_snapshots(
1488 trimmer_it
->self_msg_iter
,
1490 bt_clock_snapshot_get_value(begin_cs
),
1495 status
= BT_SELF_MESSAGE_ITERATOR_STATUS_NOMEM
;
1499 /* Replace the original message */
1500 BT_MESSAGE_MOVE_REF(msg
, new_msg
);
1503 if (G_UNLIKELY(!sstate
->inited
)) {
1504 status
= ensure_stream_state_is_inited(trimmer_it
,
1506 if (status
!= BT_SELF_MESSAGE_ITERATOR_STATUS_OK
) {
1511 push_message(trimmer_it
, msg
);
1515 case BT_MESSAGE_TYPE_STREAM_ACTIVITY_BEGINNING
:
1516 if (!trimmer_it
->end
.is_infinite
&&
1517 ns_from_origin
> trimmer_it
->end
.ns_from_origin
) {
1519 * This only happens when the message's time is
1520 * known and is greater than the trimming
1521 * range's end time. Unknown and -inf times are
1523 * `trimmer_it->end.ns_from_origin`.
1525 status
= end_iterator_streams(trimmer_it
);
1526 *reached_end
= true;
1530 if (!sstate
->inited
) {
1531 status
= ensure_stream_state_is_inited(trimmer_it
,
1534 if (status
!= BT_SELF_MESSAGE_ITERATOR_STATUS_OK
) {
1538 push_message(trimmer_it
, msg
);
1543 case BT_MESSAGE_TYPE_STREAM_ACTIVITY_END
:
1544 if (trimmer_it
->end
.is_infinite
) {
1545 push_message(trimmer_it
, msg
);
1550 if (ns_from_origin
== INT64_MIN
) {
1551 /* Unknown: push as is if stream state is inited */
1552 if (sstate
->inited
) {
1553 push_message(trimmer_it
, msg
);
1555 sstate
->last_msg_is_stream_activity_end
= true;
1557 } else if (ns_from_origin
== INT64_MAX
) {
1558 /* Infinite: use trimming range's end time */
1559 sstate
->stream_act_end_ns_from_origin
=
1560 trimmer_it
->end
.ns_from_origin
;
1562 /* Known: check if outside of trimming range */
1563 if (ns_from_origin
> trimmer_it
->end
.ns_from_origin
) {
1564 sstate
->stream_act_end_ns_from_origin
=
1565 trimmer_it
->end
.ns_from_origin
;
1566 status
= end_iterator_streams(trimmer_it
);
1567 *reached_end
= true;
1571 if (!sstate
->inited
) {
1573 * First message for this stream is a
1574 * stream activity end: we can't deduce
1575 * anything about the stream activity
1576 * beginning's time, and using this
1577 * message's time would make a useless
1578 * pair of stream activity beginning/end
1579 * with the same time. Just skip this
1580 * message and wait for something
1586 push_message(trimmer_it
, msg
);
1588 sstate
->last_msg_is_stream_activity_end
= true;
1589 sstate
->stream_act_end_ns_from_origin
= ns_from_origin
;
1593 case BT_MESSAGE_TYPE_STREAM_BEGINNING
:
1595 * We don't know what follows at this point, so just
1596 * keep this message until we know what to do with it
1597 * (it will be used in ensure_stream_state_is_inited()).
1599 BT_ASSERT(!sstate
->inited
);
1600 BT_MESSAGE_MOVE_REF(sstate
->stream_beginning_msg
, msg
);
1602 case BT_MESSAGE_TYPE_STREAM_END
:
1603 if (sstate
->inited
) {
1605 * This is the end of an inited stream: end this
1606 * stream if its stream activity end message
1607 * time is not the trimming range's end time
1608 * (which means the final stream activity end
1609 * message had an infinite time). end_stream()
1610 * will generate its own stream end message.
1612 if (trimmer_it
->end
.is_infinite
) {
1613 push_message(trimmer_it
, msg
);
1615 g_hash_table_remove(trimmer_it
->stream_states
,
1617 } else if (sstate
->stream_act_end_ns_from_origin
<
1618 trimmer_it
->end
.ns_from_origin
) {
1619 status
= end_stream(trimmer_it
, sstate
);
1620 if (status
!= BT_SELF_MESSAGE_ITERATOR_STATUS_OK
) {
1624 /* We won't need this stream state again */
1625 g_hash_table_remove(trimmer_it
->stream_states
,
1629 /* We dont't need this stream state anymore */
1630 g_hash_table_remove(trimmer_it
->stream_states
, sstate
->stream
);
1639 /* We release the message's reference whatever the outcome */
1640 bt_message_put_ref(msg
);
1641 return BT_SELF_MESSAGE_ITERATOR_STATUS_OK
;
1645 * Handles an input message. This _could_ make the iterator's output
1646 * message queue grow; this could also consume the message without
1647 * pushing anything to this queue, only modifying the stream state.
1649 * This function consumes the `msg` reference, _whatever the outcome_.
1651 * This function sets `reached_end` if handling this message made the
1652 * iterator reach the end of the trimming range. Note that the output
1653 * message queue could contain messages even if this function sets
1657 bt_self_message_iterator_status
handle_message(
1658 struct trimmer_iterator
*trimmer_it
, const bt_message
*msg
,
1661 bt_self_message_iterator_status status
;
1662 const bt_stream
*stream
= NULL
;
1663 int64_t ns_from_origin
= INT64_MIN
;
1666 struct trimmer_iterator_stream_state
*sstate
= NULL
;
1668 /* Find message's associated stream */
1669 switch (bt_message_get_type(msg
)) {
1670 case BT_MESSAGE_TYPE_EVENT
:
1671 stream
= bt_event_borrow_stream_const(
1672 bt_message_event_borrow_event_const(msg
));
1674 case BT_MESSAGE_TYPE_PACKET_BEGINNING
:
1675 stream
= bt_packet_borrow_stream_const(
1676 bt_message_packet_beginning_borrow_packet_const(msg
));
1678 case BT_MESSAGE_TYPE_PACKET_END
:
1679 stream
= bt_packet_borrow_stream_const(
1680 bt_message_packet_end_borrow_packet_const(msg
));
1682 case BT_MESSAGE_TYPE_DISCARDED_EVENTS
:
1683 stream
= bt_message_discarded_events_borrow_stream_const(msg
);
1685 case BT_MESSAGE_TYPE_DISCARDED_PACKETS
:
1686 stream
= bt_message_discarded_packets_borrow_stream_const(msg
);
1688 case BT_MESSAGE_TYPE_STREAM_ACTIVITY_BEGINNING
:
1689 stream
= bt_message_stream_activity_beginning_borrow_stream_const(msg
);
1691 case BT_MESSAGE_TYPE_STREAM_ACTIVITY_END
:
1692 stream
= bt_message_stream_activity_end_borrow_stream_const(msg
);
1694 case BT_MESSAGE_TYPE_STREAM_BEGINNING
:
1695 stream
= bt_message_stream_beginning_borrow_stream_const(msg
);
1697 case BT_MESSAGE_TYPE_STREAM_END
:
1698 stream
= bt_message_stream_end_borrow_stream_const(msg
);
1704 if (G_LIKELY(stream
)) {
1705 /* Find stream state */
1706 sstate
= g_hash_table_lookup(trimmer_it
->stream_states
,
1708 if (G_UNLIKELY(!sstate
)) {
1709 /* No stream state yet: create one now */
1710 const bt_stream_class
*sc
;
1713 * Validate right now that the stream's class
1714 * has a registered default clock class so that
1715 * an existing stream state guarantees existing
1716 * default clock snapshots for its associated
1719 * Also check that clock snapshots are always
1722 sc
= bt_stream_borrow_class_const(stream
);
1723 if (!bt_stream_class_borrow_default_clock_class_const(sc
)) {
1724 BT_LOGE("Unsupported stream: stream class does "
1725 "not have a default clock class: "
1727 "stream-id=%" PRIu64
", "
1728 "stream-name=\"%s\"",
1729 stream
, bt_stream_get_id(stream
),
1730 bt_stream_get_name(stream
));
1731 status
= BT_SELF_MESSAGE_ITERATOR_STATUS_ERROR
;
1736 * Temporary: make sure packet beginning, packet
1737 * end, discarded events, and discarded packets
1738 * messages have default clock snapshots until
1739 * the support for not having them is
1742 if (!bt_stream_class_packets_have_beginning_default_clock_snapshot(
1744 BT_LOGE("Unsupported stream: packets have "
1745 "no beginning clock snapshot: "
1747 "stream-id=%" PRIu64
", "
1748 "stream-name=\"%s\"",
1749 stream
, bt_stream_get_id(stream
),
1750 bt_stream_get_name(stream
));
1751 status
= BT_SELF_MESSAGE_ITERATOR_STATUS_ERROR
;
1755 if (!bt_stream_class_packets_have_end_default_clock_snapshot(
1757 BT_LOGE("Unsupported stream: packets have "
1758 "no end clock snapshot: "
1760 "stream-id=%" PRIu64
", "
1761 "stream-name=\"%s\"",
1762 stream
, bt_stream_get_id(stream
),
1763 bt_stream_get_name(stream
));
1764 status
= BT_SELF_MESSAGE_ITERATOR_STATUS_ERROR
;
1768 if (bt_stream_class_supports_discarded_events(sc
) &&
1769 !bt_stream_class_discarded_events_have_default_clock_snapshots(sc
)) {
1770 BT_LOGE("Unsupported stream: discarded events "
1771 "have no clock snapshots: "
1773 "stream-id=%" PRIu64
", "
1774 "stream-name=\"%s\"",
1775 stream
, bt_stream_get_id(stream
),
1776 bt_stream_get_name(stream
));
1777 status
= BT_SELF_MESSAGE_ITERATOR_STATUS_ERROR
;
1781 if (bt_stream_class_supports_discarded_packets(sc
) &&
1782 !bt_stream_class_discarded_packets_have_default_clock_snapshots(sc
)) {
1783 BT_LOGE("Unsupported stream: discarded packets "
1784 "have no clock snapshots: "
1786 "stream-id=%" PRIu64
", "
1787 "stream-name=\"%s\"",
1788 stream
, bt_stream_get_id(stream
),
1789 bt_stream_get_name(stream
));
1790 status
= BT_SELF_MESSAGE_ITERATOR_STATUS_ERROR
;
1794 sstate
= g_new0(struct trimmer_iterator_stream_state
,
1797 status
= BT_SELF_MESSAGE_ITERATOR_STATUS_NOMEM
;
1801 sstate
->stream
= stream
;
1802 sstate
->stream_act_end_ns_from_origin
= INT64_MIN
;
1803 g_hash_table_insert(trimmer_it
->stream_states
,
1804 (void *) stream
, sstate
);
1808 /* Retrieve the message's time */
1809 ret
= get_msg_ns_from_origin(msg
, &ns_from_origin
, &skip
);
1810 if (G_UNLIKELY(ret
)) {
1811 status
= BT_SELF_MESSAGE_ITERATOR_STATUS_ERROR
;
1815 if (G_LIKELY(sstate
)) {
1816 /* Message associated to a stream */
1817 status
= handle_message_with_stream_state(trimmer_it
, msg
,
1818 sstate
, ns_from_origin
, reached_end
);
1821 * handle_message_with_stream_state() unconditionally
1827 * Message not associated to a stream (message iterator
1830 if (G_UNLIKELY(ns_from_origin
> trimmer_it
->end
.ns_from_origin
)) {
1831 BT_MESSAGE_PUT_REF_AND_RESET(msg
);
1832 status
= end_iterator_streams(trimmer_it
);
1833 *reached_end
= true;
1835 push_message(trimmer_it
, msg
);
1836 status
= BT_SELF_MESSAGE_ITERATOR_STATUS_OK
;
1842 /* We release the message's reference whatever the outcome */
1843 bt_message_put_ref(msg
);
1848 void fill_message_array_from_output_messages(
1849 struct trimmer_iterator
*trimmer_it
,
1850 bt_message_array_const msgs
, uint64_t capacity
, uint64_t *count
)
1855 * Move auto-seek messages to the output array (which is this
1856 * iterator's base message array).
1858 while (capacity
> 0 && !g_queue_is_empty(trimmer_it
->output_messages
)) {
1859 msgs
[*count
] = pop_message(trimmer_it
);
1864 BT_ASSERT(*count
> 0);
1868 bt_self_message_iterator_status
state_ending(
1869 struct trimmer_iterator
*trimmer_it
,
1870 bt_message_array_const msgs
, uint64_t capacity
,
1873 bt_self_message_iterator_status status
=
1874 BT_SELF_MESSAGE_ITERATOR_STATUS_OK
;
1876 if (g_queue_is_empty(trimmer_it
->output_messages
)) {
1877 trimmer_it
->state
= TRIMMER_ITERATOR_STATE_ENDED
;
1878 status
= BT_SELF_MESSAGE_ITERATOR_STATUS_END
;
1882 fill_message_array_from_output_messages(trimmer_it
, msgs
,
1890 bt_self_message_iterator_status
state_trim(struct trimmer_iterator
*trimmer_it
,
1891 bt_message_array_const msgs
, uint64_t capacity
,
1894 bt_self_message_iterator_status status
=
1895 BT_SELF_MESSAGE_ITERATOR_STATUS_OK
;
1896 bt_message_array_const my_msgs
;
1899 bool reached_end
= false;
1901 while (g_queue_is_empty(trimmer_it
->output_messages
)) {
1902 status
= (int) bt_self_component_port_input_message_iterator_next(
1903 trimmer_it
->upstream_iter
, &my_msgs
, &my_count
);
1904 if (G_UNLIKELY(status
!= BT_SELF_MESSAGE_ITERATOR_STATUS_OK
)) {
1905 if (status
== BT_SELF_MESSAGE_ITERATOR_STATUS_END
) {
1906 status
= end_iterator_streams(trimmer_it
);
1907 if (status
!= BT_SELF_MESSAGE_ITERATOR_STATUS_OK
) {
1912 TRIMMER_ITERATOR_STATE_ENDING
;
1913 status
= state_ending(trimmer_it
, msgs
,
1920 BT_ASSERT(my_count
> 0);
1922 for (i
= 0; i
< my_count
; i
++) {
1923 status
= handle_message(trimmer_it
, my_msgs
[i
],
1927 * handle_message() unconditionally consumes the
1928 * message reference.
1932 if (G_UNLIKELY(status
!=
1933 BT_SELF_MESSAGE_ITERATOR_STATUS_OK
)) {
1934 put_messages(my_msgs
, my_count
);
1938 if (G_UNLIKELY(reached_end
)) {
1940 * This message's time was passed the
1941 * trimming time range's end time: we
1942 * are done. Their might still be
1943 * messages in the output message queue,
1944 * so move to the "ending" state and
1945 * apply it immediately since
1946 * state_trim() is called within the
1949 put_messages(my_msgs
, my_count
);
1951 TRIMMER_ITERATOR_STATE_ENDING
;
1952 status
= state_ending(trimmer_it
, msgs
,
1960 * There's at least one message in the output message queue:
1961 * move the messages to the output message array.
1963 BT_ASSERT(!g_queue_is_empty(trimmer_it
->output_messages
));
1964 fill_message_array_from_output_messages(trimmer_it
, msgs
,
1972 bt_self_message_iterator_status
trimmer_msg_iter_next(
1973 bt_self_message_iterator
*self_msg_iter
,
1974 bt_message_array_const msgs
, uint64_t capacity
,
1977 struct trimmer_iterator
*trimmer_it
=
1978 bt_self_message_iterator_get_data(self_msg_iter
);
1979 bt_self_message_iterator_status status
=
1980 BT_SELF_MESSAGE_ITERATOR_STATUS_OK
;
1982 BT_ASSERT(trimmer_it
);
1984 if (G_LIKELY(trimmer_it
->state
== TRIMMER_ITERATOR_STATE_TRIM
)) {
1985 status
= state_trim(trimmer_it
, msgs
, capacity
, count
);
1986 if (status
!= BT_SELF_MESSAGE_ITERATOR_STATUS_OK
) {
1990 switch (trimmer_it
->state
) {
1991 case TRIMMER_ITERATOR_STATE_SET_BOUNDS_NS_FROM_ORIGIN
:
1992 status
= state_set_trimmer_iterator_bounds(trimmer_it
);
1993 if (status
!= BT_SELF_MESSAGE_ITERATOR_STATUS_OK
) {
1997 status
= state_seek_initially(trimmer_it
);
1998 if (status
!= BT_SELF_MESSAGE_ITERATOR_STATUS_OK
) {
2002 status
= state_trim(trimmer_it
, msgs
, capacity
, count
);
2003 if (status
!= BT_SELF_MESSAGE_ITERATOR_STATUS_OK
) {
2008 case TRIMMER_ITERATOR_STATE_SEEK_INITIALLY
:
2009 status
= state_seek_initially(trimmer_it
);
2010 if (status
!= BT_SELF_MESSAGE_ITERATOR_STATUS_OK
) {
2014 status
= state_trim(trimmer_it
, msgs
, capacity
, count
);
2015 if (status
!= BT_SELF_MESSAGE_ITERATOR_STATUS_OK
) {
2020 case TRIMMER_ITERATOR_STATE_ENDING
:
2021 status
= state_ending(trimmer_it
, msgs
, capacity
,
2023 if (status
!= BT_SELF_MESSAGE_ITERATOR_STATUS_OK
) {
2028 case TRIMMER_ITERATOR_STATE_ENDED
:
2029 status
= BT_SELF_MESSAGE_ITERATOR_STATUS_END
;
2041 void trimmer_msg_iter_finalize(bt_self_message_iterator
*self_msg_iter
)
2043 struct trimmer_iterator
*trimmer_it
=
2044 bt_self_message_iterator_get_data(self_msg_iter
);
2046 BT_ASSERT(trimmer_it
);
2047 destroy_trimmer_iterator(trimmer_it
);