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 both stream beginning and initial stream activity
 135          * beginning messages were pushed for this stream.
 136          */
 137         bool inited;
 138
 139         /*
 140          * True if the last pushed message for this stream was a stream
 141          * activity end message.
 142          */
 143         bool last_msg_is_stream_activity_end;
 144
 145         /*
 146          * Time to use for a generated stream end activity message when
 147          * ending the stream.
 148          */
 149         int64_t stream_act_end_ns_from_origin;
 150
 151         /* Weak */
 152         const bt_stream *stream;
 153
 154         /* Owned by this (`NULL` initially and between packets) */
 155         const bt_packet *cur_packet;
 156
 157         /* Owned by this */
 158         const bt_message *stream_beginning_msg;
 159 };
 160
 161 static
 162 void destroy_trimmer_comp(struct trimmer_comp *trimmer_comp)
 163 {
 164         BT_ASSERT(trimmer_comp);
 165         g_free(trimmer_comp);
 166 }
 167
 168 static
 169 struct trimmer_comp *create_trimmer_comp(void)
 170 {
 171         return g_new0(struct trimmer_comp, 1);
 172 }
 173
 174 BT_HIDDEN
 175 void trimmer_finalize(bt_self_component_filter *self_comp)
 176 {
 177         struct trimmer_comp *trimmer_comp =
 178                 bt_self_component_get_data(
 179                         bt_self_component_filter_as_self_component(self_comp));
 180
 181         if (trimmer_comp) {
 182                 destroy_trimmer_comp(trimmer_comp);
 183         }
 184 }
 185
 186 /*
 187  * Sets the time (in ns from origin) of a trimmer bound from date and
 188  * time components.
 189  *
 190  * Returns a negative value if anything goes wrong.
 191  */
 192 static
 193 int set_bound_ns_from_origin(struct trimmer_bound *bound,
 194                 unsigned int year, unsigned int month, unsigned int day,
 195                 unsigned int hour, unsigned int minute, unsigned int second,
 196                 unsigned int ns, bool is_gmt)
 197 {
 198         int ret = 0;
 199         time_t result;
 200         struct tm tm = {
 201                 .tm_sec = second,
 202                 .tm_min = minute,
 203                 .tm_hour = hour,
 204                 .tm_mday = day,
 205                 .tm_mon = month - 1,
 206                 .tm_year = year - 1900,
 207                 .tm_isdst = -1,
 208         };
 209
 210         if (is_gmt) {
 211                 result = bt_timegm(&tm);
 212         } else {
 213                 result = mktime(&tm);
 214         }
 215
 216         if (result < 0) {
 217                 ret = -1;
 218                 goto end;
 219         }
 220
 221         BT_ASSERT(bound);
 222         bound->ns_from_origin = (int64_t) result;
 223         bound->ns_from_origin *= NS_PER_S;
 224         bound->ns_from_origin += ns;
 225         bound->is_set = true;
 226
 227 end:
 228         return ret;
 229 }
 230
 231 /*
 232  * Parses a timestamp, figuring out its format.
 233  *
 234  * Returns a negative value if anything goes wrong.
 235  *
 236  * Expected formats:
 237  *
 238  *     YYYY-MM-DD hh:mm[:ss[.ns]]
 239  *     [hh:mm:]ss[.ns]
 240  *     [-]s[.ns]
 241  *
 242  * TODO: Check overflows.
 243  */
 244 static
 245 int set_bound_from_str(struct trimmer_comp *trimmer_comp,
 246                 const char *str, struct trimmer_bound *bound, bool is_gmt)
 247 {
 248         int ret = 0;
 249         int s_ret;
 250         unsigned int year, month, day, hour, minute, second, ns;
 251         char dummy;
 252
 253         /* Try `YYYY-MM-DD hh:mm:ss.ns` format */
 254         s_ret = sscanf(str, "%u-%u-%u %u:%u:%u.%u%c", &year, &month, &day,
 255                 &hour, &minute, &second, &ns, &dummy);
 256         if (s_ret == 7) {
 257                 ret = set_bound_ns_from_origin(bound, year, month, day,
 258                         hour, minute, second, ns, is_gmt);
 259                 goto end;
 260         }
 261
 262         /* Try `YYYY-MM-DD hh:mm:ss` format */
 263         s_ret = sscanf(str, "%u-%u-%u %u:%u:%u%c", &year, &month, &day,
 264                 &hour, &minute, &second, &dummy);
 265         if (s_ret == 6) {
 266                 ret = set_bound_ns_from_origin(bound, year, month, day,
 267                         hour, minute, second, 0, is_gmt);
 268                 goto end;
 269         }
 270
 271         /* Try `YYYY-MM-DD hh:mm` format */
 272         s_ret = sscanf(str, "%u-%u-%u %u:%u%c", &year, &month, &day,
 273                 &hour, &minute, &dummy);
 274         if (s_ret == 5) {
 275                 ret = set_bound_ns_from_origin(bound, year, month, day,
 276                         hour, minute, 0, 0, is_gmt);
 277                 goto end;
 278         }
 279
 280         /* Try `YYYY-MM-DD` format */
 281         s_ret = sscanf(str, "%u-%u-%u%c", &year, &month, &day, &dummy);
 282         if (s_ret == 3) {
 283                 ret = set_bound_ns_from_origin(bound, year, month, day,
 284                         0, 0, 0, 0, is_gmt);
 285                 goto end;
 286         }
 287
 288         /* Try `hh:mm:ss.ns` format */
 289         s_ret = sscanf(str, "%u:%u:%u.%u%c", &hour, &minute, &second, &ns,
 290                 &dummy);
 291         if (s_ret == 4) {
 292                 bound->time.hour = hour;
 293                 bound->time.minute = minute;
 294                 bound->time.second = second;
 295                 bound->time.ns = ns;
 296                 goto end;
 297         }
 298
 299         /* Try `hh:mm:ss` format */
 300         s_ret = sscanf(str, "%u:%u:%u%c", &hour, &minute, &second, &dummy);
 301         if (s_ret == 3) {
 302                 bound->time.hour = hour;
 303                 bound->time.minute = minute;
 304                 bound->time.second = second;
 305                 bound->time.ns = 0;
 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.ns` format */
 319         s_ret = sscanf(str, "%u.%u%c", &second, &ns, &dummy);
 320         if (s_ret == 2) {
 321                 bound->ns_from_origin = ((int64_t) second) * NS_PER_S;
 322                 bound->ns_from_origin += (int64_t) ns;
 323                 bound->is_set = true;
 324                 goto end;
 325         }
 326
 327         /* Try `-s` format */
 328         s_ret = sscanf(str, "-%u%c", &second, &dummy);
 329         if (s_ret == 1) {
 330                 bound->ns_from_origin = -((int64_t) second) * NS_PER_S;
 331                 bound->is_set = true;
 332                 goto end;
 333         }
 334
 335         /* Try `s` format */
 336         s_ret = sscanf(str, "%u%c", &second, &dummy);
 337         if (s_ret == 1) {
 338                 bound->ns_from_origin = (int64_t) second * NS_PER_S;
 339                 bound->is_set = true;
 340                 goto end;
 341         }
 342
 343         BT_COMP_LOGE("Invalid date/time format: param=\"%s\"", str);
 344         ret = -1;
 345
 346 end:
 347         return ret;
 348 }
 349
 350 /*
 351  * Sets a trimmer bound's properties from a parameter string/integer
 352  * value.
 353  *
 354  * Returns a negative value if anything goes wrong.
 355  */
 356 static
 357 int set_bound_from_param(struct trimmer_comp *trimmer_comp,
 358                 const char *param_name, const bt_value *param,
 359                 struct trimmer_bound *bound, bool is_gmt)
 360 {
 361         int ret;
 362         const char *arg;
 363         char tmp_arg[64];
 364
 365         if (bt_value_is_signed_integer(param)) {
 366                 int64_t value = bt_value_signed_integer_get(param);
 367
 368                 /*
 369                  * Just convert it to a temporary string to handle
 370                  * everything the same way.
 371                  */
 372                 sprintf(tmp_arg, "%" PRId64, value);
 373                 arg = tmp_arg;
 374         } else if (bt_value_is_string(param)) {
 375                 arg = bt_value_string_get(param);
 376         } else {
 377                 BT_COMP_LOGE("`%s` parameter must be an integer or a string value.",
 378                         param_name);
 379                 ret = -1;
 380                 goto end;
 381         }
 382
 383         ret = set_bound_from_str(trimmer_comp, arg, bound, is_gmt);
 384
 385 end:
 386         return ret;
 387 }
 388
 389 static
 390 int validate_trimmer_bounds(struct trimmer_comp *trimmer_comp,
 391                 struct trimmer_bound *begin, struct trimmer_bound *end)
 392 {
 393         int ret = 0;
 394
 395         BT_ASSERT(begin->is_set);
 396         BT_ASSERT(end->is_set);
 397
 398         if (!begin->is_infinite && !end->is_infinite &&
 399                         begin->ns_from_origin > end->ns_from_origin) {
 400                 BT_COMP_LOGE("Trimming time range's beginning time is greater than end time: "
 401                         "begin-ns-from-origin=%" PRId64 ", "
 402                         "end-ns-from-origin=%" PRId64,
 403                         begin->ns_from_origin,
 404                         end->ns_from_origin);
 405                 ret = -1;
 406                 goto end;
 407         }
 408
 409         if (!begin->is_infinite && begin->ns_from_origin == INT64_MIN) {
 410                 BT_COMP_LOGE("Invalid trimming time range's beginning time: "
 411                         "ns-from-origin=%" PRId64,
 412                         begin->ns_from_origin);
 413                 ret = -1;
 414                 goto end;
 415         }
 416
 417         if (!end->is_infinite && end->ns_from_origin == INT64_MIN) {
 418                 BT_COMP_LOGE("Invalid trimming time range's end time: "
 419                         "ns-from-origin=%" PRId64,
 420                         end->ns_from_origin);
 421                 ret = -1;
 422                 goto end;
 423         }
 424
 425 end:
 426         return ret;
 427 }
 428
 429 static
 430 int init_trimmer_comp_from_params(struct trimmer_comp *trimmer_comp,
 431                 const bt_value *params)
 432 {
 433         const bt_value *value;
 434         int ret = 0;
 435
 436         BT_ASSERT(params);
 437         value = bt_value_map_borrow_entry_value_const(params, "gmt");
 438         if (value) {
 439                 trimmer_comp->is_gmt = (bool) bt_value_bool_get(value);
 440         }
 441
 442         value = bt_value_map_borrow_entry_value_const(params, "begin");
 443         if (value) {
 444                 if (set_bound_from_param(trimmer_comp, "begin", value,
 445                                 &trimmer_comp->begin, trimmer_comp->is_gmt)) {
 446                         /* set_bound_from_param() logs errors */
 447                         ret = BT_SELF_COMPONENT_STATUS_ERROR;
 448                         goto end;
 449                 }
 450         } else {
 451                 trimmer_comp->begin.is_infinite = true;
 452                 trimmer_comp->begin.is_set = true;
 453         }
 454
 455         value = bt_value_map_borrow_entry_value_const(params, "end");
 456         if (value) {
 457                 if (set_bound_from_param(trimmer_comp, "end", value,
 458                                 &trimmer_comp->end, trimmer_comp->is_gmt)) {
 459                         /* set_bound_from_param() logs errors */
 460                         ret = BT_SELF_COMPONENT_STATUS_ERROR;
 461                         goto end;
 462                 }
 463         } else {
 464                 trimmer_comp->end.is_infinite = true;
 465                 trimmer_comp->end.is_set = true;
 466         }
 467
 468 end:
 469         if (trimmer_comp->begin.is_set && trimmer_comp->end.is_set) {
 470                 /* validate_trimmer_bounds() logs errors */
 471                 ret = validate_trimmer_bounds(trimmer_comp,
 472                         &trimmer_comp->begin, &trimmer_comp->end);
 473         }
 474
 475         return ret;
 476 }
 477
 478 bt_self_component_status trimmer_init(bt_self_component_filter *self_comp_flt,
 479                 const bt_value *params, void *init_data)
 480 {
 481         int ret;
 482         bt_self_component_status status;
 483         struct trimmer_comp *trimmer_comp = create_trimmer_comp();
 484         bt_self_component *self_comp =
 485                 bt_self_component_filter_as_self_component(self_comp_flt);
 486         if (!trimmer_comp) {
 487                 status = BT_SELF_COMPONENT_STATUS_NOMEM;
 488                 goto error;
 489         }
 490
 491         trimmer_comp->log_level = bt_component_get_logging_level(
 492                 bt_self_component_as_component(self_comp));
 493         trimmer_comp->self_comp = self_comp;
 494         status = bt_self_component_filter_add_input_port(
 495                 self_comp_flt, in_port_name, NULL, NULL);
 496         if (status != BT_SELF_COMPONENT_STATUS_OK) {
 497                 goto error;
 498         }
 499
 500         status = bt_self_component_filter_add_output_port(
 501                 self_comp_flt, "out", NULL, NULL);
 502         if (status != BT_SELF_COMPONENT_STATUS_OK) {
 503                 goto error;
 504         }
 505
 506         ret = init_trimmer_comp_from_params(trimmer_comp, params);
 507         if (ret) {
 508                 status = BT_SELF_COMPONENT_STATUS_ERROR;
 509                 goto error;
 510         }
 511
 512         bt_self_component_set_data(self_comp, trimmer_comp);
 513         goto end;
 514
 515 error:
 516         if (status == BT_SELF_COMPONENT_STATUS_OK) {
 517                 status = BT_SELF_COMPONENT_STATUS_ERROR;
 518         }
 519
 520         if (trimmer_comp) {
 521                 destroy_trimmer_comp(trimmer_comp);
 522         }
 523
 524 end:
 525         return status;
 526 }
 527
 528 static
 529 void destroy_trimmer_iterator(struct trimmer_iterator *trimmer_it)
 530 {
 531         BT_ASSERT(trimmer_it);
 532         bt_self_component_port_input_message_iterator_put_ref(
 533                 trimmer_it->upstream_iter);
 534
 535         if (trimmer_it->output_messages) {
 536                 g_queue_free(trimmer_it->output_messages);
 537         }
 538
 539         if (trimmer_it->stream_states) {
 540                 g_hash_table_destroy(trimmer_it->stream_states);
 541         }
 542
 543         g_free(trimmer_it);
 544 }
 545
 546 static
 547 void destroy_trimmer_iterator_stream_state(
 548                 struct trimmer_iterator_stream_state *sstate)
 549 {
 550         BT_ASSERT(sstate);
 551         BT_PACKET_PUT_REF_AND_RESET(sstate->cur_packet);
 552         BT_MESSAGE_PUT_REF_AND_RESET(sstate->stream_beginning_msg);
 553         g_free(sstate);
 554 }
 555
 556 BT_HIDDEN
 557 bt_self_message_iterator_status trimmer_msg_iter_init(
 558                 bt_self_message_iterator *self_msg_iter,
 559                 bt_self_component_filter *self_comp,
 560                 bt_self_component_port_output *port)
 561 {
 562         bt_self_message_iterator_status status =
 563                 BT_SELF_MESSAGE_ITERATOR_STATUS_OK;
 564         struct trimmer_iterator *trimmer_it;
 565
 566         trimmer_it = g_new0(struct trimmer_iterator, 1);
 567         if (!trimmer_it) {
 568                 status = BT_SELF_MESSAGE_ITERATOR_STATUS_NOMEM;
 569                 goto end;
 570         }
 571
 572         trimmer_it->trimmer_comp = bt_self_component_get_data(
 573                 bt_self_component_filter_as_self_component(self_comp));
 574         BT_ASSERT(trimmer_it->trimmer_comp);
 575
 576         if (trimmer_it->trimmer_comp->begin.is_set &&
 577                         trimmer_it->trimmer_comp->end.is_set) {
 578                 /*
 579                  * Both trimming time range's bounds are set, so skip
 580                  * the
 581                  * `TRIMMER_ITERATOR_STATE_SET_BOUNDS_NS_FROM_ORIGIN`
 582                  * phase.
 583                  */
 584                 trimmer_it->state = TRIMMER_ITERATOR_STATE_SEEK_INITIALLY;
 585         }
 586
 587         trimmer_it->begin = trimmer_it->trimmer_comp->begin;
 588         trimmer_it->end = trimmer_it->trimmer_comp->end;
 589         trimmer_it->upstream_iter =
 590                 bt_self_component_port_input_message_iterator_create(
 591                         bt_self_component_filter_borrow_input_port_by_name(
 592                                 self_comp, in_port_name));
 593         if (!trimmer_it->upstream_iter) {
 594                 status = BT_SELF_MESSAGE_ITERATOR_STATUS_ERROR;
 595                 goto end;
 596         }
 597
 598         trimmer_it->output_messages = g_queue_new();
 599         if (!trimmer_it->output_messages) {
 600                 status = BT_SELF_MESSAGE_ITERATOR_STATUS_NOMEM;
 601                 goto end;
 602         }
 603
 604         trimmer_it->stream_states = g_hash_table_new_full(g_direct_hash,
 605                 g_direct_equal, NULL,
 606                 (GDestroyNotify) destroy_trimmer_iterator_stream_state);
 607         if (!trimmer_it->stream_states) {
 608                 status = BT_SELF_MESSAGE_ITERATOR_STATUS_NOMEM;
 609                 goto end;
 610         }
 611
 612         trimmer_it->self_msg_iter = self_msg_iter;
 613         bt_self_message_iterator_set_data(self_msg_iter, trimmer_it);
 614
 615 end:
 616         if (status != BT_SELF_MESSAGE_ITERATOR_STATUS_OK && trimmer_it) {
 617                 destroy_trimmer_iterator(trimmer_it);
 618         }
 619
 620         return status;
 621 }
 622
 623 static inline
 624 int get_msg_ns_from_origin(const bt_message *msg, int64_t *ns_from_origin,
 625                 bool *skip)
 626 {
 627         const bt_clock_class *clock_class = NULL;
 628         const bt_clock_snapshot *clock_snapshot = NULL;
 629         bt_message_stream_activity_clock_snapshot_state sa_cs_state;
 630         int ret = 0;
 631
 632         BT_ASSERT(msg);
 633         BT_ASSERT(ns_from_origin);
 634         BT_ASSERT(skip);
 635
 636         switch (bt_message_get_type(msg)) {
 637         case BT_MESSAGE_TYPE_EVENT:
 638                 clock_class =
 639                         bt_message_event_borrow_stream_class_default_clock_class_const(
 640                                 msg);
 641                 if (G_UNLIKELY(!clock_class)) {
 642                         goto error;
 643                 }
 644
 645                 clock_snapshot = bt_message_event_borrow_default_clock_snapshot_const(
 646                         msg);
 647                 break;
 648         case BT_MESSAGE_TYPE_PACKET_BEGINNING:
 649                 clock_class =
 650                         bt_message_packet_beginning_borrow_stream_class_default_clock_class_const(
 651                                 msg);
 652                 if (G_UNLIKELY(!clock_class)) {
 653                         goto error;
 654                 }
 655
 656                 clock_snapshot = bt_message_packet_beginning_borrow_default_clock_snapshot_const(
 657                         msg);
 658                 break;
 659         case BT_MESSAGE_TYPE_PACKET_END:
 660                 clock_class =
 661                         bt_message_packet_end_borrow_stream_class_default_clock_class_const(
 662                                 msg);
 663                 if (G_UNLIKELY(!clock_class)) {
 664                         goto error;
 665                 }
 666
 667                 clock_snapshot = bt_message_packet_end_borrow_default_clock_snapshot_const(
 668                         msg);
 669                 break;
 670         case BT_MESSAGE_TYPE_DISCARDED_EVENTS:
 671                 clock_class =
 672                         bt_message_discarded_events_borrow_stream_class_default_clock_class_const(
 673                                 msg);
 674                 if (G_UNLIKELY(!clock_class)) {
 675                         goto error;
 676                 }
 677
 678                 clock_snapshot = bt_message_discarded_events_borrow_beginning_default_clock_snapshot_const(
 679                         msg);
 680                 break;
 681         case BT_MESSAGE_TYPE_DISCARDED_PACKETS:
 682                 clock_class =
 683                         bt_message_discarded_packets_borrow_stream_class_default_clock_class_const(
 684                                 msg);
 685                 if (G_UNLIKELY(!clock_class)) {
 686                         goto error;
 687                 }
 688
 689                 clock_snapshot = bt_message_discarded_packets_borrow_beginning_default_clock_snapshot_const(
 690                         msg);
 691                 break;
 692         case BT_MESSAGE_TYPE_STREAM_ACTIVITY_BEGINNING:
 693                 clock_class =
 694                         bt_message_stream_activity_beginning_borrow_stream_class_default_clock_class_const(
 695                                 msg);
 696                 if (G_UNLIKELY(!clock_class)) {
 697                         goto error;
 698                 }
 699
 700                 sa_cs_state = bt_message_stream_activity_beginning_borrow_default_clock_snapshot_const(
 701                         msg, &clock_snapshot);
 702                 if (sa_cs_state == BT_MESSAGE_STREAM_ACTIVITY_CLOCK_SNAPSHOT_STATE_UNKNOWN ||
 703                                 sa_cs_state == BT_MESSAGE_STREAM_ACTIVITY_CLOCK_SNAPSHOT_STATE_INFINITE) {
 704                         /* Lowest possible time to always include them */
 705                         *ns_from_origin = INT64_MIN;
 706                         goto no_clock_snapshot;
 707                 }
 708
 709                 break;
 710         case BT_MESSAGE_TYPE_STREAM_ACTIVITY_END:
 711                 clock_class =
 712                         bt_message_stream_activity_end_borrow_stream_class_default_clock_class_const(
 713                                 msg);
 714                 if (G_UNLIKELY(!clock_class)) {
 715                         goto error;
 716                 }
 717
 718                 sa_cs_state = bt_message_stream_activity_end_borrow_default_clock_snapshot_const(
 719                         msg, &clock_snapshot);
 720                 if (sa_cs_state == BT_MESSAGE_STREAM_ACTIVITY_CLOCK_SNAPSHOT_STATE_UNKNOWN) {
 721                         /* Lowest time to always include it */
 722                         *ns_from_origin = INT64_MIN;
 723                         goto no_clock_snapshot;
 724                 } else if (sa_cs_state == BT_MESSAGE_STREAM_ACTIVITY_CLOCK_SNAPSHOT_STATE_INFINITE) {
 725                         /* Greatest time to always exclude it */
 726                         *ns_from_origin = INT64_MAX;
 727                         goto no_clock_snapshot;
 728                 }
 729
 730                 break;
 731         case BT_MESSAGE_TYPE_MESSAGE_ITERATOR_INACTIVITY:
 732                 clock_snapshot =
 733                         bt_message_message_iterator_inactivity_borrow_default_clock_snapshot_const(
 734                                 msg);
 735                 break;
 736         default:
 737                 goto no_clock_snapshot;
 738         }
 739
 740         ret = bt_clock_snapshot_get_ns_from_origin(clock_snapshot,
 741                 ns_from_origin);
 742         if (G_UNLIKELY(ret)) {
 743                 goto error;
 744         }
 745
 746         goto end;
 747
 748 no_clock_snapshot:
 749         *skip = true;
 750         goto end;
 751
 752 error:
 753         ret = -1;
 754
 755 end:
 756         return ret;
 757 }
 758
 759 static inline
 760 void put_messages(bt_message_array_const msgs, uint64_t count)
 761 {
 762         uint64_t i;
 763
 764         for (i = 0; i < count; i++) {
 765                 BT_MESSAGE_PUT_REF_AND_RESET(msgs[i]);
 766         }
 767 }
 768
 769 static inline
 770 int set_trimmer_iterator_bound(struct trimmer_iterator *trimmer_it,
 771                 struct trimmer_bound *bound, int64_t ns_from_origin,
 772                 bool is_gmt)
 773 {
 774         struct trimmer_comp *trimmer_comp = trimmer_it->trimmer_comp;
 775         struct tm tm;
 776         time_t time_seconds = (time_t) (ns_from_origin / NS_PER_S);
 777         int ret = 0;
 778
 779         BT_ASSERT(!bound->is_set);
 780         errno = 0;
 781
 782         /* We only need to extract the date from this time */
 783         if (is_gmt) {
 784                 bt_gmtime_r(&time_seconds, &tm);
 785         } else {
 786                 bt_localtime_r(&time_seconds, &tm);
 787         }
 788
 789         if (errno) {
 790                 BT_COMP_LOGE_ERRNO("Cannot convert timestamp to date and time",
 791                         "ts=%" PRId64, (int64_t) time_seconds);
 792                 ret = -1;
 793                 goto end;
 794         }
 795
 796         ret = set_bound_ns_from_origin(bound, tm.tm_year + 1900, tm.tm_mon + 1,
 797                 tm.tm_mday, bound->time.hour, bound->time.minute,
 798                 bound->time.second, bound->time.ns, is_gmt);
 799
 800 end:
 801         return ret;
 802 }
 803
 804 static
 805 bt_self_message_iterator_status state_set_trimmer_iterator_bounds(
 806                 struct trimmer_iterator *trimmer_it)
 807 {
 808         bt_message_iterator_status upstream_iter_status =
 809                 BT_MESSAGE_ITERATOR_STATUS_OK;
 810         struct trimmer_comp *trimmer_comp = trimmer_it->trimmer_comp;
 811         bt_message_array_const msgs;
 812         uint64_t count = 0;
 813         int64_t ns_from_origin = INT64_MIN;
 814         uint64_t i;
 815         int ret;
 816
 817         BT_ASSERT(!trimmer_it->begin.is_set ||
 818                 !trimmer_it->end.is_set);
 819
 820         while (true) {
 821                 upstream_iter_status =
 822                         bt_self_component_port_input_message_iterator_next(
 823                                 trimmer_it->upstream_iter, &msgs, &count);
 824                 if (upstream_iter_status != BT_MESSAGE_ITERATOR_STATUS_OK) {
 825                         goto end;
 826                 }
 827
 828                 for (i = 0; i < count; i++) {
 829                         const bt_message *msg = msgs[i];
 830                         bool skip = false;
 831                         int ret;
 832
 833                         ret = get_msg_ns_from_origin(msg, &ns_from_origin,
 834                                 &skip);
 835                         if (ret) {
 836                                 goto error;
 837                         }
 838
 839                         if (skip) {
 840                                 continue;
 841                         }
 842
 843                         BT_ASSERT(ns_from_origin != INT64_MIN &&
 844                                 ns_from_origin != INT64_MAX);
 845                         put_messages(msgs, count);
 846                         goto found;
 847                 }
 848
 849                 put_messages(msgs, count);
 850         }
 851
 852 found:
 853         if (!trimmer_it->begin.is_set) {
 854                 BT_ASSERT(!trimmer_it->begin.is_infinite);
 855                 ret = set_trimmer_iterator_bound(trimmer_it, &trimmer_it->begin,
 856                         ns_from_origin, trimmer_comp->is_gmt);
 857                 if (ret) {
 858                         goto error;
 859                 }
 860         }
 861
 862         if (!trimmer_it->end.is_set) {
 863                 BT_ASSERT(!trimmer_it->end.is_infinite);
 864                 ret = set_trimmer_iterator_bound(trimmer_it, &trimmer_it->end,
 865                         ns_from_origin, trimmer_comp->is_gmt);
 866                 if (ret) {
 867                         goto error;
 868                 }
 869         }
 870
 871         ret = validate_trimmer_bounds(trimmer_it->trimmer_comp,
 872                 &trimmer_it->begin, &trimmer_it->end);
 873         if (ret) {
 874                 goto error;
 875         }
 876
 877         goto end;
 878
 879 error:
 880         put_messages(msgs, count);
 881         upstream_iter_status = BT_MESSAGE_ITERATOR_STATUS_ERROR;
 882
 883 end:
 884         return (int) upstream_iter_status;
 885 }
 886
 887 static
 888 bt_self_message_iterator_status state_seek_initially(
 889                 struct trimmer_iterator *trimmer_it)
 890 {
 891         struct trimmer_comp *trimmer_comp = trimmer_it->trimmer_comp;
 892         bt_self_message_iterator_status status =
 893                 BT_SELF_MESSAGE_ITERATOR_STATUS_OK;
 894
 895         BT_ASSERT(trimmer_it->begin.is_set);
 896
 897         if (trimmer_it->begin.is_infinite) {
 898                 if (!bt_self_component_port_input_message_iterator_can_seek_beginning(
 899                                 trimmer_it->upstream_iter)) {
 900                         BT_COMP_LOGE_STR("Cannot make upstream message iterator initially seek its beginning.");
 901                         status = BT_SELF_MESSAGE_ITERATOR_STATUS_ERROR;
 902                         goto end;
 903                 }
 904
 905                 status = (int) bt_self_component_port_input_message_iterator_seek_beginning(
 906                         trimmer_it->upstream_iter);
 907         } else {
 908                 if (!bt_self_component_port_input_message_iterator_can_seek_ns_from_origin(
 909                                 trimmer_it->upstream_iter,
 910                                 trimmer_it->begin.ns_from_origin)) {
 911                         BT_COMP_LOGE("Cannot make upstream message iterator initially seek: "
 912                                 "seek-ns-from-origin=%" PRId64,
 913                                 trimmer_it->begin.ns_from_origin);
 914                         status = BT_SELF_MESSAGE_ITERATOR_STATUS_ERROR;
 915                         goto end;
 916                 }
 917
 918                 status = (int) bt_self_component_port_input_message_iterator_seek_ns_from_origin(
 919                         trimmer_it->upstream_iter, trimmer_it->begin.ns_from_origin);
 920         }
 921
 922         if (status == BT_SELF_MESSAGE_ITERATOR_STATUS_OK) {
 923                 trimmer_it->state = TRIMMER_ITERATOR_STATE_TRIM;
 924         }
 925
 926 end:
 927         return status;
 928 }
 929
 930 static inline
 931 void push_message(struct trimmer_iterator *trimmer_it, const bt_message *msg)
 932 {
 933         g_queue_push_head(trimmer_it->output_messages, (void *) msg);
 934 }
 935
 936 static inline
 937 const bt_message *pop_message(struct trimmer_iterator *trimmer_it)
 938 {
 939         return g_queue_pop_tail(trimmer_it->output_messages);
 940 }
 941
 942 static inline
 943 int clock_raw_value_from_ns_from_origin(const bt_clock_class *clock_class,
 944                 int64_t ns_from_origin, uint64_t *raw_value)
 945 {
 946
 947         int64_t cc_offset_s;
 948         uint64_t cc_offset_cycles;
 949         uint64_t cc_freq;
 950
 951         bt_clock_class_get_offset(clock_class, &cc_offset_s, &cc_offset_cycles);
 952         cc_freq = bt_clock_class_get_frequency(clock_class);
 953         return bt_common_clock_value_from_ns_from_origin(cc_offset_s,
 954                 cc_offset_cycles, cc_freq, ns_from_origin, raw_value);
 955 }
 956
 957 static inline
 958 bt_self_message_iterator_status end_stream(struct trimmer_iterator *trimmer_it,
 959                 struct trimmer_iterator_stream_state *sstate)
 960 {
 961         bt_self_message_iterator_status status =
 962                 BT_SELF_MESSAGE_ITERATOR_STATUS_OK;
 963         uint64_t raw_value;
 964         const bt_clock_class *clock_class;
 965         int ret;
 966         bt_message *msg = NULL;
 967
 968         BT_ASSERT(!trimmer_it->end.is_infinite);
 969
 970         if (!sstate->stream) {
 971                 goto end;
 972         }
 973
 974         if (sstate->cur_packet) {
 975                 /*
 976                  * The last message could not have been a stream
 977                  * activity end message if we have a current packet.
 978                  */
 979                 BT_ASSERT(!sstate->last_msg_is_stream_activity_end);
 980
 981                 /*
 982                  * Create and push a packet end message, making its time
 983                  * the trimming range's end time.
 984                  */
 985                 clock_class = bt_stream_class_borrow_default_clock_class_const(
 986                         bt_stream_borrow_class_const(sstate->stream));
 987                 BT_ASSERT(clock_class);
 988                 ret = clock_raw_value_from_ns_from_origin(clock_class,
 989                         trimmer_it->end.ns_from_origin, &raw_value);
 990                 if (ret) {
 991                         status = BT_SELF_MESSAGE_ITERATOR_STATUS_ERROR;
 992                         goto end;
 993                 }
 994
 995                 msg = bt_message_packet_end_create_with_default_clock_snapshot(
 996                         trimmer_it->self_msg_iter, sstate->cur_packet,
 997                         raw_value);
 998                 if (!msg) {
 999                         status = BT_SELF_MESSAGE_ITERATOR_STATUS_NOMEM;
1000                         goto end;
1001                 }
1002
1003                 push_message(trimmer_it, msg);
1004                 msg = NULL;
1005                 BT_PACKET_PUT_REF_AND_RESET(sstate->cur_packet);
1006
1007                 /*
1008                  * Because we generated a packet end message, set the
1009                  * stream activity end message's time to use to the
1010                  * trimming range's end time (this packet end message's
1011                  * time).
1012                  */
1013                 sstate->stream_act_end_ns_from_origin =
1014                         trimmer_it->end.ns_from_origin;
1015         }
1016
1017         if (!sstate->last_msg_is_stream_activity_end) {
1018                 /* Create and push a stream activity end message */
1019                 msg = bt_message_stream_activity_end_create(
1020                         trimmer_it->self_msg_iter, sstate->stream);
1021                 if (!msg) {
1022                         status = BT_SELF_MESSAGE_ITERATOR_STATUS_NOMEM;
1023                         goto end;
1024                 }
1025
1026                 clock_class = bt_stream_class_borrow_default_clock_class_const(
1027                         bt_stream_borrow_class_const(sstate->stream));
1028                 BT_ASSERT(clock_class);
1029
1030                 if (sstate->stream_act_end_ns_from_origin == INT64_MIN) {
1031                         /*
1032                          * We received at least what is necessary to
1033                          * have a stream state (stream beginning and
1034                          * stream activity beginning messages), but
1035                          * nothing else: use the trimmer range's end
1036                          * time.
1037                          */
1038                         sstate->stream_act_end_ns_from_origin =
1039                                 trimmer_it->end.ns_from_origin;
1040                 }
1041
1042                 ret = clock_raw_value_from_ns_from_origin(clock_class,
1043                         sstate->stream_act_end_ns_from_origin, &raw_value);
1044                 if (ret) {
1045                         status = BT_SELF_MESSAGE_ITERATOR_STATUS_ERROR;
1046                         goto end;
1047                 }
1048
1049                 bt_message_stream_activity_end_set_default_clock_snapshot(
1050                         msg, raw_value);
1051                 push_message(trimmer_it, msg);
1052                 msg = NULL;
1053         }
1054
1055         /* Create and push a stream end message */
1056         msg = bt_message_stream_end_create(trimmer_it->self_msg_iter,
1057                 sstate->stream);
1058         if (!msg) {
1059                 status = BT_SELF_MESSAGE_ITERATOR_STATUS_NOMEM;
1060                 goto end;
1061         }
1062
1063         push_message(trimmer_it, msg);
1064         msg = NULL;
1065
1066         /*
1067          * Just to make sure that we don't use this stream state again
1068          * in the future without an obvious error.
1069          */
1070         sstate->stream = NULL;
1071
1072 end:
1073         bt_message_put_ref(msg);
1074         return status;
1075 }
1076
1077 static inline
1078 bt_self_message_iterator_status end_iterator_streams(
1079                 struct trimmer_iterator *trimmer_it)
1080 {
1081         bt_self_message_iterator_status status =
1082                 BT_SELF_MESSAGE_ITERATOR_STATUS_OK;
1083         GHashTableIter iter;
1084         gpointer key, sstate;
1085
1086         if (trimmer_it->end.is_infinite) {
1087                 /*
1088                  * An infinite trimming range's end time guarantees that
1089                  * we received (and pushed) all the appropriate end
1090                  * messages.
1091                  */
1092                 goto remove_all;
1093         }
1094
1095         /*
1096          * End each stream and then remove them from the hash table of
1097          * stream states to release unneeded references.
1098          */
1099         g_hash_table_iter_init(&iter, trimmer_it->stream_states);
1100
1101         while (g_hash_table_iter_next(&iter, &key, &sstate)) {
1102                 status = end_stream(trimmer_it, sstate);
1103                 if (status) {
1104                         goto end;
1105                 }
1106         }
1107
1108 remove_all:
1109         g_hash_table_remove_all(trimmer_it->stream_states);
1110
1111 end:
1112         return status;
1113 }
1114
1115 static inline
1116 bt_self_message_iterator_status create_stream_beginning_activity_message(
1117                 struct trimmer_iterator *trimmer_it,
1118                 const bt_stream *stream,
1119                 const bt_clock_class *clock_class, bt_message **msg)
1120 {
1121         bt_message *local_msg;
1122         bt_self_message_iterator_status status =
1123                 BT_SELF_MESSAGE_ITERATOR_STATUS_OK;
1124
1125         BT_ASSERT(msg);
1126         BT_ASSERT(!trimmer_it->begin.is_infinite);
1127
1128         local_msg = bt_message_stream_activity_beginning_create(
1129                 trimmer_it->self_msg_iter, stream);
1130         if (!local_msg) {
1131                 status = BT_SELF_MESSAGE_ITERATOR_STATUS_NOMEM;
1132                 goto end;
1133         }
1134
1135         if (clock_class) {
1136                 int ret;
1137                 uint64_t raw_value;
1138
1139                 ret = clock_raw_value_from_ns_from_origin(clock_class,
1140                         trimmer_it->begin.ns_from_origin, &raw_value);
1141                 if (ret) {
1142                         status = BT_SELF_MESSAGE_ITERATOR_STATUS_ERROR;
1143                         bt_message_put_ref(local_msg);
1144                         goto end;
1145                 }
1146
1147                 bt_message_stream_activity_beginning_set_default_clock_snapshot(
1148                         local_msg, raw_value);
1149         }
1150
1151         BT_MESSAGE_MOVE_REF(*msg, local_msg);
1152
1153 end:
1154         return status;
1155 }
1156
1157 /*
1158  * Makes sure to initialize a stream state, pushing the appropriate
1159  * initial messages.
1160  *
1161  * `stream_act_beginning_msg` is an initial stream activity beginning
1162  * message to potentially use, depending on its clock snapshot state.
1163  * This function consumes `stream_act_beginning_msg` unconditionally.
1164  */
1165 static inline
1166 bt_self_message_iterator_status ensure_stream_state_is_inited(
1167                 struct trimmer_iterator *trimmer_it,
1168                 struct trimmer_iterator_stream_state *sstate,
1169                 const bt_message *stream_act_beginning_msg)
1170 {
1171         bt_self_message_iterator_status status =
1172                 BT_SELF_MESSAGE_ITERATOR_STATUS_OK;
1173         bt_message *new_msg = NULL;
1174         const bt_clock_class *clock_class =
1175                 bt_stream_class_borrow_default_clock_class_const(
1176                         bt_stream_borrow_class_const(sstate->stream));
1177
1178         BT_ASSERT(!sstate->inited);
1179
1180         if (!sstate->stream_beginning_msg) {
1181                 /* No initial stream beginning message: create one */
1182                 sstate->stream_beginning_msg =
1183                         bt_message_stream_beginning_create(
1184                                 trimmer_it->self_msg_iter, sstate->stream);
1185                 if (!sstate->stream_beginning_msg) {
1186                         status = BT_SELF_MESSAGE_ITERATOR_STATUS_NOMEM;
1187                         goto end;
1188                 }
1189         }
1190
1191         /* Push initial stream beginning message */
1192         BT_ASSERT(sstate->stream_beginning_msg);
1193         push_message(trimmer_it, sstate->stream_beginning_msg);
1194         sstate->stream_beginning_msg = NULL;
1195
1196         if (stream_act_beginning_msg) {
1197                 /*
1198                  * Initial stream activity beginning message exists: if
1199                  * its time is -inf, then create and push a new one
1200                  * having the trimming range's beginning time. Otherwise
1201                  * push it as is (known and unknown).
1202                  */
1203                 const bt_clock_snapshot *cs;
1204                 bt_message_stream_activity_clock_snapshot_state sa_cs_state;
1205
1206                 sa_cs_state = bt_message_stream_activity_beginning_borrow_default_clock_snapshot_const(
1207                         stream_act_beginning_msg, &cs);
1208                 if (sa_cs_state == BT_MESSAGE_STREAM_ACTIVITY_CLOCK_SNAPSHOT_STATE_INFINITE &&
1209                                 !trimmer_it->begin.is_infinite) {
1210                         /*
1211                          * -inf time: use trimming range's beginning
1212                          * time (which is not -inf).
1213                          */
1214                         status = create_stream_beginning_activity_message(
1215                                 trimmer_it, sstate->stream, clock_class,
1216                                 &new_msg);
1217                         if (status != BT_SELF_MESSAGE_ITERATOR_STATUS_OK) {
1218                                 goto end;
1219                         }
1220
1221                         push_message(trimmer_it, new_msg);
1222                         new_msg = NULL;
1223                 } else {
1224                         /* Known/unknown: push as is */
1225                         push_message(trimmer_it, stream_act_beginning_msg);
1226                         stream_act_beginning_msg = NULL;
1227                 }
1228         } else {
1229                 BT_ASSERT(!trimmer_it->begin.is_infinite);
1230
1231                 /*
1232                  * No stream beginning activity message: create and push
1233                  * a new message.
1234                  */
1235                 status = create_stream_beginning_activity_message(
1236                         trimmer_it, sstate->stream, clock_class, &new_msg);
1237                 if (status != BT_SELF_MESSAGE_ITERATOR_STATUS_OK) {
1238                         goto end;
1239                 }
1240
1241                 push_message(trimmer_it, new_msg);
1242                 new_msg = NULL;
1243         }
1244
1245         sstate->inited = true;
1246
1247 end:
1248         bt_message_put_ref(new_msg);
1249         bt_message_put_ref(stream_act_beginning_msg);
1250         return status;
1251 }
1252
1253 static inline
1254 bt_self_message_iterator_status ensure_cur_packet_exists(
1255         struct trimmer_iterator *trimmer_it,
1256         struct trimmer_iterator_stream_state *sstate,
1257         const bt_packet *packet)
1258 {
1259         bt_self_message_iterator_status status =
1260                 BT_SELF_MESSAGE_ITERATOR_STATUS_OK;
1261         int ret;
1262         const bt_clock_class *clock_class =
1263                 bt_stream_class_borrow_default_clock_class_const(
1264                         bt_stream_borrow_class_const(sstate->stream));
1265         bt_message *msg = NULL;
1266         uint64_t raw_value;
1267
1268         BT_ASSERT(!trimmer_it->begin.is_infinite);
1269         BT_ASSERT(!sstate->cur_packet);
1270
1271         /*
1272          * Create and push an initial packet beginning message,
1273          * making its time the trimming range's beginning time.
1274          */
1275         ret = clock_raw_value_from_ns_from_origin(clock_class,
1276                 trimmer_it->begin.ns_from_origin, &raw_value);
1277         if (ret) {
1278                 status = BT_SELF_MESSAGE_ITERATOR_STATUS_ERROR;
1279                 goto end;
1280         }
1281
1282         msg = bt_message_packet_beginning_create_with_default_clock_snapshot(
1283                 trimmer_it->self_msg_iter, packet, raw_value);
1284         if (!msg) {
1285                 status = BT_SELF_MESSAGE_ITERATOR_STATUS_NOMEM;
1286                 goto end;
1287         }
1288
1289         push_message(trimmer_it, msg);
1290         msg = NULL;
1291
1292         /* Set packet as this stream's current packet */
1293         sstate->cur_packet = packet;
1294         bt_packet_get_ref(sstate->cur_packet);
1295
1296 end:
1297         bt_message_put_ref(msg);
1298         return status;
1299 }
1300
1301 /*
1302  * Handles a message which is associated to a given stream state. This
1303  * _could_ make the iterator's output message queue grow; this could
1304  * also consume the message without pushing anything to this queue, only
1305  * modifying the stream state.
1306  *
1307  * This function consumes the `msg` reference, _whatever the outcome_.
1308  *
1309  * `ns_from_origin` is the message's time, as given by
1310  * get_msg_ns_from_origin().
1311  *
1312  * This function sets `reached_end` if handling this message made the
1313  * iterator reach the end of the trimming range. Note that the output
1314  * message queue could contain messages even if this function sets
1315  * `reached_end`.
1316  */
1317 static inline
1318 bt_self_message_iterator_status handle_message_with_stream_state(
1319                 struct trimmer_iterator *trimmer_it, const bt_message *msg,
1320                 struct trimmer_iterator_stream_state *sstate,
1321                 int64_t ns_from_origin, bool *reached_end)
1322 {
1323         bt_self_message_iterator_status status =
1324                 BT_SELF_MESSAGE_ITERATOR_STATUS_OK;
1325         bt_message_type msg_type = bt_message_get_type(msg);
1326         int ret;
1327
1328         switch (msg_type) {
1329         case BT_MESSAGE_TYPE_EVENT:
1330                 if (G_UNLIKELY(!trimmer_it->end.is_infinite &&
1331                                 ns_from_origin > trimmer_it->end.ns_from_origin)) {
1332                         status = end_iterator_streams(trimmer_it);
1333                         *reached_end = true;
1334                         break;
1335                 }
1336
1337                 if (G_UNLIKELY(!sstate->inited)) {
1338                         status = ensure_stream_state_is_inited(trimmer_it,
1339                                 sstate, NULL);
1340                         if (status != BT_SELF_MESSAGE_ITERATOR_STATUS_OK) {
1341                                 goto end;
1342                         }
1343                 }
1344
1345                 if (G_UNLIKELY(!sstate->cur_packet)) {
1346                         const bt_event *event =
1347                                 bt_message_event_borrow_event_const(msg);
1348                         const bt_packet *packet = bt_event_borrow_packet_const(
1349                                 event);
1350
1351                         status = ensure_cur_packet_exists(trimmer_it, sstate,
1352                                 packet);
1353                         if (status != BT_SELF_MESSAGE_ITERATOR_STATUS_OK) {
1354                                 goto end;
1355                         }
1356                 }
1357
1358                 BT_ASSERT(sstate->cur_packet);
1359                 push_message(trimmer_it, msg);
1360                 msg = NULL;
1361                 break;
1362         case BT_MESSAGE_TYPE_PACKET_BEGINNING:
1363                 if (G_UNLIKELY(!trimmer_it->end.is_infinite &&
1364                                 ns_from_origin > trimmer_it->end.ns_from_origin)) {
1365                         status = end_iterator_streams(trimmer_it);
1366                         *reached_end = true;
1367                         break;
1368                 }
1369
1370                 if (G_UNLIKELY(!sstate->inited)) {
1371                         status = ensure_stream_state_is_inited(trimmer_it,
1372                                 sstate, NULL);
1373                         if (status != BT_SELF_MESSAGE_ITERATOR_STATUS_OK) {
1374                                 goto end;
1375                         }
1376                 }
1377
1378                 BT_ASSERT(!sstate->cur_packet);
1379                 sstate->cur_packet =
1380                         bt_message_packet_beginning_borrow_packet_const(msg);
1381                 bt_packet_get_ref(sstate->cur_packet);
1382                 push_message(trimmer_it, msg);
1383                 msg = NULL;
1384                 break;
1385         case BT_MESSAGE_TYPE_PACKET_END:
1386                 sstate->stream_act_end_ns_from_origin = ns_from_origin;
1387
1388                 if (G_UNLIKELY(!trimmer_it->end.is_infinite &&
1389                                 ns_from_origin > trimmer_it->end.ns_from_origin)) {
1390                         status = end_iterator_streams(trimmer_it);
1391                         *reached_end = true;
1392                         break;
1393                 }
1394
1395                 if (G_UNLIKELY(!sstate->inited)) {
1396                         status = ensure_stream_state_is_inited(trimmer_it,
1397                                 sstate, NULL);
1398                         if (status != BT_SELF_MESSAGE_ITERATOR_STATUS_OK) {
1399                                 goto end;
1400                         }
1401                 }
1402
1403                 if (G_UNLIKELY(!sstate->cur_packet)) {
1404                         const bt_packet *packet =
1405                                 bt_message_packet_end_borrow_packet_const(msg);
1406
1407                         status = ensure_cur_packet_exists(trimmer_it, sstate,
1408                                 packet);
1409                         if (status != BT_SELF_MESSAGE_ITERATOR_STATUS_OK) {
1410                                 goto end;
1411                         }
1412                 }
1413
1414                 BT_ASSERT(sstate->cur_packet);
1415                 BT_PACKET_PUT_REF_AND_RESET(sstate->cur_packet);
1416                 push_message(trimmer_it, msg);
1417                 msg = NULL;
1418                 break;
1419         case BT_MESSAGE_TYPE_DISCARDED_EVENTS:
1420         case BT_MESSAGE_TYPE_DISCARDED_PACKETS:
1421         {
1422                 /*
1423                  * `ns_from_origin` is the message's time range's
1424                  * beginning time here.
1425                  */
1426                 int64_t end_ns_from_origin;
1427                 const bt_clock_snapshot *end_cs;
1428
1429                 if (bt_message_get_type(msg) ==
1430                                 BT_MESSAGE_TYPE_DISCARDED_EVENTS) {
1431                         /*
1432                          * Safe to ignore the return value because we
1433                          * know there's a default clock and it's always
1434                          * known.
1435                          */
1436                         end_cs = bt_message_discarded_events_borrow_end_default_clock_snapshot_const(
1437                                 msg);
1438                 } else {
1439                         /*
1440                          * Safe to ignore the return value because we
1441                          * know there's a default clock and it's always
1442                          * known.
1443                          */
1444                         end_cs = bt_message_discarded_packets_borrow_end_default_clock_snapshot_const(
1445                                 msg);
1446                 }
1447
1448                 if (bt_clock_snapshot_get_ns_from_origin(end_cs,
1449                                 &end_ns_from_origin)) {
1450                         status = BT_SELF_MESSAGE_ITERATOR_STATUS_ERROR;
1451                         goto end;
1452                 }
1453
1454                 sstate->stream_act_end_ns_from_origin = end_ns_from_origin;
1455
1456                 if (!trimmer_it->end.is_infinite &&
1457                                 ns_from_origin > trimmer_it->end.ns_from_origin) {
1458                         status = end_iterator_streams(trimmer_it);
1459                         *reached_end = true;
1460                         break;
1461                 }
1462
1463                 if (!trimmer_it->end.is_infinite &&
1464                                 end_ns_from_origin > trimmer_it->end.ns_from_origin) {
1465                         /*
1466                          * This message's end time is outside the
1467                          * trimming time range: replace it with a new
1468                          * message having an end time equal to the
1469                          * trimming time range's end and without a
1470                          * count.
1471                          */
1472                         const bt_clock_class *clock_class =
1473                                 bt_clock_snapshot_borrow_clock_class_const(
1474                                         end_cs);
1475                         const bt_clock_snapshot *begin_cs;
1476                         bt_message *new_msg;
1477                         uint64_t end_raw_value;
1478
1479                         ret = clock_raw_value_from_ns_from_origin(clock_class,
1480                                 trimmer_it->end.ns_from_origin, &end_raw_value);
1481                         if (ret) {
1482                                 status = BT_SELF_MESSAGE_ITERATOR_STATUS_ERROR;
1483                                 goto end;
1484                         }
1485
1486                         if (msg_type == BT_MESSAGE_TYPE_DISCARDED_EVENTS) {
1487                                 begin_cs = bt_message_discarded_events_borrow_beginning_default_clock_snapshot_const(
1488                                         msg);
1489                                 new_msg = bt_message_discarded_events_create_with_default_clock_snapshots(
1490                                         trimmer_it->self_msg_iter,
1491                                         sstate->stream,
1492                                         bt_clock_snapshot_get_value(begin_cs),
1493                                         end_raw_value);
1494                         } else {
1495                                 begin_cs = bt_message_discarded_packets_borrow_beginning_default_clock_snapshot_const(
1496                                         msg);
1497                                 new_msg = bt_message_discarded_packets_create_with_default_clock_snapshots(
1498                                         trimmer_it->self_msg_iter,
1499                                         sstate->stream,
1500                                         bt_clock_snapshot_get_value(begin_cs),
1501                                         end_raw_value);
1502                         }
1503
1504                         if (!new_msg) {
1505                                 status = BT_SELF_MESSAGE_ITERATOR_STATUS_NOMEM;
1506                                 goto end;
1507                         }
1508
1509                         /* Replace the original message */
1510                         BT_MESSAGE_MOVE_REF(msg, new_msg);
1511                 }
1512
1513                 if (G_UNLIKELY(!sstate->inited)) {
1514                         status = ensure_stream_state_is_inited(trimmer_it,
1515                                 sstate, NULL);
1516                         if (status != BT_SELF_MESSAGE_ITERATOR_STATUS_OK) {
1517                                 goto end;
1518                         }
1519                 }
1520
1521                 push_message(trimmer_it, msg);
1522                 msg = NULL;
1523                 break;
1524         }
1525         case BT_MESSAGE_TYPE_STREAM_ACTIVITY_BEGINNING:
1526                 if (!trimmer_it->end.is_infinite &&
1527                                 ns_from_origin > trimmer_it->end.ns_from_origin) {
1528                         /*
1529                          * This only happens when the message's time is
1530                          * known and is greater than the trimming
1531                          * range's end time. Unknown and -inf times are
1532                          * always less than
1533                          * `trimmer_it->end.ns_from_origin`.
1534                          */
1535                         status = end_iterator_streams(trimmer_it);
1536                         *reached_end = true;
1537                         break;
1538                 }
1539
1540                 if (!sstate->inited) {
1541                         status = ensure_stream_state_is_inited(trimmer_it,
1542                                 sstate, msg);
1543                         msg = NULL;
1544                         if (status != BT_SELF_MESSAGE_ITERATOR_STATUS_OK) {
1545                                 goto end;
1546                         }
1547                 } else {
1548                         push_message(trimmer_it, msg);
1549                         msg = NULL;
1550                 }
1551
1552                 break;
1553         case BT_MESSAGE_TYPE_STREAM_ACTIVITY_END:
1554                 if (trimmer_it->end.is_infinite) {
1555                         push_message(trimmer_it, msg);
1556                         msg = NULL;
1557                         break;
1558                 }
1559
1560                 if (ns_from_origin == INT64_MIN) {
1561                         /* Unknown: push as is if stream state is inited */
1562                         if (sstate->inited) {
1563                                 push_message(trimmer_it, msg);
1564                                 msg = NULL;
1565                                 sstate->last_msg_is_stream_activity_end = true;
1566                         }
1567                 } else if (ns_from_origin == INT64_MAX) {
1568                         /* Infinite: use trimming range's end time */
1569                         sstate->stream_act_end_ns_from_origin =
1570                                 trimmer_it->end.ns_from_origin;
1571                 } else {
1572                         /* Known: check if outside of trimming range */
1573                         if (ns_from_origin > trimmer_it->end.ns_from_origin) {
1574                                 sstate->stream_act_end_ns_from_origin =
1575                                         trimmer_it->end.ns_from_origin;
1576                                 status = end_iterator_streams(trimmer_it);
1577                                 *reached_end = true;
1578                                 break;
1579                         }
1580
1581                         if (!sstate->inited) {
1582                                 /*
1583                                  * First message for this stream is a
1584                                  * stream activity end: we can't deduce
1585                                  * anything about the stream activity
1586                                  * beginning's time, and using this
1587                                  * message's time would make a useless
1588                                  * pair of stream activity beginning/end
1589                                  * with the same time. Just skip this
1590                                  * message and wait for something
1591                                  * useful.
1592                                  */
1593                                 break;
1594                         }
1595
1596                         push_message(trimmer_it, msg);
1597                         msg = NULL;
1598                         sstate->last_msg_is_stream_activity_end = true;
1599                         sstate->stream_act_end_ns_from_origin = ns_from_origin;
1600                 }
1601
1602                 break;
1603         case BT_MESSAGE_TYPE_STREAM_BEGINNING:
1604                 /*
1605                  * We don't know what follows at this point, so just
1606                  * keep this message until we know what to do with it
1607                  * (it will be used in ensure_stream_state_is_inited()).
1608                  */
1609                 BT_ASSERT(!sstate->inited);
1610                 BT_MESSAGE_MOVE_REF(sstate->stream_beginning_msg, msg);
1611                 break;
1612         case BT_MESSAGE_TYPE_STREAM_END:
1613                 if (sstate->inited) {
1614                         /*
1615                          * This is the end of an inited stream: end this
1616                          * stream if its stream activity end message
1617                          * time is not the trimming range's end time
1618                          * (which means the final stream activity end
1619                          * message had an infinite time). end_stream()
1620                          * will generate its own stream end message.
1621                          */
1622                         if (trimmer_it->end.is_infinite) {
1623                                 push_message(trimmer_it, msg);
1624                                 msg = NULL;
1625                                 g_hash_table_remove(trimmer_it->stream_states,
1626                                         sstate->stream);
1627                         } else if (sstate->stream_act_end_ns_from_origin <
1628                                         trimmer_it->end.ns_from_origin) {
1629                                 status = end_stream(trimmer_it, sstate);
1630                                 if (status != BT_SELF_MESSAGE_ITERATOR_STATUS_OK) {
1631                                         goto end;
1632                                 }
1633
1634                                 /* We won't need this stream state again */
1635                                 g_hash_table_remove(trimmer_it->stream_states,
1636                                         sstate->stream);
1637                         }
1638                 } else {
1639                         /* We dont't need this stream state anymore */
1640                         g_hash_table_remove(trimmer_it->stream_states, sstate->stream);
1641                 }
1642
1643                 break;
1644         default:
1645                 break;
1646         }
1647
1648 end:
1649         /* We release the message's reference whatever the outcome */
1650         bt_message_put_ref(msg);
1651         return BT_SELF_MESSAGE_ITERATOR_STATUS_OK;
1652 }
1653
1654 /*
1655  * Handles an input message. This _could_ make the iterator's output
1656  * message queue grow; this could also consume the message without
1657  * pushing anything to this queue, only modifying the stream state.
1658  *
1659  * This function consumes the `msg` reference, _whatever the outcome_.
1660  *
1661  * This function sets `reached_end` if handling this message made the
1662  * iterator reach the end of the trimming range. Note that the output
1663  * message queue could contain messages even if this function sets
1664  * `reached_end`.
1665  */
1666 static inline
1667 bt_self_message_iterator_status handle_message(
1668                 struct trimmer_iterator *trimmer_it, const bt_message *msg,
1669                 bool *reached_end)
1670 {
1671         bt_self_message_iterator_status status;
1672         const bt_stream *stream = NULL;
1673         int64_t ns_from_origin = INT64_MIN;
1674         bool skip;
1675         int ret;
1676         struct trimmer_iterator_stream_state *sstate = NULL;
1677         struct trimmer_comp *trimmer_comp = trimmer_it->trimmer_comp;
1678
1679         /* Find message's associated stream */
1680         switch (bt_message_get_type(msg)) {
1681         case BT_MESSAGE_TYPE_EVENT:
1682                 stream = bt_event_borrow_stream_const(
1683                         bt_message_event_borrow_event_const(msg));
1684                 break;
1685         case BT_MESSAGE_TYPE_PACKET_BEGINNING:
1686                 stream = bt_packet_borrow_stream_const(
1687                         bt_message_packet_beginning_borrow_packet_const(msg));
1688                 break;
1689         case BT_MESSAGE_TYPE_PACKET_END:
1690                 stream = bt_packet_borrow_stream_const(
1691                         bt_message_packet_end_borrow_packet_const(msg));
1692                 break;
1693         case BT_MESSAGE_TYPE_DISCARDED_EVENTS:
1694                 stream = bt_message_discarded_events_borrow_stream_const(msg);
1695                 break;
1696         case BT_MESSAGE_TYPE_DISCARDED_PACKETS:
1697                 stream = bt_message_discarded_packets_borrow_stream_const(msg);
1698                 break;
1699         case BT_MESSAGE_TYPE_STREAM_ACTIVITY_BEGINNING:
1700                 stream = bt_message_stream_activity_beginning_borrow_stream_const(msg);
1701                 break;
1702         case BT_MESSAGE_TYPE_STREAM_ACTIVITY_END:
1703                 stream = bt_message_stream_activity_end_borrow_stream_const(msg);
1704                 break;
1705         case BT_MESSAGE_TYPE_STREAM_BEGINNING:
1706                 stream = bt_message_stream_beginning_borrow_stream_const(msg);
1707                 break;
1708         case BT_MESSAGE_TYPE_STREAM_END:
1709                 stream = bt_message_stream_end_borrow_stream_const(msg);
1710                 break;
1711         default:
1712                 break;
1713         }
1714
1715         if (G_LIKELY(stream)) {
1716                 /* Find stream state */
1717                 sstate = g_hash_table_lookup(trimmer_it->stream_states,
1718                         stream);
1719                 if (G_UNLIKELY(!sstate)) {
1720                         /* No stream state yet: create one now */
1721                         const bt_stream_class *sc;
1722
1723                         /*
1724                          * Validate right now that the stream's class
1725                          * has a registered default clock class so that
1726                          * an existing stream state guarantees existing
1727                          * default clock snapshots for its associated
1728                          * messages.
1729                          *
1730                          * Also check that clock snapshots are always
1731                          * known.
1732                          */
1733                         sc = bt_stream_borrow_class_const(stream);
1734                         if (!bt_stream_class_borrow_default_clock_class_const(sc)) {
1735                                 BT_COMP_LOGE("Unsupported stream: stream class does "
1736                                         "not have a default clock class: "
1737                                         "stream-addr=%p, "
1738                                         "stream-id=%" PRIu64 ", "
1739                                         "stream-name=\"%s\"",
1740                                         stream, bt_stream_get_id(stream),
1741                                         bt_stream_get_name(stream));
1742                                 status = BT_SELF_MESSAGE_ITERATOR_STATUS_ERROR;
1743                                 goto end;
1744                         }
1745
1746                         /*
1747                          * Temporary: make sure packet beginning, packet
1748                          * end, discarded events, and discarded packets
1749                          * messages have default clock snapshots until
1750                          * the support for not having them is
1751                          * implemented.
1752                          */
1753                         if (!bt_stream_class_packets_have_beginning_default_clock_snapshot(
1754                                         sc)) {
1755                                 BT_COMP_LOGE("Unsupported stream: packets have "
1756                                         "no beginning clock snapshot: "
1757                                         "stream-addr=%p, "
1758                                         "stream-id=%" PRIu64 ", "
1759                                         "stream-name=\"%s\"",
1760                                         stream, bt_stream_get_id(stream),
1761                                         bt_stream_get_name(stream));
1762                                 status = BT_SELF_MESSAGE_ITERATOR_STATUS_ERROR;
1763                                 goto end;
1764                         }
1765
1766                         if (!bt_stream_class_packets_have_end_default_clock_snapshot(
1767                                         sc)) {
1768                                 BT_COMP_LOGE("Unsupported stream: packets have "
1769                                         "no end clock snapshot: "
1770                                         "stream-addr=%p, "
1771                                         "stream-id=%" PRIu64 ", "
1772                                         "stream-name=\"%s\"",
1773                                         stream, bt_stream_get_id(stream),
1774                                         bt_stream_get_name(stream));
1775                                 status = BT_SELF_MESSAGE_ITERATOR_STATUS_ERROR;
1776                                 goto end;
1777                         }
1778
1779                         if (bt_stream_class_supports_discarded_events(sc) &&
1780                                         !bt_stream_class_discarded_events_have_default_clock_snapshots(sc)) {
1781                                 BT_COMP_LOGE("Unsupported stream: discarded events "
1782                                         "have no clock snapshots: "
1783                                         "stream-addr=%p, "
1784                                         "stream-id=%" PRIu64 ", "
1785                                         "stream-name=\"%s\"",
1786                                         stream, bt_stream_get_id(stream),
1787                                         bt_stream_get_name(stream));
1788                                 status = BT_SELF_MESSAGE_ITERATOR_STATUS_ERROR;
1789                                 goto end;
1790                         }
1791
1792                         if (bt_stream_class_supports_discarded_packets(sc) &&
1793                                         !bt_stream_class_discarded_packets_have_default_clock_snapshots(sc)) {
1794                                 BT_COMP_LOGE("Unsupported stream: discarded packets "
1795                                         "have no clock snapshots: "
1796                                         "stream-addr=%p, "
1797                                         "stream-id=%" PRIu64 ", "
1798                                         "stream-name=\"%s\"",
1799                                         stream, bt_stream_get_id(stream),
1800                                         bt_stream_get_name(stream));
1801                                 status = BT_SELF_MESSAGE_ITERATOR_STATUS_ERROR;
1802                                 goto end;
1803                         }
1804
1805                         sstate = g_new0(struct trimmer_iterator_stream_state,
1806                                 1);
1807                         if (!sstate) {
1808                                 status = BT_SELF_MESSAGE_ITERATOR_STATUS_NOMEM;
1809                                 goto end;
1810                         }
1811
1812                         sstate->stream = stream;
1813                         sstate->stream_act_end_ns_from_origin = INT64_MIN;
1814                         g_hash_table_insert(trimmer_it->stream_states,
1815                                 (void *) stream, sstate);
1816                 }
1817         }
1818
1819         /* Retrieve the message's time */
1820         ret = get_msg_ns_from_origin(msg, &ns_from_origin, &skip);
1821         if (G_UNLIKELY(ret)) {
1822                 status = BT_SELF_MESSAGE_ITERATOR_STATUS_ERROR;
1823                 goto end;
1824         }
1825
1826         if (G_LIKELY(sstate)) {
1827                 /* Message associated to a stream */
1828                 status = handle_message_with_stream_state(trimmer_it, msg,
1829                         sstate, ns_from_origin, reached_end);
1830
1831                 /*
1832                  * handle_message_with_stream_state() unconditionally
1833                  * consumes `msg`.
1834                  */
1835                 msg = NULL;
1836         } else {
1837                 /*
1838                  * Message not associated to a stream (message iterator
1839                  * inactivity).
1840                  */
1841                 if (G_UNLIKELY(ns_from_origin > trimmer_it->end.ns_from_origin)) {
1842                         BT_MESSAGE_PUT_REF_AND_RESET(msg);
1843                         status = end_iterator_streams(trimmer_it);
1844                         *reached_end = true;
1845                 } else {
1846                         push_message(trimmer_it, msg);
1847                         status = BT_SELF_MESSAGE_ITERATOR_STATUS_OK;
1848                         msg = NULL;
1849                 }
1850         }
1851
1852 end:
1853         /* We release the message's reference whatever the outcome */
1854         bt_message_put_ref(msg);
1855         return status;
1856 }
1857
1858 static inline
1859 void fill_message_array_from_output_messages(
1860                 struct trimmer_iterator *trimmer_it,
1861                 bt_message_array_const msgs, uint64_t capacity, uint64_t *count)
1862 {
1863         *count = 0;
1864
1865         /*
1866          * Move auto-seek messages to the output array (which is this
1867          * iterator's base message array).
1868          */
1869         while (capacity > 0 && !g_queue_is_empty(trimmer_it->output_messages)) {
1870                 msgs[*count] = pop_message(trimmer_it);
1871                 capacity--;
1872                 (*count)++;
1873         }
1874
1875         BT_ASSERT(*count > 0);
1876 }
1877
1878 static inline
1879 bt_self_message_iterator_status state_ending(
1880                 struct trimmer_iterator *trimmer_it,
1881                 bt_message_array_const msgs, uint64_t capacity,
1882                 uint64_t *count)
1883 {
1884         bt_self_message_iterator_status status =
1885                 BT_SELF_MESSAGE_ITERATOR_STATUS_OK;
1886
1887         if (g_queue_is_empty(trimmer_it->output_messages)) {
1888                 trimmer_it->state = TRIMMER_ITERATOR_STATE_ENDED;
1889                 status = BT_SELF_MESSAGE_ITERATOR_STATUS_END;
1890                 goto end;
1891         }
1892
1893         fill_message_array_from_output_messages(trimmer_it, msgs,
1894                 capacity, count);
1895
1896 end:
1897         return status;
1898 }
1899
1900 static inline
1901 bt_self_message_iterator_status state_trim(struct trimmer_iterator *trimmer_it,
1902                 bt_message_array_const msgs, uint64_t capacity,
1903                 uint64_t *count)
1904 {
1905         bt_self_message_iterator_status status =
1906                 BT_SELF_MESSAGE_ITERATOR_STATUS_OK;
1907         bt_message_array_const my_msgs;
1908         uint64_t my_count;
1909         uint64_t i;
1910         bool reached_end = false;
1911
1912         while (g_queue_is_empty(trimmer_it->output_messages)) {
1913                 status = (int) bt_self_component_port_input_message_iterator_next(
1914                         trimmer_it->upstream_iter, &my_msgs, &my_count);
1915                 if (G_UNLIKELY(status != BT_SELF_MESSAGE_ITERATOR_STATUS_OK)) {
1916                         if (status == BT_SELF_MESSAGE_ITERATOR_STATUS_END) {
1917                                 status = end_iterator_streams(trimmer_it);
1918                                 if (status != BT_SELF_MESSAGE_ITERATOR_STATUS_OK) {
1919                                         goto end;
1920                                 }
1921
1922                                 trimmer_it->state =
1923                                         TRIMMER_ITERATOR_STATE_ENDING;
1924                                 status = state_ending(trimmer_it, msgs,
1925                                         capacity, count);
1926                         }
1927
1928                         goto end;
1929                 }
1930
1931                 BT_ASSERT(my_count > 0);
1932
1933                 for (i = 0; i < my_count; i++) {
1934                         status = handle_message(trimmer_it, my_msgs[i],
1935                                 &reached_end);
1936
1937                         /*
1938                          * handle_message() unconditionally consumes the
1939                          * message reference.
1940                          */
1941                         my_msgs[i] = NULL;
1942
1943                         if (G_UNLIKELY(status !=
1944                                         BT_SELF_MESSAGE_ITERATOR_STATUS_OK)) {
1945                                 put_messages(my_msgs, my_count);
1946                                 goto end;
1947                         }
1948
1949                         if (G_UNLIKELY(reached_end)) {
1950                                 /*
1951                                  * This message's time was passed the
1952                                  * trimming time range's end time: we
1953                                  * are done. Their might still be
1954                                  * messages in the output message queue,
1955                                  * so move to the "ending" state and
1956                                  * apply it immediately since
1957                                  * state_trim() is called within the
1958                                  * "next" method.
1959                                  */
1960                                 put_messages(my_msgs, my_count);
1961                                 trimmer_it->state =
1962                                         TRIMMER_ITERATOR_STATE_ENDING;
1963                                 status = state_ending(trimmer_it, msgs,
1964                                         capacity, count);
1965                                 goto end;
1966                         }
1967                 }
1968         }
1969
1970         /*
1971          * There's at least one message in the output message queue:
1972          * move the messages to the output message array.
1973          */
1974         BT_ASSERT(!g_queue_is_empty(trimmer_it->output_messages));
1975         fill_message_array_from_output_messages(trimmer_it, msgs,
1976                 capacity, count);
1977
1978 end:
1979         return status;
1980 }
1981
1982 BT_HIDDEN
1983 bt_self_message_iterator_status trimmer_msg_iter_next(
1984                 bt_self_message_iterator *self_msg_iter,
1985                 bt_message_array_const msgs, uint64_t capacity,
1986                 uint64_t *count)
1987 {
1988         struct trimmer_iterator *trimmer_it =
1989                 bt_self_message_iterator_get_data(self_msg_iter);
1990         bt_self_message_iterator_status status =
1991                 BT_SELF_MESSAGE_ITERATOR_STATUS_OK;
1992
1993         BT_ASSERT(trimmer_it);
1994
1995         if (G_LIKELY(trimmer_it->state == TRIMMER_ITERATOR_STATE_TRIM)) {
1996                 status = state_trim(trimmer_it, msgs, capacity, count);
1997                 if (status != BT_SELF_MESSAGE_ITERATOR_STATUS_OK) {
1998                         goto end;
1999                 }
2000         } else {
2001                 switch (trimmer_it->state) {
2002                 case TRIMMER_ITERATOR_STATE_SET_BOUNDS_NS_FROM_ORIGIN:
2003                         status = state_set_trimmer_iterator_bounds(trimmer_it);
2004                         if (status != BT_SELF_MESSAGE_ITERATOR_STATUS_OK) {
2005                                 goto end;
2006                         }
2007
2008                         status = state_seek_initially(trimmer_it);
2009                         if (status != BT_SELF_MESSAGE_ITERATOR_STATUS_OK) {
2010                                 goto end;
2011                         }
2012
2013                         status = state_trim(trimmer_it, msgs, capacity, count);
2014                         if (status != BT_SELF_MESSAGE_ITERATOR_STATUS_OK) {
2015                                 goto end;
2016                         }
2017
2018                         break;
2019                 case TRIMMER_ITERATOR_STATE_SEEK_INITIALLY:
2020                         status = state_seek_initially(trimmer_it);
2021                         if (status != BT_SELF_MESSAGE_ITERATOR_STATUS_OK) {
2022                                 goto end;
2023                         }
2024
2025                         status = state_trim(trimmer_it, msgs, capacity, count);
2026                         if (status != BT_SELF_MESSAGE_ITERATOR_STATUS_OK) {
2027                                 goto end;
2028                         }
2029
2030                         break;
2031                 case TRIMMER_ITERATOR_STATE_ENDING:
2032                         status = state_ending(trimmer_it, msgs, capacity,
2033                                 count);
2034                         if (status != BT_SELF_MESSAGE_ITERATOR_STATUS_OK) {
2035                                 goto end;
2036                         }
2037
2038                         break;
2039                 case TRIMMER_ITERATOR_STATE_ENDED:
2040                         status = BT_SELF_MESSAGE_ITERATOR_STATUS_END;
2041                         break;
2042                 default:
2043                         abort();
2044                 }
2045         }
2046
2047 end:
2048         return status;
2049 }
2050
2051 BT_HIDDEN
2052 void trimmer_msg_iter_finalize(bt_self_message_iterator *self_msg_iter)
2053 {
2054         struct trimmer_iterator *trimmer_it =
2055                 bt_self_message_iterator_get_data(self_msg_iter);
2056
2057         BT_ASSERT(trimmer_it);
2058         destroy_trimmer_iterator(trimmer_it);
2059 }