projects / babeltrace.git / blob
? search:
summary | shortlog | log | commit | commitdiff | tree
blame | history | raw | HEAD
lib: strictly type function return status enumerations
[babeltrace.git] / src / plugins / utils / trimmer / trimmer.c
1 /*
2 * Copyright 2016 Jérémie Galarneau <jeremie.galarneau@efficios.com>
3 * Copyright 2019 Philippe Proulx <pproulx@efficios.com>
4 *
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:
11 *
12 * The above copyright notice and this permission notice shall be included in
13 * all copies or substantial portions of the Software.
14 *
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
21 * SOFTWARE.
22 */
23
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"
28
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"
34 #include <stdint.h>
35 #include <inttypes.h>
36 #include <glib.h>
37
38 #include "trimmer.h"
39
40 #define NS_PER_S INT64_C(1000000000)
41
42 static const char * const in_port_name = "in";
43
44 struct trimmer_time {
45 unsigned int hour, minute, second, ns;
46 };
47
48 struct trimmer_bound {
49 /*
50 * Nanoseconds from origin, valid if `is_set` is set and
51 * `is_infinite` is false.
52 */
53 int64_t ns_from_origin;
54
55 /* True if this bound's full time (`ns_from_origin`) is set */
56 bool is_set;
57
58 /*
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.
62 */
63 bool is_infinite;
64
65 /*
66 * This bound's time without the date; this time is used to set
67 * `ns_from_origin` once we know the date.
68 */
69 struct trimmer_time time;
70 };
71
72 struct trimmer_comp {
73 struct trimmer_bound begin, end;
74 bool is_gmt;
75 bt_logging_level log_level;
76 bt_self_component *self_comp;
77 };
78
79 enum trimmer_iterator_state {
80 /*
81 * Find the first message's date and set the bounds's times
82 * accordingly.
83 */
84 TRIMMER_ITERATOR_STATE_SET_BOUNDS_NS_FROM_ORIGIN,
85
86 /*
87 * Initially seek to the trimming range's beginning time.
88 */
89 TRIMMER_ITERATOR_STATE_SEEK_INITIALLY,
90
91 /*
92 * Fill the output message queue for as long as received input
93 * messages are within the trimming time range.
94 */
95 TRIMMER_ITERATOR_STATE_TRIM,
96
97 /* Flush the remaining messages in the output message queue */
98 TRIMMER_ITERATOR_STATE_ENDING,
99
100 /* Trimming operation and message iterator is ended */
101 TRIMMER_ITERATOR_STATE_ENDED,
102 };
103
104 struct trimmer_iterator {
105 /* Weak */
106 struct trimmer_comp *trimmer_comp;
107
108 /* Weak */
109 bt_self_message_iterator *self_msg_iter;
110
111 enum trimmer_iterator_state state;
112
113 /* Owned by this */
114 bt_self_component_port_input_message_iterator *upstream_iter;
115 struct trimmer_bound begin, end;
116
117 /*
118 * Queue of `const bt_message *` (owned by the queue).
119 *
120 * This is where the trimming operation pushes the messages to
121 * output by this message iterator.
122 */
123 GQueue *output_messages;
124
125 /*
126 * Hash table of `bt_stream *` (weak) to
127 * `struct trimmer_iterator_stream_state *` (owned by the HT).
128 */
129 GHashTable *stream_states;
130 };
131
132 struct trimmer_iterator_stream_state {
133 /*
134 * True if the last pushed message for this stream was a stream
135 * activity end message.
136 */
137 bool last_msg_is_stream_activity_end;
138
139 /*
140 * Time to use for a generated stream end activity message when
141 * ending the stream.
142 */
143 int64_t stream_act_end_ns_from_origin;
144
145 /* Weak */
146 const bt_stream *stream;
147
148 /* Owned by this (`NULL` initially and between packets) */
149 const bt_packet *cur_packet;
150 };
151
152 static
153 void destroy_trimmer_comp(struct trimmer_comp *trimmer_comp)
154 {
155 BT_ASSERT(trimmer_comp);
156 g_free(trimmer_comp);
157 }
158
159 static
160 struct trimmer_comp *create_trimmer_comp(void)
161 {
162 return g_new0(struct trimmer_comp, 1);
163 }
164
165 BT_HIDDEN
166 void trimmer_finalize(bt_self_component_filter *self_comp)
167 {
168 struct trimmer_comp *trimmer_comp =
169 bt_self_component_get_data(
170 bt_self_component_filter_as_self_component(self_comp));
171
172 if (trimmer_comp) {
173 destroy_trimmer_comp(trimmer_comp);
174 }
175 }
176
177 /*
178 * Sets the time (in ns from origin) of a trimmer bound from date and
179 * time components.
180 *
181 * Returns a negative value if anything goes wrong.
182 */
183 static
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)
188 {
189 int ret = 0;
190 time_t result;
191 struct tm tm = {
192 .tm_sec = second,
193 .tm_min = minute,
194 .tm_hour = hour,
195 .tm_mday = day,
196 .tm_mon = month - 1,
197 .tm_year = year - 1900,
198 .tm_isdst = -1,
199 };
200
201 if (is_gmt) {
202 result = bt_timegm(&tm);
203 } else {
204 result = mktime(&tm);
205 }
206
207 if (result < 0) {
208 ret = -1;
209 goto end;
210 }
211
212 BT_ASSERT(bound);
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;
217
218 end:
219 return ret;
220 }
221
222 /*
223 * Parses a timestamp, figuring out its format.
224 *
225 * Returns a negative value if anything goes wrong.
226 *
227 * Expected formats:
228 *
229 * YYYY-MM-DD hh:mm[:ss[.ns]]
230 * [hh:mm:]ss[.ns]
231 * [-]s[.ns]
232 *
233 * TODO: Check overflows.
234 */
235 static
236 int set_bound_from_str(struct trimmer_comp *trimmer_comp,
237 const char *str, struct trimmer_bound *bound, bool is_gmt)
238 {
239 int ret = 0;
240 int s_ret;
241 unsigned int year, month, day, hour, minute, second, ns;
242 char dummy;
243
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);
247 if (s_ret == 7) {
248 ret = set_bound_ns_from_origin(bound, year, month, day,
249 hour, minute, second, ns, is_gmt);
250 goto end;
251 }
252
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);
256 if (s_ret == 6) {
257 ret = set_bound_ns_from_origin(bound, year, month, day,
258 hour, minute, second, 0, is_gmt);
259 goto end;
260 }
261
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);
265 if (s_ret == 5) {
266 ret = set_bound_ns_from_origin(bound, year, month, day,
267 hour, minute, 0, 0, is_gmt);
268 goto end;
269 }
270
271 /* Try `YYYY-MM-DD` format */
272 s_ret = sscanf(str, "%u-%u-%u%c", &year, &month, &day, &dummy);
273 if (s_ret == 3) {
274 ret = set_bound_ns_from_origin(bound, year, month, day,
275 0, 0, 0, 0, is_gmt);
276 goto end;
277 }
278
279 /* Try `hh:mm:ss.ns` format */
280 s_ret = sscanf(str, "%u:%u:%u.%u%c", &hour, &minute, &second, &ns,
281 &dummy);
282 if (s_ret == 4) {
283 bound->time.hour = hour;
284 bound->time.minute = minute;
285 bound->time.second = second;
286 bound->time.ns = ns;
287 goto end;
288 }
289
290 /* Try `hh:mm:ss` format */
291 s_ret = sscanf(str, "%u:%u:%u%c", &hour, &minute, &second, &dummy);
292 if (s_ret == 3) {
293 bound->time.hour = hour;
294 bound->time.minute = minute;
295 bound->time.second = second;
296 bound->time.ns = 0;
297 goto end;
298 }
299
300 /* Try `-s.ns` format */
301 s_ret = sscanf(str, "-%u.%u%c", &second, &ns, &dummy);
302 if (s_ret == 2) {
303 bound->ns_from_origin = -((int64_t) second) * NS_PER_S;
304 bound->ns_from_origin -= (int64_t) ns;
305 bound->is_set = true;
306 goto end;
307 }
308
309 /* Try `s.ns` format */
310 s_ret = sscanf(str, "%u.%u%c", &second, &ns, &dummy);
311 if (s_ret == 2) {
312 bound->ns_from_origin = ((int64_t) second) * NS_PER_S;
313 bound->ns_from_origin += (int64_t) ns;
314 bound->is_set = true;
315 goto end;
316 }
317
318 /* Try `-s` format */
319 s_ret = sscanf(str, "-%u%c", &second, &dummy);
320 if (s_ret == 1) {
321 bound->ns_from_origin = -((int64_t) second) * NS_PER_S;
322 bound->is_set = true;
323 goto end;
324 }
325
326 /* Try `s` format */
327 s_ret = sscanf(str, "%u%c", &second, &dummy);
328 if (s_ret == 1) {
329 bound->ns_from_origin = (int64_t) second * NS_PER_S;
330 bound->is_set = true;
331 goto end;
332 }
333
334 BT_COMP_LOGE("Invalid date/time format: param=\"%s\"", str);
335 ret = -1;
336
337 end:
338 return ret;
339 }
340
341 /*
342 * Sets a trimmer bound's properties from a parameter string/integer
343 * value.
344 *
345 * Returns a negative value if anything goes wrong.
346 */
347 static
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)
351 {
352 int ret;
353 const char *arg;
354 char tmp_arg[64];
355
356 if (bt_value_is_signed_integer(param)) {
357 int64_t value = bt_value_signed_integer_get(param);
358
359 /*
360 * Just convert it to a temporary string to handle
361 * everything the same way.
362 */
363 sprintf(tmp_arg, "%" PRId64, value);
364 arg = tmp_arg;
365 } else if (bt_value_is_string(param)) {
366 arg = bt_value_string_get(param);
367 } else {
368 BT_COMP_LOGE("`%s` parameter must be an integer or a string value.",
369 param_name);
370 ret = -1;
371 goto end;
372 }
373
374 ret = set_bound_from_str(trimmer_comp, arg, bound, is_gmt);
375
376 end:
377 return ret;
378 }
379
380 static
381 int validate_trimmer_bounds(struct trimmer_comp *trimmer_comp,
382 struct trimmer_bound *begin, struct trimmer_bound *end)
383 {
384 int ret = 0;
385
386 BT_ASSERT(begin->is_set);
387 BT_ASSERT(end->is_set);
388
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);
396 ret = -1;
397 goto end;
398 }
399
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);
404 ret = -1;
405 goto end;
406 }
407
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);
412 ret = -1;
413 goto end;
414 }
415
416 end:
417 return ret;
418 }
419
420 static
421 int init_trimmer_comp_from_params(struct trimmer_comp *trimmer_comp,
422 const bt_value *params)
423 {
424 const bt_value *value;
425 int ret = 0;
426
427 BT_ASSERT(params);
428 value = bt_value_map_borrow_entry_value_const(params, "gmt");
429 if (value) {
430 trimmer_comp->is_gmt = (bool) bt_value_bool_get(value);
431 }
432
433 value = bt_value_map_borrow_entry_value_const(params, "begin");
434 if (value) {
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 */
438 ret = -1;
439 goto end;
440 }
441 } else {
442 trimmer_comp->begin.is_infinite = true;
443 trimmer_comp->begin.is_set = true;
444 }
445
446 value = bt_value_map_borrow_entry_value_const(params, "end");
447 if (value) {
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 */
451 ret = -1;
452 goto end;
453 }
454 } else {
455 trimmer_comp->end.is_infinite = true;
456 trimmer_comp->end.is_set = true;
457 }
458
459 end:
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);
464 }
465
466 return ret;
467 }
468
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)
472 {
473 int ret;
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);
480 if (!trimmer_comp) {
481 status = BT_COMPONENT_CLASS_INIT_METHOD_STATUS_MEMORY_ERROR;
482 goto error;
483 }
484
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;
493 goto error;
494 case BT_SELF_COMPONENT_ADD_PORT_STATUS_MEMORY_ERROR:
495 status = BT_COMPONENT_CLASS_INIT_METHOD_STATUS_MEMORY_ERROR;
496 goto error;
497 default:
498 break;
499 }
500
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;
506 goto error;
507 case BT_SELF_COMPONENT_ADD_PORT_STATUS_MEMORY_ERROR:
508 status = BT_COMPONENT_CLASS_INIT_METHOD_STATUS_MEMORY_ERROR;
509 goto error;
510 default:
511 break;
512 }
513
514 ret = init_trimmer_comp_from_params(trimmer_comp, params);
515 if (ret) {
516 status = BT_COMPONENT_CLASS_INIT_METHOD_STATUS_ERROR;
517 goto error;
518 }
519
520 bt_self_component_set_data(self_comp, trimmer_comp);
521 goto end;
522
523 error:
524 if (status == BT_COMPONENT_CLASS_INIT_METHOD_STATUS_OK) {
525 status = BT_COMPONENT_CLASS_INIT_METHOD_STATUS_ERROR;
526 }
527
528 if (trimmer_comp) {
529 destroy_trimmer_comp(trimmer_comp);
530 }
531
532 end:
533 return status;
534 }
535
536 static
537 void destroy_trimmer_iterator(struct trimmer_iterator *trimmer_it)
538 {
539 BT_ASSERT(trimmer_it);
540 bt_self_component_port_input_message_iterator_put_ref(
541 trimmer_it->upstream_iter);
542
543 if (trimmer_it->output_messages) {
544 g_queue_free(trimmer_it->output_messages);
545 }
546
547 if (trimmer_it->stream_states) {
548 g_hash_table_destroy(trimmer_it->stream_states);
549 }
550
551 g_free(trimmer_it);
552 }
553
554 static
555 void destroy_trimmer_iterator_stream_state(
556 struct trimmer_iterator_stream_state *sstate)
557 {
558 BT_ASSERT(sstate);
559 BT_PACKET_PUT_REF_AND_RESET(sstate->cur_packet);
560 g_free(sstate);
561 }
562
563 BT_HIDDEN
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)
568 {
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;
572
573 trimmer_it = g_new0(struct trimmer_iterator, 1);
574 if (!trimmer_it) {
575 status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_INIT_METHOD_STATUS_MEMORY_ERROR;
576 goto end;
577 }
578
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);
582
583 if (trimmer_it->trimmer_comp->begin.is_set &&
584 trimmer_it->trimmer_comp->end.is_set) {
585 /*
586 * Both trimming time range's bounds are set, so skip
587 * the
588 * `TRIMMER_ITERATOR_STATE_SET_BOUNDS_NS_FROM_ORIGIN`
589 * phase.
590 */
591 trimmer_it->state = TRIMMER_ITERATOR_STATE_SEEK_INITIALLY;
592 }
593
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;
602 goto end;
603 }
604
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;
608 goto end;
609 }
610
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;
616 goto end;
617 }
618
619 trimmer_it->self_msg_iter = self_msg_iter;
620 bt_self_message_iterator_set_data(self_msg_iter, trimmer_it);
621
622 end:
623 if (status != BT_COMPONENT_CLASS_MESSAGE_ITERATOR_INIT_METHOD_STATUS_OK && trimmer_it) {
624 destroy_trimmer_iterator(trimmer_it);
625 }
626
627 return status;
628 }
629
630 static inline
631 int get_msg_ns_from_origin(const bt_message *msg, int64_t *ns_from_origin,
632 bool *skip)
633 {
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;
637 int ret = 0;
638
639 BT_ASSERT(msg);
640 BT_ASSERT(ns_from_origin);
641 BT_ASSERT(skip);
642
643 switch (bt_message_get_type(msg)) {
644 case BT_MESSAGE_TYPE_EVENT:
645 clock_class =
646 bt_message_event_borrow_stream_class_default_clock_class_const(
647 msg);
648 if (G_UNLIKELY(!clock_class)) {
649 goto error;
650 }
651
652 clock_snapshot = bt_message_event_borrow_default_clock_snapshot_const(
653 msg);
654 break;
655 case BT_MESSAGE_TYPE_PACKET_BEGINNING:
656 clock_class =
657 bt_message_packet_beginning_borrow_stream_class_default_clock_class_const(
658 msg);
659 if (G_UNLIKELY(!clock_class)) {
660 goto error;
661 }
662
663 clock_snapshot = bt_message_packet_beginning_borrow_default_clock_snapshot_const(
664 msg);
665 break;
666 case BT_MESSAGE_TYPE_PACKET_END:
667 clock_class =
668 bt_message_packet_end_borrow_stream_class_default_clock_class_const(
669 msg);
670 if (G_UNLIKELY(!clock_class)) {
671 goto error;
672 }
673
674 clock_snapshot = bt_message_packet_end_borrow_default_clock_snapshot_const(
675 msg);
676 break;
677 case BT_MESSAGE_TYPE_DISCARDED_EVENTS:
678 clock_class =
679 bt_message_discarded_events_borrow_stream_class_default_clock_class_const(
680 msg);
681 if (G_UNLIKELY(!clock_class)) {
682 goto error;
683 }
684
685 clock_snapshot = bt_message_discarded_events_borrow_beginning_default_clock_snapshot_const(
686 msg);
687 break;
688 case BT_MESSAGE_TYPE_DISCARDED_PACKETS:
689 clock_class =
690 bt_message_discarded_packets_borrow_stream_class_default_clock_class_const(
691 msg);
692 if (G_UNLIKELY(!clock_class)) {
693 goto error;
694 }
695
696 clock_snapshot = bt_message_discarded_packets_borrow_beginning_default_clock_snapshot_const(
697 msg);
698 break;
699 case BT_MESSAGE_TYPE_STREAM_ACTIVITY_BEGINNING:
700 clock_class =
701 bt_message_stream_activity_beginning_borrow_stream_class_default_clock_class_const(
702 msg);
703 if (G_UNLIKELY(!clock_class)) {
704 goto error;
705 }
706
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;
714 }
715
716 break;
717 case BT_MESSAGE_TYPE_STREAM_ACTIVITY_END:
718 clock_class =
719 bt_message_stream_activity_end_borrow_stream_class_default_clock_class_const(
720 msg);
721 if (G_UNLIKELY(!clock_class)) {
722 goto error;
723 }
724
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;
735 }
736
737 break;
738 case BT_MESSAGE_TYPE_MESSAGE_ITERATOR_INACTIVITY:
739 clock_snapshot =
740 bt_message_message_iterator_inactivity_borrow_default_clock_snapshot_const(
741 msg);
742 break;
743 default:
744 goto no_clock_snapshot;
745 }
746
747 ret = bt_clock_snapshot_get_ns_from_origin(clock_snapshot,
748 ns_from_origin);
749 if (G_UNLIKELY(ret)) {
750 goto error;
751 }
752
753 goto end;
754
755 no_clock_snapshot:
756 *skip = true;
757 goto end;
758
759 error:
760 ret = -1;
761
762 end:
763 return ret;
764 }
765
766 static inline
767 void put_messages(bt_message_array_const msgs, uint64_t count)
768 {
769 uint64_t i;
770
771 for (i = 0; i < count; i++) {
772 BT_MESSAGE_PUT_REF_AND_RESET(msgs[i]);
773 }
774 }
775
776 static inline
777 int set_trimmer_iterator_bound(struct trimmer_iterator *trimmer_it,
778 struct trimmer_bound *bound, int64_t ns_from_origin,
779 bool is_gmt)
780 {
781 struct trimmer_comp *trimmer_comp = trimmer_it->trimmer_comp;
782 struct tm tm;
783 time_t time_seconds = (time_t) (ns_from_origin / NS_PER_S);
784 int ret = 0;
785
786 BT_ASSERT(!bound->is_set);
787 errno = 0;
788
789 /* We only need to extract the date from this time */
790 if (is_gmt) {
791 bt_gmtime_r(&time_seconds, &tm);
792 } else {
793 bt_localtime_r(&time_seconds, &tm);
794 }
795
796 if (errno) {
797 BT_COMP_LOGE_ERRNO("Cannot convert timestamp to date and time",
798 "ts=%" PRId64, (int64_t) time_seconds);
799 ret = -1;
800 goto end;
801 }
802
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);
806
807 end:
808 return ret;
809 }
810
811 static
812 bt_component_class_message_iterator_next_method_status
813 state_set_trimmer_iterator_bounds(
814 struct trimmer_iterator *trimmer_it)
815 {
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;
820 uint64_t count = 0;
821 int64_t ns_from_origin = INT64_MIN;
822 uint64_t i;
823 int ret;
824
825 BT_ASSERT(!trimmer_it->begin.is_set ||
826 !trimmer_it->end.is_set);
827
828 while (true) {
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) {
833 goto end;
834 }
835
836 for (i = 0; i < count; i++) {
837 const bt_message *msg = msgs[i];
838 bool skip = false;
839 int ret;
840
841 ret = get_msg_ns_from_origin(msg, &ns_from_origin,
842 &skip);
843 if (ret) {
844 goto error;
845 }
846
847 if (skip) {
848 continue;
849 }
850
851 BT_ASSERT(ns_from_origin != INT64_MIN &&
852 ns_from_origin != INT64_MAX);
853 put_messages(msgs, count);
854 goto found;
855 }
856
857 put_messages(msgs, count);
858 }
859
860 found:
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);
865 if (ret) {
866 goto error;
867 }
868 }
869
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);
874 if (ret) {
875 goto error;
876 }
877 }
878
879 ret = validate_trimmer_bounds(trimmer_it->trimmer_comp,
880 &trimmer_it->begin, &trimmer_it->end);
881 if (ret) {
882 goto error;
883 }
884
885 goto end;
886
887 error:
888 put_messages(msgs, count);
889 upstream_iter_status =
890 BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_ERROR;
891
892 end:
893 return (int) upstream_iter_status;
894 }
895
896 static
897 bt_component_class_message_iterator_next_method_status state_seek_initially(
898 struct trimmer_iterator *trimmer_it)
899 {
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;
903
904 BT_ASSERT(trimmer_it->begin.is_set);
905
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;
911 goto end;
912 }
913
914 status = (int) bt_self_component_port_input_message_iterator_seek_beginning(
915 trimmer_it->upstream_iter);
916 } else {
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;
924 goto end;
925 }
926
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);
929 }
930
931 if (status == BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK) {
932 trimmer_it->state = TRIMMER_ITERATOR_STATE_TRIM;
933 }
934
935 end:
936 return status;
937 }
938
939 static inline
940 void push_message(struct trimmer_iterator *trimmer_it, const bt_message *msg)
941 {
942 g_queue_push_head(trimmer_it->output_messages, (void *) msg);
943 }
944
945 static inline
946 const bt_message *pop_message(struct trimmer_iterator *trimmer_it)
947 {
948 return g_queue_pop_tail(trimmer_it->output_messages);
949 }
950
951 static inline
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)
954 {
955
956 int64_t cc_offset_s;
957 uint64_t cc_offset_cycles;
958 uint64_t cc_freq;
959
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);
964 }
965
966 static inline
967 bt_component_class_message_iterator_next_method_status
968 end_stream(struct trimmer_iterator *trimmer_it,
969 struct trimmer_iterator_stream_state *sstate)
970 {
971 bt_component_class_message_iterator_next_method_status status =
972 BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK;
973 uint64_t raw_value;
974 const bt_clock_class *clock_class;
975 int ret;
976 bt_message *msg = NULL;
977
978 BT_ASSERT(!trimmer_it->end.is_infinite);
979
980 if (!sstate->stream) {
981 goto end;
982 }
983
984 if (sstate->cur_packet) {
985 /*
986 * The last message could not have been a stream
987 * activity end message if we have a current packet.
988 */
989 BT_ASSERT(!sstate->last_msg_is_stream_activity_end);
990
991 /*
992 * Create and push a packet end message, making its time
993 * the trimming range's end time.
994 */
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);
1000 if (ret) {
1001 status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_ERROR;
1002 goto end;
1003 }
1004
1005 msg = bt_message_packet_end_create_with_default_clock_snapshot(
1006 trimmer_it->self_msg_iter, sstate->cur_packet,
1007 raw_value);
1008 if (!msg) {
1009 status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_MEMORY_ERROR;
1010 goto end;
1011 }
1012
1013 push_message(trimmer_it, msg);
1014 msg = NULL;
1015 BT_PACKET_PUT_REF_AND_RESET(sstate->cur_packet);
1016
1017 /*
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
1021 * time).
1022 */
1023 sstate->stream_act_end_ns_from_origin =
1024 trimmer_it->end.ns_from_origin;
1025 }
1026
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);
1031 if (!msg) {
1032 status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_MEMORY_ERROR;
1033 goto end;
1034 }
1035
1036 clock_class = bt_stream_class_borrow_default_clock_class_const(
1037 bt_stream_borrow_class_const(sstate->stream));
1038 BT_ASSERT(clock_class);
1039
1040 if (sstate->stream_act_end_ns_from_origin == INT64_MIN) {
1041 /*
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
1046 * time.
1047 */
1048 sstate->stream_act_end_ns_from_origin =
1049 trimmer_it->end.ns_from_origin;
1050 }
1051
1052 ret = clock_raw_value_from_ns_from_origin(clock_class,
1053 sstate->stream_act_end_ns_from_origin, &raw_value);
1054 if (ret) {
1055 status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_ERROR;
1056 goto end;
1057 }
1058
1059 bt_message_stream_activity_end_set_default_clock_snapshot(
1060 msg, raw_value);
1061 push_message(trimmer_it, msg);
1062 msg = NULL;
1063 }
1064
1065 /* Create and push a stream end message */
1066 msg = bt_message_stream_end_create(trimmer_it->self_msg_iter,
1067 sstate->stream);
1068 if (!msg) {
1069 status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_MEMORY_ERROR;
1070 goto end;
1071 }
1072
1073 push_message(trimmer_it, msg);
1074 msg = NULL;
1075
1076 /*
1077 * Just to make sure that we don't use this stream state again
1078 * in the future without an obvious error.
1079 */
1080 sstate->stream = NULL;
1081
1082 end:
1083 bt_message_put_ref(msg);
1084 return status;
1085 }
1086
1087 static inline
1088 bt_component_class_message_iterator_next_method_status end_iterator_streams(
1089 struct trimmer_iterator *trimmer_it)
1090 {
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;
1095
1096 if (trimmer_it->end.is_infinite) {
1097 /*
1098 * An infinite trimming range's end time guarantees that
1099 * we received (and pushed) all the appropriate end
1100 * messages.
1101 */
1102 goto remove_all;
1103 }
1104
1105 /*
1106 * End each stream and then remove them from the hash table of
1107 * stream states to release unneeded references.
1108 */
1109 g_hash_table_iter_init(&iter, trimmer_it->stream_states);
1110
1111 while (g_hash_table_iter_next(&iter, &key, &sstate)) {
1112 status = end_stream(trimmer_it, sstate);
1113 if (status) {
1114 goto end;
1115 }
1116 }
1117
1118 remove_all:
1119 g_hash_table_remove_all(trimmer_it->stream_states);
1120
1121 end:
1122 return status;
1123 }
1124
1125 static inline
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)
1131 {
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;
1135
1136 BT_ASSERT(msg);
1137 BT_ASSERT(!trimmer_it->begin.is_infinite);
1138
1139 local_msg = bt_message_stream_activity_beginning_create(
1140 trimmer_it->self_msg_iter, stream);
1141 if (!local_msg) {
1142 status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_MEMORY_ERROR;
1143 goto end;
1144 }
1145
1146 if (clock_class) {
1147 int ret;
1148 uint64_t raw_value;
1149
1150 ret = clock_raw_value_from_ns_from_origin(clock_class,
1151 trimmer_it->begin.ns_from_origin, &raw_value);
1152 if (ret) {
1153 status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_ERROR;
1154 bt_message_put_ref(local_msg);
1155 goto end;
1156 }
1157
1158 bt_message_stream_activity_beginning_set_default_clock_snapshot(
1159 local_msg, raw_value);
1160 }
1161
1162 BT_MESSAGE_MOVE_REF(*msg, local_msg);
1163
1164 end:
1165 return status;
1166 }
1167
1168 /*
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.
1173 *
1174 * This function consumes the `msg` reference, _whatever the outcome_.
1175 *
1176 * `ns_from_origin` is the message's time, as given by
1177 * get_msg_ns_from_origin().
1178 *
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
1182 * `reached_end`.
1183 */
1184 static inline
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)
1190 {
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);
1194 int ret;
1195
1196 switch (msg_type) {
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;
1202 break;
1203 }
1204
1205 BT_ASSERT(sstate->cur_packet);
1206 push_message(trimmer_it, msg);
1207 msg = NULL;
1208 break;
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;
1214 break;
1215 }
1216
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);
1222 msg = NULL;
1223 break;
1224 case BT_MESSAGE_TYPE_PACKET_END:
1225 sstate->stream_act_end_ns_from_origin = ns_from_origin;
1226
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;
1231 break;
1232 }
1233
1234 BT_ASSERT(sstate->cur_packet);
1235 BT_PACKET_PUT_REF_AND_RESET(sstate->cur_packet);
1236 push_message(trimmer_it, msg);
1237 msg = NULL;
1238 break;
1239 case BT_MESSAGE_TYPE_DISCARDED_EVENTS:
1240 case BT_MESSAGE_TYPE_DISCARDED_PACKETS:
1241 {
1242 /*
1243 * `ns_from_origin` is the message's time range's
1244 * beginning time here.
1245 */
1246 int64_t end_ns_from_origin;
1247 const bt_clock_snapshot *end_cs;
1248
1249 if (bt_message_get_type(msg) ==
1250 BT_MESSAGE_TYPE_DISCARDED_EVENTS) {
1251 /*
1252 * Safe to ignore the return value because we
1253 * know there's a default clock and it's always
1254 * known.
1255 */
1256 end_cs = bt_message_discarded_events_borrow_end_default_clock_snapshot_const(
1257 msg);
1258 } else {
1259 /*
1260 * Safe to ignore the return value because we
1261 * know there's a default clock and it's always
1262 * known.
1263 */
1264 end_cs = bt_message_discarded_packets_borrow_end_default_clock_snapshot_const(
1265 msg);
1266 }
1267
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;
1271 goto end;
1272 }
1273
1274 sstate->stream_act_end_ns_from_origin = end_ns_from_origin;
1275
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;
1280 break;
1281 }
1282
1283 if (!trimmer_it->end.is_infinite &&
1284 end_ns_from_origin > trimmer_it->end.ns_from_origin) {
1285 /*
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
1290 * count.
1291 */
1292 const bt_clock_class *clock_class =
1293 bt_clock_snapshot_borrow_clock_class_const(
1294 end_cs);
1295 const bt_clock_snapshot *begin_cs;
1296 bt_message *new_msg;
1297 uint64_t end_raw_value;
1298
1299 ret = clock_raw_value_from_ns_from_origin(clock_class,
1300 trimmer_it->end.ns_from_origin, &end_raw_value);
1301 if (ret) {
1302 status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_ERROR;
1303 goto end;
1304 }
1305
1306 if (msg_type == BT_MESSAGE_TYPE_DISCARDED_EVENTS) {
1307 begin_cs = bt_message_discarded_events_borrow_beginning_default_clock_snapshot_const(
1308 msg);
1309 new_msg = bt_message_discarded_events_create_with_default_clock_snapshots(
1310 trimmer_it->self_msg_iter,
1311 sstate->stream,
1312 bt_clock_snapshot_get_value(begin_cs),
1313 end_raw_value);
1314 } else {
1315 begin_cs = bt_message_discarded_packets_borrow_beginning_default_clock_snapshot_const(
1316 msg);
1317 new_msg = bt_message_discarded_packets_create_with_default_clock_snapshots(
1318 trimmer_it->self_msg_iter,
1319 sstate->stream,
1320 bt_clock_snapshot_get_value(begin_cs),
1321 end_raw_value);
1322 }
1323
1324 if (!new_msg) {
1325 status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_MEMORY_ERROR;
1326 goto end;
1327 }
1328
1329 /* Replace the original message */
1330 BT_MESSAGE_MOVE_REF(msg, new_msg);
1331 }
1332
1333 push_message(trimmer_it, msg);
1334 msg = NULL;
1335 break;
1336 }
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) {
1340 /*
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
1344 * always less than
1345 * `trimmer_it->end.ns_from_origin`.
1346 */
1347 status = end_iterator_streams(trimmer_it);
1348 *reached_end = true;
1349 break;
1350 }
1351
1352 push_message(trimmer_it, msg);
1353 msg = NULL;
1354 break;
1355 case BT_MESSAGE_TYPE_STREAM_ACTIVITY_END:
1356 if (trimmer_it->end.is_infinite) {
1357 push_message(trimmer_it, msg);
1358 msg = NULL;
1359 break;
1360 }
1361
1362 if (ns_from_origin == INT64_MIN) {
1363 /* Unknown: consider it to be in the trimmer window. */
1364 push_message(trimmer_it, msg);
1365 msg = NULL;
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;
1371 } else {
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;
1378 break;
1379 }
1380
1381 push_message(trimmer_it, msg);
1382 msg = NULL;
1383 sstate->last_msg_is_stream_activity_end = true;
1384 sstate->stream_act_end_ns_from_origin = ns_from_origin;
1385 }
1386
1387 break;
1388 case BT_MESSAGE_TYPE_STREAM_BEGINNING:
1389 push_message(trimmer_it, msg);
1390 msg = NULL;
1391 break;
1392 case BT_MESSAGE_TYPE_STREAM_END:
1393 /*
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.
1400 */
1401 if (trimmer_it->end.is_infinite) {
1402 push_message(trimmer_it, msg);
1403 msg = NULL;
1404
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) {
1411 goto end;
1412 }
1413
1414 /* We won't need this stream state again */
1415 g_hash_table_remove(trimmer_it->stream_states, sstate->stream);
1416 }
1417 break;
1418 default:
1419 break;
1420 }
1421
1422 end:
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;
1426 }
1427
1428 /*
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.
1432 *
1433 * This function consumes the `msg` reference, _whatever the outcome_.
1434 *
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
1438 * `reached_end`.
1439 */
1440 static inline
1441 bt_component_class_message_iterator_next_method_status handle_message(
1442 struct trimmer_iterator *trimmer_it, const bt_message *msg,
1443 bool *reached_end)
1444 {
1445 bt_component_class_message_iterator_next_method_status status;
1446 const bt_stream *stream = NULL;
1447 int64_t ns_from_origin = INT64_MIN;
1448 bool skip;
1449 int ret;
1450 struct trimmer_iterator_stream_state *sstate = NULL;
1451 struct trimmer_comp *trimmer_comp = trimmer_it->trimmer_comp;
1452
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));
1458 break;
1459 case BT_MESSAGE_TYPE_PACKET_BEGINNING:
1460 stream = bt_packet_borrow_stream_const(
1461 bt_message_packet_beginning_borrow_packet_const(msg));
1462 break;
1463 case BT_MESSAGE_TYPE_PACKET_END:
1464 stream = bt_packet_borrow_stream_const(
1465 bt_message_packet_end_borrow_packet_const(msg));
1466 break;
1467 case BT_MESSAGE_TYPE_DISCARDED_EVENTS:
1468 stream = bt_message_discarded_events_borrow_stream_const(msg);
1469 break;
1470 case BT_MESSAGE_TYPE_DISCARDED_PACKETS:
1471 stream = bt_message_discarded_packets_borrow_stream_const(msg);
1472 break;
1473 case BT_MESSAGE_TYPE_STREAM_ACTIVITY_BEGINNING:
1474 stream = bt_message_stream_activity_beginning_borrow_stream_const(msg);
1475 break;
1476 case BT_MESSAGE_TYPE_STREAM_ACTIVITY_END:
1477 stream = bt_message_stream_activity_end_borrow_stream_const(msg);
1478 break;
1479 case BT_MESSAGE_TYPE_STREAM_BEGINNING:
1480 stream = bt_message_stream_beginning_borrow_stream_const(msg);
1481 break;
1482 case BT_MESSAGE_TYPE_STREAM_END:
1483 stream = bt_message_stream_end_borrow_stream_const(msg);
1484 break;
1485 default:
1486 break;
1487 }
1488
1489 if (G_LIKELY(stream)) {
1490 /* Find stream state */
1491 sstate = g_hash_table_lookup(trimmer_it->stream_states,
1492 stream);
1493 if (G_UNLIKELY(!sstate)) {
1494 /* No stream state yet: create one now */
1495 const bt_stream_class *sc;
1496
1497 /*
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
1502 * messages.
1503 *
1504 * Also check that clock snapshots are always
1505 * known.
1506 */
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: "
1511 "stream-addr=%p, "
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;
1517 goto end;
1518 }
1519
1520 /*
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
1525 * implemented.
1526 */
1527 if (!bt_stream_class_packets_have_beginning_default_clock_snapshot(
1528 sc)) {
1529 BT_COMP_LOGE("Unsupported stream: packets have "
1530 "no beginning clock snapshot: "
1531 "stream-addr=%p, "
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;
1537 goto end;
1538 }
1539
1540 if (!bt_stream_class_packets_have_end_default_clock_snapshot(
1541 sc)) {
1542 BT_COMP_LOGE("Unsupported stream: packets have "
1543 "no end clock snapshot: "
1544 "stream-addr=%p, "
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;
1550 goto end;
1551 }
1552
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: "
1557 "stream-addr=%p, "
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;
1563 goto end;
1564 }
1565
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: "
1570 "stream-addr=%p, "
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;
1576 goto end;
1577 }
1578
1579 sstate = g_new0(struct trimmer_iterator_stream_state,
1580 1);
1581 if (!sstate) {
1582 status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_MEMORY_ERROR;
1583 goto end;
1584 }
1585
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);
1590 }
1591 }
1592
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;
1597 goto end;
1598 }
1599
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);
1604
1605 /*
1606 * handle_message_with_stream_state() unconditionally
1607 * consumes `msg`.
1608 */
1609 msg = NULL;
1610 } else {
1611 /*
1612 * Message not associated to a stream (message iterator
1613 * inactivity).
1614 */
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;
1619 } else {
1620 push_message(trimmer_it, msg);
1621 status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK;
1622 msg = NULL;
1623 }
1624 }
1625
1626 end:
1627 /* We release the message's reference whatever the outcome */
1628 bt_message_put_ref(msg);
1629 return status;
1630 }
1631
1632 static inline
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)
1636 {
1637 *count = 0;
1638
1639 /*
1640 * Move auto-seek messages to the output array (which is this
1641 * iterator's base message array).
1642 */
1643 while (capacity > 0 && !g_queue_is_empty(trimmer_it->output_messages)) {
1644 msgs[*count] = pop_message(trimmer_it);
1645 capacity--;
1646 (*count)++;
1647 }
1648
1649 BT_ASSERT(*count > 0);
1650 }
1651
1652 static inline
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,
1656 uint64_t *count)
1657 {
1658 bt_component_class_message_iterator_next_method_status status =
1659 BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK;
1660
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;
1664 goto end;
1665 }
1666
1667 fill_message_array_from_output_messages(trimmer_it, msgs,
1668 capacity, count);
1669
1670 end:
1671 return status;
1672 }
1673
1674 static inline
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,
1678 uint64_t *count)
1679 {
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;
1683 uint64_t my_count;
1684 uint64_t i;
1685 bool reached_end = false;
1686
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) {
1694 goto end;
1695 }
1696
1697 trimmer_it->state =
1698 TRIMMER_ITERATOR_STATE_ENDING;
1699 status = state_ending(trimmer_it, msgs,
1700 capacity, count);
1701 }
1702
1703 goto end;
1704 }
1705
1706 BT_ASSERT(my_count > 0);
1707
1708 for (i = 0; i < my_count; i++) {
1709 status = handle_message(trimmer_it, my_msgs[i],
1710 &reached_end);
1711
1712 /*
1713 * handle_message() unconditionally consumes the
1714 * message reference.
1715 */
1716 my_msgs[i] = NULL;
1717
1718 if (G_UNLIKELY(status !=
1719 BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK)) {
1720 put_messages(my_msgs, my_count);
1721 goto end;
1722 }
1723
1724 if (G_UNLIKELY(reached_end)) {
1725 /*
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
1733 * "next" method.
1734 */
1735 put_messages(my_msgs, my_count);
1736 trimmer_it->state =
1737 TRIMMER_ITERATOR_STATE_ENDING;
1738 status = state_ending(trimmer_it, msgs,
1739 capacity, count);
1740 goto end;
1741 }
1742 }
1743 }
1744
1745 /*
1746 * There's at least one message in the output message queue:
1747 * move the messages to the output message array.
1748 */
1749 BT_ASSERT(!g_queue_is_empty(trimmer_it->output_messages));
1750 fill_message_array_from_output_messages(trimmer_it, msgs,
1751 capacity, count);
1752
1753 end:
1754 return status;
1755 }
1756
1757 BT_HIDDEN
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,
1761 uint64_t *count)
1762 {
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;
1767
1768 BT_ASSERT(trimmer_it);
1769
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) {
1773 goto end;
1774 }
1775 } else {
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) {
1780 goto end;
1781 }
1782
1783 status = state_seek_initially(trimmer_it);
1784 if (status != BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK) {
1785 goto end;
1786 }
1787
1788 status = state_trim(trimmer_it, msgs, capacity, count);
1789 if (status != BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK) {
1790 goto end;
1791 }
1792
1793 break;
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) {
1797 goto end;
1798 }
1799
1800 status = state_trim(trimmer_it, msgs, capacity, count);
1801 if (status != BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK) {
1802 goto end;
1803 }
1804
1805 break;
1806 case TRIMMER_ITERATOR_STATE_ENDING:
1807 status = state_ending(trimmer_it, msgs, capacity,
1808 count);
1809 if (status != BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK) {
1810 goto end;
1811 }
1812
1813 break;
1814 case TRIMMER_ITERATOR_STATE_ENDED:
1815 status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_END;
1816 break;
1817 default:
1818 abort();
1819 }
1820 }
1821
1822 end:
1823 return status;
1824 }
1825
1826 BT_HIDDEN
1827 void trimmer_msg_iter_finalize(bt_self_message_iterator *self_msg_iter)
1828 {
1829 struct trimmer_iterator *trimmer_it =
1830 bt_self_message_iterator_get_data(self_msg_iter);
1831
1832 BT_ASSERT(trimmer_it);
1833 destroy_trimmer_iterator(trimmer_it);
1834 }
BabelTrace - Trace Format Babel Tower
This page took 0.067624 seconds and 4 git commands to generate.