2 * Copyright 2016 Jérémie Galarneau <jeremie.galarneau@efficios.com>
3 * Copyright 2019 Philippe Proulx <pproulx@efficios.com>
5 * Permission is hereby granted, free of charge, to any person obtaining a copy
6 * of this software and associated documentation files (the "Software"), to deal
7 * in the Software without restriction, including without limitation the rights
8 * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
9 * copies of the Software, and to permit persons to whom the Software is
10 * furnished to do so, subject to the following conditions:
12 * The above copyright notice and this permission notice shall be included in
13 * all copies or substantial portions of the Software.
15 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
16 * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
17 * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
18 * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
19 * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
20 * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
24 #define BT_COMP_LOG_SELF_COMP (trimmer_comp->self_comp)
25 #define BT_LOG_OUTPUT_LEVEL (trimmer_comp->log_level)
26 #define BT_LOG_TAG "PLUGIN/FLT.UTILS.TRIMMER"
27 #include "plugins/comp-logging.h"
29 #include "compat/utc.h"
30 #include "compat/time.h"
31 #include <babeltrace2/babeltrace.h>
32 #include "common/common.h"
33 #include "common/assert.h"
40 #define NS_PER_S INT64_C(1000000000)
42 static const char * const in_port_name
= "in";
45 unsigned int hour
, minute
, second
, ns
;
48 struct trimmer_bound
{
50 * Nanoseconds from origin, valid if `is_set` is set and
51 * `is_infinite` is false.
53 int64_t ns_from_origin
;
55 /* True if this bound's full time (`ns_from_origin`) is set */
59 * True if this bound represents the infinity (negative or
60 * positive depending on which bound it is). If this is true,
61 * then we don't care about `ns_from_origin` above.
66 * This bound's time without the date; this time is used to set
67 * `ns_from_origin` once we know the date.
69 struct trimmer_time time
;
73 struct trimmer_bound begin
, end
;
75 bt_logging_level log_level
;
76 bt_self_component
*self_comp
;
79 enum trimmer_iterator_state
{
81 * Find the first message's date and set the bounds's times
84 TRIMMER_ITERATOR_STATE_SET_BOUNDS_NS_FROM_ORIGIN
,
87 * Initially seek to the trimming range's beginning time.
89 TRIMMER_ITERATOR_STATE_SEEK_INITIALLY
,
92 * Fill the output message queue for as long as received input
93 * messages are within the trimming time range.
95 TRIMMER_ITERATOR_STATE_TRIM
,
97 /* Flush the remaining messages in the output message queue */
98 TRIMMER_ITERATOR_STATE_ENDING
,
100 /* Trimming operation and message iterator is ended */
101 TRIMMER_ITERATOR_STATE_ENDED
,
104 struct trimmer_iterator
{
106 struct trimmer_comp
*trimmer_comp
;
109 bt_self_message_iterator
*self_msg_iter
;
111 enum trimmer_iterator_state state
;
114 bt_self_component_port_input_message_iterator
*upstream_iter
;
115 struct trimmer_bound begin
, end
;
118 * Queue of `const bt_message *` (owned by the queue).
120 * This is where the trimming operation pushes the messages to
121 * output by this message iterator.
123 GQueue
*output_messages
;
126 * Hash table of `bt_stream *` (weak) to
127 * `struct trimmer_iterator_stream_state *` (owned by the HT).
129 GHashTable
*stream_states
;
132 struct trimmer_iterator_stream_state
{
134 * True if both stream beginning and initial stream activity
135 * beginning messages were pushed for this stream.
140 * True if the last pushed message for this stream was a stream
141 * activity end message.
143 bool last_msg_is_stream_activity_end
;
146 * Time to use for a generated stream end activity message when
149 int64_t stream_act_end_ns_from_origin
;
152 const bt_stream
*stream
;
154 /* Owned by this (`NULL` initially and between packets) */
155 const bt_packet
*cur_packet
;
158 const bt_message
*stream_beginning_msg
;
162 void destroy_trimmer_comp(struct trimmer_comp
*trimmer_comp
)
164 BT_ASSERT(trimmer_comp
);
165 g_free(trimmer_comp
);
169 struct trimmer_comp
*create_trimmer_comp(void)
171 return g_new0(struct trimmer_comp
, 1);
175 void trimmer_finalize(bt_self_component_filter
*self_comp
)
177 struct trimmer_comp
*trimmer_comp
=
178 bt_self_component_get_data(
179 bt_self_component_filter_as_self_component(self_comp
));
182 destroy_trimmer_comp(trimmer_comp
);
187 * Sets the time (in ns from origin) of a trimmer bound from date and
190 * Returns a negative value if anything goes wrong.
193 int set_bound_ns_from_origin(struct trimmer_bound
*bound
,
194 unsigned int year
, unsigned int month
, unsigned int day
,
195 unsigned int hour
, unsigned int minute
, unsigned int second
,
196 unsigned int ns
, bool is_gmt
)
206 .tm_year
= year
- 1900,
211 result
= bt_timegm(&tm
);
213 result
= mktime(&tm
);
222 bound
->ns_from_origin
= (int64_t) result
;
223 bound
->ns_from_origin
*= NS_PER_S
;
224 bound
->ns_from_origin
+= ns
;
225 bound
->is_set
= true;
232 * Parses a timestamp, figuring out its format.
234 * Returns a negative value if anything goes wrong.
238 * YYYY-MM-DD hh:mm[:ss[.ns]]
242 * TODO: Check overflows.
245 int set_bound_from_str(struct trimmer_comp
*trimmer_comp
,
246 const char *str
, struct trimmer_bound
*bound
, bool is_gmt
)
250 unsigned int year
, month
, day
, hour
, minute
, second
, ns
;
253 /* Try `YYYY-MM-DD hh:mm:ss.ns` format */
254 s_ret
= sscanf(str
, "%u-%u-%u %u:%u:%u.%u%c", &year
, &month
, &day
,
255 &hour
, &minute
, &second
, &ns
, &dummy
);
257 ret
= set_bound_ns_from_origin(bound
, year
, month
, day
,
258 hour
, minute
, second
, ns
, is_gmt
);
262 /* Try `YYYY-MM-DD hh:mm:ss` format */
263 s_ret
= sscanf(str
, "%u-%u-%u %u:%u:%u%c", &year
, &month
, &day
,
264 &hour
, &minute
, &second
, &dummy
);
266 ret
= set_bound_ns_from_origin(bound
, year
, month
, day
,
267 hour
, minute
, second
, 0, is_gmt
);
271 /* Try `YYYY-MM-DD hh:mm` format */
272 s_ret
= sscanf(str
, "%u-%u-%u %u:%u%c", &year
, &month
, &day
,
273 &hour
, &minute
, &dummy
);
275 ret
= set_bound_ns_from_origin(bound
, year
, month
, day
,
276 hour
, minute
, 0, 0, is_gmt
);
280 /* Try `YYYY-MM-DD` format */
281 s_ret
= sscanf(str
, "%u-%u-%u%c", &year
, &month
, &day
, &dummy
);
283 ret
= set_bound_ns_from_origin(bound
, year
, month
, day
,
288 /* Try `hh:mm:ss.ns` format */
289 s_ret
= sscanf(str
, "%u:%u:%u.%u%c", &hour
, &minute
, &second
, &ns
,
292 bound
->time
.hour
= hour
;
293 bound
->time
.minute
= minute
;
294 bound
->time
.second
= second
;
299 /* Try `hh:mm:ss` format */
300 s_ret
= sscanf(str
, "%u:%u:%u%c", &hour
, &minute
, &second
, &dummy
);
302 bound
->time
.hour
= hour
;
303 bound
->time
.minute
= minute
;
304 bound
->time
.second
= second
;
309 /* Try `-s.ns` format */
310 s_ret
= sscanf(str
, "-%u.%u%c", &second
, &ns
, &dummy
);
312 bound
->ns_from_origin
= -((int64_t) second
) * NS_PER_S
;
313 bound
->ns_from_origin
-= (int64_t) ns
;
314 bound
->is_set
= true;
318 /* Try `s.ns` format */
319 s_ret
= sscanf(str
, "%u.%u%c", &second
, &ns
, &dummy
);
321 bound
->ns_from_origin
= ((int64_t) second
) * NS_PER_S
;
322 bound
->ns_from_origin
+= (int64_t) ns
;
323 bound
->is_set
= true;
327 /* Try `-s` format */
328 s_ret
= sscanf(str
, "-%u%c", &second
, &dummy
);
330 bound
->ns_from_origin
= -((int64_t) second
) * NS_PER_S
;
331 bound
->is_set
= true;
336 s_ret
= sscanf(str
, "%u%c", &second
, &dummy
);
338 bound
->ns_from_origin
= (int64_t) second
* NS_PER_S
;
339 bound
->is_set
= true;
343 BT_COMP_LOGE("Invalid date/time format: param=\"%s\"", str
);
351 * Sets a trimmer bound's properties from a parameter string/integer
354 * Returns a negative value if anything goes wrong.
357 int set_bound_from_param(struct trimmer_comp
*trimmer_comp
,
358 const char *param_name
, const bt_value
*param
,
359 struct trimmer_bound
*bound
, bool is_gmt
)
365 if (bt_value_is_signed_integer(param
)) {
366 int64_t value
= bt_value_signed_integer_get(param
);
369 * Just convert it to a temporary string to handle
370 * everything the same way.
372 sprintf(tmp_arg
, "%" PRId64
, value
);
374 } else if (bt_value_is_string(param
)) {
375 arg
= bt_value_string_get(param
);
377 BT_COMP_LOGE("`%s` parameter must be an integer or a string value.",
383 ret
= set_bound_from_str(trimmer_comp
, arg
, bound
, is_gmt
);
390 int validate_trimmer_bounds(struct trimmer_comp
*trimmer_comp
,
391 struct trimmer_bound
*begin
, struct trimmer_bound
*end
)
395 BT_ASSERT(begin
->is_set
);
396 BT_ASSERT(end
->is_set
);
398 if (!begin
->is_infinite
&& !end
->is_infinite
&&
399 begin
->ns_from_origin
> end
->ns_from_origin
) {
400 BT_COMP_LOGE("Trimming time range's beginning time is greater than end time: "
401 "begin-ns-from-origin=%" PRId64
", "
402 "end-ns-from-origin=%" PRId64
,
403 begin
->ns_from_origin
,
404 end
->ns_from_origin
);
409 if (!begin
->is_infinite
&& begin
->ns_from_origin
== INT64_MIN
) {
410 BT_COMP_LOGE("Invalid trimming time range's beginning time: "
411 "ns-from-origin=%" PRId64
,
412 begin
->ns_from_origin
);
417 if (!end
->is_infinite
&& end
->ns_from_origin
== INT64_MIN
) {
418 BT_COMP_LOGE("Invalid trimming time range's end time: "
419 "ns-from-origin=%" PRId64
,
420 end
->ns_from_origin
);
430 int init_trimmer_comp_from_params(struct trimmer_comp
*trimmer_comp
,
431 const bt_value
*params
)
433 const bt_value
*value
;
437 value
= bt_value_map_borrow_entry_value_const(params
, "gmt");
439 trimmer_comp
->is_gmt
= (bool) bt_value_bool_get(value
);
442 value
= bt_value_map_borrow_entry_value_const(params
, "begin");
444 if (set_bound_from_param(trimmer_comp
, "begin", value
,
445 &trimmer_comp
->begin
, trimmer_comp
->is_gmt
)) {
446 /* set_bound_from_param() logs errors */
447 ret
= BT_SELF_COMPONENT_STATUS_ERROR
;
451 trimmer_comp
->begin
.is_infinite
= true;
452 trimmer_comp
->begin
.is_set
= true;
455 value
= bt_value_map_borrow_entry_value_const(params
, "end");
457 if (set_bound_from_param(trimmer_comp
, "end", value
,
458 &trimmer_comp
->end
, trimmer_comp
->is_gmt
)) {
459 /* set_bound_from_param() logs errors */
460 ret
= BT_SELF_COMPONENT_STATUS_ERROR
;
464 trimmer_comp
->end
.is_infinite
= true;
465 trimmer_comp
->end
.is_set
= true;
469 if (trimmer_comp
->begin
.is_set
&& trimmer_comp
->end
.is_set
) {
470 /* validate_trimmer_bounds() logs errors */
471 ret
= validate_trimmer_bounds(trimmer_comp
,
472 &trimmer_comp
->begin
, &trimmer_comp
->end
);
478 bt_self_component_status
trimmer_init(bt_self_component_filter
*self_comp_flt
,
479 const bt_value
*params
, void *init_data
)
482 bt_self_component_status status
;
483 struct trimmer_comp
*trimmer_comp
= create_trimmer_comp();
484 bt_self_component
*self_comp
=
485 bt_self_component_filter_as_self_component(self_comp_flt
);
487 status
= BT_SELF_COMPONENT_STATUS_NOMEM
;
491 trimmer_comp
->log_level
= bt_component_get_logging_level(
492 bt_self_component_as_component(self_comp
));
493 trimmer_comp
->self_comp
= self_comp
;
494 status
= bt_self_component_filter_add_input_port(
495 self_comp_flt
, in_port_name
, NULL
, NULL
);
496 if (status
!= BT_SELF_COMPONENT_STATUS_OK
) {
500 status
= bt_self_component_filter_add_output_port(
501 self_comp_flt
, "out", NULL
, NULL
);
502 if (status
!= BT_SELF_COMPONENT_STATUS_OK
) {
506 ret
= init_trimmer_comp_from_params(trimmer_comp
, params
);
508 status
= BT_SELF_COMPONENT_STATUS_ERROR
;
512 bt_self_component_set_data(self_comp
, trimmer_comp
);
516 if (status
== BT_SELF_COMPONENT_STATUS_OK
) {
517 status
= BT_SELF_COMPONENT_STATUS_ERROR
;
521 destroy_trimmer_comp(trimmer_comp
);
529 void destroy_trimmer_iterator(struct trimmer_iterator
*trimmer_it
)
531 BT_ASSERT(trimmer_it
);
532 bt_self_component_port_input_message_iterator_put_ref(
533 trimmer_it
->upstream_iter
);
535 if (trimmer_it
->output_messages
) {
536 g_queue_free(trimmer_it
->output_messages
);
539 if (trimmer_it
->stream_states
) {
540 g_hash_table_destroy(trimmer_it
->stream_states
);
547 void destroy_trimmer_iterator_stream_state(
548 struct trimmer_iterator_stream_state
*sstate
)
551 BT_PACKET_PUT_REF_AND_RESET(sstate
->cur_packet
);
552 BT_MESSAGE_PUT_REF_AND_RESET(sstate
->stream_beginning_msg
);
557 bt_self_message_iterator_status
trimmer_msg_iter_init(
558 bt_self_message_iterator
*self_msg_iter
,
559 bt_self_component_filter
*self_comp
,
560 bt_self_component_port_output
*port
)
562 bt_self_message_iterator_status status
=
563 BT_SELF_MESSAGE_ITERATOR_STATUS_OK
;
564 struct trimmer_iterator
*trimmer_it
;
566 trimmer_it
= g_new0(struct trimmer_iterator
, 1);
568 status
= BT_SELF_MESSAGE_ITERATOR_STATUS_NOMEM
;
572 trimmer_it
->trimmer_comp
= bt_self_component_get_data(
573 bt_self_component_filter_as_self_component(self_comp
));
574 BT_ASSERT(trimmer_it
->trimmer_comp
);
576 if (trimmer_it
->trimmer_comp
->begin
.is_set
&&
577 trimmer_it
->trimmer_comp
->end
.is_set
) {
579 * Both trimming time range's bounds are set, so skip
581 * `TRIMMER_ITERATOR_STATE_SET_BOUNDS_NS_FROM_ORIGIN`
584 trimmer_it
->state
= TRIMMER_ITERATOR_STATE_SEEK_INITIALLY
;
587 trimmer_it
->begin
= trimmer_it
->trimmer_comp
->begin
;
588 trimmer_it
->end
= trimmer_it
->trimmer_comp
->end
;
589 trimmer_it
->upstream_iter
=
590 bt_self_component_port_input_message_iterator_create(
591 bt_self_component_filter_borrow_input_port_by_name(
592 self_comp
, in_port_name
));
593 if (!trimmer_it
->upstream_iter
) {
594 status
= BT_SELF_MESSAGE_ITERATOR_STATUS_ERROR
;
598 trimmer_it
->output_messages
= g_queue_new();
599 if (!trimmer_it
->output_messages
) {
600 status
= BT_SELF_MESSAGE_ITERATOR_STATUS_NOMEM
;
604 trimmer_it
->stream_states
= g_hash_table_new_full(g_direct_hash
,
605 g_direct_equal
, NULL
,
606 (GDestroyNotify
) destroy_trimmer_iterator_stream_state
);
607 if (!trimmer_it
->stream_states
) {
608 status
= BT_SELF_MESSAGE_ITERATOR_STATUS_NOMEM
;
612 trimmer_it
->self_msg_iter
= self_msg_iter
;
613 bt_self_message_iterator_set_data(self_msg_iter
, trimmer_it
);
616 if (status
!= BT_SELF_MESSAGE_ITERATOR_STATUS_OK
&& trimmer_it
) {
617 destroy_trimmer_iterator(trimmer_it
);
624 int get_msg_ns_from_origin(const bt_message
*msg
, int64_t *ns_from_origin
,
627 const bt_clock_class
*clock_class
= NULL
;
628 const bt_clock_snapshot
*clock_snapshot
= NULL
;
629 bt_message_stream_activity_clock_snapshot_state sa_cs_state
;
633 BT_ASSERT(ns_from_origin
);
636 switch (bt_message_get_type(msg
)) {
637 case BT_MESSAGE_TYPE_EVENT
:
639 bt_message_event_borrow_stream_class_default_clock_class_const(
641 if (G_UNLIKELY(!clock_class
)) {
645 clock_snapshot
= bt_message_event_borrow_default_clock_snapshot_const(
648 case BT_MESSAGE_TYPE_PACKET_BEGINNING
:
650 bt_message_packet_beginning_borrow_stream_class_default_clock_class_const(
652 if (G_UNLIKELY(!clock_class
)) {
656 clock_snapshot
= bt_message_packet_beginning_borrow_default_clock_snapshot_const(
659 case BT_MESSAGE_TYPE_PACKET_END
:
661 bt_message_packet_end_borrow_stream_class_default_clock_class_const(
663 if (G_UNLIKELY(!clock_class
)) {
667 clock_snapshot
= bt_message_packet_end_borrow_default_clock_snapshot_const(
670 case BT_MESSAGE_TYPE_DISCARDED_EVENTS
:
672 bt_message_discarded_events_borrow_stream_class_default_clock_class_const(
674 if (G_UNLIKELY(!clock_class
)) {
678 clock_snapshot
= bt_message_discarded_events_borrow_beginning_default_clock_snapshot_const(
681 case BT_MESSAGE_TYPE_DISCARDED_PACKETS
:
683 bt_message_discarded_packets_borrow_stream_class_default_clock_class_const(
685 if (G_UNLIKELY(!clock_class
)) {
689 clock_snapshot
= bt_message_discarded_packets_borrow_beginning_default_clock_snapshot_const(
692 case BT_MESSAGE_TYPE_STREAM_ACTIVITY_BEGINNING
:
694 bt_message_stream_activity_beginning_borrow_stream_class_default_clock_class_const(
696 if (G_UNLIKELY(!clock_class
)) {
700 sa_cs_state
= bt_message_stream_activity_beginning_borrow_default_clock_snapshot_const(
701 msg
, &clock_snapshot
);
702 if (sa_cs_state
== BT_MESSAGE_STREAM_ACTIVITY_CLOCK_SNAPSHOT_STATE_UNKNOWN
||
703 sa_cs_state
== BT_MESSAGE_STREAM_ACTIVITY_CLOCK_SNAPSHOT_STATE_INFINITE
) {
704 /* Lowest possible time to always include them */
705 *ns_from_origin
= INT64_MIN
;
706 goto no_clock_snapshot
;
710 case BT_MESSAGE_TYPE_STREAM_ACTIVITY_END
:
712 bt_message_stream_activity_end_borrow_stream_class_default_clock_class_const(
714 if (G_UNLIKELY(!clock_class
)) {
718 sa_cs_state
= bt_message_stream_activity_end_borrow_default_clock_snapshot_const(
719 msg
, &clock_snapshot
);
720 if (sa_cs_state
== BT_MESSAGE_STREAM_ACTIVITY_CLOCK_SNAPSHOT_STATE_UNKNOWN
) {
721 /* Lowest time to always include it */
722 *ns_from_origin
= INT64_MIN
;
723 goto no_clock_snapshot
;
724 } else if (sa_cs_state
== BT_MESSAGE_STREAM_ACTIVITY_CLOCK_SNAPSHOT_STATE_INFINITE
) {
725 /* Greatest time to always exclude it */
726 *ns_from_origin
= INT64_MAX
;
727 goto no_clock_snapshot
;
731 case BT_MESSAGE_TYPE_MESSAGE_ITERATOR_INACTIVITY
:
733 bt_message_message_iterator_inactivity_borrow_default_clock_snapshot_const(
737 goto no_clock_snapshot
;
740 ret
= bt_clock_snapshot_get_ns_from_origin(clock_snapshot
,
742 if (G_UNLIKELY(ret
)) {
760 void put_messages(bt_message_array_const msgs
, uint64_t count
)
764 for (i
= 0; i
< count
; i
++) {
765 BT_MESSAGE_PUT_REF_AND_RESET(msgs
[i
]);
770 int set_trimmer_iterator_bound(struct trimmer_iterator
*trimmer_it
,
771 struct trimmer_bound
*bound
, int64_t ns_from_origin
,
774 struct trimmer_comp
*trimmer_comp
= trimmer_it
->trimmer_comp
;
776 time_t time_seconds
= (time_t) (ns_from_origin
/ NS_PER_S
);
779 BT_ASSERT(!bound
->is_set
);
782 /* We only need to extract the date from this time */
784 bt_gmtime_r(&time_seconds
, &tm
);
786 bt_localtime_r(&time_seconds
, &tm
);
790 BT_COMP_LOGE_ERRNO("Cannot convert timestamp to date and time",
791 "ts=%" PRId64
, (int64_t) time_seconds
);
796 ret
= set_bound_ns_from_origin(bound
, tm
.tm_year
+ 1900, tm
.tm_mon
+ 1,
797 tm
.tm_mday
, bound
->time
.hour
, bound
->time
.minute
,
798 bound
->time
.second
, bound
->time
.ns
, is_gmt
);
805 bt_self_message_iterator_status
state_set_trimmer_iterator_bounds(
806 struct trimmer_iterator
*trimmer_it
)
808 bt_message_iterator_status upstream_iter_status
=
809 BT_MESSAGE_ITERATOR_STATUS_OK
;
810 struct trimmer_comp
*trimmer_comp
= trimmer_it
->trimmer_comp
;
811 bt_message_array_const msgs
;
813 int64_t ns_from_origin
= INT64_MIN
;
817 BT_ASSERT(!trimmer_it
->begin
.is_set
||
818 !trimmer_it
->end
.is_set
);
821 upstream_iter_status
=
822 bt_self_component_port_input_message_iterator_next(
823 trimmer_it
->upstream_iter
, &msgs
, &count
);
824 if (upstream_iter_status
!= BT_MESSAGE_ITERATOR_STATUS_OK
) {
828 for (i
= 0; i
< count
; i
++) {
829 const bt_message
*msg
= msgs
[i
];
833 ret
= get_msg_ns_from_origin(msg
, &ns_from_origin
,
843 BT_ASSERT(ns_from_origin
!= INT64_MIN
&&
844 ns_from_origin
!= INT64_MAX
);
845 put_messages(msgs
, count
);
849 put_messages(msgs
, count
);
853 if (!trimmer_it
->begin
.is_set
) {
854 BT_ASSERT(!trimmer_it
->begin
.is_infinite
);
855 ret
= set_trimmer_iterator_bound(trimmer_it
, &trimmer_it
->begin
,
856 ns_from_origin
, trimmer_comp
->is_gmt
);
862 if (!trimmer_it
->end
.is_set
) {
863 BT_ASSERT(!trimmer_it
->end
.is_infinite
);
864 ret
= set_trimmer_iterator_bound(trimmer_it
, &trimmer_it
->end
,
865 ns_from_origin
, trimmer_comp
->is_gmt
);
871 ret
= validate_trimmer_bounds(trimmer_it
->trimmer_comp
,
872 &trimmer_it
->begin
, &trimmer_it
->end
);
880 put_messages(msgs
, count
);
881 upstream_iter_status
= BT_MESSAGE_ITERATOR_STATUS_ERROR
;
884 return (int) upstream_iter_status
;
888 bt_self_message_iterator_status
state_seek_initially(
889 struct trimmer_iterator
*trimmer_it
)
891 struct trimmer_comp
*trimmer_comp
= trimmer_it
->trimmer_comp
;
892 bt_self_message_iterator_status status
=
893 BT_SELF_MESSAGE_ITERATOR_STATUS_OK
;
895 BT_ASSERT(trimmer_it
->begin
.is_set
);
897 if (trimmer_it
->begin
.is_infinite
) {
898 if (!bt_self_component_port_input_message_iterator_can_seek_beginning(
899 trimmer_it
->upstream_iter
)) {
900 BT_COMP_LOGE_STR("Cannot make upstream message iterator initially seek its beginning.");
901 status
= BT_SELF_MESSAGE_ITERATOR_STATUS_ERROR
;
905 status
= (int) bt_self_component_port_input_message_iterator_seek_beginning(
906 trimmer_it
->upstream_iter
);
908 if (!bt_self_component_port_input_message_iterator_can_seek_ns_from_origin(
909 trimmer_it
->upstream_iter
,
910 trimmer_it
->begin
.ns_from_origin
)) {
911 BT_COMP_LOGE("Cannot make upstream message iterator initially seek: "
912 "seek-ns-from-origin=%" PRId64
,
913 trimmer_it
->begin
.ns_from_origin
);
914 status
= BT_SELF_MESSAGE_ITERATOR_STATUS_ERROR
;
918 status
= (int) bt_self_component_port_input_message_iterator_seek_ns_from_origin(
919 trimmer_it
->upstream_iter
, trimmer_it
->begin
.ns_from_origin
);
922 if (status
== BT_SELF_MESSAGE_ITERATOR_STATUS_OK
) {
923 trimmer_it
->state
= TRIMMER_ITERATOR_STATE_TRIM
;
931 void push_message(struct trimmer_iterator
*trimmer_it
, const bt_message
*msg
)
933 g_queue_push_head(trimmer_it
->output_messages
, (void *) msg
);
937 const bt_message
*pop_message(struct trimmer_iterator
*trimmer_it
)
939 return g_queue_pop_tail(trimmer_it
->output_messages
);
943 int clock_raw_value_from_ns_from_origin(const bt_clock_class
*clock_class
,
944 int64_t ns_from_origin
, uint64_t *raw_value
)
948 uint64_t cc_offset_cycles
;
951 bt_clock_class_get_offset(clock_class
, &cc_offset_s
, &cc_offset_cycles
);
952 cc_freq
= bt_clock_class_get_frequency(clock_class
);
953 return bt_common_clock_value_from_ns_from_origin(cc_offset_s
,
954 cc_offset_cycles
, cc_freq
, ns_from_origin
, raw_value
);
958 bt_self_message_iterator_status
end_stream(struct trimmer_iterator
*trimmer_it
,
959 struct trimmer_iterator_stream_state
*sstate
)
961 bt_self_message_iterator_status status
=
962 BT_SELF_MESSAGE_ITERATOR_STATUS_OK
;
964 const bt_clock_class
*clock_class
;
966 bt_message
*msg
= NULL
;
968 BT_ASSERT(!trimmer_it
->end
.is_infinite
);
970 if (!sstate
->stream
) {
974 if (sstate
->cur_packet
) {
976 * The last message could not have been a stream
977 * activity end message if we have a current packet.
979 BT_ASSERT(!sstate
->last_msg_is_stream_activity_end
);
982 * Create and push a packet end message, making its time
983 * the trimming range's end time.
985 clock_class
= bt_stream_class_borrow_default_clock_class_const(
986 bt_stream_borrow_class_const(sstate
->stream
));
987 BT_ASSERT(clock_class
);
988 ret
= clock_raw_value_from_ns_from_origin(clock_class
,
989 trimmer_it
->end
.ns_from_origin
, &raw_value
);
991 status
= BT_SELF_MESSAGE_ITERATOR_STATUS_ERROR
;
995 msg
= bt_message_packet_end_create_with_default_clock_snapshot(
996 trimmer_it
->self_msg_iter
, sstate
->cur_packet
,
999 status
= BT_SELF_MESSAGE_ITERATOR_STATUS_NOMEM
;
1003 push_message(trimmer_it
, msg
);
1005 BT_PACKET_PUT_REF_AND_RESET(sstate
->cur_packet
);
1008 * Because we generated a packet end message, set the
1009 * stream activity end message's time to use to the
1010 * trimming range's end time (this packet end message's
1013 sstate
->stream_act_end_ns_from_origin
=
1014 trimmer_it
->end
.ns_from_origin
;
1017 if (!sstate
->last_msg_is_stream_activity_end
) {
1018 /* Create and push a stream activity end message */
1019 msg
= bt_message_stream_activity_end_create(
1020 trimmer_it
->self_msg_iter
, sstate
->stream
);
1022 status
= BT_SELF_MESSAGE_ITERATOR_STATUS_NOMEM
;
1026 clock_class
= bt_stream_class_borrow_default_clock_class_const(
1027 bt_stream_borrow_class_const(sstate
->stream
));
1028 BT_ASSERT(clock_class
);
1030 if (sstate
->stream_act_end_ns_from_origin
== INT64_MIN
) {
1032 * We received at least what is necessary to
1033 * have a stream state (stream beginning and
1034 * stream activity beginning messages), but
1035 * nothing else: use the trimmer range's end
1038 sstate
->stream_act_end_ns_from_origin
=
1039 trimmer_it
->end
.ns_from_origin
;
1042 ret
= clock_raw_value_from_ns_from_origin(clock_class
,
1043 sstate
->stream_act_end_ns_from_origin
, &raw_value
);
1045 status
= BT_SELF_MESSAGE_ITERATOR_STATUS_ERROR
;
1049 bt_message_stream_activity_end_set_default_clock_snapshot(
1051 push_message(trimmer_it
, msg
);
1055 /* Create and push a stream end message */
1056 msg
= bt_message_stream_end_create(trimmer_it
->self_msg_iter
,
1059 status
= BT_SELF_MESSAGE_ITERATOR_STATUS_NOMEM
;
1063 push_message(trimmer_it
, msg
);
1067 * Just to make sure that we don't use this stream state again
1068 * in the future without an obvious error.
1070 sstate
->stream
= NULL
;
1073 bt_message_put_ref(msg
);
1078 bt_self_message_iterator_status
end_iterator_streams(
1079 struct trimmer_iterator
*trimmer_it
)
1081 bt_self_message_iterator_status status
=
1082 BT_SELF_MESSAGE_ITERATOR_STATUS_OK
;
1083 GHashTableIter iter
;
1084 gpointer key
, sstate
;
1086 if (trimmer_it
->end
.is_infinite
) {
1088 * An infinite trimming range's end time guarantees that
1089 * we received (and pushed) all the appropriate end
1096 * End each stream and then remove them from the hash table of
1097 * stream states to release unneeded references.
1099 g_hash_table_iter_init(&iter
, trimmer_it
->stream_states
);
1101 while (g_hash_table_iter_next(&iter
, &key
, &sstate
)) {
1102 status
= end_stream(trimmer_it
, sstate
);
1109 g_hash_table_remove_all(trimmer_it
->stream_states
);
1116 bt_self_message_iterator_status
create_stream_beginning_activity_message(
1117 struct trimmer_iterator
*trimmer_it
,
1118 const bt_stream
*stream
,
1119 const bt_clock_class
*clock_class
, bt_message
**msg
)
1121 bt_message
*local_msg
;
1122 bt_self_message_iterator_status status
=
1123 BT_SELF_MESSAGE_ITERATOR_STATUS_OK
;
1126 BT_ASSERT(!trimmer_it
->begin
.is_infinite
);
1128 local_msg
= bt_message_stream_activity_beginning_create(
1129 trimmer_it
->self_msg_iter
, stream
);
1131 status
= BT_SELF_MESSAGE_ITERATOR_STATUS_NOMEM
;
1139 ret
= clock_raw_value_from_ns_from_origin(clock_class
,
1140 trimmer_it
->begin
.ns_from_origin
, &raw_value
);
1142 status
= BT_SELF_MESSAGE_ITERATOR_STATUS_ERROR
;
1143 bt_message_put_ref(local_msg
);
1147 bt_message_stream_activity_beginning_set_default_clock_snapshot(
1148 local_msg
, raw_value
);
1151 BT_MESSAGE_MOVE_REF(*msg
, local_msg
);
1158 * Makes sure to initialize a stream state, pushing the appropriate
1161 * `stream_act_beginning_msg` is an initial stream activity beginning
1162 * message to potentially use, depending on its clock snapshot state.
1163 * This function consumes `stream_act_beginning_msg` unconditionally.
1166 bt_self_message_iterator_status
ensure_stream_state_is_inited(
1167 struct trimmer_iterator
*trimmer_it
,
1168 struct trimmer_iterator_stream_state
*sstate
,
1169 const bt_message
*stream_act_beginning_msg
)
1171 bt_self_message_iterator_status status
=
1172 BT_SELF_MESSAGE_ITERATOR_STATUS_OK
;
1173 bt_message
*new_msg
= NULL
;
1174 const bt_clock_class
*clock_class
=
1175 bt_stream_class_borrow_default_clock_class_const(
1176 bt_stream_borrow_class_const(sstate
->stream
));
1178 BT_ASSERT(!sstate
->inited
);
1180 if (!sstate
->stream_beginning_msg
) {
1181 /* No initial stream beginning message: create one */
1182 sstate
->stream_beginning_msg
=
1183 bt_message_stream_beginning_create(
1184 trimmer_it
->self_msg_iter
, sstate
->stream
);
1185 if (!sstate
->stream_beginning_msg
) {
1186 status
= BT_SELF_MESSAGE_ITERATOR_STATUS_NOMEM
;
1191 /* Push initial stream beginning message */
1192 BT_ASSERT(sstate
->stream_beginning_msg
);
1193 push_message(trimmer_it
, sstate
->stream_beginning_msg
);
1194 sstate
->stream_beginning_msg
= NULL
;
1196 if (stream_act_beginning_msg
) {
1198 * Initial stream activity beginning message exists: if
1199 * its time is -inf, then create and push a new one
1200 * having the trimming range's beginning time. Otherwise
1201 * push it as is (known and unknown).
1203 const bt_clock_snapshot
*cs
;
1204 bt_message_stream_activity_clock_snapshot_state sa_cs_state
;
1206 sa_cs_state
= bt_message_stream_activity_beginning_borrow_default_clock_snapshot_const(
1207 stream_act_beginning_msg
, &cs
);
1208 if (sa_cs_state
== BT_MESSAGE_STREAM_ACTIVITY_CLOCK_SNAPSHOT_STATE_INFINITE
&&
1209 !trimmer_it
->begin
.is_infinite
) {
1211 * -inf time: use trimming range's beginning
1212 * time (which is not -inf).
1214 status
= create_stream_beginning_activity_message(
1215 trimmer_it
, sstate
->stream
, clock_class
,
1217 if (status
!= BT_SELF_MESSAGE_ITERATOR_STATUS_OK
) {
1221 push_message(trimmer_it
, new_msg
);
1224 /* Known/unknown: push as is */
1225 push_message(trimmer_it
, stream_act_beginning_msg
);
1226 stream_act_beginning_msg
= NULL
;
1229 BT_ASSERT(!trimmer_it
->begin
.is_infinite
);
1232 * No stream beginning activity message: create and push
1235 status
= create_stream_beginning_activity_message(
1236 trimmer_it
, sstate
->stream
, clock_class
, &new_msg
);
1237 if (status
!= BT_SELF_MESSAGE_ITERATOR_STATUS_OK
) {
1241 push_message(trimmer_it
, new_msg
);
1245 sstate
->inited
= true;
1248 bt_message_put_ref(new_msg
);
1249 bt_message_put_ref(stream_act_beginning_msg
);
1254 bt_self_message_iterator_status
ensure_cur_packet_exists(
1255 struct trimmer_iterator
*trimmer_it
,
1256 struct trimmer_iterator_stream_state
*sstate
,
1257 const bt_packet
*packet
)
1259 bt_self_message_iterator_status status
=
1260 BT_SELF_MESSAGE_ITERATOR_STATUS_OK
;
1262 const bt_clock_class
*clock_class
=
1263 bt_stream_class_borrow_default_clock_class_const(
1264 bt_stream_borrow_class_const(sstate
->stream
));
1265 bt_message
*msg
= NULL
;
1268 BT_ASSERT(!trimmer_it
->begin
.is_infinite
);
1269 BT_ASSERT(!sstate
->cur_packet
);
1272 * Create and push an initial packet beginning message,
1273 * making its time the trimming range's beginning time.
1275 ret
= clock_raw_value_from_ns_from_origin(clock_class
,
1276 trimmer_it
->begin
.ns_from_origin
, &raw_value
);
1278 status
= BT_SELF_MESSAGE_ITERATOR_STATUS_ERROR
;
1282 msg
= bt_message_packet_beginning_create_with_default_clock_snapshot(
1283 trimmer_it
->self_msg_iter
, packet
, raw_value
);
1285 status
= BT_SELF_MESSAGE_ITERATOR_STATUS_NOMEM
;
1289 push_message(trimmer_it
, msg
);
1292 /* Set packet as this stream's current packet */
1293 sstate
->cur_packet
= packet
;
1294 bt_packet_get_ref(sstate
->cur_packet
);
1297 bt_message_put_ref(msg
);
1302 * Handles a message which is associated to a given stream state. This
1303 * _could_ make the iterator's output message queue grow; this could
1304 * also consume the message without pushing anything to this queue, only
1305 * modifying the stream state.
1307 * This function consumes the `msg` reference, _whatever the outcome_.
1309 * `ns_from_origin` is the message's time, as given by
1310 * get_msg_ns_from_origin().
1312 * This function sets `reached_end` if handling this message made the
1313 * iterator reach the end of the trimming range. Note that the output
1314 * message queue could contain messages even if this function sets
1318 bt_self_message_iterator_status
handle_message_with_stream_state(
1319 struct trimmer_iterator
*trimmer_it
, const bt_message
*msg
,
1320 struct trimmer_iterator_stream_state
*sstate
,
1321 int64_t ns_from_origin
, bool *reached_end
)
1323 bt_self_message_iterator_status status
=
1324 BT_SELF_MESSAGE_ITERATOR_STATUS_OK
;
1325 bt_message_type msg_type
= bt_message_get_type(msg
);
1329 case BT_MESSAGE_TYPE_EVENT
:
1330 if (G_UNLIKELY(!trimmer_it
->end
.is_infinite
&&
1331 ns_from_origin
> trimmer_it
->end
.ns_from_origin
)) {
1332 status
= end_iterator_streams(trimmer_it
);
1333 *reached_end
= true;
1337 if (G_UNLIKELY(!sstate
->inited
)) {
1338 status
= ensure_stream_state_is_inited(trimmer_it
,
1340 if (status
!= BT_SELF_MESSAGE_ITERATOR_STATUS_OK
) {
1345 if (G_UNLIKELY(!sstate
->cur_packet
)) {
1346 const bt_event
*event
=
1347 bt_message_event_borrow_event_const(msg
);
1348 const bt_packet
*packet
= bt_event_borrow_packet_const(
1351 status
= ensure_cur_packet_exists(trimmer_it
, sstate
,
1353 if (status
!= BT_SELF_MESSAGE_ITERATOR_STATUS_OK
) {
1358 BT_ASSERT(sstate
->cur_packet
);
1359 push_message(trimmer_it
, msg
);
1362 case BT_MESSAGE_TYPE_PACKET_BEGINNING
:
1363 if (G_UNLIKELY(!trimmer_it
->end
.is_infinite
&&
1364 ns_from_origin
> trimmer_it
->end
.ns_from_origin
)) {
1365 status
= end_iterator_streams(trimmer_it
);
1366 *reached_end
= true;
1370 if (G_UNLIKELY(!sstate
->inited
)) {
1371 status
= ensure_stream_state_is_inited(trimmer_it
,
1373 if (status
!= BT_SELF_MESSAGE_ITERATOR_STATUS_OK
) {
1378 BT_ASSERT(!sstate
->cur_packet
);
1379 sstate
->cur_packet
=
1380 bt_message_packet_beginning_borrow_packet_const(msg
);
1381 bt_packet_get_ref(sstate
->cur_packet
);
1382 push_message(trimmer_it
, msg
);
1385 case BT_MESSAGE_TYPE_PACKET_END
:
1386 sstate
->stream_act_end_ns_from_origin
= ns_from_origin
;
1388 if (G_UNLIKELY(!trimmer_it
->end
.is_infinite
&&
1389 ns_from_origin
> trimmer_it
->end
.ns_from_origin
)) {
1390 status
= end_iterator_streams(trimmer_it
);
1391 *reached_end
= true;
1395 if (G_UNLIKELY(!sstate
->inited
)) {
1396 status
= ensure_stream_state_is_inited(trimmer_it
,
1398 if (status
!= BT_SELF_MESSAGE_ITERATOR_STATUS_OK
) {
1403 if (G_UNLIKELY(!sstate
->cur_packet
)) {
1404 const bt_packet
*packet
=
1405 bt_message_packet_end_borrow_packet_const(msg
);
1407 status
= ensure_cur_packet_exists(trimmer_it
, sstate
,
1409 if (status
!= BT_SELF_MESSAGE_ITERATOR_STATUS_OK
) {
1414 BT_ASSERT(sstate
->cur_packet
);
1415 BT_PACKET_PUT_REF_AND_RESET(sstate
->cur_packet
);
1416 push_message(trimmer_it
, msg
);
1419 case BT_MESSAGE_TYPE_DISCARDED_EVENTS
:
1420 case BT_MESSAGE_TYPE_DISCARDED_PACKETS
:
1423 * `ns_from_origin` is the message's time range's
1424 * beginning time here.
1426 int64_t end_ns_from_origin
;
1427 const bt_clock_snapshot
*end_cs
;
1429 if (bt_message_get_type(msg
) ==
1430 BT_MESSAGE_TYPE_DISCARDED_EVENTS
) {
1432 * Safe to ignore the return value because we
1433 * know there's a default clock and it's always
1436 end_cs
= bt_message_discarded_events_borrow_end_default_clock_snapshot_const(
1440 * Safe to ignore the return value because we
1441 * know there's a default clock and it's always
1444 end_cs
= bt_message_discarded_packets_borrow_end_default_clock_snapshot_const(
1448 if (bt_clock_snapshot_get_ns_from_origin(end_cs
,
1449 &end_ns_from_origin
)) {
1450 status
= BT_SELF_MESSAGE_ITERATOR_STATUS_ERROR
;
1454 sstate
->stream_act_end_ns_from_origin
= end_ns_from_origin
;
1456 if (!trimmer_it
->end
.is_infinite
&&
1457 ns_from_origin
> trimmer_it
->end
.ns_from_origin
) {
1458 status
= end_iterator_streams(trimmer_it
);
1459 *reached_end
= true;
1463 if (!trimmer_it
->end
.is_infinite
&&
1464 end_ns_from_origin
> trimmer_it
->end
.ns_from_origin
) {
1466 * This message's end time is outside the
1467 * trimming time range: replace it with a new
1468 * message having an end time equal to the
1469 * trimming time range's end and without a
1472 const bt_clock_class
*clock_class
=
1473 bt_clock_snapshot_borrow_clock_class_const(
1475 const bt_clock_snapshot
*begin_cs
;
1476 bt_message
*new_msg
;
1477 uint64_t end_raw_value
;
1479 ret
= clock_raw_value_from_ns_from_origin(clock_class
,
1480 trimmer_it
->end
.ns_from_origin
, &end_raw_value
);
1482 status
= BT_SELF_MESSAGE_ITERATOR_STATUS_ERROR
;
1486 if (msg_type
== BT_MESSAGE_TYPE_DISCARDED_EVENTS
) {
1487 begin_cs
= bt_message_discarded_events_borrow_beginning_default_clock_snapshot_const(
1489 new_msg
= bt_message_discarded_events_create_with_default_clock_snapshots(
1490 trimmer_it
->self_msg_iter
,
1492 bt_clock_snapshot_get_value(begin_cs
),
1495 begin_cs
= bt_message_discarded_packets_borrow_beginning_default_clock_snapshot_const(
1497 new_msg
= bt_message_discarded_packets_create_with_default_clock_snapshots(
1498 trimmer_it
->self_msg_iter
,
1500 bt_clock_snapshot_get_value(begin_cs
),
1505 status
= BT_SELF_MESSAGE_ITERATOR_STATUS_NOMEM
;
1509 /* Replace the original message */
1510 BT_MESSAGE_MOVE_REF(msg
, new_msg
);
1513 if (G_UNLIKELY(!sstate
->inited
)) {
1514 status
= ensure_stream_state_is_inited(trimmer_it
,
1516 if (status
!= BT_SELF_MESSAGE_ITERATOR_STATUS_OK
) {
1521 push_message(trimmer_it
, msg
);
1525 case BT_MESSAGE_TYPE_STREAM_ACTIVITY_BEGINNING
:
1526 if (!trimmer_it
->end
.is_infinite
&&
1527 ns_from_origin
> trimmer_it
->end
.ns_from_origin
) {
1529 * This only happens when the message's time is
1530 * known and is greater than the trimming
1531 * range's end time. Unknown and -inf times are
1533 * `trimmer_it->end.ns_from_origin`.
1535 status
= end_iterator_streams(trimmer_it
);
1536 *reached_end
= true;
1540 if (!sstate
->inited
) {
1541 status
= ensure_stream_state_is_inited(trimmer_it
,
1544 if (status
!= BT_SELF_MESSAGE_ITERATOR_STATUS_OK
) {
1548 push_message(trimmer_it
, msg
);
1553 case BT_MESSAGE_TYPE_STREAM_ACTIVITY_END
:
1554 if (trimmer_it
->end
.is_infinite
) {
1555 push_message(trimmer_it
, msg
);
1560 if (ns_from_origin
== INT64_MIN
) {
1561 /* Unknown: push as is if stream state is inited */
1562 if (sstate
->inited
) {
1563 push_message(trimmer_it
, msg
);
1565 sstate
->last_msg_is_stream_activity_end
= true;
1567 } else if (ns_from_origin
== INT64_MAX
) {
1568 /* Infinite: use trimming range's end time */
1569 sstate
->stream_act_end_ns_from_origin
=
1570 trimmer_it
->end
.ns_from_origin
;
1572 /* Known: check if outside of trimming range */
1573 if (ns_from_origin
> trimmer_it
->end
.ns_from_origin
) {
1574 sstate
->stream_act_end_ns_from_origin
=
1575 trimmer_it
->end
.ns_from_origin
;
1576 status
= end_iterator_streams(trimmer_it
);
1577 *reached_end
= true;
1581 if (!sstate
->inited
) {
1583 * First message for this stream is a
1584 * stream activity end: we can't deduce
1585 * anything about the stream activity
1586 * beginning's time, and using this
1587 * message's time would make a useless
1588 * pair of stream activity beginning/end
1589 * with the same time. Just skip this
1590 * message and wait for something
1596 push_message(trimmer_it
, msg
);
1598 sstate
->last_msg_is_stream_activity_end
= true;
1599 sstate
->stream_act_end_ns_from_origin
= ns_from_origin
;
1603 case BT_MESSAGE_TYPE_STREAM_BEGINNING
:
1605 * We don't know what follows at this point, so just
1606 * keep this message until we know what to do with it
1607 * (it will be used in ensure_stream_state_is_inited()).
1609 BT_ASSERT(!sstate
->inited
);
1610 BT_MESSAGE_MOVE_REF(sstate
->stream_beginning_msg
, msg
);
1612 case BT_MESSAGE_TYPE_STREAM_END
:
1613 if (sstate
->inited
) {
1615 * This is the end of an inited stream: end this
1616 * stream if its stream activity end message
1617 * time is not the trimming range's end time
1618 * (which means the final stream activity end
1619 * message had an infinite time). end_stream()
1620 * will generate its own stream end message.
1622 if (trimmer_it
->end
.is_infinite
) {
1623 push_message(trimmer_it
, msg
);
1625 g_hash_table_remove(trimmer_it
->stream_states
,
1627 } else if (sstate
->stream_act_end_ns_from_origin
<
1628 trimmer_it
->end
.ns_from_origin
) {
1629 status
= end_stream(trimmer_it
, sstate
);
1630 if (status
!= BT_SELF_MESSAGE_ITERATOR_STATUS_OK
) {
1634 /* We won't need this stream state again */
1635 g_hash_table_remove(trimmer_it
->stream_states
,
1639 /* We dont't need this stream state anymore */
1640 g_hash_table_remove(trimmer_it
->stream_states
, sstate
->stream
);
1649 /* We release the message's reference whatever the outcome */
1650 bt_message_put_ref(msg
);
1651 return BT_SELF_MESSAGE_ITERATOR_STATUS_OK
;
1655 * Handles an input message. This _could_ make the iterator's output
1656 * message queue grow; this could also consume the message without
1657 * pushing anything to this queue, only modifying the stream state.
1659 * This function consumes the `msg` reference, _whatever the outcome_.
1661 * This function sets `reached_end` if handling this message made the
1662 * iterator reach the end of the trimming range. Note that the output
1663 * message queue could contain messages even if this function sets
1667 bt_self_message_iterator_status
handle_message(
1668 struct trimmer_iterator
*trimmer_it
, const bt_message
*msg
,
1671 bt_self_message_iterator_status status
;
1672 const bt_stream
*stream
= NULL
;
1673 int64_t ns_from_origin
= INT64_MIN
;
1676 struct trimmer_iterator_stream_state
*sstate
= NULL
;
1677 struct trimmer_comp
*trimmer_comp
= trimmer_it
->trimmer_comp
;
1679 /* Find message's associated stream */
1680 switch (bt_message_get_type(msg
)) {
1681 case BT_MESSAGE_TYPE_EVENT
:
1682 stream
= bt_event_borrow_stream_const(
1683 bt_message_event_borrow_event_const(msg
));
1685 case BT_MESSAGE_TYPE_PACKET_BEGINNING
:
1686 stream
= bt_packet_borrow_stream_const(
1687 bt_message_packet_beginning_borrow_packet_const(msg
));
1689 case BT_MESSAGE_TYPE_PACKET_END
:
1690 stream
= bt_packet_borrow_stream_const(
1691 bt_message_packet_end_borrow_packet_const(msg
));
1693 case BT_MESSAGE_TYPE_DISCARDED_EVENTS
:
1694 stream
= bt_message_discarded_events_borrow_stream_const(msg
);
1696 case BT_MESSAGE_TYPE_DISCARDED_PACKETS
:
1697 stream
= bt_message_discarded_packets_borrow_stream_const(msg
);
1699 case BT_MESSAGE_TYPE_STREAM_ACTIVITY_BEGINNING
:
1700 stream
= bt_message_stream_activity_beginning_borrow_stream_const(msg
);
1702 case BT_MESSAGE_TYPE_STREAM_ACTIVITY_END
:
1703 stream
= bt_message_stream_activity_end_borrow_stream_const(msg
);
1705 case BT_MESSAGE_TYPE_STREAM_BEGINNING
:
1706 stream
= bt_message_stream_beginning_borrow_stream_const(msg
);
1708 case BT_MESSAGE_TYPE_STREAM_END
:
1709 stream
= bt_message_stream_end_borrow_stream_const(msg
);
1715 if (G_LIKELY(stream
)) {
1716 /* Find stream state */
1717 sstate
= g_hash_table_lookup(trimmer_it
->stream_states
,
1719 if (G_UNLIKELY(!sstate
)) {
1720 /* No stream state yet: create one now */
1721 const bt_stream_class
*sc
;
1724 * Validate right now that the stream's class
1725 * has a registered default clock class so that
1726 * an existing stream state guarantees existing
1727 * default clock snapshots for its associated
1730 * Also check that clock snapshots are always
1733 sc
= bt_stream_borrow_class_const(stream
);
1734 if (!bt_stream_class_borrow_default_clock_class_const(sc
)) {
1735 BT_COMP_LOGE("Unsupported stream: stream class does "
1736 "not have a default clock class: "
1738 "stream-id=%" PRIu64
", "
1739 "stream-name=\"%s\"",
1740 stream
, bt_stream_get_id(stream
),
1741 bt_stream_get_name(stream
));
1742 status
= BT_SELF_MESSAGE_ITERATOR_STATUS_ERROR
;
1747 * Temporary: make sure packet beginning, packet
1748 * end, discarded events, and discarded packets
1749 * messages have default clock snapshots until
1750 * the support for not having them is
1753 if (!bt_stream_class_packets_have_beginning_default_clock_snapshot(
1755 BT_COMP_LOGE("Unsupported stream: packets have "
1756 "no beginning clock snapshot: "
1758 "stream-id=%" PRIu64
", "
1759 "stream-name=\"%s\"",
1760 stream
, bt_stream_get_id(stream
),
1761 bt_stream_get_name(stream
));
1762 status
= BT_SELF_MESSAGE_ITERATOR_STATUS_ERROR
;
1766 if (!bt_stream_class_packets_have_end_default_clock_snapshot(
1768 BT_COMP_LOGE("Unsupported stream: packets have "
1769 "no end clock snapshot: "
1771 "stream-id=%" PRIu64
", "
1772 "stream-name=\"%s\"",
1773 stream
, bt_stream_get_id(stream
),
1774 bt_stream_get_name(stream
));
1775 status
= BT_SELF_MESSAGE_ITERATOR_STATUS_ERROR
;
1779 if (bt_stream_class_supports_discarded_events(sc
) &&
1780 !bt_stream_class_discarded_events_have_default_clock_snapshots(sc
)) {
1781 BT_COMP_LOGE("Unsupported stream: discarded events "
1782 "have no clock snapshots: "
1784 "stream-id=%" PRIu64
", "
1785 "stream-name=\"%s\"",
1786 stream
, bt_stream_get_id(stream
),
1787 bt_stream_get_name(stream
));
1788 status
= BT_SELF_MESSAGE_ITERATOR_STATUS_ERROR
;
1792 if (bt_stream_class_supports_discarded_packets(sc
) &&
1793 !bt_stream_class_discarded_packets_have_default_clock_snapshots(sc
)) {
1794 BT_COMP_LOGE("Unsupported stream: discarded packets "
1795 "have no clock snapshots: "
1797 "stream-id=%" PRIu64
", "
1798 "stream-name=\"%s\"",
1799 stream
, bt_stream_get_id(stream
),
1800 bt_stream_get_name(stream
));
1801 status
= BT_SELF_MESSAGE_ITERATOR_STATUS_ERROR
;
1805 sstate
= g_new0(struct trimmer_iterator_stream_state
,
1808 status
= BT_SELF_MESSAGE_ITERATOR_STATUS_NOMEM
;
1812 sstate
->stream
= stream
;
1813 sstate
->stream_act_end_ns_from_origin
= INT64_MIN
;
1814 g_hash_table_insert(trimmer_it
->stream_states
,
1815 (void *) stream
, sstate
);
1819 /* Retrieve the message's time */
1820 ret
= get_msg_ns_from_origin(msg
, &ns_from_origin
, &skip
);
1821 if (G_UNLIKELY(ret
)) {
1822 status
= BT_SELF_MESSAGE_ITERATOR_STATUS_ERROR
;
1826 if (G_LIKELY(sstate
)) {
1827 /* Message associated to a stream */
1828 status
= handle_message_with_stream_state(trimmer_it
, msg
,
1829 sstate
, ns_from_origin
, reached_end
);
1832 * handle_message_with_stream_state() unconditionally
1838 * Message not associated to a stream (message iterator
1841 if (G_UNLIKELY(ns_from_origin
> trimmer_it
->end
.ns_from_origin
)) {
1842 BT_MESSAGE_PUT_REF_AND_RESET(msg
);
1843 status
= end_iterator_streams(trimmer_it
);
1844 *reached_end
= true;
1846 push_message(trimmer_it
, msg
);
1847 status
= BT_SELF_MESSAGE_ITERATOR_STATUS_OK
;
1853 /* We release the message's reference whatever the outcome */
1854 bt_message_put_ref(msg
);
1859 void fill_message_array_from_output_messages(
1860 struct trimmer_iterator
*trimmer_it
,
1861 bt_message_array_const msgs
, uint64_t capacity
, uint64_t *count
)
1866 * Move auto-seek messages to the output array (which is this
1867 * iterator's base message array).
1869 while (capacity
> 0 && !g_queue_is_empty(trimmer_it
->output_messages
)) {
1870 msgs
[*count
] = pop_message(trimmer_it
);
1875 BT_ASSERT(*count
> 0);
1879 bt_self_message_iterator_status
state_ending(
1880 struct trimmer_iterator
*trimmer_it
,
1881 bt_message_array_const msgs
, uint64_t capacity
,
1884 bt_self_message_iterator_status status
=
1885 BT_SELF_MESSAGE_ITERATOR_STATUS_OK
;
1887 if (g_queue_is_empty(trimmer_it
->output_messages
)) {
1888 trimmer_it
->state
= TRIMMER_ITERATOR_STATE_ENDED
;
1889 status
= BT_SELF_MESSAGE_ITERATOR_STATUS_END
;
1893 fill_message_array_from_output_messages(trimmer_it
, msgs
,
1901 bt_self_message_iterator_status
state_trim(struct trimmer_iterator
*trimmer_it
,
1902 bt_message_array_const msgs
, uint64_t capacity
,
1905 bt_self_message_iterator_status status
=
1906 BT_SELF_MESSAGE_ITERATOR_STATUS_OK
;
1907 bt_message_array_const my_msgs
;
1910 bool reached_end
= false;
1912 while (g_queue_is_empty(trimmer_it
->output_messages
)) {
1913 status
= (int) bt_self_component_port_input_message_iterator_next(
1914 trimmer_it
->upstream_iter
, &my_msgs
, &my_count
);
1915 if (G_UNLIKELY(status
!= BT_SELF_MESSAGE_ITERATOR_STATUS_OK
)) {
1916 if (status
== BT_SELF_MESSAGE_ITERATOR_STATUS_END
) {
1917 status
= end_iterator_streams(trimmer_it
);
1918 if (status
!= BT_SELF_MESSAGE_ITERATOR_STATUS_OK
) {
1923 TRIMMER_ITERATOR_STATE_ENDING
;
1924 status
= state_ending(trimmer_it
, msgs
,
1931 BT_ASSERT(my_count
> 0);
1933 for (i
= 0; i
< my_count
; i
++) {
1934 status
= handle_message(trimmer_it
, my_msgs
[i
],
1938 * handle_message() unconditionally consumes the
1939 * message reference.
1943 if (G_UNLIKELY(status
!=
1944 BT_SELF_MESSAGE_ITERATOR_STATUS_OK
)) {
1945 put_messages(my_msgs
, my_count
);
1949 if (G_UNLIKELY(reached_end
)) {
1951 * This message's time was passed the
1952 * trimming time range's end time: we
1953 * are done. Their might still be
1954 * messages in the output message queue,
1955 * so move to the "ending" state and
1956 * apply it immediately since
1957 * state_trim() is called within the
1960 put_messages(my_msgs
, my_count
);
1962 TRIMMER_ITERATOR_STATE_ENDING
;
1963 status
= state_ending(trimmer_it
, msgs
,
1971 * There's at least one message in the output message queue:
1972 * move the messages to the output message array.
1974 BT_ASSERT(!g_queue_is_empty(trimmer_it
->output_messages
));
1975 fill_message_array_from_output_messages(trimmer_it
, msgs
,
1983 bt_self_message_iterator_status
trimmer_msg_iter_next(
1984 bt_self_message_iterator
*self_msg_iter
,
1985 bt_message_array_const msgs
, uint64_t capacity
,
1988 struct trimmer_iterator
*trimmer_it
=
1989 bt_self_message_iterator_get_data(self_msg_iter
);
1990 bt_self_message_iterator_status status
=
1991 BT_SELF_MESSAGE_ITERATOR_STATUS_OK
;
1993 BT_ASSERT(trimmer_it
);
1995 if (G_LIKELY(trimmer_it
->state
== TRIMMER_ITERATOR_STATE_TRIM
)) {
1996 status
= state_trim(trimmer_it
, msgs
, capacity
, count
);
1997 if (status
!= BT_SELF_MESSAGE_ITERATOR_STATUS_OK
) {
2001 switch (trimmer_it
->state
) {
2002 case TRIMMER_ITERATOR_STATE_SET_BOUNDS_NS_FROM_ORIGIN
:
2003 status
= state_set_trimmer_iterator_bounds(trimmer_it
);
2004 if (status
!= BT_SELF_MESSAGE_ITERATOR_STATUS_OK
) {
2008 status
= state_seek_initially(trimmer_it
);
2009 if (status
!= BT_SELF_MESSAGE_ITERATOR_STATUS_OK
) {
2013 status
= state_trim(trimmer_it
, msgs
, capacity
, count
);
2014 if (status
!= BT_SELF_MESSAGE_ITERATOR_STATUS_OK
) {
2019 case TRIMMER_ITERATOR_STATE_SEEK_INITIALLY
:
2020 status
= state_seek_initially(trimmer_it
);
2021 if (status
!= BT_SELF_MESSAGE_ITERATOR_STATUS_OK
) {
2025 status
= state_trim(trimmer_it
, msgs
, capacity
, count
);
2026 if (status
!= BT_SELF_MESSAGE_ITERATOR_STATUS_OK
) {
2031 case TRIMMER_ITERATOR_STATE_ENDING
:
2032 status
= state_ending(trimmer_it
, msgs
, capacity
,
2034 if (status
!= BT_SELF_MESSAGE_ITERATOR_STATUS_OK
) {
2039 case TRIMMER_ITERATOR_STATE_ENDED
:
2040 status
= BT_SELF_MESSAGE_ITERATOR_STATUS_END
;
2052 void trimmer_msg_iter_finalize(bt_self_message_iterator
*self_msg_iter
)
2054 struct trimmer_iterator
*trimmer_it
=
2055 bt_self_message_iterator_get_data(self_msg_iter
);
2057 BT_ASSERT(trimmer_it
);
2058 destroy_trimmer_iterator(trimmer_it
);