2 * Copyright 2016 Jérémie Galarneau <jeremie.galarneau@efficios.com>
3 * Copyright 2019 Philippe Proulx <pproulx@efficios.com>
5 * Permission is hereby granted, free of charge, to any person obtaining a copy
6 * of this software and associated documentation files (the "Software"), to deal
7 * in the Software without restriction, including without limitation the rights
8 * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
9 * copies of the Software, and to permit persons to whom the Software is
10 * furnished to do so, subject to the following conditions:
12 * The above copyright notice and this permission notice shall be included in
13 * all copies or substantial portions of the Software.
15 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
16 * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
17 * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
18 * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
19 * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
20 * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
24 #define BT_LOG_OUTPUT_LEVEL (trimmer_comp->log_level)
25 #define BT_LOG_TAG "PLUGIN/FLT.UTILS.TRIMMER"
26 #include "logging/log.h"
28 #include "compat/utc.h"
29 #include "compat/time.h"
30 #include <babeltrace2/babeltrace.h>
31 #include "common/common.h"
32 #include "common/assert.h"
39 #define NS_PER_S INT64_C(1000000000)
41 static const char * const in_port_name
= "in";
44 unsigned int hour
, minute
, second
, ns
;
47 struct trimmer_bound
{
49 * Nanoseconds from origin, valid if `is_set` is set and
50 * `is_infinite` is false.
52 int64_t ns_from_origin
;
54 /* True if this bound's full time (`ns_from_origin`) is set */
58 * True if this bound represents the infinity (negative or
59 * positive depending on which bound it is). If this is true,
60 * then we don't care about `ns_from_origin` above.
65 * This bound's time without the date; this time is used to set
66 * `ns_from_origin` once we know the date.
68 struct trimmer_time time
;
72 struct trimmer_bound begin
, end
;
74 bt_logging_level log_level
;
77 enum trimmer_iterator_state
{
79 * Find the first message's date and set the bounds's times
82 TRIMMER_ITERATOR_STATE_SET_BOUNDS_NS_FROM_ORIGIN
,
85 * Initially seek to the trimming range's beginning time.
87 TRIMMER_ITERATOR_STATE_SEEK_INITIALLY
,
90 * Fill the output message queue for as long as received input
91 * messages are within the trimming time range.
93 TRIMMER_ITERATOR_STATE_TRIM
,
95 /* Flush the remaining messages in the output message queue */
96 TRIMMER_ITERATOR_STATE_ENDING
,
98 /* Trimming operation and message iterator is ended */
99 TRIMMER_ITERATOR_STATE_ENDED
,
102 struct trimmer_iterator
{
104 struct trimmer_comp
*trimmer_comp
;
107 bt_self_message_iterator
*self_msg_iter
;
109 enum trimmer_iterator_state state
;
112 bt_self_component_port_input_message_iterator
*upstream_iter
;
113 struct trimmer_bound begin
, end
;
116 * Queue of `const bt_message *` (owned by the queue).
118 * This is where the trimming operation pushes the messages to
119 * output by this message iterator.
121 GQueue
*output_messages
;
124 * Hash table of `bt_stream *` (weak) to
125 * `struct trimmer_iterator_stream_state *` (owned by the HT).
127 GHashTable
*stream_states
;
130 struct trimmer_iterator_stream_state
{
132 * True if both stream beginning and initial stream activity
133 * beginning messages were pushed for this stream.
138 * True if the last pushed message for this stream was a stream
139 * activity end message.
141 bool last_msg_is_stream_activity_end
;
144 * Time to use for a generated stream end activity message when
147 int64_t stream_act_end_ns_from_origin
;
150 const bt_stream
*stream
;
152 /* Owned by this (`NULL` initially and between packets) */
153 const bt_packet
*cur_packet
;
156 const bt_message
*stream_beginning_msg
;
160 void destroy_trimmer_comp(struct trimmer_comp
*trimmer_comp
)
162 BT_ASSERT(trimmer_comp
);
163 g_free(trimmer_comp
);
167 struct trimmer_comp
*create_trimmer_comp(void)
169 return g_new0(struct trimmer_comp
, 1);
173 void trimmer_finalize(bt_self_component_filter
*self_comp
)
175 struct trimmer_comp
*trimmer_comp
=
176 bt_self_component_get_data(
177 bt_self_component_filter_as_self_component(self_comp
));
180 destroy_trimmer_comp(trimmer_comp
);
185 * Sets the time (in ns from origin) of a trimmer bound from date and
188 * Returns a negative value if anything goes wrong.
191 int set_bound_ns_from_origin(struct trimmer_bound
*bound
,
192 unsigned int year
, unsigned int month
, unsigned int day
,
193 unsigned int hour
, unsigned int minute
, unsigned int second
,
194 unsigned int ns
, bool is_gmt
)
204 .tm_year
= year
- 1900,
209 result
= bt_timegm(&tm
);
211 result
= mktime(&tm
);
220 bound
->ns_from_origin
= (int64_t) result
;
221 bound
->ns_from_origin
*= NS_PER_S
;
222 bound
->ns_from_origin
+= ns
;
223 bound
->is_set
= true;
230 * Parses a timestamp, figuring out its format.
232 * Returns a negative value if anything goes wrong.
236 * YYYY-MM-DD hh:mm[:ss[.ns]]
240 * TODO: Check overflows.
243 int set_bound_from_str(struct trimmer_comp
*trimmer_comp
,
244 const char *str
, struct trimmer_bound
*bound
, bool is_gmt
)
248 unsigned int year
, month
, day
, hour
, minute
, second
, ns
;
251 /* Try `YYYY-MM-DD hh:mm:ss.ns` format */
252 s_ret
= sscanf(str
, "%u-%u-%u %u:%u:%u.%u%c", &year
, &month
, &day
,
253 &hour
, &minute
, &second
, &ns
, &dummy
);
255 ret
= set_bound_ns_from_origin(bound
, year
, month
, day
,
256 hour
, minute
, second
, ns
, is_gmt
);
260 /* Try `YYYY-MM-DD hh:mm:ss` format */
261 s_ret
= sscanf(str
, "%u-%u-%u %u:%u:%u%c", &year
, &month
, &day
,
262 &hour
, &minute
, &second
, &dummy
);
264 ret
= set_bound_ns_from_origin(bound
, year
, month
, day
,
265 hour
, minute
, second
, 0, is_gmt
);
269 /* Try `YYYY-MM-DD hh:mm` format */
270 s_ret
= sscanf(str
, "%u-%u-%u %u:%u%c", &year
, &month
, &day
,
271 &hour
, &minute
, &dummy
);
273 ret
= set_bound_ns_from_origin(bound
, year
, month
, day
,
274 hour
, minute
, 0, 0, is_gmt
);
278 /* Try `YYYY-MM-DD` format */
279 s_ret
= sscanf(str
, "%u-%u-%u%c", &year
, &month
, &day
, &dummy
);
281 ret
= set_bound_ns_from_origin(bound
, year
, month
, day
,
286 /* Try `hh:mm:ss.ns` format */
287 s_ret
= sscanf(str
, "%u:%u:%u.%u%c", &hour
, &minute
, &second
, &ns
,
290 bound
->time
.hour
= hour
;
291 bound
->time
.minute
= minute
;
292 bound
->time
.second
= second
;
297 /* Try `hh:mm:ss` format */
298 s_ret
= sscanf(str
, "%u:%u:%u%c", &hour
, &minute
, &second
, &dummy
);
300 bound
->time
.hour
= hour
;
301 bound
->time
.minute
= minute
;
302 bound
->time
.second
= second
;
307 /* Try `-s.ns` format */
308 s_ret
= sscanf(str
, "-%u.%u%c", &second
, &ns
, &dummy
);
310 bound
->ns_from_origin
= -((int64_t) second
) * NS_PER_S
;
311 bound
->ns_from_origin
-= (int64_t) ns
;
312 bound
->is_set
= true;
316 /* Try `s.ns` format */
317 s_ret
= sscanf(str
, "%u.%u%c", &second
, &ns
, &dummy
);
319 bound
->ns_from_origin
= ((int64_t) second
) * NS_PER_S
;
320 bound
->ns_from_origin
+= (int64_t) ns
;
321 bound
->is_set
= true;
325 /* Try `-s` format */
326 s_ret
= sscanf(str
, "-%u%c", &second
, &dummy
);
328 bound
->ns_from_origin
= -((int64_t) second
) * NS_PER_S
;
329 bound
->is_set
= true;
334 s_ret
= sscanf(str
, "%u%c", &second
, &dummy
);
336 bound
->ns_from_origin
= (int64_t) second
* NS_PER_S
;
337 bound
->is_set
= true;
341 BT_LOGE("Invalid date/time format: param=\"%s\"", str
);
349 * Sets a trimmer bound's properties from a parameter string/integer
352 * Returns a negative value if anything goes wrong.
355 int set_bound_from_param(struct trimmer_comp
*trimmer_comp
,
356 const char *param_name
, const bt_value
*param
,
357 struct trimmer_bound
*bound
, bool is_gmt
)
363 if (bt_value_is_signed_integer(param
)) {
364 int64_t value
= bt_value_signed_integer_get(param
);
367 * Just convert it to a temporary string to handle
368 * everything the same way.
370 sprintf(tmp_arg
, "%" PRId64
, value
);
372 } else if (bt_value_is_string(param
)) {
373 arg
= bt_value_string_get(param
);
375 BT_LOGE("`%s` parameter must be an integer or a string value.",
381 ret
= set_bound_from_str(trimmer_comp
, arg
, bound
, is_gmt
);
388 int validate_trimmer_bounds(struct trimmer_comp
*trimmer_comp
,
389 struct trimmer_bound
*begin
, struct trimmer_bound
*end
)
393 BT_ASSERT(begin
->is_set
);
394 BT_ASSERT(end
->is_set
);
396 if (!begin
->is_infinite
&& !end
->is_infinite
&&
397 begin
->ns_from_origin
> end
->ns_from_origin
) {
398 BT_LOGE("Trimming time range's beginning time is greater than end time: "
399 "begin-ns-from-origin=%" PRId64
", "
400 "end-ns-from-origin=%" PRId64
,
401 begin
->ns_from_origin
,
402 end
->ns_from_origin
);
407 if (!begin
->is_infinite
&& begin
->ns_from_origin
== INT64_MIN
) {
408 BT_LOGE("Invalid trimming time range's beginning time: "
409 "ns-from-origin=%" PRId64
,
410 begin
->ns_from_origin
);
415 if (!end
->is_infinite
&& end
->ns_from_origin
== INT64_MIN
) {
416 BT_LOGE("Invalid trimming time range's end time: "
417 "ns-from-origin=%" PRId64
,
418 end
->ns_from_origin
);
428 int init_trimmer_comp_from_params(struct trimmer_comp
*trimmer_comp
,
429 const bt_value
*params
)
431 const bt_value
*value
;
435 value
= bt_value_map_borrow_entry_value_const(params
, "gmt");
437 trimmer_comp
->is_gmt
= (bool) bt_value_bool_get(value
);
440 value
= bt_value_map_borrow_entry_value_const(params
, "begin");
442 if (set_bound_from_param(trimmer_comp
, "begin", value
,
443 &trimmer_comp
->begin
, trimmer_comp
->is_gmt
)) {
444 /* set_bound_from_param() logs errors */
445 ret
= BT_SELF_COMPONENT_STATUS_ERROR
;
449 trimmer_comp
->begin
.is_infinite
= true;
450 trimmer_comp
->begin
.is_set
= true;
453 value
= bt_value_map_borrow_entry_value_const(params
, "end");
455 if (set_bound_from_param(trimmer_comp
, "end", value
,
456 &trimmer_comp
->end
, trimmer_comp
->is_gmt
)) {
457 /* set_bound_from_param() logs errors */
458 ret
= BT_SELF_COMPONENT_STATUS_ERROR
;
462 trimmer_comp
->end
.is_infinite
= true;
463 trimmer_comp
->end
.is_set
= true;
467 if (trimmer_comp
->begin
.is_set
&& trimmer_comp
->end
.is_set
) {
468 /* validate_trimmer_bounds() logs errors */
469 ret
= validate_trimmer_bounds(trimmer_comp
,
470 &trimmer_comp
->begin
, &trimmer_comp
->end
);
476 bt_self_component_status
trimmer_init(bt_self_component_filter
*self_comp
,
477 const bt_value
*params
, void *init_data
)
480 bt_self_component_status status
;
481 struct trimmer_comp
*trimmer_comp
= create_trimmer_comp();
484 status
= BT_SELF_COMPONENT_STATUS_NOMEM
;
488 trimmer_comp
->log_level
= bt_component_get_logging_level(
489 bt_self_component_as_component(
490 bt_self_component_filter_as_self_component(self_comp
)));
491 status
= bt_self_component_filter_add_input_port(
492 self_comp
, in_port_name
, NULL
, NULL
);
493 if (status
!= BT_SELF_COMPONENT_STATUS_OK
) {
497 status
= bt_self_component_filter_add_output_port(
498 self_comp
, "out", NULL
, NULL
);
499 if (status
!= BT_SELF_COMPONENT_STATUS_OK
) {
503 ret
= init_trimmer_comp_from_params(trimmer_comp
, params
);
505 status
= BT_SELF_COMPONENT_STATUS_ERROR
;
509 bt_self_component_set_data(
510 bt_self_component_filter_as_self_component(self_comp
),
515 if (status
== BT_SELF_COMPONENT_STATUS_OK
) {
516 status
= BT_SELF_COMPONENT_STATUS_ERROR
;
520 destroy_trimmer_comp(trimmer_comp
);
528 void destroy_trimmer_iterator(struct trimmer_iterator
*trimmer_it
)
530 BT_ASSERT(trimmer_it
);
531 bt_self_component_port_input_message_iterator_put_ref(
532 trimmer_it
->upstream_iter
);
534 if (trimmer_it
->output_messages
) {
535 g_queue_free(trimmer_it
->output_messages
);
538 if (trimmer_it
->stream_states
) {
539 g_hash_table_destroy(trimmer_it
->stream_states
);
546 void destroy_trimmer_iterator_stream_state(
547 struct trimmer_iterator_stream_state
*sstate
)
550 BT_PACKET_PUT_REF_AND_RESET(sstate
->cur_packet
);
551 BT_MESSAGE_PUT_REF_AND_RESET(sstate
->stream_beginning_msg
);
556 bt_self_message_iterator_status
trimmer_msg_iter_init(
557 bt_self_message_iterator
*self_msg_iter
,
558 bt_self_component_filter
*self_comp
,
559 bt_self_component_port_output
*port
)
561 bt_self_message_iterator_status status
=
562 BT_SELF_MESSAGE_ITERATOR_STATUS_OK
;
563 struct trimmer_iterator
*trimmer_it
;
565 trimmer_it
= g_new0(struct trimmer_iterator
, 1);
567 status
= BT_SELF_MESSAGE_ITERATOR_STATUS_NOMEM
;
571 trimmer_it
->trimmer_comp
= bt_self_component_get_data(
572 bt_self_component_filter_as_self_component(self_comp
));
573 BT_ASSERT(trimmer_it
->trimmer_comp
);
575 if (trimmer_it
->trimmer_comp
->begin
.is_set
&&
576 trimmer_it
->trimmer_comp
->end
.is_set
) {
578 * Both trimming time range's bounds are set, so skip
580 * `TRIMMER_ITERATOR_STATE_SET_BOUNDS_NS_FROM_ORIGIN`
583 trimmer_it
->state
= TRIMMER_ITERATOR_STATE_SEEK_INITIALLY
;
586 trimmer_it
->begin
= trimmer_it
->trimmer_comp
->begin
;
587 trimmer_it
->end
= trimmer_it
->trimmer_comp
->end
;
588 trimmer_it
->upstream_iter
=
589 bt_self_component_port_input_message_iterator_create(
590 bt_self_component_filter_borrow_input_port_by_name(
591 self_comp
, in_port_name
));
592 if (!trimmer_it
->upstream_iter
) {
593 status
= BT_SELF_MESSAGE_ITERATOR_STATUS_ERROR
;
597 trimmer_it
->output_messages
= g_queue_new();
598 if (!trimmer_it
->output_messages
) {
599 status
= BT_SELF_MESSAGE_ITERATOR_STATUS_NOMEM
;
603 trimmer_it
->stream_states
= g_hash_table_new_full(g_direct_hash
,
604 g_direct_equal
, NULL
,
605 (GDestroyNotify
) destroy_trimmer_iterator_stream_state
);
606 if (!trimmer_it
->stream_states
) {
607 status
= BT_SELF_MESSAGE_ITERATOR_STATUS_NOMEM
;
611 trimmer_it
->self_msg_iter
= self_msg_iter
;
612 bt_self_message_iterator_set_data(self_msg_iter
, trimmer_it
);
615 if (status
!= BT_SELF_MESSAGE_ITERATOR_STATUS_OK
&& trimmer_it
) {
616 destroy_trimmer_iterator(trimmer_it
);
623 int get_msg_ns_from_origin(const bt_message
*msg
, int64_t *ns_from_origin
,
626 const bt_clock_class
*clock_class
= NULL
;
627 const bt_clock_snapshot
*clock_snapshot
= NULL
;
628 bt_message_stream_activity_clock_snapshot_state sa_cs_state
;
632 BT_ASSERT(ns_from_origin
);
635 switch (bt_message_get_type(msg
)) {
636 case BT_MESSAGE_TYPE_EVENT
:
638 bt_message_event_borrow_stream_class_default_clock_class_const(
640 if (G_UNLIKELY(!clock_class
)) {
644 clock_snapshot
= bt_message_event_borrow_default_clock_snapshot_const(
647 case BT_MESSAGE_TYPE_PACKET_BEGINNING
:
649 bt_message_packet_beginning_borrow_stream_class_default_clock_class_const(
651 if (G_UNLIKELY(!clock_class
)) {
655 clock_snapshot
= bt_message_packet_beginning_borrow_default_clock_snapshot_const(
658 case BT_MESSAGE_TYPE_PACKET_END
:
660 bt_message_packet_end_borrow_stream_class_default_clock_class_const(
662 if (G_UNLIKELY(!clock_class
)) {
666 clock_snapshot
= bt_message_packet_end_borrow_default_clock_snapshot_const(
669 case BT_MESSAGE_TYPE_DISCARDED_EVENTS
:
671 bt_message_discarded_events_borrow_stream_class_default_clock_class_const(
673 if (G_UNLIKELY(!clock_class
)) {
677 clock_snapshot
= bt_message_discarded_events_borrow_beginning_default_clock_snapshot_const(
680 case BT_MESSAGE_TYPE_DISCARDED_PACKETS
:
682 bt_message_discarded_packets_borrow_stream_class_default_clock_class_const(
684 if (G_UNLIKELY(!clock_class
)) {
688 clock_snapshot
= bt_message_discarded_packets_borrow_beginning_default_clock_snapshot_const(
691 case BT_MESSAGE_TYPE_STREAM_ACTIVITY_BEGINNING
:
693 bt_message_stream_activity_beginning_borrow_stream_class_default_clock_class_const(
695 if (G_UNLIKELY(!clock_class
)) {
699 sa_cs_state
= bt_message_stream_activity_beginning_borrow_default_clock_snapshot_const(
700 msg
, &clock_snapshot
);
701 if (sa_cs_state
== BT_MESSAGE_STREAM_ACTIVITY_CLOCK_SNAPSHOT_STATE_UNKNOWN
||
702 sa_cs_state
== BT_MESSAGE_STREAM_ACTIVITY_CLOCK_SNAPSHOT_STATE_INFINITE
) {
703 /* Lowest possible time to always include them */
704 *ns_from_origin
= INT64_MIN
;
705 goto no_clock_snapshot
;
709 case BT_MESSAGE_TYPE_STREAM_ACTIVITY_END
:
711 bt_message_stream_activity_end_borrow_stream_class_default_clock_class_const(
713 if (G_UNLIKELY(!clock_class
)) {
717 sa_cs_state
= bt_message_stream_activity_end_borrow_default_clock_snapshot_const(
718 msg
, &clock_snapshot
);
719 if (sa_cs_state
== BT_MESSAGE_STREAM_ACTIVITY_CLOCK_SNAPSHOT_STATE_UNKNOWN
) {
720 /* Lowest time to always include it */
721 *ns_from_origin
= INT64_MIN
;
722 goto no_clock_snapshot
;
723 } else if (sa_cs_state
== BT_MESSAGE_STREAM_ACTIVITY_CLOCK_SNAPSHOT_STATE_INFINITE
) {
724 /* Greatest time to always exclude it */
725 *ns_from_origin
= INT64_MAX
;
726 goto no_clock_snapshot
;
730 case BT_MESSAGE_TYPE_MESSAGE_ITERATOR_INACTIVITY
:
732 bt_message_message_iterator_inactivity_borrow_default_clock_snapshot_const(
736 goto no_clock_snapshot
;
739 ret
= bt_clock_snapshot_get_ns_from_origin(clock_snapshot
,
741 if (G_UNLIKELY(ret
)) {
759 void put_messages(bt_message_array_const msgs
, uint64_t count
)
763 for (i
= 0; i
< count
; i
++) {
764 BT_MESSAGE_PUT_REF_AND_RESET(msgs
[i
]);
769 int set_trimmer_iterator_bound(struct trimmer_iterator
*trimmer_it
,
770 struct trimmer_bound
*bound
, int64_t ns_from_origin
,
773 struct trimmer_comp
*trimmer_comp
= trimmer_it
->trimmer_comp
;
775 time_t time_seconds
= (time_t) (ns_from_origin
/ NS_PER_S
);
778 BT_ASSERT(!bound
->is_set
);
781 /* We only need to extract the date from this time */
783 bt_gmtime_r(&time_seconds
, &tm
);
785 bt_localtime_r(&time_seconds
, &tm
);
789 BT_LOGE_ERRNO("Cannot convert timestamp to date and time",
790 "ts=%" PRId64
, (int64_t) time_seconds
);
795 ret
= set_bound_ns_from_origin(bound
, tm
.tm_year
+ 1900, tm
.tm_mon
+ 1,
796 tm
.tm_mday
, bound
->time
.hour
, bound
->time
.minute
,
797 bound
->time
.second
, bound
->time
.ns
, is_gmt
);
804 bt_self_message_iterator_status
state_set_trimmer_iterator_bounds(
805 struct trimmer_iterator
*trimmer_it
)
807 bt_message_iterator_status upstream_iter_status
=
808 BT_MESSAGE_ITERATOR_STATUS_OK
;
809 struct trimmer_comp
*trimmer_comp
= trimmer_it
->trimmer_comp
;
810 bt_message_array_const msgs
;
812 int64_t ns_from_origin
= INT64_MIN
;
816 BT_ASSERT(!trimmer_it
->begin
.is_set
||
817 !trimmer_it
->end
.is_set
);
820 upstream_iter_status
=
821 bt_self_component_port_input_message_iterator_next(
822 trimmer_it
->upstream_iter
, &msgs
, &count
);
823 if (upstream_iter_status
!= BT_MESSAGE_ITERATOR_STATUS_OK
) {
827 for (i
= 0; i
< count
; i
++) {
828 const bt_message
*msg
= msgs
[i
];
832 ret
= get_msg_ns_from_origin(msg
, &ns_from_origin
,
842 BT_ASSERT(ns_from_origin
!= INT64_MIN
&&
843 ns_from_origin
!= INT64_MAX
);
844 put_messages(msgs
, count
);
848 put_messages(msgs
, count
);
852 if (!trimmer_it
->begin
.is_set
) {
853 BT_ASSERT(!trimmer_it
->begin
.is_infinite
);
854 ret
= set_trimmer_iterator_bound(trimmer_it
, &trimmer_it
->begin
,
855 ns_from_origin
, trimmer_comp
->is_gmt
);
861 if (!trimmer_it
->end
.is_set
) {
862 BT_ASSERT(!trimmer_it
->end
.is_infinite
);
863 ret
= set_trimmer_iterator_bound(trimmer_it
, &trimmer_it
->end
,
864 ns_from_origin
, trimmer_comp
->is_gmt
);
870 ret
= validate_trimmer_bounds(trimmer_it
->trimmer_comp
,
871 &trimmer_it
->begin
, &trimmer_it
->end
);
879 put_messages(msgs
, count
);
880 upstream_iter_status
= BT_MESSAGE_ITERATOR_STATUS_ERROR
;
883 return (int) upstream_iter_status
;
887 bt_self_message_iterator_status
state_seek_initially(
888 struct trimmer_iterator
*trimmer_it
)
890 struct trimmer_comp
*trimmer_comp
= trimmer_it
->trimmer_comp
;
891 bt_self_message_iterator_status status
=
892 BT_SELF_MESSAGE_ITERATOR_STATUS_OK
;
894 BT_ASSERT(trimmer_it
->begin
.is_set
);
896 if (trimmer_it
->begin
.is_infinite
) {
897 if (!bt_self_component_port_input_message_iterator_can_seek_beginning(
898 trimmer_it
->upstream_iter
)) {
899 BT_LOGE_STR("Cannot make upstream message iterator initially seek its beginning.");
900 status
= BT_SELF_MESSAGE_ITERATOR_STATUS_ERROR
;
904 status
= (int) bt_self_component_port_input_message_iterator_seek_beginning(
905 trimmer_it
->upstream_iter
);
907 if (!bt_self_component_port_input_message_iterator_can_seek_ns_from_origin(
908 trimmer_it
->upstream_iter
,
909 trimmer_it
->begin
.ns_from_origin
)) {
910 BT_LOGE("Cannot make upstream message iterator initially seek: "
911 "seek-ns-from-origin=%" PRId64
,
912 trimmer_it
->begin
.ns_from_origin
);
913 status
= BT_SELF_MESSAGE_ITERATOR_STATUS_ERROR
;
917 status
= (int) bt_self_component_port_input_message_iterator_seek_ns_from_origin(
918 trimmer_it
->upstream_iter
, trimmer_it
->begin
.ns_from_origin
);
921 if (status
== BT_SELF_MESSAGE_ITERATOR_STATUS_OK
) {
922 trimmer_it
->state
= TRIMMER_ITERATOR_STATE_TRIM
;
930 void push_message(struct trimmer_iterator
*trimmer_it
, const bt_message
*msg
)
932 g_queue_push_head(trimmer_it
->output_messages
, (void *) msg
);
936 const bt_message
*pop_message(struct trimmer_iterator
*trimmer_it
)
938 return g_queue_pop_tail(trimmer_it
->output_messages
);
942 int clock_raw_value_from_ns_from_origin(const bt_clock_class
*clock_class
,
943 int64_t ns_from_origin
, uint64_t *raw_value
)
947 uint64_t cc_offset_cycles
;
950 bt_clock_class_get_offset(clock_class
, &cc_offset_s
, &cc_offset_cycles
);
951 cc_freq
= bt_clock_class_get_frequency(clock_class
);
952 return bt_common_clock_value_from_ns_from_origin(cc_offset_s
,
953 cc_offset_cycles
, cc_freq
, ns_from_origin
, raw_value
);
957 bt_self_message_iterator_status
end_stream(struct trimmer_iterator
*trimmer_it
,
958 struct trimmer_iterator_stream_state
*sstate
)
960 bt_self_message_iterator_status status
=
961 BT_SELF_MESSAGE_ITERATOR_STATUS_OK
;
963 const bt_clock_class
*clock_class
;
965 bt_message
*msg
= NULL
;
967 BT_ASSERT(!trimmer_it
->end
.is_infinite
);
969 if (!sstate
->stream
) {
973 if (sstate
->cur_packet
) {
975 * The last message could not have been a stream
976 * activity end message if we have a current packet.
978 BT_ASSERT(!sstate
->last_msg_is_stream_activity_end
);
981 * Create and push a packet end message, making its time
982 * the trimming range's end time.
984 clock_class
= bt_stream_class_borrow_default_clock_class_const(
985 bt_stream_borrow_class_const(sstate
->stream
));
986 BT_ASSERT(clock_class
);
987 ret
= clock_raw_value_from_ns_from_origin(clock_class
,
988 trimmer_it
->end
.ns_from_origin
, &raw_value
);
990 status
= BT_SELF_MESSAGE_ITERATOR_STATUS_ERROR
;
994 msg
= bt_message_packet_end_create_with_default_clock_snapshot(
995 trimmer_it
->self_msg_iter
, sstate
->cur_packet
,
998 status
= BT_SELF_MESSAGE_ITERATOR_STATUS_NOMEM
;
1002 push_message(trimmer_it
, msg
);
1004 BT_PACKET_PUT_REF_AND_RESET(sstate
->cur_packet
);
1007 * Because we generated a packet end message, set the
1008 * stream activity end message's time to use to the
1009 * trimming range's end time (this packet end message's
1012 sstate
->stream_act_end_ns_from_origin
=
1013 trimmer_it
->end
.ns_from_origin
;
1016 if (!sstate
->last_msg_is_stream_activity_end
) {
1017 /* Create and push a stream activity end message */
1018 msg
= bt_message_stream_activity_end_create(
1019 trimmer_it
->self_msg_iter
, sstate
->stream
);
1021 status
= BT_SELF_MESSAGE_ITERATOR_STATUS_NOMEM
;
1025 clock_class
= bt_stream_class_borrow_default_clock_class_const(
1026 bt_stream_borrow_class_const(sstate
->stream
));
1027 BT_ASSERT(clock_class
);
1029 if (sstate
->stream_act_end_ns_from_origin
== INT64_MIN
) {
1031 * We received at least what is necessary to
1032 * have a stream state (stream beginning and
1033 * stream activity beginning messages), but
1034 * nothing else: use the trimmer range's end
1037 sstate
->stream_act_end_ns_from_origin
=
1038 trimmer_it
->end
.ns_from_origin
;
1041 ret
= clock_raw_value_from_ns_from_origin(clock_class
,
1042 sstate
->stream_act_end_ns_from_origin
, &raw_value
);
1044 status
= BT_SELF_MESSAGE_ITERATOR_STATUS_ERROR
;
1048 bt_message_stream_activity_end_set_default_clock_snapshot(
1050 push_message(trimmer_it
, msg
);
1054 /* Create and push a stream end message */
1055 msg
= bt_message_stream_end_create(trimmer_it
->self_msg_iter
,
1058 status
= BT_SELF_MESSAGE_ITERATOR_STATUS_NOMEM
;
1062 push_message(trimmer_it
, msg
);
1066 * Just to make sure that we don't use this stream state again
1067 * in the future without an obvious error.
1069 sstate
->stream
= NULL
;
1072 bt_message_put_ref(msg
);
1077 bt_self_message_iterator_status
end_iterator_streams(
1078 struct trimmer_iterator
*trimmer_it
)
1080 bt_self_message_iterator_status status
=
1081 BT_SELF_MESSAGE_ITERATOR_STATUS_OK
;
1082 GHashTableIter iter
;
1083 gpointer key
, sstate
;
1085 if (trimmer_it
->end
.is_infinite
) {
1087 * An infinite trimming range's end time guarantees that
1088 * we received (and pushed) all the appropriate end
1095 * End each stream and then remove them from the hash table of
1096 * stream states to release unneeded references.
1098 g_hash_table_iter_init(&iter
, trimmer_it
->stream_states
);
1100 while (g_hash_table_iter_next(&iter
, &key
, &sstate
)) {
1101 status
= end_stream(trimmer_it
, sstate
);
1108 g_hash_table_remove_all(trimmer_it
->stream_states
);
1115 bt_self_message_iterator_status
create_stream_beginning_activity_message(
1116 struct trimmer_iterator
*trimmer_it
,
1117 const bt_stream
*stream
,
1118 const bt_clock_class
*clock_class
, bt_message
**msg
)
1120 bt_message
*local_msg
;
1121 bt_self_message_iterator_status status
=
1122 BT_SELF_MESSAGE_ITERATOR_STATUS_OK
;
1125 BT_ASSERT(!trimmer_it
->begin
.is_infinite
);
1127 local_msg
= bt_message_stream_activity_beginning_create(
1128 trimmer_it
->self_msg_iter
, stream
);
1130 status
= BT_SELF_MESSAGE_ITERATOR_STATUS_NOMEM
;
1138 ret
= clock_raw_value_from_ns_from_origin(clock_class
,
1139 trimmer_it
->begin
.ns_from_origin
, &raw_value
);
1141 status
= BT_SELF_MESSAGE_ITERATOR_STATUS_ERROR
;
1142 bt_message_put_ref(local_msg
);
1146 bt_message_stream_activity_beginning_set_default_clock_snapshot(
1147 local_msg
, raw_value
);
1150 BT_MESSAGE_MOVE_REF(*msg
, local_msg
);
1157 * Makes sure to initialize a stream state, pushing the appropriate
1160 * `stream_act_beginning_msg` is an initial stream activity beginning
1161 * message to potentially use, depending on its clock snapshot state.
1162 * This function consumes `stream_act_beginning_msg` unconditionally.
1165 bt_self_message_iterator_status
ensure_stream_state_is_inited(
1166 struct trimmer_iterator
*trimmer_it
,
1167 struct trimmer_iterator_stream_state
*sstate
,
1168 const bt_message
*stream_act_beginning_msg
)
1170 bt_self_message_iterator_status status
=
1171 BT_SELF_MESSAGE_ITERATOR_STATUS_OK
;
1172 bt_message
*new_msg
= NULL
;
1173 const bt_clock_class
*clock_class
=
1174 bt_stream_class_borrow_default_clock_class_const(
1175 bt_stream_borrow_class_const(sstate
->stream
));
1177 BT_ASSERT(!sstate
->inited
);
1179 if (!sstate
->stream_beginning_msg
) {
1180 /* No initial stream beginning message: create one */
1181 sstate
->stream_beginning_msg
=
1182 bt_message_stream_beginning_create(
1183 trimmer_it
->self_msg_iter
, sstate
->stream
);
1184 if (!sstate
->stream_beginning_msg
) {
1185 status
= BT_SELF_MESSAGE_ITERATOR_STATUS_NOMEM
;
1190 /* Push initial stream beginning message */
1191 BT_ASSERT(sstate
->stream_beginning_msg
);
1192 push_message(trimmer_it
, sstate
->stream_beginning_msg
);
1193 sstate
->stream_beginning_msg
= NULL
;
1195 if (stream_act_beginning_msg
) {
1197 * Initial stream activity beginning message exists: if
1198 * its time is -inf, then create and push a new one
1199 * having the trimming range's beginning time. Otherwise
1200 * push it as is (known and unknown).
1202 const bt_clock_snapshot
*cs
;
1203 bt_message_stream_activity_clock_snapshot_state sa_cs_state
;
1205 sa_cs_state
= bt_message_stream_activity_beginning_borrow_default_clock_snapshot_const(
1206 stream_act_beginning_msg
, &cs
);
1207 if (sa_cs_state
== BT_MESSAGE_STREAM_ACTIVITY_CLOCK_SNAPSHOT_STATE_INFINITE
&&
1208 !trimmer_it
->begin
.is_infinite
) {
1210 * -inf time: use trimming range's beginning
1211 * time (which is not -inf).
1213 status
= create_stream_beginning_activity_message(
1214 trimmer_it
, sstate
->stream
, clock_class
,
1216 if (status
!= BT_SELF_MESSAGE_ITERATOR_STATUS_OK
) {
1220 push_message(trimmer_it
, new_msg
);
1223 /* Known/unknown: push as is */
1224 push_message(trimmer_it
, stream_act_beginning_msg
);
1225 stream_act_beginning_msg
= NULL
;
1228 BT_ASSERT(!trimmer_it
->begin
.is_infinite
);
1231 * No stream beginning activity message: create and push
1234 status
= create_stream_beginning_activity_message(
1235 trimmer_it
, sstate
->stream
, clock_class
, &new_msg
);
1236 if (status
!= BT_SELF_MESSAGE_ITERATOR_STATUS_OK
) {
1240 push_message(trimmer_it
, new_msg
);
1244 sstate
->inited
= true;
1247 bt_message_put_ref(new_msg
);
1248 bt_message_put_ref(stream_act_beginning_msg
);
1253 bt_self_message_iterator_status
ensure_cur_packet_exists(
1254 struct trimmer_iterator
*trimmer_it
,
1255 struct trimmer_iterator_stream_state
*sstate
,
1256 const bt_packet
*packet
)
1258 bt_self_message_iterator_status status
=
1259 BT_SELF_MESSAGE_ITERATOR_STATUS_OK
;
1261 const bt_clock_class
*clock_class
=
1262 bt_stream_class_borrow_default_clock_class_const(
1263 bt_stream_borrow_class_const(sstate
->stream
));
1264 bt_message
*msg
= NULL
;
1267 BT_ASSERT(!trimmer_it
->begin
.is_infinite
);
1268 BT_ASSERT(!sstate
->cur_packet
);
1271 * Create and push an initial packet beginning message,
1272 * making its time the trimming range's beginning time.
1274 ret
= clock_raw_value_from_ns_from_origin(clock_class
,
1275 trimmer_it
->begin
.ns_from_origin
, &raw_value
);
1277 status
= BT_SELF_MESSAGE_ITERATOR_STATUS_ERROR
;
1281 msg
= bt_message_packet_beginning_create_with_default_clock_snapshot(
1282 trimmer_it
->self_msg_iter
, packet
, raw_value
);
1284 status
= BT_SELF_MESSAGE_ITERATOR_STATUS_NOMEM
;
1288 push_message(trimmer_it
, msg
);
1291 /* Set packet as this stream's current packet */
1292 sstate
->cur_packet
= packet
;
1293 bt_packet_get_ref(sstate
->cur_packet
);
1296 bt_message_put_ref(msg
);
1301 * Handles a message which is associated to a given stream state. This
1302 * _could_ make the iterator's output message queue grow; this could
1303 * also consume the message without pushing anything to this queue, only
1304 * modifying the stream state.
1306 * This function consumes the `msg` reference, _whatever the outcome_.
1308 * `ns_from_origin` is the message's time, as given by
1309 * get_msg_ns_from_origin().
1311 * This function sets `reached_end` if handling this message made the
1312 * iterator reach the end of the trimming range. Note that the output
1313 * message queue could contain messages even if this function sets
1317 bt_self_message_iterator_status
handle_message_with_stream_state(
1318 struct trimmer_iterator
*trimmer_it
, const bt_message
*msg
,
1319 struct trimmer_iterator_stream_state
*sstate
,
1320 int64_t ns_from_origin
, bool *reached_end
)
1322 bt_self_message_iterator_status status
=
1323 BT_SELF_MESSAGE_ITERATOR_STATUS_OK
;
1324 bt_message_type msg_type
= bt_message_get_type(msg
);
1328 case BT_MESSAGE_TYPE_EVENT
:
1329 if (G_UNLIKELY(!trimmer_it
->end
.is_infinite
&&
1330 ns_from_origin
> trimmer_it
->end
.ns_from_origin
)) {
1331 status
= end_iterator_streams(trimmer_it
);
1332 *reached_end
= true;
1336 if (G_UNLIKELY(!sstate
->inited
)) {
1337 status
= ensure_stream_state_is_inited(trimmer_it
,
1339 if (status
!= BT_SELF_MESSAGE_ITERATOR_STATUS_OK
) {
1344 if (G_UNLIKELY(!sstate
->cur_packet
)) {
1345 const bt_event
*event
=
1346 bt_message_event_borrow_event_const(msg
);
1347 const bt_packet
*packet
= bt_event_borrow_packet_const(
1350 status
= ensure_cur_packet_exists(trimmer_it
, sstate
,
1352 if (status
!= BT_SELF_MESSAGE_ITERATOR_STATUS_OK
) {
1357 BT_ASSERT(sstate
->cur_packet
);
1358 push_message(trimmer_it
, msg
);
1361 case BT_MESSAGE_TYPE_PACKET_BEGINNING
:
1362 if (G_UNLIKELY(!trimmer_it
->end
.is_infinite
&&
1363 ns_from_origin
> trimmer_it
->end
.ns_from_origin
)) {
1364 status
= end_iterator_streams(trimmer_it
);
1365 *reached_end
= true;
1369 if (G_UNLIKELY(!sstate
->inited
)) {
1370 status
= ensure_stream_state_is_inited(trimmer_it
,
1372 if (status
!= BT_SELF_MESSAGE_ITERATOR_STATUS_OK
) {
1377 BT_ASSERT(!sstate
->cur_packet
);
1378 sstate
->cur_packet
=
1379 bt_message_packet_beginning_borrow_packet_const(msg
);
1380 bt_packet_get_ref(sstate
->cur_packet
);
1381 push_message(trimmer_it
, msg
);
1384 case BT_MESSAGE_TYPE_PACKET_END
:
1385 sstate
->stream_act_end_ns_from_origin
= ns_from_origin
;
1387 if (G_UNLIKELY(!trimmer_it
->end
.is_infinite
&&
1388 ns_from_origin
> trimmer_it
->end
.ns_from_origin
)) {
1389 status
= end_iterator_streams(trimmer_it
);
1390 *reached_end
= true;
1394 if (G_UNLIKELY(!sstate
->inited
)) {
1395 status
= ensure_stream_state_is_inited(trimmer_it
,
1397 if (status
!= BT_SELF_MESSAGE_ITERATOR_STATUS_OK
) {
1402 if (G_UNLIKELY(!sstate
->cur_packet
)) {
1403 const bt_packet
*packet
=
1404 bt_message_packet_end_borrow_packet_const(msg
);
1406 status
= ensure_cur_packet_exists(trimmer_it
, sstate
,
1408 if (status
!= BT_SELF_MESSAGE_ITERATOR_STATUS_OK
) {
1413 BT_ASSERT(sstate
->cur_packet
);
1414 BT_PACKET_PUT_REF_AND_RESET(sstate
->cur_packet
);
1415 push_message(trimmer_it
, msg
);
1418 case BT_MESSAGE_TYPE_DISCARDED_EVENTS
:
1419 case BT_MESSAGE_TYPE_DISCARDED_PACKETS
:
1422 * `ns_from_origin` is the message's time range's
1423 * beginning time here.
1425 int64_t end_ns_from_origin
;
1426 const bt_clock_snapshot
*end_cs
;
1428 if (bt_message_get_type(msg
) ==
1429 BT_MESSAGE_TYPE_DISCARDED_EVENTS
) {
1431 * Safe to ignore the return value because we
1432 * know there's a default clock and it's always
1435 end_cs
= bt_message_discarded_events_borrow_end_default_clock_snapshot_const(
1439 * Safe to ignore the return value because we
1440 * know there's a default clock and it's always
1443 end_cs
= bt_message_discarded_packets_borrow_end_default_clock_snapshot_const(
1447 if (bt_clock_snapshot_get_ns_from_origin(end_cs
,
1448 &end_ns_from_origin
)) {
1449 status
= BT_SELF_MESSAGE_ITERATOR_STATUS_ERROR
;
1453 sstate
->stream_act_end_ns_from_origin
= end_ns_from_origin
;
1455 if (!trimmer_it
->end
.is_infinite
&&
1456 ns_from_origin
> trimmer_it
->end
.ns_from_origin
) {
1457 status
= end_iterator_streams(trimmer_it
);
1458 *reached_end
= true;
1462 if (!trimmer_it
->end
.is_infinite
&&
1463 end_ns_from_origin
> trimmer_it
->end
.ns_from_origin
) {
1465 * This message's end time is outside the
1466 * trimming time range: replace it with a new
1467 * message having an end time equal to the
1468 * trimming time range's end and without a
1471 const bt_clock_class
*clock_class
=
1472 bt_clock_snapshot_borrow_clock_class_const(
1474 const bt_clock_snapshot
*begin_cs
;
1475 bt_message
*new_msg
;
1476 uint64_t end_raw_value
;
1478 ret
= clock_raw_value_from_ns_from_origin(clock_class
,
1479 trimmer_it
->end
.ns_from_origin
, &end_raw_value
);
1481 status
= BT_SELF_MESSAGE_ITERATOR_STATUS_ERROR
;
1485 if (msg_type
== BT_MESSAGE_TYPE_DISCARDED_EVENTS
) {
1486 begin_cs
= bt_message_discarded_events_borrow_beginning_default_clock_snapshot_const(
1488 new_msg
= bt_message_discarded_events_create_with_default_clock_snapshots(
1489 trimmer_it
->self_msg_iter
,
1491 bt_clock_snapshot_get_value(begin_cs
),
1494 begin_cs
= bt_message_discarded_packets_borrow_beginning_default_clock_snapshot_const(
1496 new_msg
= bt_message_discarded_packets_create_with_default_clock_snapshots(
1497 trimmer_it
->self_msg_iter
,
1499 bt_clock_snapshot_get_value(begin_cs
),
1504 status
= BT_SELF_MESSAGE_ITERATOR_STATUS_NOMEM
;
1508 /* Replace the original message */
1509 BT_MESSAGE_MOVE_REF(msg
, new_msg
);
1512 if (G_UNLIKELY(!sstate
->inited
)) {
1513 status
= ensure_stream_state_is_inited(trimmer_it
,
1515 if (status
!= BT_SELF_MESSAGE_ITERATOR_STATUS_OK
) {
1520 push_message(trimmer_it
, msg
);
1524 case BT_MESSAGE_TYPE_STREAM_ACTIVITY_BEGINNING
:
1525 if (!trimmer_it
->end
.is_infinite
&&
1526 ns_from_origin
> trimmer_it
->end
.ns_from_origin
) {
1528 * This only happens when the message's time is
1529 * known and is greater than the trimming
1530 * range's end time. Unknown and -inf times are
1532 * `trimmer_it->end.ns_from_origin`.
1534 status
= end_iterator_streams(trimmer_it
);
1535 *reached_end
= true;
1539 if (!sstate
->inited
) {
1540 status
= ensure_stream_state_is_inited(trimmer_it
,
1543 if (status
!= BT_SELF_MESSAGE_ITERATOR_STATUS_OK
) {
1547 push_message(trimmer_it
, msg
);
1552 case BT_MESSAGE_TYPE_STREAM_ACTIVITY_END
:
1553 if (trimmer_it
->end
.is_infinite
) {
1554 push_message(trimmer_it
, msg
);
1559 if (ns_from_origin
== INT64_MIN
) {
1560 /* Unknown: push as is if stream state is inited */
1561 if (sstate
->inited
) {
1562 push_message(trimmer_it
, msg
);
1564 sstate
->last_msg_is_stream_activity_end
= true;
1566 } else if (ns_from_origin
== INT64_MAX
) {
1567 /* Infinite: use trimming range's end time */
1568 sstate
->stream_act_end_ns_from_origin
=
1569 trimmer_it
->end
.ns_from_origin
;
1571 /* Known: check if outside of trimming range */
1572 if (ns_from_origin
> trimmer_it
->end
.ns_from_origin
) {
1573 sstate
->stream_act_end_ns_from_origin
=
1574 trimmer_it
->end
.ns_from_origin
;
1575 status
= end_iterator_streams(trimmer_it
);
1576 *reached_end
= true;
1580 if (!sstate
->inited
) {
1582 * First message for this stream is a
1583 * stream activity end: we can't deduce
1584 * anything about the stream activity
1585 * beginning's time, and using this
1586 * message's time would make a useless
1587 * pair of stream activity beginning/end
1588 * with the same time. Just skip this
1589 * message and wait for something
1595 push_message(trimmer_it
, msg
);
1597 sstate
->last_msg_is_stream_activity_end
= true;
1598 sstate
->stream_act_end_ns_from_origin
= ns_from_origin
;
1602 case BT_MESSAGE_TYPE_STREAM_BEGINNING
:
1604 * We don't know what follows at this point, so just
1605 * keep this message until we know what to do with it
1606 * (it will be used in ensure_stream_state_is_inited()).
1608 BT_ASSERT(!sstate
->inited
);
1609 BT_MESSAGE_MOVE_REF(sstate
->stream_beginning_msg
, msg
);
1611 case BT_MESSAGE_TYPE_STREAM_END
:
1612 if (sstate
->inited
) {
1614 * This is the end of an inited stream: end this
1615 * stream if its stream activity end message
1616 * time is not the trimming range's end time
1617 * (which means the final stream activity end
1618 * message had an infinite time). end_stream()
1619 * will generate its own stream end message.
1621 if (trimmer_it
->end
.is_infinite
) {
1622 push_message(trimmer_it
, msg
);
1624 g_hash_table_remove(trimmer_it
->stream_states
,
1626 } else if (sstate
->stream_act_end_ns_from_origin
<
1627 trimmer_it
->end
.ns_from_origin
) {
1628 status
= end_stream(trimmer_it
, sstate
);
1629 if (status
!= BT_SELF_MESSAGE_ITERATOR_STATUS_OK
) {
1633 /* We won't need this stream state again */
1634 g_hash_table_remove(trimmer_it
->stream_states
,
1638 /* We dont't need this stream state anymore */
1639 g_hash_table_remove(trimmer_it
->stream_states
, sstate
->stream
);
1648 /* We release the message's reference whatever the outcome */
1649 bt_message_put_ref(msg
);
1650 return BT_SELF_MESSAGE_ITERATOR_STATUS_OK
;
1654 * Handles an input message. This _could_ make the iterator's output
1655 * message queue grow; this could also consume the message without
1656 * pushing anything to this queue, only modifying the stream state.
1658 * This function consumes the `msg` reference, _whatever the outcome_.
1660 * This function sets `reached_end` if handling this message made the
1661 * iterator reach the end of the trimming range. Note that the output
1662 * message queue could contain messages even if this function sets
1666 bt_self_message_iterator_status
handle_message(
1667 struct trimmer_iterator
*trimmer_it
, const bt_message
*msg
,
1670 bt_self_message_iterator_status status
;
1671 const bt_stream
*stream
= NULL
;
1672 int64_t ns_from_origin
= INT64_MIN
;
1675 struct trimmer_iterator_stream_state
*sstate
= NULL
;
1676 struct trimmer_comp
*trimmer_comp
= trimmer_it
->trimmer_comp
;
1678 /* Find message's associated stream */
1679 switch (bt_message_get_type(msg
)) {
1680 case BT_MESSAGE_TYPE_EVENT
:
1681 stream
= bt_event_borrow_stream_const(
1682 bt_message_event_borrow_event_const(msg
));
1684 case BT_MESSAGE_TYPE_PACKET_BEGINNING
:
1685 stream
= bt_packet_borrow_stream_const(
1686 bt_message_packet_beginning_borrow_packet_const(msg
));
1688 case BT_MESSAGE_TYPE_PACKET_END
:
1689 stream
= bt_packet_borrow_stream_const(
1690 bt_message_packet_end_borrow_packet_const(msg
));
1692 case BT_MESSAGE_TYPE_DISCARDED_EVENTS
:
1693 stream
= bt_message_discarded_events_borrow_stream_const(msg
);
1695 case BT_MESSAGE_TYPE_DISCARDED_PACKETS
:
1696 stream
= bt_message_discarded_packets_borrow_stream_const(msg
);
1698 case BT_MESSAGE_TYPE_STREAM_ACTIVITY_BEGINNING
:
1699 stream
= bt_message_stream_activity_beginning_borrow_stream_const(msg
);
1701 case BT_MESSAGE_TYPE_STREAM_ACTIVITY_END
:
1702 stream
= bt_message_stream_activity_end_borrow_stream_const(msg
);
1704 case BT_MESSAGE_TYPE_STREAM_BEGINNING
:
1705 stream
= bt_message_stream_beginning_borrow_stream_const(msg
);
1707 case BT_MESSAGE_TYPE_STREAM_END
:
1708 stream
= bt_message_stream_end_borrow_stream_const(msg
);
1714 if (G_LIKELY(stream
)) {
1715 /* Find stream state */
1716 sstate
= g_hash_table_lookup(trimmer_it
->stream_states
,
1718 if (G_UNLIKELY(!sstate
)) {
1719 /* No stream state yet: create one now */
1720 const bt_stream_class
*sc
;
1723 * Validate right now that the stream's class
1724 * has a registered default clock class so that
1725 * an existing stream state guarantees existing
1726 * default clock snapshots for its associated
1729 * Also check that clock snapshots are always
1732 sc
= bt_stream_borrow_class_const(stream
);
1733 if (!bt_stream_class_borrow_default_clock_class_const(sc
)) {
1734 BT_LOGE("Unsupported stream: stream class does "
1735 "not have a default clock class: "
1737 "stream-id=%" PRIu64
", "
1738 "stream-name=\"%s\"",
1739 stream
, bt_stream_get_id(stream
),
1740 bt_stream_get_name(stream
));
1741 status
= BT_SELF_MESSAGE_ITERATOR_STATUS_ERROR
;
1746 * Temporary: make sure packet beginning, packet
1747 * end, discarded events, and discarded packets
1748 * messages have default clock snapshots until
1749 * the support for not having them is
1752 if (!bt_stream_class_packets_have_beginning_default_clock_snapshot(
1754 BT_LOGE("Unsupported stream: packets have "
1755 "no beginning clock snapshot: "
1757 "stream-id=%" PRIu64
", "
1758 "stream-name=\"%s\"",
1759 stream
, bt_stream_get_id(stream
),
1760 bt_stream_get_name(stream
));
1761 status
= BT_SELF_MESSAGE_ITERATOR_STATUS_ERROR
;
1765 if (!bt_stream_class_packets_have_end_default_clock_snapshot(
1767 BT_LOGE("Unsupported stream: packets have "
1768 "no end clock snapshot: "
1770 "stream-id=%" PRIu64
", "
1771 "stream-name=\"%s\"",
1772 stream
, bt_stream_get_id(stream
),
1773 bt_stream_get_name(stream
));
1774 status
= BT_SELF_MESSAGE_ITERATOR_STATUS_ERROR
;
1778 if (bt_stream_class_supports_discarded_events(sc
) &&
1779 !bt_stream_class_discarded_events_have_default_clock_snapshots(sc
)) {
1780 BT_LOGE("Unsupported stream: discarded events "
1781 "have no clock snapshots: "
1783 "stream-id=%" PRIu64
", "
1784 "stream-name=\"%s\"",
1785 stream
, bt_stream_get_id(stream
),
1786 bt_stream_get_name(stream
));
1787 status
= BT_SELF_MESSAGE_ITERATOR_STATUS_ERROR
;
1791 if (bt_stream_class_supports_discarded_packets(sc
) &&
1792 !bt_stream_class_discarded_packets_have_default_clock_snapshots(sc
)) {
1793 BT_LOGE("Unsupported stream: discarded packets "
1794 "have no clock snapshots: "
1796 "stream-id=%" PRIu64
", "
1797 "stream-name=\"%s\"",
1798 stream
, bt_stream_get_id(stream
),
1799 bt_stream_get_name(stream
));
1800 status
= BT_SELF_MESSAGE_ITERATOR_STATUS_ERROR
;
1804 sstate
= g_new0(struct trimmer_iterator_stream_state
,
1807 status
= BT_SELF_MESSAGE_ITERATOR_STATUS_NOMEM
;
1811 sstate
->stream
= stream
;
1812 sstate
->stream_act_end_ns_from_origin
= INT64_MIN
;
1813 g_hash_table_insert(trimmer_it
->stream_states
,
1814 (void *) stream
, sstate
);
1818 /* Retrieve the message's time */
1819 ret
= get_msg_ns_from_origin(msg
, &ns_from_origin
, &skip
);
1820 if (G_UNLIKELY(ret
)) {
1821 status
= BT_SELF_MESSAGE_ITERATOR_STATUS_ERROR
;
1825 if (G_LIKELY(sstate
)) {
1826 /* Message associated to a stream */
1827 status
= handle_message_with_stream_state(trimmer_it
, msg
,
1828 sstate
, ns_from_origin
, reached_end
);
1831 * handle_message_with_stream_state() unconditionally
1837 * Message not associated to a stream (message iterator
1840 if (G_UNLIKELY(ns_from_origin
> trimmer_it
->end
.ns_from_origin
)) {
1841 BT_MESSAGE_PUT_REF_AND_RESET(msg
);
1842 status
= end_iterator_streams(trimmer_it
);
1843 *reached_end
= true;
1845 push_message(trimmer_it
, msg
);
1846 status
= BT_SELF_MESSAGE_ITERATOR_STATUS_OK
;
1852 /* We release the message's reference whatever the outcome */
1853 bt_message_put_ref(msg
);
1858 void fill_message_array_from_output_messages(
1859 struct trimmer_iterator
*trimmer_it
,
1860 bt_message_array_const msgs
, uint64_t capacity
, uint64_t *count
)
1865 * Move auto-seek messages to the output array (which is this
1866 * iterator's base message array).
1868 while (capacity
> 0 && !g_queue_is_empty(trimmer_it
->output_messages
)) {
1869 msgs
[*count
] = pop_message(trimmer_it
);
1874 BT_ASSERT(*count
> 0);
1878 bt_self_message_iterator_status
state_ending(
1879 struct trimmer_iterator
*trimmer_it
,
1880 bt_message_array_const msgs
, uint64_t capacity
,
1883 bt_self_message_iterator_status status
=
1884 BT_SELF_MESSAGE_ITERATOR_STATUS_OK
;
1886 if (g_queue_is_empty(trimmer_it
->output_messages
)) {
1887 trimmer_it
->state
= TRIMMER_ITERATOR_STATE_ENDED
;
1888 status
= BT_SELF_MESSAGE_ITERATOR_STATUS_END
;
1892 fill_message_array_from_output_messages(trimmer_it
, msgs
,
1900 bt_self_message_iterator_status
state_trim(struct trimmer_iterator
*trimmer_it
,
1901 bt_message_array_const msgs
, uint64_t capacity
,
1904 bt_self_message_iterator_status status
=
1905 BT_SELF_MESSAGE_ITERATOR_STATUS_OK
;
1906 bt_message_array_const my_msgs
;
1909 bool reached_end
= false;
1911 while (g_queue_is_empty(trimmer_it
->output_messages
)) {
1912 status
= (int) bt_self_component_port_input_message_iterator_next(
1913 trimmer_it
->upstream_iter
, &my_msgs
, &my_count
);
1914 if (G_UNLIKELY(status
!= BT_SELF_MESSAGE_ITERATOR_STATUS_OK
)) {
1915 if (status
== BT_SELF_MESSAGE_ITERATOR_STATUS_END
) {
1916 status
= end_iterator_streams(trimmer_it
);
1917 if (status
!= BT_SELF_MESSAGE_ITERATOR_STATUS_OK
) {
1922 TRIMMER_ITERATOR_STATE_ENDING
;
1923 status
= state_ending(trimmer_it
, msgs
,
1930 BT_ASSERT(my_count
> 0);
1932 for (i
= 0; i
< my_count
; i
++) {
1933 status
= handle_message(trimmer_it
, my_msgs
[i
],
1937 * handle_message() unconditionally consumes the
1938 * message reference.
1942 if (G_UNLIKELY(status
!=
1943 BT_SELF_MESSAGE_ITERATOR_STATUS_OK
)) {
1944 put_messages(my_msgs
, my_count
);
1948 if (G_UNLIKELY(reached_end
)) {
1950 * This message's time was passed the
1951 * trimming time range's end time: we
1952 * are done. Their might still be
1953 * messages in the output message queue,
1954 * so move to the "ending" state and
1955 * apply it immediately since
1956 * state_trim() is called within the
1959 put_messages(my_msgs
, my_count
);
1961 TRIMMER_ITERATOR_STATE_ENDING
;
1962 status
= state_ending(trimmer_it
, msgs
,
1970 * There's at least one message in the output message queue:
1971 * move the messages to the output message array.
1973 BT_ASSERT(!g_queue_is_empty(trimmer_it
->output_messages
));
1974 fill_message_array_from_output_messages(trimmer_it
, msgs
,
1982 bt_self_message_iterator_status
trimmer_msg_iter_next(
1983 bt_self_message_iterator
*self_msg_iter
,
1984 bt_message_array_const msgs
, uint64_t capacity
,
1987 struct trimmer_iterator
*trimmer_it
=
1988 bt_self_message_iterator_get_data(self_msg_iter
);
1989 bt_self_message_iterator_status status
=
1990 BT_SELF_MESSAGE_ITERATOR_STATUS_OK
;
1992 BT_ASSERT(trimmer_it
);
1994 if (G_LIKELY(trimmer_it
->state
== TRIMMER_ITERATOR_STATE_TRIM
)) {
1995 status
= state_trim(trimmer_it
, msgs
, capacity
, count
);
1996 if (status
!= BT_SELF_MESSAGE_ITERATOR_STATUS_OK
) {
2000 switch (trimmer_it
->state
) {
2001 case TRIMMER_ITERATOR_STATE_SET_BOUNDS_NS_FROM_ORIGIN
:
2002 status
= state_set_trimmer_iterator_bounds(trimmer_it
);
2003 if (status
!= BT_SELF_MESSAGE_ITERATOR_STATUS_OK
) {
2007 status
= state_seek_initially(trimmer_it
);
2008 if (status
!= BT_SELF_MESSAGE_ITERATOR_STATUS_OK
) {
2012 status
= state_trim(trimmer_it
, msgs
, capacity
, count
);
2013 if (status
!= BT_SELF_MESSAGE_ITERATOR_STATUS_OK
) {
2018 case TRIMMER_ITERATOR_STATE_SEEK_INITIALLY
:
2019 status
= state_seek_initially(trimmer_it
);
2020 if (status
!= BT_SELF_MESSAGE_ITERATOR_STATUS_OK
) {
2024 status
= state_trim(trimmer_it
, msgs
, capacity
, count
);
2025 if (status
!= BT_SELF_MESSAGE_ITERATOR_STATUS_OK
) {
2030 case TRIMMER_ITERATOR_STATE_ENDING
:
2031 status
= state_ending(trimmer_it
, msgs
, capacity
,
2033 if (status
!= BT_SELF_MESSAGE_ITERATOR_STATUS_OK
) {
2038 case TRIMMER_ITERATOR_STATE_ENDED
:
2039 status
= BT_SELF_MESSAGE_ITERATOR_STATUS_END
;
2051 void trimmer_msg_iter_finalize(bt_self_message_iterator
*self_msg_iter
)
2053 struct trimmer_iterator
*trimmer_it
=
2054 bt_self_message_iterator_get_data(self_msg_iter
);
2056 BT_ASSERT(trimmer_it
);
2057 destroy_trimmer_iterator(trimmer_it
);