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 the last pushed message for this stream was a stream
135 * activity end message.
137 bool last_msg_is_stream_activity_end
;
140 * Time to use for a generated stream end activity message when
143 int64_t stream_act_end_ns_from_origin
;
146 const bt_stream
*stream
;
148 /* Owned by this (`NULL` initially and between packets) */
149 const bt_packet
*cur_packet
;
153 void destroy_trimmer_comp(struct trimmer_comp
*trimmer_comp
)
155 BT_ASSERT(trimmer_comp
);
156 g_free(trimmer_comp
);
160 struct trimmer_comp
*create_trimmer_comp(void)
162 return g_new0(struct trimmer_comp
, 1);
166 void trimmer_finalize(bt_self_component_filter
*self_comp
)
168 struct trimmer_comp
*trimmer_comp
=
169 bt_self_component_get_data(
170 bt_self_component_filter_as_self_component(self_comp
));
173 destroy_trimmer_comp(trimmer_comp
);
178 * Sets the time (in ns from origin) of a trimmer bound from date and
181 * Returns a negative value if anything goes wrong.
184 int set_bound_ns_from_origin(struct trimmer_bound
*bound
,
185 unsigned int year
, unsigned int month
, unsigned int day
,
186 unsigned int hour
, unsigned int minute
, unsigned int second
,
187 unsigned int ns
, bool is_gmt
)
197 .tm_year
= year
- 1900,
202 result
= bt_timegm(&tm
);
204 result
= mktime(&tm
);
213 bound
->ns_from_origin
= (int64_t) result
;
214 bound
->ns_from_origin
*= NS_PER_S
;
215 bound
->ns_from_origin
+= ns
;
216 bound
->is_set
= true;
223 * Parses a timestamp, figuring out its format.
225 * Returns a negative value if anything goes wrong.
229 * YYYY-MM-DD hh:mm[:ss[.ns]]
233 * TODO: Check overflows.
236 int set_bound_from_str(struct trimmer_comp
*trimmer_comp
,
237 const char *str
, struct trimmer_bound
*bound
, bool is_gmt
)
241 unsigned int year
, month
, day
, hour
, minute
, second
, ns
;
244 /* Try `YYYY-MM-DD hh:mm:ss.ns` format */
245 s_ret
= sscanf(str
, "%u-%u-%u %u:%u:%u.%u%c", &year
, &month
, &day
,
246 &hour
, &minute
, &second
, &ns
, &dummy
);
248 ret
= set_bound_ns_from_origin(bound
, year
, month
, day
,
249 hour
, minute
, second
, ns
, is_gmt
);
253 /* Try `YYYY-MM-DD hh:mm:ss` format */
254 s_ret
= sscanf(str
, "%u-%u-%u %u:%u:%u%c", &year
, &month
, &day
,
255 &hour
, &minute
, &second
, &dummy
);
257 ret
= set_bound_ns_from_origin(bound
, year
, month
, day
,
258 hour
, minute
, second
, 0, is_gmt
);
262 /* Try `YYYY-MM-DD hh:mm` format */
263 s_ret
= sscanf(str
, "%u-%u-%u %u:%u%c", &year
, &month
, &day
,
264 &hour
, &minute
, &dummy
);
266 ret
= set_bound_ns_from_origin(bound
, year
, month
, day
,
267 hour
, minute
, 0, 0, is_gmt
);
271 /* Try `YYYY-MM-DD` format */
272 s_ret
= sscanf(str
, "%u-%u-%u%c", &year
, &month
, &day
, &dummy
);
274 ret
= set_bound_ns_from_origin(bound
, year
, month
, day
,
279 /* Try `hh:mm:ss.ns` format */
280 s_ret
= sscanf(str
, "%u:%u:%u.%u%c", &hour
, &minute
, &second
, &ns
,
283 bound
->time
.hour
= hour
;
284 bound
->time
.minute
= minute
;
285 bound
->time
.second
= second
;
290 /* Try `hh:mm:ss` format */
291 s_ret
= sscanf(str
, "%u:%u:%u%c", &hour
, &minute
, &second
, &dummy
);
293 bound
->time
.hour
= hour
;
294 bound
->time
.minute
= minute
;
295 bound
->time
.second
= second
;
300 /* Try `-s.ns` format */
301 s_ret
= sscanf(str
, "-%u.%u%c", &second
, &ns
, &dummy
);
303 bound
->ns_from_origin
= -((int64_t) second
) * NS_PER_S
;
304 bound
->ns_from_origin
-= (int64_t) ns
;
305 bound
->is_set
= true;
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` format */
319 s_ret
= sscanf(str
, "-%u%c", &second
, &dummy
);
321 bound
->ns_from_origin
= -((int64_t) second
) * NS_PER_S
;
322 bound
->is_set
= true;
327 s_ret
= sscanf(str
, "%u%c", &second
, &dummy
);
329 bound
->ns_from_origin
= (int64_t) second
* NS_PER_S
;
330 bound
->is_set
= true;
334 BT_COMP_LOGE("Invalid date/time format: param=\"%s\"", str
);
342 * Sets a trimmer bound's properties from a parameter string/integer
345 * Returns a negative value if anything goes wrong.
348 int set_bound_from_param(struct trimmer_comp
*trimmer_comp
,
349 const char *param_name
, const bt_value
*param
,
350 struct trimmer_bound
*bound
, bool is_gmt
)
356 if (bt_value_is_signed_integer(param
)) {
357 int64_t value
= bt_value_signed_integer_get(param
);
360 * Just convert it to a temporary string to handle
361 * everything the same way.
363 sprintf(tmp_arg
, "%" PRId64
, value
);
365 } else if (bt_value_is_string(param
)) {
366 arg
= bt_value_string_get(param
);
368 BT_COMP_LOGE("`%s` parameter must be an integer or a string value.",
374 ret
= set_bound_from_str(trimmer_comp
, arg
, bound
, is_gmt
);
381 int validate_trimmer_bounds(struct trimmer_comp
*trimmer_comp
,
382 struct trimmer_bound
*begin
, struct trimmer_bound
*end
)
386 BT_ASSERT(begin
->is_set
);
387 BT_ASSERT(end
->is_set
);
389 if (!begin
->is_infinite
&& !end
->is_infinite
&&
390 begin
->ns_from_origin
> end
->ns_from_origin
) {
391 BT_COMP_LOGE("Trimming time range's beginning time is greater than end time: "
392 "begin-ns-from-origin=%" PRId64
", "
393 "end-ns-from-origin=%" PRId64
,
394 begin
->ns_from_origin
,
395 end
->ns_from_origin
);
400 if (!begin
->is_infinite
&& begin
->ns_from_origin
== INT64_MIN
) {
401 BT_COMP_LOGE("Invalid trimming time range's beginning time: "
402 "ns-from-origin=%" PRId64
,
403 begin
->ns_from_origin
);
408 if (!end
->is_infinite
&& end
->ns_from_origin
== INT64_MIN
) {
409 BT_COMP_LOGE("Invalid trimming time range's end time: "
410 "ns-from-origin=%" PRId64
,
411 end
->ns_from_origin
);
421 int init_trimmer_comp_from_params(struct trimmer_comp
*trimmer_comp
,
422 const bt_value
*params
)
424 const bt_value
*value
;
428 value
= bt_value_map_borrow_entry_value_const(params
, "gmt");
430 trimmer_comp
->is_gmt
= (bool) bt_value_bool_get(value
);
433 value
= bt_value_map_borrow_entry_value_const(params
, "begin");
435 if (set_bound_from_param(trimmer_comp
, "begin", value
,
436 &trimmer_comp
->begin
, trimmer_comp
->is_gmt
)) {
437 /* set_bound_from_param() logs errors */
442 trimmer_comp
->begin
.is_infinite
= true;
443 trimmer_comp
->begin
.is_set
= true;
446 value
= bt_value_map_borrow_entry_value_const(params
, "end");
448 if (set_bound_from_param(trimmer_comp
, "end", value
,
449 &trimmer_comp
->end
, trimmer_comp
->is_gmt
)) {
450 /* set_bound_from_param() logs errors */
455 trimmer_comp
->end
.is_infinite
= true;
456 trimmer_comp
->end
.is_set
= true;
460 if (trimmer_comp
->begin
.is_set
&& trimmer_comp
->end
.is_set
) {
461 /* validate_trimmer_bounds() logs errors */
462 ret
= validate_trimmer_bounds(trimmer_comp
,
463 &trimmer_comp
->begin
, &trimmer_comp
->end
);
469 bt_component_class_init_method_status
trimmer_init(
470 bt_self_component_filter
*self_comp_flt
,
471 const bt_value
*params
, void *init_data
)
474 bt_component_class_init_method_status status
=
475 BT_COMPONENT_CLASS_INIT_METHOD_STATUS_OK
;
476 bt_self_component_add_port_status add_port_status
;
477 struct trimmer_comp
*trimmer_comp
= create_trimmer_comp();
478 bt_self_component
*self_comp
=
479 bt_self_component_filter_as_self_component(self_comp_flt
);
481 status
= BT_COMPONENT_CLASS_INIT_METHOD_STATUS_MEMORY_ERROR
;
485 trimmer_comp
->log_level
= bt_component_get_logging_level(
486 bt_self_component_as_component(self_comp
));
487 trimmer_comp
->self_comp
= self_comp
;
488 add_port_status
= bt_self_component_filter_add_input_port(
489 self_comp_flt
, in_port_name
, NULL
, NULL
);
490 switch (add_port_status
) {
491 case BT_SELF_COMPONENT_ADD_PORT_STATUS_ERROR
:
492 status
= BT_COMPONENT_CLASS_INIT_METHOD_STATUS_ERROR
;
494 case BT_SELF_COMPONENT_ADD_PORT_STATUS_MEMORY_ERROR
:
495 status
= BT_COMPONENT_CLASS_INIT_METHOD_STATUS_MEMORY_ERROR
;
501 add_port_status
= bt_self_component_filter_add_output_port(
502 self_comp_flt
, "out", NULL
, NULL
);
503 switch (add_port_status
) {
504 case BT_SELF_COMPONENT_ADD_PORT_STATUS_ERROR
:
505 status
= BT_COMPONENT_CLASS_INIT_METHOD_STATUS_ERROR
;
507 case BT_SELF_COMPONENT_ADD_PORT_STATUS_MEMORY_ERROR
:
508 status
= BT_COMPONENT_CLASS_INIT_METHOD_STATUS_MEMORY_ERROR
;
514 ret
= init_trimmer_comp_from_params(trimmer_comp
, params
);
516 status
= BT_COMPONENT_CLASS_INIT_METHOD_STATUS_ERROR
;
520 bt_self_component_set_data(self_comp
, trimmer_comp
);
524 if (status
== BT_COMPONENT_CLASS_INIT_METHOD_STATUS_OK
) {
525 status
= BT_COMPONENT_CLASS_INIT_METHOD_STATUS_ERROR
;
529 destroy_trimmer_comp(trimmer_comp
);
537 void destroy_trimmer_iterator(struct trimmer_iterator
*trimmer_it
)
539 BT_ASSERT(trimmer_it
);
540 bt_self_component_port_input_message_iterator_put_ref(
541 trimmer_it
->upstream_iter
);
543 if (trimmer_it
->output_messages
) {
544 g_queue_free(trimmer_it
->output_messages
);
547 if (trimmer_it
->stream_states
) {
548 g_hash_table_destroy(trimmer_it
->stream_states
);
555 void destroy_trimmer_iterator_stream_state(
556 struct trimmer_iterator_stream_state
*sstate
)
559 BT_PACKET_PUT_REF_AND_RESET(sstate
->cur_packet
);
564 bt_component_class_message_iterator_init_method_status
trimmer_msg_iter_init(
565 bt_self_message_iterator
*self_msg_iter
,
566 bt_self_component_filter
*self_comp
,
567 bt_self_component_port_output
*port
)
569 bt_component_class_message_iterator_init_method_status status
=
570 BT_COMPONENT_CLASS_MESSAGE_ITERATOR_INIT_METHOD_STATUS_OK
;
571 struct trimmer_iterator
*trimmer_it
;
573 trimmer_it
= g_new0(struct trimmer_iterator
, 1);
575 status
= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_INIT_METHOD_STATUS_MEMORY_ERROR
;
579 trimmer_it
->trimmer_comp
= bt_self_component_get_data(
580 bt_self_component_filter_as_self_component(self_comp
));
581 BT_ASSERT(trimmer_it
->trimmer_comp
);
583 if (trimmer_it
->trimmer_comp
->begin
.is_set
&&
584 trimmer_it
->trimmer_comp
->end
.is_set
) {
586 * Both trimming time range's bounds are set, so skip
588 * `TRIMMER_ITERATOR_STATE_SET_BOUNDS_NS_FROM_ORIGIN`
591 trimmer_it
->state
= TRIMMER_ITERATOR_STATE_SEEK_INITIALLY
;
594 trimmer_it
->begin
= trimmer_it
->trimmer_comp
->begin
;
595 trimmer_it
->end
= trimmer_it
->trimmer_comp
->end
;
596 trimmer_it
->upstream_iter
=
597 bt_self_component_port_input_message_iterator_create(
598 bt_self_component_filter_borrow_input_port_by_name(
599 self_comp
, in_port_name
));
600 if (!trimmer_it
->upstream_iter
) {
601 status
= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_INIT_METHOD_STATUS_ERROR
;
605 trimmer_it
->output_messages
= g_queue_new();
606 if (!trimmer_it
->output_messages
) {
607 status
= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_INIT_METHOD_STATUS_MEMORY_ERROR
;
611 trimmer_it
->stream_states
= g_hash_table_new_full(g_direct_hash
,
612 g_direct_equal
, NULL
,
613 (GDestroyNotify
) destroy_trimmer_iterator_stream_state
);
614 if (!trimmer_it
->stream_states
) {
615 status
= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_INIT_METHOD_STATUS_MEMORY_ERROR
;
619 trimmer_it
->self_msg_iter
= self_msg_iter
;
620 bt_self_message_iterator_set_data(self_msg_iter
, trimmer_it
);
623 if (status
!= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_INIT_METHOD_STATUS_OK
&& trimmer_it
) {
624 destroy_trimmer_iterator(trimmer_it
);
631 int get_msg_ns_from_origin(const bt_message
*msg
, int64_t *ns_from_origin
,
634 const bt_clock_class
*clock_class
= NULL
;
635 const bt_clock_snapshot
*clock_snapshot
= NULL
;
636 bt_message_stream_activity_clock_snapshot_state sa_cs_state
;
640 BT_ASSERT(ns_from_origin
);
643 switch (bt_message_get_type(msg
)) {
644 case BT_MESSAGE_TYPE_EVENT
:
646 bt_message_event_borrow_stream_class_default_clock_class_const(
648 if (G_UNLIKELY(!clock_class
)) {
652 clock_snapshot
= bt_message_event_borrow_default_clock_snapshot_const(
655 case BT_MESSAGE_TYPE_PACKET_BEGINNING
:
657 bt_message_packet_beginning_borrow_stream_class_default_clock_class_const(
659 if (G_UNLIKELY(!clock_class
)) {
663 clock_snapshot
= bt_message_packet_beginning_borrow_default_clock_snapshot_const(
666 case BT_MESSAGE_TYPE_PACKET_END
:
668 bt_message_packet_end_borrow_stream_class_default_clock_class_const(
670 if (G_UNLIKELY(!clock_class
)) {
674 clock_snapshot
= bt_message_packet_end_borrow_default_clock_snapshot_const(
677 case BT_MESSAGE_TYPE_DISCARDED_EVENTS
:
679 bt_message_discarded_events_borrow_stream_class_default_clock_class_const(
681 if (G_UNLIKELY(!clock_class
)) {
685 clock_snapshot
= bt_message_discarded_events_borrow_beginning_default_clock_snapshot_const(
688 case BT_MESSAGE_TYPE_DISCARDED_PACKETS
:
690 bt_message_discarded_packets_borrow_stream_class_default_clock_class_const(
692 if (G_UNLIKELY(!clock_class
)) {
696 clock_snapshot
= bt_message_discarded_packets_borrow_beginning_default_clock_snapshot_const(
699 case BT_MESSAGE_TYPE_STREAM_ACTIVITY_BEGINNING
:
701 bt_message_stream_activity_beginning_borrow_stream_class_default_clock_class_const(
703 if (G_UNLIKELY(!clock_class
)) {
707 sa_cs_state
= bt_message_stream_activity_beginning_borrow_default_clock_snapshot_const(
708 msg
, &clock_snapshot
);
709 if (sa_cs_state
== BT_MESSAGE_STREAM_ACTIVITY_CLOCK_SNAPSHOT_STATE_UNKNOWN
||
710 sa_cs_state
== BT_MESSAGE_STREAM_ACTIVITY_CLOCK_SNAPSHOT_STATE_INFINITE
) {
711 /* Lowest possible time to always include them */
712 *ns_from_origin
= INT64_MIN
;
713 goto no_clock_snapshot
;
717 case BT_MESSAGE_TYPE_STREAM_ACTIVITY_END
:
719 bt_message_stream_activity_end_borrow_stream_class_default_clock_class_const(
721 if (G_UNLIKELY(!clock_class
)) {
725 sa_cs_state
= bt_message_stream_activity_end_borrow_default_clock_snapshot_const(
726 msg
, &clock_snapshot
);
727 if (sa_cs_state
== BT_MESSAGE_STREAM_ACTIVITY_CLOCK_SNAPSHOT_STATE_UNKNOWN
) {
728 /* Lowest time to always include it */
729 *ns_from_origin
= INT64_MIN
;
730 goto no_clock_snapshot
;
731 } else if (sa_cs_state
== BT_MESSAGE_STREAM_ACTIVITY_CLOCK_SNAPSHOT_STATE_INFINITE
) {
732 /* Greatest time to always exclude it */
733 *ns_from_origin
= INT64_MAX
;
734 goto no_clock_snapshot
;
738 case BT_MESSAGE_TYPE_MESSAGE_ITERATOR_INACTIVITY
:
740 bt_message_message_iterator_inactivity_borrow_default_clock_snapshot_const(
744 goto no_clock_snapshot
;
747 ret
= bt_clock_snapshot_get_ns_from_origin(clock_snapshot
,
749 if (G_UNLIKELY(ret
)) {
767 void put_messages(bt_message_array_const msgs
, uint64_t count
)
771 for (i
= 0; i
< count
; i
++) {
772 BT_MESSAGE_PUT_REF_AND_RESET(msgs
[i
]);
777 int set_trimmer_iterator_bound(struct trimmer_iterator
*trimmer_it
,
778 struct trimmer_bound
*bound
, int64_t ns_from_origin
,
781 struct trimmer_comp
*trimmer_comp
= trimmer_it
->trimmer_comp
;
783 time_t time_seconds
= (time_t) (ns_from_origin
/ NS_PER_S
);
786 BT_ASSERT(!bound
->is_set
);
789 /* We only need to extract the date from this time */
791 bt_gmtime_r(&time_seconds
, &tm
);
793 bt_localtime_r(&time_seconds
, &tm
);
797 BT_COMP_LOGE_ERRNO("Cannot convert timestamp to date and time",
798 "ts=%" PRId64
, (int64_t) time_seconds
);
803 ret
= set_bound_ns_from_origin(bound
, tm
.tm_year
+ 1900, tm
.tm_mon
+ 1,
804 tm
.tm_mday
, bound
->time
.hour
, bound
->time
.minute
,
805 bound
->time
.second
, bound
->time
.ns
, is_gmt
);
812 bt_component_class_message_iterator_next_method_status
813 state_set_trimmer_iterator_bounds(
814 struct trimmer_iterator
*trimmer_it
)
816 bt_message_iterator_next_status upstream_iter_status
=
817 BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK
;
818 struct trimmer_comp
*trimmer_comp
= trimmer_it
->trimmer_comp
;
819 bt_message_array_const msgs
;
821 int64_t ns_from_origin
= INT64_MIN
;
825 BT_ASSERT(!trimmer_it
->begin
.is_set
||
826 !trimmer_it
->end
.is_set
);
829 upstream_iter_status
=
830 bt_self_component_port_input_message_iterator_next(
831 trimmer_it
->upstream_iter
, &msgs
, &count
);
832 if (upstream_iter_status
!= BT_MESSAGE_ITERATOR_NEXT_STATUS_OK
) {
836 for (i
= 0; i
< count
; i
++) {
837 const bt_message
*msg
= msgs
[i
];
841 ret
= get_msg_ns_from_origin(msg
, &ns_from_origin
,
851 BT_ASSERT(ns_from_origin
!= INT64_MIN
&&
852 ns_from_origin
!= INT64_MAX
);
853 put_messages(msgs
, count
);
857 put_messages(msgs
, count
);
861 if (!trimmer_it
->begin
.is_set
) {
862 BT_ASSERT(!trimmer_it
->begin
.is_infinite
);
863 ret
= set_trimmer_iterator_bound(trimmer_it
, &trimmer_it
->begin
,
864 ns_from_origin
, trimmer_comp
->is_gmt
);
870 if (!trimmer_it
->end
.is_set
) {
871 BT_ASSERT(!trimmer_it
->end
.is_infinite
);
872 ret
= set_trimmer_iterator_bound(trimmer_it
, &trimmer_it
->end
,
873 ns_from_origin
, trimmer_comp
->is_gmt
);
879 ret
= validate_trimmer_bounds(trimmer_it
->trimmer_comp
,
880 &trimmer_it
->begin
, &trimmer_it
->end
);
888 put_messages(msgs
, count
);
889 upstream_iter_status
=
890 BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_ERROR
;
893 return (int) upstream_iter_status
;
897 bt_component_class_message_iterator_next_method_status
state_seek_initially(
898 struct trimmer_iterator
*trimmer_it
)
900 struct trimmer_comp
*trimmer_comp
= trimmer_it
->trimmer_comp
;
901 bt_component_class_message_iterator_next_method_status status
=
902 BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK
;
904 BT_ASSERT(trimmer_it
->begin
.is_set
);
906 if (trimmer_it
->begin
.is_infinite
) {
907 if (!bt_self_component_port_input_message_iterator_can_seek_beginning(
908 trimmer_it
->upstream_iter
)) {
909 BT_COMP_LOGE_STR("Cannot make upstream message iterator initially seek its beginning.");
910 status
= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_ERROR
;
914 status
= (int) bt_self_component_port_input_message_iterator_seek_beginning(
915 trimmer_it
->upstream_iter
);
917 if (!bt_self_component_port_input_message_iterator_can_seek_ns_from_origin(
918 trimmer_it
->upstream_iter
,
919 trimmer_it
->begin
.ns_from_origin
)) {
920 BT_COMP_LOGE("Cannot make upstream message iterator initially seek: "
921 "seek-ns-from-origin=%" PRId64
,
922 trimmer_it
->begin
.ns_from_origin
);
923 status
= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_ERROR
;
927 status
= (int) bt_self_component_port_input_message_iterator_seek_ns_from_origin(
928 trimmer_it
->upstream_iter
, trimmer_it
->begin
.ns_from_origin
);
931 if (status
== BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK
) {
932 trimmer_it
->state
= TRIMMER_ITERATOR_STATE_TRIM
;
940 void push_message(struct trimmer_iterator
*trimmer_it
, const bt_message
*msg
)
942 g_queue_push_head(trimmer_it
->output_messages
, (void *) msg
);
946 const bt_message
*pop_message(struct trimmer_iterator
*trimmer_it
)
948 return g_queue_pop_tail(trimmer_it
->output_messages
);
952 int clock_raw_value_from_ns_from_origin(const bt_clock_class
*clock_class
,
953 int64_t ns_from_origin
, uint64_t *raw_value
)
957 uint64_t cc_offset_cycles
;
960 bt_clock_class_get_offset(clock_class
, &cc_offset_s
, &cc_offset_cycles
);
961 cc_freq
= bt_clock_class_get_frequency(clock_class
);
962 return bt_common_clock_value_from_ns_from_origin(cc_offset_s
,
963 cc_offset_cycles
, cc_freq
, ns_from_origin
, raw_value
);
967 bt_component_class_message_iterator_next_method_status
968 end_stream(struct trimmer_iterator
*trimmer_it
,
969 struct trimmer_iterator_stream_state
*sstate
)
971 bt_component_class_message_iterator_next_method_status status
=
972 BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK
;
974 const bt_clock_class
*clock_class
;
976 bt_message
*msg
= NULL
;
978 BT_ASSERT(!trimmer_it
->end
.is_infinite
);
980 if (!sstate
->stream
) {
984 if (sstate
->cur_packet
) {
986 * The last message could not have been a stream
987 * activity end message if we have a current packet.
989 BT_ASSERT(!sstate
->last_msg_is_stream_activity_end
);
992 * Create and push a packet end message, making its time
993 * the trimming range's end time.
995 clock_class
= bt_stream_class_borrow_default_clock_class_const(
996 bt_stream_borrow_class_const(sstate
->stream
));
997 BT_ASSERT(clock_class
);
998 ret
= clock_raw_value_from_ns_from_origin(clock_class
,
999 trimmer_it
->end
.ns_from_origin
, &raw_value
);
1001 status
= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_ERROR
;
1005 msg
= bt_message_packet_end_create_with_default_clock_snapshot(
1006 trimmer_it
->self_msg_iter
, sstate
->cur_packet
,
1009 status
= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_MEMORY_ERROR
;
1013 push_message(trimmer_it
, msg
);
1015 BT_PACKET_PUT_REF_AND_RESET(sstate
->cur_packet
);
1018 * Because we generated a packet end message, set the
1019 * stream activity end message's time to use to the
1020 * trimming range's end time (this packet end message's
1023 sstate
->stream_act_end_ns_from_origin
=
1024 trimmer_it
->end
.ns_from_origin
;
1027 if (!sstate
->last_msg_is_stream_activity_end
) {
1028 /* Create and push a stream activity end message */
1029 msg
= bt_message_stream_activity_end_create(
1030 trimmer_it
->self_msg_iter
, sstate
->stream
);
1032 status
= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_MEMORY_ERROR
;
1036 clock_class
= bt_stream_class_borrow_default_clock_class_const(
1037 bt_stream_borrow_class_const(sstate
->stream
));
1038 BT_ASSERT(clock_class
);
1040 if (sstate
->stream_act_end_ns_from_origin
== INT64_MIN
) {
1042 * We received at least what is necessary to
1043 * have a stream state (stream beginning and
1044 * stream activity beginning messages), but
1045 * nothing else: use the trimmer range's end
1048 sstate
->stream_act_end_ns_from_origin
=
1049 trimmer_it
->end
.ns_from_origin
;
1052 ret
= clock_raw_value_from_ns_from_origin(clock_class
,
1053 sstate
->stream_act_end_ns_from_origin
, &raw_value
);
1055 status
= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_ERROR
;
1059 bt_message_stream_activity_end_set_default_clock_snapshot(
1061 push_message(trimmer_it
, msg
);
1065 /* Create and push a stream end message */
1066 msg
= bt_message_stream_end_create(trimmer_it
->self_msg_iter
,
1069 status
= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_MEMORY_ERROR
;
1073 push_message(trimmer_it
, msg
);
1077 * Just to make sure that we don't use this stream state again
1078 * in the future without an obvious error.
1080 sstate
->stream
= NULL
;
1083 bt_message_put_ref(msg
);
1088 bt_component_class_message_iterator_next_method_status
end_iterator_streams(
1089 struct trimmer_iterator
*trimmer_it
)
1091 bt_component_class_message_iterator_next_method_status status
=
1092 BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK
;
1093 GHashTableIter iter
;
1094 gpointer key
, sstate
;
1096 if (trimmer_it
->end
.is_infinite
) {
1098 * An infinite trimming range's end time guarantees that
1099 * we received (and pushed) all the appropriate end
1106 * End each stream and then remove them from the hash table of
1107 * stream states to release unneeded references.
1109 g_hash_table_iter_init(&iter
, trimmer_it
->stream_states
);
1111 while (g_hash_table_iter_next(&iter
, &key
, &sstate
)) {
1112 status
= end_stream(trimmer_it
, sstate
);
1119 g_hash_table_remove_all(trimmer_it
->stream_states
);
1126 bt_component_class_message_iterator_next_method_status
1127 create_stream_beginning_activity_message(
1128 struct trimmer_iterator
*trimmer_it
,
1129 const bt_stream
*stream
,
1130 const bt_clock_class
*clock_class
, bt_message
**msg
)
1132 bt_message
*local_msg
;
1133 bt_component_class_message_iterator_next_method_status status
=
1134 BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK
;
1137 BT_ASSERT(!trimmer_it
->begin
.is_infinite
);
1139 local_msg
= bt_message_stream_activity_beginning_create(
1140 trimmer_it
->self_msg_iter
, stream
);
1142 status
= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_MEMORY_ERROR
;
1150 ret
= clock_raw_value_from_ns_from_origin(clock_class
,
1151 trimmer_it
->begin
.ns_from_origin
, &raw_value
);
1153 status
= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_ERROR
;
1154 bt_message_put_ref(local_msg
);
1158 bt_message_stream_activity_beginning_set_default_clock_snapshot(
1159 local_msg
, raw_value
);
1162 BT_MESSAGE_MOVE_REF(*msg
, local_msg
);
1169 * Handles a message which is associated to a given stream state. This
1170 * _could_ make the iterator's output message queue grow; this could
1171 * also consume the message without pushing anything to this queue, only
1172 * modifying the stream state.
1174 * This function consumes the `msg` reference, _whatever the outcome_.
1176 * `ns_from_origin` is the message's time, as given by
1177 * get_msg_ns_from_origin().
1179 * This function sets `reached_end` if handling this message made the
1180 * iterator reach the end of the trimming range. Note that the output
1181 * message queue could contain messages even if this function sets
1185 bt_component_class_message_iterator_next_method_status
1186 handle_message_with_stream_state(
1187 struct trimmer_iterator
*trimmer_it
, const bt_message
*msg
,
1188 struct trimmer_iterator_stream_state
*sstate
,
1189 int64_t ns_from_origin
, bool *reached_end
)
1191 bt_component_class_message_iterator_next_method_status status
=
1192 BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK
;
1193 bt_message_type msg_type
= bt_message_get_type(msg
);
1197 case BT_MESSAGE_TYPE_EVENT
:
1198 if (G_UNLIKELY(!trimmer_it
->end
.is_infinite
&&
1199 ns_from_origin
> trimmer_it
->end
.ns_from_origin
)) {
1200 status
= end_iterator_streams(trimmer_it
);
1201 *reached_end
= true;
1205 BT_ASSERT(sstate
->cur_packet
);
1206 push_message(trimmer_it
, msg
);
1209 case BT_MESSAGE_TYPE_PACKET_BEGINNING
:
1210 if (G_UNLIKELY(!trimmer_it
->end
.is_infinite
&&
1211 ns_from_origin
> trimmer_it
->end
.ns_from_origin
)) {
1212 status
= end_iterator_streams(trimmer_it
);
1213 *reached_end
= true;
1217 BT_ASSERT(!sstate
->cur_packet
);
1218 sstate
->cur_packet
=
1219 bt_message_packet_beginning_borrow_packet_const(msg
);
1220 bt_packet_get_ref(sstate
->cur_packet
);
1221 push_message(trimmer_it
, msg
);
1224 case BT_MESSAGE_TYPE_PACKET_END
:
1225 sstate
->stream_act_end_ns_from_origin
= ns_from_origin
;
1227 if (G_UNLIKELY(!trimmer_it
->end
.is_infinite
&&
1228 ns_from_origin
> trimmer_it
->end
.ns_from_origin
)) {
1229 status
= end_iterator_streams(trimmer_it
);
1230 *reached_end
= true;
1234 BT_ASSERT(sstate
->cur_packet
);
1235 BT_PACKET_PUT_REF_AND_RESET(sstate
->cur_packet
);
1236 push_message(trimmer_it
, msg
);
1239 case BT_MESSAGE_TYPE_DISCARDED_EVENTS
:
1240 case BT_MESSAGE_TYPE_DISCARDED_PACKETS
:
1243 * `ns_from_origin` is the message's time range's
1244 * beginning time here.
1246 int64_t end_ns_from_origin
;
1247 const bt_clock_snapshot
*end_cs
;
1249 if (bt_message_get_type(msg
) ==
1250 BT_MESSAGE_TYPE_DISCARDED_EVENTS
) {
1252 * Safe to ignore the return value because we
1253 * know there's a default clock and it's always
1256 end_cs
= bt_message_discarded_events_borrow_end_default_clock_snapshot_const(
1260 * Safe to ignore the return value because we
1261 * know there's a default clock and it's always
1264 end_cs
= bt_message_discarded_packets_borrow_end_default_clock_snapshot_const(
1268 if (bt_clock_snapshot_get_ns_from_origin(end_cs
,
1269 &end_ns_from_origin
)) {
1270 status
= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_ERROR
;
1274 sstate
->stream_act_end_ns_from_origin
= end_ns_from_origin
;
1276 if (!trimmer_it
->end
.is_infinite
&&
1277 ns_from_origin
> trimmer_it
->end
.ns_from_origin
) {
1278 status
= end_iterator_streams(trimmer_it
);
1279 *reached_end
= true;
1283 if (!trimmer_it
->end
.is_infinite
&&
1284 end_ns_from_origin
> trimmer_it
->end
.ns_from_origin
) {
1286 * This message's end time is outside the
1287 * trimming time range: replace it with a new
1288 * message having an end time equal to the
1289 * trimming time range's end and without a
1292 const bt_clock_class
*clock_class
=
1293 bt_clock_snapshot_borrow_clock_class_const(
1295 const bt_clock_snapshot
*begin_cs
;
1296 bt_message
*new_msg
;
1297 uint64_t end_raw_value
;
1299 ret
= clock_raw_value_from_ns_from_origin(clock_class
,
1300 trimmer_it
->end
.ns_from_origin
, &end_raw_value
);
1302 status
= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_ERROR
;
1306 if (msg_type
== BT_MESSAGE_TYPE_DISCARDED_EVENTS
) {
1307 begin_cs
= bt_message_discarded_events_borrow_beginning_default_clock_snapshot_const(
1309 new_msg
= bt_message_discarded_events_create_with_default_clock_snapshots(
1310 trimmer_it
->self_msg_iter
,
1312 bt_clock_snapshot_get_value(begin_cs
),
1315 begin_cs
= bt_message_discarded_packets_borrow_beginning_default_clock_snapshot_const(
1317 new_msg
= bt_message_discarded_packets_create_with_default_clock_snapshots(
1318 trimmer_it
->self_msg_iter
,
1320 bt_clock_snapshot_get_value(begin_cs
),
1325 status
= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_MEMORY_ERROR
;
1329 /* Replace the original message */
1330 BT_MESSAGE_MOVE_REF(msg
, new_msg
);
1333 push_message(trimmer_it
, msg
);
1337 case BT_MESSAGE_TYPE_STREAM_ACTIVITY_BEGINNING
:
1338 if (!trimmer_it
->end
.is_infinite
&&
1339 ns_from_origin
> trimmer_it
->end
.ns_from_origin
) {
1341 * This only happens when the message's time is
1342 * known and is greater than the trimming
1343 * range's end time. Unknown and -inf times are
1345 * `trimmer_it->end.ns_from_origin`.
1347 status
= end_iterator_streams(trimmer_it
);
1348 *reached_end
= true;
1352 push_message(trimmer_it
, msg
);
1355 case BT_MESSAGE_TYPE_STREAM_ACTIVITY_END
:
1356 if (trimmer_it
->end
.is_infinite
) {
1357 push_message(trimmer_it
, msg
);
1362 if (ns_from_origin
== INT64_MIN
) {
1363 /* Unknown: consider it to be in the trimmer window. */
1364 push_message(trimmer_it
, msg
);
1366 sstate
->last_msg_is_stream_activity_end
= true;
1367 } else if (ns_from_origin
== INT64_MAX
) {
1368 /* Infinite: use trimming range's end time */
1369 sstate
->stream_act_end_ns_from_origin
=
1370 trimmer_it
->end
.ns_from_origin
;
1372 /* Known: check if outside of trimming range */
1373 if (ns_from_origin
> trimmer_it
->end
.ns_from_origin
) {
1374 sstate
->stream_act_end_ns_from_origin
=
1375 trimmer_it
->end
.ns_from_origin
;
1376 status
= end_iterator_streams(trimmer_it
);
1377 *reached_end
= true;
1381 push_message(trimmer_it
, msg
);
1383 sstate
->last_msg_is_stream_activity_end
= true;
1384 sstate
->stream_act_end_ns_from_origin
= ns_from_origin
;
1388 case BT_MESSAGE_TYPE_STREAM_BEGINNING
:
1389 push_message(trimmer_it
, msg
);
1392 case BT_MESSAGE_TYPE_STREAM_END
:
1394 * This is the end of a stream: end this
1395 * stream if its stream activity end message
1396 * time is not the trimming range's end time
1397 * (which means the final stream activity end
1398 * message had an infinite time). end_stream()
1399 * will generate its own stream end message.
1401 if (trimmer_it
->end
.is_infinite
) {
1402 push_message(trimmer_it
, msg
);
1405 /* We won't need this stream state again */
1406 g_hash_table_remove(trimmer_it
->stream_states
, sstate
->stream
);
1407 } else if (sstate
->stream_act_end_ns_from_origin
<
1408 trimmer_it
->end
.ns_from_origin
) {
1409 status
= end_stream(trimmer_it
, sstate
);
1410 if (status
!= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK
) {
1414 /* We won't need this stream state again */
1415 g_hash_table_remove(trimmer_it
->stream_states
, sstate
->stream
);
1423 /* We release the message's reference whatever the outcome */
1424 bt_message_put_ref(msg
);
1425 return BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK
;
1429 * Handles an input message. This _could_ make the iterator's output
1430 * message queue grow; this could also consume the message without
1431 * pushing anything to this queue, only modifying the stream state.
1433 * This function consumes the `msg` reference, _whatever the outcome_.
1435 * This function sets `reached_end` if handling this message made the
1436 * iterator reach the end of the trimming range. Note that the output
1437 * message queue could contain messages even if this function sets
1441 bt_component_class_message_iterator_next_method_status
handle_message(
1442 struct trimmer_iterator
*trimmer_it
, const bt_message
*msg
,
1445 bt_component_class_message_iterator_next_method_status status
;
1446 const bt_stream
*stream
= NULL
;
1447 int64_t ns_from_origin
= INT64_MIN
;
1450 struct trimmer_iterator_stream_state
*sstate
= NULL
;
1451 struct trimmer_comp
*trimmer_comp
= trimmer_it
->trimmer_comp
;
1453 /* Find message's associated stream */
1454 switch (bt_message_get_type(msg
)) {
1455 case BT_MESSAGE_TYPE_EVENT
:
1456 stream
= bt_event_borrow_stream_const(
1457 bt_message_event_borrow_event_const(msg
));
1459 case BT_MESSAGE_TYPE_PACKET_BEGINNING
:
1460 stream
= bt_packet_borrow_stream_const(
1461 bt_message_packet_beginning_borrow_packet_const(msg
));
1463 case BT_MESSAGE_TYPE_PACKET_END
:
1464 stream
= bt_packet_borrow_stream_const(
1465 bt_message_packet_end_borrow_packet_const(msg
));
1467 case BT_MESSAGE_TYPE_DISCARDED_EVENTS
:
1468 stream
= bt_message_discarded_events_borrow_stream_const(msg
);
1470 case BT_MESSAGE_TYPE_DISCARDED_PACKETS
:
1471 stream
= bt_message_discarded_packets_borrow_stream_const(msg
);
1473 case BT_MESSAGE_TYPE_STREAM_ACTIVITY_BEGINNING
:
1474 stream
= bt_message_stream_activity_beginning_borrow_stream_const(msg
);
1476 case BT_MESSAGE_TYPE_STREAM_ACTIVITY_END
:
1477 stream
= bt_message_stream_activity_end_borrow_stream_const(msg
);
1479 case BT_MESSAGE_TYPE_STREAM_BEGINNING
:
1480 stream
= bt_message_stream_beginning_borrow_stream_const(msg
);
1482 case BT_MESSAGE_TYPE_STREAM_END
:
1483 stream
= bt_message_stream_end_borrow_stream_const(msg
);
1489 if (G_LIKELY(stream
)) {
1490 /* Find stream state */
1491 sstate
= g_hash_table_lookup(trimmer_it
->stream_states
,
1493 if (G_UNLIKELY(!sstate
)) {
1494 /* No stream state yet: create one now */
1495 const bt_stream_class
*sc
;
1498 * Validate right now that the stream's class
1499 * has a registered default clock class so that
1500 * an existing stream state guarantees existing
1501 * default clock snapshots for its associated
1504 * Also check that clock snapshots are always
1507 sc
= bt_stream_borrow_class_const(stream
);
1508 if (!bt_stream_class_borrow_default_clock_class_const(sc
)) {
1509 BT_COMP_LOGE("Unsupported stream: stream class does "
1510 "not have a default clock class: "
1512 "stream-id=%" PRIu64
", "
1513 "stream-name=\"%s\"",
1514 stream
, bt_stream_get_id(stream
),
1515 bt_stream_get_name(stream
));
1516 status
= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_ERROR
;
1521 * Temporary: make sure packet beginning, packet
1522 * end, discarded events, and discarded packets
1523 * messages have default clock snapshots until
1524 * the support for not having them is
1527 if (!bt_stream_class_packets_have_beginning_default_clock_snapshot(
1529 BT_COMP_LOGE("Unsupported stream: packets have "
1530 "no beginning clock snapshot: "
1532 "stream-id=%" PRIu64
", "
1533 "stream-name=\"%s\"",
1534 stream
, bt_stream_get_id(stream
),
1535 bt_stream_get_name(stream
));
1536 status
= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_ERROR
;
1540 if (!bt_stream_class_packets_have_end_default_clock_snapshot(
1542 BT_COMP_LOGE("Unsupported stream: packets have "
1543 "no end clock snapshot: "
1545 "stream-id=%" PRIu64
", "
1546 "stream-name=\"%s\"",
1547 stream
, bt_stream_get_id(stream
),
1548 bt_stream_get_name(stream
));
1549 status
= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_ERROR
;
1553 if (bt_stream_class_supports_discarded_events(sc
) &&
1554 !bt_stream_class_discarded_events_have_default_clock_snapshots(sc
)) {
1555 BT_COMP_LOGE("Unsupported stream: discarded events "
1556 "have no clock snapshots: "
1558 "stream-id=%" PRIu64
", "
1559 "stream-name=\"%s\"",
1560 stream
, bt_stream_get_id(stream
),
1561 bt_stream_get_name(stream
));
1562 status
= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_ERROR
;
1566 if (bt_stream_class_supports_discarded_packets(sc
) &&
1567 !bt_stream_class_discarded_packets_have_default_clock_snapshots(sc
)) {
1568 BT_COMP_LOGE("Unsupported stream: discarded packets "
1569 "have no clock snapshots: "
1571 "stream-id=%" PRIu64
", "
1572 "stream-name=\"%s\"",
1573 stream
, bt_stream_get_id(stream
),
1574 bt_stream_get_name(stream
));
1575 status
= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_ERROR
;
1579 sstate
= g_new0(struct trimmer_iterator_stream_state
,
1582 status
= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_MEMORY_ERROR
;
1586 sstate
->stream
= stream
;
1587 sstate
->stream_act_end_ns_from_origin
= INT64_MIN
;
1588 g_hash_table_insert(trimmer_it
->stream_states
,
1589 (void *) stream
, sstate
);
1593 /* Retrieve the message's time */
1594 ret
= get_msg_ns_from_origin(msg
, &ns_from_origin
, &skip
);
1595 if (G_UNLIKELY(ret
)) {
1596 status
= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_ERROR
;
1600 if (G_LIKELY(sstate
)) {
1601 /* Message associated to a stream */
1602 status
= handle_message_with_stream_state(trimmer_it
, msg
,
1603 sstate
, ns_from_origin
, reached_end
);
1606 * handle_message_with_stream_state() unconditionally
1612 * Message not associated to a stream (message iterator
1615 if (G_UNLIKELY(ns_from_origin
> trimmer_it
->end
.ns_from_origin
)) {
1616 BT_MESSAGE_PUT_REF_AND_RESET(msg
);
1617 status
= end_iterator_streams(trimmer_it
);
1618 *reached_end
= true;
1620 push_message(trimmer_it
, msg
);
1621 status
= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK
;
1627 /* We release the message's reference whatever the outcome */
1628 bt_message_put_ref(msg
);
1633 void fill_message_array_from_output_messages(
1634 struct trimmer_iterator
*trimmer_it
,
1635 bt_message_array_const msgs
, uint64_t capacity
, uint64_t *count
)
1640 * Move auto-seek messages to the output array (which is this
1641 * iterator's base message array).
1643 while (capacity
> 0 && !g_queue_is_empty(trimmer_it
->output_messages
)) {
1644 msgs
[*count
] = pop_message(trimmer_it
);
1649 BT_ASSERT(*count
> 0);
1653 bt_component_class_message_iterator_next_method_status
state_ending(
1654 struct trimmer_iterator
*trimmer_it
,
1655 bt_message_array_const msgs
, uint64_t capacity
,
1658 bt_component_class_message_iterator_next_method_status status
=
1659 BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK
;
1661 if (g_queue_is_empty(trimmer_it
->output_messages
)) {
1662 trimmer_it
->state
= TRIMMER_ITERATOR_STATE_ENDED
;
1663 status
= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_END
;
1667 fill_message_array_from_output_messages(trimmer_it
, msgs
,
1675 bt_component_class_message_iterator_next_method_status
1676 state_trim(struct trimmer_iterator
*trimmer_it
,
1677 bt_message_array_const msgs
, uint64_t capacity
,
1680 bt_component_class_message_iterator_next_method_status status
=
1681 BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK
;
1682 bt_message_array_const my_msgs
;
1685 bool reached_end
= false;
1687 while (g_queue_is_empty(trimmer_it
->output_messages
)) {
1688 status
= (int) bt_self_component_port_input_message_iterator_next(
1689 trimmer_it
->upstream_iter
, &my_msgs
, &my_count
);
1690 if (G_UNLIKELY(status
!= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK
)) {
1691 if (status
== BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_END
) {
1692 status
= end_iterator_streams(trimmer_it
);
1693 if (status
!= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK
) {
1698 TRIMMER_ITERATOR_STATE_ENDING
;
1699 status
= state_ending(trimmer_it
, msgs
,
1706 BT_ASSERT(my_count
> 0);
1708 for (i
= 0; i
< my_count
; i
++) {
1709 status
= handle_message(trimmer_it
, my_msgs
[i
],
1713 * handle_message() unconditionally consumes the
1714 * message reference.
1718 if (G_UNLIKELY(status
!=
1719 BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK
)) {
1720 put_messages(my_msgs
, my_count
);
1724 if (G_UNLIKELY(reached_end
)) {
1726 * This message's time was passed the
1727 * trimming time range's end time: we
1728 * are done. Their might still be
1729 * messages in the output message queue,
1730 * so move to the "ending" state and
1731 * apply it immediately since
1732 * state_trim() is called within the
1735 put_messages(my_msgs
, my_count
);
1737 TRIMMER_ITERATOR_STATE_ENDING
;
1738 status
= state_ending(trimmer_it
, msgs
,
1746 * There's at least one message in the output message queue:
1747 * move the messages to the output message array.
1749 BT_ASSERT(!g_queue_is_empty(trimmer_it
->output_messages
));
1750 fill_message_array_from_output_messages(trimmer_it
, msgs
,
1758 bt_component_class_message_iterator_next_method_status
trimmer_msg_iter_next(
1759 bt_self_message_iterator
*self_msg_iter
,
1760 bt_message_array_const msgs
, uint64_t capacity
,
1763 struct trimmer_iterator
*trimmer_it
=
1764 bt_self_message_iterator_get_data(self_msg_iter
);
1765 bt_component_class_message_iterator_next_method_status status
=
1766 BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK
;
1768 BT_ASSERT(trimmer_it
);
1770 if (G_LIKELY(trimmer_it
->state
== TRIMMER_ITERATOR_STATE_TRIM
)) {
1771 status
= state_trim(trimmer_it
, msgs
, capacity
, count
);
1772 if (status
!= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK
) {
1776 switch (trimmer_it
->state
) {
1777 case TRIMMER_ITERATOR_STATE_SET_BOUNDS_NS_FROM_ORIGIN
:
1778 status
= state_set_trimmer_iterator_bounds(trimmer_it
);
1779 if (status
!= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK
) {
1783 status
= state_seek_initially(trimmer_it
);
1784 if (status
!= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK
) {
1788 status
= state_trim(trimmer_it
, msgs
, capacity
, count
);
1789 if (status
!= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK
) {
1794 case TRIMMER_ITERATOR_STATE_SEEK_INITIALLY
:
1795 status
= state_seek_initially(trimmer_it
);
1796 if (status
!= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK
) {
1800 status
= state_trim(trimmer_it
, msgs
, capacity
, count
);
1801 if (status
!= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK
) {
1806 case TRIMMER_ITERATOR_STATE_ENDING
:
1807 status
= state_ending(trimmer_it
, msgs
, capacity
,
1809 if (status
!= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK
) {
1814 case TRIMMER_ITERATOR_STATE_ENDED
:
1815 status
= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_END
;
1827 void trimmer_msg_iter_finalize(bt_self_message_iterator
*self_msg_iter
)
1829 struct trimmer_iterator
*trimmer_it
=
1830 bt_self_message_iterator_get_data(self_msg_iter
);
1832 BT_ASSERT(trimmer_it
);
1833 destroy_trimmer_iterator(trimmer_it
);