src/lib/graph/iterator.c

   1 /*
   2  * Copyright 2017-2018 Philippe Proulx <pproulx@efficios.com>
   3  * Copyright 2015 Jérémie Galarneau <jeremie.galarneau@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_LOG_TAG "LIB/MSG-ITER"
  25 #include "lib/logging.h"
  26
  27 #include "compat/compiler.h"
  28 #include "compat/glib.h"
  29 #include "lib/trace-ir/clock-class.h"
  30 #include "lib/trace-ir/clock-snapshot.h"
  31 #include <babeltrace2/trace-ir/field.h>
  32 #include <babeltrace2/trace-ir/event-const.h>
  33 #include "lib/trace-ir/event.h"
  34 #include <babeltrace2/trace-ir/packet-const.h>
  35 #include "lib/trace-ir/packet.h"
  36 #include "lib/trace-ir/stream.h"
  37 #include <babeltrace2/trace-ir/clock-class-const.h>
  38 #include <babeltrace2/trace-ir/stream-class-const.h>
  39 #include <babeltrace2/trace-ir/stream-const.h>
  40 #include <babeltrace2/graph/connection-const.h>
  41 #include <babeltrace2/graph/component-const.h>
  42 #include <babeltrace2/graph/component-sink-const.h>
  43 #include <babeltrace2/graph/message-const.h>
  44 #include <babeltrace2/graph/message-iterator.h>
  45 #include <babeltrace2/graph/self-component-port-input-message-iterator.h>
  46 #include <babeltrace2/graph/port-output-message-iterator.h>
  47 #include <babeltrace2/graph/message-event-const.h>
  48 #include <babeltrace2/graph/message-message-iterator-inactivity-const.h>
  49 #include <babeltrace2/graph/message-packet-beginning.h>
  50 #include <babeltrace2/graph/message-packet-beginning-const.h>
  51 #include <babeltrace2/graph/message-packet-end-const.h>
  52 #include <babeltrace2/graph/message-stream-activity-beginning.h>
  53 #include <babeltrace2/graph/message-stream-activity-beginning-const.h>
  54 #include <babeltrace2/graph/message-stream-activity-end-const.h>
  55 #include <babeltrace2/graph/message-stream-beginning.h>
  56 #include <babeltrace2/graph/message-stream-beginning-const.h>
  57 #include <babeltrace2/graph/message-stream-end-const.h>
  58 #include <babeltrace2/graph/port-const.h>
  59 #include <babeltrace2/graph/graph.h>
  60 #include <babeltrace2/graph/graph-const.h>
  61 #include <babeltrace2/types.h>
  62 #include "common/assert.h"
  63 #include "lib/assert-pre.h"
  64 #include "lib/assert-post.h"
  65 #include <stdint.h>
  66 #include <inttypes.h>
  67 #include <stdlib.h>
  68
  69 #include "component-class.h"
  70 #include "component-class-sink-colander.h"
  71 #include "component.h"
  72 #include "component-sink.h"
  73 #include "component-source.h"
  74 #include "connection.h"
  75 #include "graph.h"
  76 #include "message/discarded-items.h"
  77 #include "message/event.h"
  78 #include "message/iterator.h"
  79 #include "message/message.h"
  80 #include "message/message-iterator-inactivity.h"
  81 #include "message/stream.h"
  82 #include "message/packet.h"
  83 #include "message/stream-activity.h"
  84 #include "lib/func-status.h"
  85
  86 /*
  87  * TODO: Use graph's state (number of active iterators, etc.) and
  88  * possibly system specifications to make a better guess than this.
  89  */
  90 #define MSG_BATCH_SIZE  15
  91
  92 #define BT_ASSERT_PRE_ITER_HAS_STATE_TO_SEEK(_iter)                     \
  93         BT_ASSERT_PRE((_iter)->state == BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_ACTIVE || \
  94                 (_iter)->state == BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_ENDED || \
  95                 (_iter)->state == BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_LAST_SEEKING_RETURNED_AGAIN || \
  96                 (_iter)->state == BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_LAST_SEEKING_RETURNED_ERROR, \
  97                 "Message iterator is in the wrong state: %!+i", _iter)
  98
  99 static inline
 100 void set_self_comp_port_input_msg_iterator_state(
 101                 struct bt_self_component_port_input_message_iterator *iterator,
 102                 enum bt_self_component_port_input_message_iterator_state state)
 103 {
 104         BT_ASSERT(iterator);
 105         BT_LIB_LOGD("Updating message iterator's state: new-state=%s",
 106                 bt_self_component_port_input_message_iterator_state_string(state));
 107         iterator->state = state;
 108 }
 109
 110 static
 111 void destroy_base_message_iterator(struct bt_object *obj)
 112 {
 113         struct bt_message_iterator *iterator = (void *) obj;
 114
 115         BT_ASSERT(iterator);
 116
 117         if (iterator->msgs) {
 118                 g_ptr_array_free(iterator->msgs, TRUE);
 119                 iterator->msgs = NULL;
 120         }
 121
 122         g_free(iterator);
 123 }
 124
 125 static
 126 void bt_self_component_port_input_message_iterator_destroy(struct bt_object *obj)
 127 {
 128         struct bt_self_component_port_input_message_iterator *iterator;
 129
 130         BT_ASSERT(obj);
 131
 132         /*
 133          * The message iterator's reference count is 0 if we're
 134          * here. Increment it to avoid a double-destroy (possibly
 135          * infinitely recursive). This could happen for example if the
 136          * message iterator's finalization function does
 137          * bt_object_get_ref() (or anything that causes
 138          * bt_object_get_ref() to be called) on itself (ref. count goes
 139          * from 0 to 1), and then bt_object_put_ref(): the reference
 140          * count would go from 1 to 0 again and this function would be
 141          * called again.
 142          */
 143         obj->ref_count++;
 144         iterator = (void *) obj;
 145         BT_LIB_LOGI("Destroying self component input port message iterator object: "
 146                 "%!+i", iterator);
 147         bt_self_component_port_input_message_iterator_try_finalize(iterator);
 148
 149         if (iterator->connection) {
 150                 /*
 151                  * Remove ourself from the originating connection so
 152                  * that it does not try to finalize a dangling pointer
 153                  * later.
 154                  */
 155                 bt_connection_remove_iterator(iterator->connection, iterator);
 156                 iterator->connection = NULL;
 157         }
 158
 159         if (iterator->auto_seek.msgs) {
 160                 while (!g_queue_is_empty(iterator->auto_seek.msgs)) {
 161                         bt_object_put_no_null_check(
 162                                 g_queue_pop_tail(iterator->auto_seek.msgs));
 163                 }
 164
 165                 g_queue_free(iterator->auto_seek.msgs);
 166                 iterator->auto_seek.msgs = NULL;
 167         }
 168
 169         destroy_base_message_iterator(obj);
 170 }
 171
 172 BT_HIDDEN
 173 void bt_self_component_port_input_message_iterator_try_finalize(
 174                 struct bt_self_component_port_input_message_iterator *iterator)
 175 {
 176         typedef void (*method_t)(void *);
 177
 178         struct bt_component_class *comp_class = NULL;
 179         method_t method = NULL;
 180
 181         BT_ASSERT(iterator);
 182
 183         switch (iterator->state) {
 184         case BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_NON_INITIALIZED:
 185                 /* Skip user finalization if user initialization failed */
 186                 BT_LIB_LOGD("Not finalizing non-initialized message iterator: "
 187                         "%!+i", iterator);
 188                 goto end;
 189         case BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_FINALIZED:
 190                 /* Already finalized */
 191                 BT_LIB_LOGD("Not finalizing message iterator: already finalized: "
 192                         "%!+i", iterator);
 193                 goto end;
 194         case BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_FINALIZING:
 195                 /* Finalizing */
 196                 BT_LIB_LOGF("Message iterator is already being finalized: "
 197                         "%!+i", iterator);
 198                 abort();
 199         default:
 200                 break;
 201         }
 202
 203         BT_LIB_LOGD("Finalizing message iterator: %!+i", iterator);
 204         set_self_comp_port_input_msg_iterator_state(iterator,
 205                 BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_FINALIZING);
 206         BT_ASSERT(iterator->upstream_component);
 207         comp_class = iterator->upstream_component->class;
 208
 209         /* Call user-defined destroy method */
 210         switch (comp_class->type) {
 211         case BT_COMPONENT_CLASS_TYPE_SOURCE:
 212         {
 213                 struct bt_component_class_source *src_comp_cls =
 214                         (void *) comp_class;
 215
 216                 method = (method_t) src_comp_cls->methods.msg_iter_finalize;
 217                 break;
 218         }
 219         case BT_COMPONENT_CLASS_TYPE_FILTER:
 220         {
 221                 struct bt_component_class_filter *flt_comp_cls =
 222                         (void *) comp_class;
 223
 224                 method = (method_t) flt_comp_cls->methods.msg_iter_finalize;
 225                 break;
 226         }
 227         default:
 228                 /* Unreachable */
 229                 abort();
 230         }
 231
 232         if (method) {
 233                 BT_LIB_LOGD("Calling user's finalization method: %!+i",
 234                         iterator);
 235                 method(iterator);
 236         }
 237
 238         iterator->upstream_component = NULL;
 239         iterator->upstream_port = NULL;
 240         set_self_comp_port_input_msg_iterator_state(iterator,
 241                 BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_FINALIZED);
 242         BT_LIB_LOGD("Finalized message iterator: %!+i", iterator);
 243
 244 end:
 245         return;
 246 }
 247
 248 BT_HIDDEN
 249 void bt_self_component_port_input_message_iterator_set_connection(
 250                 struct bt_self_component_port_input_message_iterator *iterator,
 251                 struct bt_connection *connection)
 252 {
 253         BT_ASSERT(iterator);
 254         iterator->connection = connection;
 255         BT_LIB_LOGI("Set message iterator's connection: "
 256                 "%![iter-]+i, %![conn-]+x", iterator, connection);
 257 }
 258
 259 static
 260 int init_message_iterator(struct bt_message_iterator *iterator,
 261                 enum bt_message_iterator_type type,
 262                 bt_object_release_func destroy)
 263 {
 264         int ret = 0;
 265
 266         bt_object_init_shared(&iterator->base, destroy);
 267         iterator->type = type;
 268         iterator->msgs = g_ptr_array_new();
 269         if (!iterator->msgs) {
 270                 BT_LIB_LOGE_APPEND_CAUSE("Failed to allocate a GPtrArray.");
 271                 ret = -1;
 272                 goto end;
 273         }
 274
 275         g_ptr_array_set_size(iterator->msgs, MSG_BATCH_SIZE);
 276
 277 end:
 278         return ret;
 279 }
 280
 281 static
 282 bt_bool can_seek_ns_from_origin_true(
 283                 struct bt_self_component_port_input_message_iterator *iterator,
 284                 int64_t ns_from_origin)
 285 {
 286         return BT_TRUE;
 287 }
 288
 289 static
 290 bt_bool can_seek_beginning_true(
 291                 struct bt_self_component_port_input_message_iterator *iterator)
 292 {
 293         return BT_TRUE;
 294 }
 295
 296 static
 297 struct bt_self_component_port_input_message_iterator *
 298 bt_self_component_port_input_message_iterator_create_initial(
 299                 struct bt_component *upstream_comp,
 300                 struct bt_port *upstream_port)
 301 {
 302         int ret;
 303         struct bt_self_component_port_input_message_iterator *iterator = NULL;
 304
 305         BT_ASSERT(upstream_comp);
 306         BT_ASSERT(upstream_port);
 307         BT_ASSERT(bt_port_is_connected(upstream_port));
 308         BT_LIB_LOGI("Creating initial message iterator on self component input port: "
 309                 "%![up-comp-]+c, %![up-port-]+p", upstream_comp, upstream_port);
 310         BT_ASSERT(bt_component_get_class_type(upstream_comp) ==
 311                 BT_COMPONENT_CLASS_TYPE_SOURCE ||
 312                 bt_component_get_class_type(upstream_comp) ==
 313                 BT_COMPONENT_CLASS_TYPE_FILTER);
 314         iterator = g_new0(
 315                 struct bt_self_component_port_input_message_iterator, 1);
 316         if (!iterator) {
 317                 BT_LIB_LOGE_APPEND_CAUSE(
 318                         "Failed to allocate one self component input port "
 319                         "message iterator.");
 320                 goto error;
 321         }
 322
 323         ret = init_message_iterator((void *) iterator,
 324                 BT_MESSAGE_ITERATOR_TYPE_SELF_COMPONENT_PORT_INPUT,
 325                 bt_self_component_port_input_message_iterator_destroy);
 326         if (ret) {
 327                 /* init_message_iterator() logs errors */
 328                 goto error;
 329         }
 330
 331         iterator->last_ns_from_origin = INT64_MIN;
 332
 333         iterator->auto_seek.msgs = g_queue_new();
 334         if (!iterator->auto_seek.msgs) {
 335                 BT_LIB_LOGE_APPEND_CAUSE("Failed to allocate a GQueue.");
 336                 ret = -1;
 337                 goto end;
 338         }
 339
 340         iterator->upstream_component = upstream_comp;
 341         iterator->upstream_port = upstream_port;
 342         iterator->connection = iterator->upstream_port->connection;
 343         iterator->graph = bt_component_borrow_graph(upstream_comp);
 344         set_self_comp_port_input_msg_iterator_state(iterator,
 345                 BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_NON_INITIALIZED);
 346
 347         switch (iterator->upstream_component->class->type) {
 348         case BT_COMPONENT_CLASS_TYPE_SOURCE:
 349         {
 350                 struct bt_component_class_source *src_comp_cls =
 351                         (void *) iterator->upstream_component->class;
 352
 353                 iterator->methods.next =
 354                         (bt_self_component_port_input_message_iterator_next_method)
 355                                 src_comp_cls->methods.msg_iter_next;
 356                 iterator->methods.seek_ns_from_origin =
 357                         (bt_self_component_port_input_message_iterator_seek_ns_from_origin_method)
 358                                 src_comp_cls->methods.msg_iter_seek_ns_from_origin;
 359                 iterator->methods.seek_beginning =
 360                         (bt_self_component_port_input_message_iterator_seek_beginning_method)
 361                                 src_comp_cls->methods.msg_iter_seek_beginning;
 362                 iterator->methods.can_seek_ns_from_origin =
 363                         (bt_self_component_port_input_message_iterator_can_seek_ns_from_origin_method)
 364                                 src_comp_cls->methods.msg_iter_can_seek_ns_from_origin;
 365                 iterator->methods.can_seek_beginning =
 366                         (bt_self_component_port_input_message_iterator_can_seek_beginning_method)
 367                                 src_comp_cls->methods.msg_iter_can_seek_beginning;
 368                 break;
 369         }
 370         case BT_COMPONENT_CLASS_TYPE_FILTER:
 371         {
 372                 struct bt_component_class_filter *flt_comp_cls =
 373                         (void *) iterator->upstream_component->class;
 374
 375                 iterator->methods.next =
 376                         (bt_self_component_port_input_message_iterator_next_method)
 377                                 flt_comp_cls->methods.msg_iter_next;
 378                 iterator->methods.seek_ns_from_origin =
 379                         (bt_self_component_port_input_message_iterator_seek_ns_from_origin_method)
 380                                 flt_comp_cls->methods.msg_iter_seek_ns_from_origin;
 381                 iterator->methods.seek_beginning =
 382                         (bt_self_component_port_input_message_iterator_seek_beginning_method)
 383                                 flt_comp_cls->methods.msg_iter_seek_beginning;
 384                 iterator->methods.can_seek_ns_from_origin =
 385                         (bt_self_component_port_input_message_iterator_can_seek_ns_from_origin_method)
 386                                 flt_comp_cls->methods.msg_iter_can_seek_ns_from_origin;
 387                 iterator->methods.can_seek_beginning =
 388                         (bt_self_component_port_input_message_iterator_can_seek_beginning_method)
 389                                 flt_comp_cls->methods.msg_iter_can_seek_beginning;
 390                 break;
 391         }
 392         default:
 393                 abort();
 394         }
 395
 396         if (iterator->methods.seek_ns_from_origin &&
 397                         !iterator->methods.can_seek_ns_from_origin) {
 398                 iterator->methods.can_seek_ns_from_origin =
 399                         (bt_self_component_port_input_message_iterator_can_seek_ns_from_origin_method)
 400                                 can_seek_ns_from_origin_true;
 401         }
 402
 403         if (iterator->methods.seek_beginning &&
 404                         !iterator->methods.can_seek_beginning) {
 405                 iterator->methods.can_seek_beginning =
 406                         (bt_self_component_port_input_message_iterator_seek_beginning_method)
 407                                 can_seek_beginning_true;
 408         }
 409
 410         BT_LIB_LOGI("Created initial message iterator on self component input port: "
 411                 "%![up-port-]+p, %![up-comp-]+c, %![iter-]+i",
 412                 upstream_port, upstream_comp, iterator);
 413         goto end;
 414
 415 error:
 416         BT_OBJECT_PUT_REF_AND_RESET(iterator);
 417
 418 end:
 419         return iterator;
 420 }
 421
 422 struct bt_self_component_port_input_message_iterator *
 423 bt_self_component_port_input_message_iterator_create(
 424                 struct bt_self_component_port_input *self_port)
 425 {
 426         typedef enum bt_component_class_message_iterator_init_method_status (*init_method_t)(
 427                         void *, void *, void *);
 428
 429         init_method_t init_method = NULL;
 430         struct bt_self_component_port_input_message_iterator *iterator =
 431                 NULL;
 432         struct bt_port *port = (void *) self_port;
 433         struct bt_port *upstream_port;
 434         struct bt_component *comp;
 435         struct bt_component *upstream_comp;
 436         struct bt_component_class *upstream_comp_cls;
 437
 438         BT_ASSERT_PRE_NON_NULL(port, "Port");
 439         comp = bt_port_borrow_component_inline(port);
 440         BT_ASSERT_PRE(bt_port_is_connected(port),
 441                 "Port is not connected: %![port-]+p", port);
 442         BT_ASSERT_PRE(comp, "Port is not part of a component: %![port-]+p",
 443                 port);
 444         BT_ASSERT_PRE(!bt_component_graph_is_canceled(comp),
 445                 "Port's component's graph is canceled: "
 446                 "%![port-]+p, %![comp-]+c", port, comp);
 447         BT_ASSERT(port->connection);
 448         upstream_port = port->connection->upstream_port;
 449         BT_ASSERT(upstream_port);
 450         upstream_comp = bt_port_borrow_component_inline(upstream_port);
 451         BT_ASSERT(upstream_comp);
 452         BT_ASSERT_PRE(
 453                 bt_component_borrow_graph(upstream_comp)->config_state !=
 454                         BT_GRAPH_CONFIGURATION_STATE_CONFIGURING,
 455                 "Graph is not configured: %!+g",
 456                 bt_component_borrow_graph(upstream_comp));
 457         upstream_comp_cls = upstream_comp->class;
 458         BT_ASSERT(upstream_comp->class->type ==
 459                 BT_COMPONENT_CLASS_TYPE_SOURCE ||
 460                 upstream_comp->class->type ==
 461                 BT_COMPONENT_CLASS_TYPE_FILTER);
 462         iterator = bt_self_component_port_input_message_iterator_create_initial(
 463                 upstream_comp, upstream_port);
 464         if (!iterator) {
 465                 BT_LIB_LOGE_APPEND_CAUSE(
 466                         "Cannot create self component input port message iterator.");
 467                 goto error;
 468         }
 469
 470         switch (upstream_comp_cls->type) {
 471         case BT_COMPONENT_CLASS_TYPE_SOURCE:
 472         {
 473                 struct bt_component_class_source *src_comp_cls =
 474                         (void *) upstream_comp_cls;
 475
 476                 init_method =
 477                         (init_method_t) src_comp_cls->methods.msg_iter_init;
 478                 break;
 479         }
 480         case BT_COMPONENT_CLASS_TYPE_FILTER:
 481         {
 482                 struct bt_component_class_filter *flt_comp_cls =
 483                         (void *) upstream_comp_cls;
 484
 485                 init_method =
 486                         (init_method_t) flt_comp_cls->methods.msg_iter_init;
 487                 break;
 488         }
 489         default:
 490                 /* Unreachable */
 491                 abort();
 492         }
 493
 494         if (init_method) {
 495                 enum bt_component_class_message_iterator_init_method_status iter_status;
 496
 497                 BT_LIB_LOGD("Calling user's initialization method: %!+i", iterator);
 498                 iter_status = init_method(iterator, upstream_comp,
 499                         upstream_port);
 500                 BT_LOGD("User method returned: status=%s",
 501                         bt_common_func_status_string(iter_status));
 502                 if (iter_status != BT_FUNC_STATUS_OK) {
 503                         BT_LIB_LOGW_APPEND_CAUSE(
 504                                 "Component input port message iterator initialization method failed: "
 505                                 "%![iter-]+i, status=%s",
 506                                 iterator,
 507                                 bt_common_func_status_string(iter_status));
 508                         goto error;
 509                 }
 510         }
 511
 512         set_self_comp_port_input_msg_iterator_state(iterator,
 513                 BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_ACTIVE);
 514         g_ptr_array_add(port->connection->iterators, iterator);
 515         BT_LIB_LOGI("Created message iterator on self component input port: "
 516                 "%![up-port-]+p, %![up-comp-]+c, %![iter-]+i",
 517                 upstream_port, upstream_comp, iterator);
 518         goto end;
 519
 520 error:
 521         BT_OBJECT_PUT_REF_AND_RESET(iterator);
 522
 523 end:
 524         return iterator;
 525 }
 526
 527 void *bt_self_message_iterator_get_data(
 528                 const struct bt_self_message_iterator *self_iterator)
 529 {
 530         struct bt_self_component_port_input_message_iterator *iterator =
 531                 (void *) self_iterator;
 532
 533         BT_ASSERT_PRE_NON_NULL(iterator, "Message iterator");
 534         return iterator->user_data;
 535 }
 536
 537 void bt_self_message_iterator_set_data(
 538                 struct bt_self_message_iterator *self_iterator, void *data)
 539 {
 540         struct bt_self_component_port_input_message_iterator *iterator =
 541                 (void *) self_iterator;
 542
 543         BT_ASSERT_PRE_NON_NULL(iterator, "Message iterator");
 544         iterator->user_data = data;
 545         BT_LIB_LOGD("Set message iterator's user data: "
 546                 "%!+i, user-data-addr=%p", iterator, data);
 547 }
 548
 549 /*
 550  * Validate that the default clock snapshot in `msg` doesn't make us go back in
 551  * time.
 552  */
 553
 554 BT_ASSERT_POST_FUNC
 555 static
 556 bool clock_snapshots_are_monotonic_one(
 557                 struct bt_self_component_port_input_message_iterator *iterator,
 558                 const bt_message *msg)
 559 {
 560         const struct bt_clock_snapshot *clock_snapshot = NULL;
 561         bt_message_type message_type = bt_message_get_type(msg);
 562         int64_t ns_from_origin;
 563         enum bt_clock_snapshot_get_ns_from_origin_status clock_snapshot_status;
 564
 565         /*
 566          * The default is true: if we can't figure out the clock snapshot
 567          * (or there is none), assume it is fine.
 568          */
 569         bool result = true;
 570
 571         switch (message_type) {
 572         case BT_MESSAGE_TYPE_EVENT:
 573         {
 574                 struct bt_message_event *event_msg = (struct bt_message_event *) msg;
 575                 clock_snapshot = event_msg->default_cs;
 576                 break;
 577         }
 578         case BT_MESSAGE_TYPE_MESSAGE_ITERATOR_INACTIVITY:
 579         {
 580                 struct bt_message_message_iterator_inactivity *inactivity_msg =
 581                         (struct bt_message_message_iterator_inactivity *) msg;
 582                 clock_snapshot = inactivity_msg->default_cs;
 583                 break;
 584         }
 585         case BT_MESSAGE_TYPE_PACKET_BEGINNING:
 586         case BT_MESSAGE_TYPE_PACKET_END:
 587         {
 588                 struct bt_message_packet *packet_msg = (struct bt_message_packet *) msg;
 589                 clock_snapshot = packet_msg->default_cs;
 590                 break;
 591         }
 592         case BT_MESSAGE_TYPE_STREAM_ACTIVITY_BEGINNING:
 593         case BT_MESSAGE_TYPE_STREAM_ACTIVITY_END:
 594         {
 595                 struct bt_message_stream_activity *str_act_msg =
 596                         (struct bt_message_stream_activity *) msg;
 597
 598                 if (str_act_msg->default_cs_state == BT_MESSAGE_STREAM_ACTIVITY_CLOCK_SNAPSHOT_STATE_KNOWN) {
 599                         clock_snapshot = str_act_msg->default_cs;
 600                 }
 601                 break;
 602         }
 603         case BT_MESSAGE_TYPE_STREAM_BEGINNING:
 604         case BT_MESSAGE_TYPE_STREAM_END:
 605                 /* These messages don't have clock snapshots. */
 606                 goto end;
 607         case BT_MESSAGE_TYPE_DISCARDED_EVENTS:
 608         case BT_MESSAGE_TYPE_DISCARDED_PACKETS:
 609         {
 610                 struct bt_message_discarded_items *discarded_msg =
 611                         (struct bt_message_discarded_items *) msg;
 612
 613                 clock_snapshot = discarded_msg->default_begin_cs;
 614                 break;
 615         }
 616         }
 617
 618         if (!clock_snapshot) {
 619                 goto end;
 620         }
 621
 622         clock_snapshot_status = bt_clock_snapshot_get_ns_from_origin(clock_snapshot, &ns_from_origin);
 623         if (clock_snapshot_status != BT_FUNC_STATUS_OK) {
 624                 goto end;
 625         }
 626
 627         result = ns_from_origin >= iterator->last_ns_from_origin;
 628         iterator->last_ns_from_origin = ns_from_origin;
 629 end:
 630         return result;
 631 }
 632
 633 BT_ASSERT_POST_FUNC
 634 static
 635 bool clock_snapshots_are_monotonic(
 636                 struct bt_self_component_port_input_message_iterator *iterator,
 637                 bt_message_array_const msgs, uint64_t msg_count)
 638 {
 639         uint64_t i;
 640         bool result;
 641
 642         for (i = 0; i < msg_count; i++) {
 643                 if (!clock_snapshots_are_monotonic_one(iterator, msgs[i])) {
 644                         result = false;
 645                         goto end;
 646                 }
 647         }
 648
 649         result = true;
 650
 651 end:
 652         return result;
 653 }
 654
 655 /*
 656  * When a new stream begins, verify that the clock class tied to this
 657  * stream is compatible with what we've seen before.
 658  */
 659
 660 BT_ASSERT_POST_FUNC
 661 static
 662 bool clock_classes_are_compatible_one(struct bt_self_component_port_input_message_iterator *iterator,
 663                 const struct bt_message *msg)
 664 {
 665         enum bt_message_type message_type = bt_message_get_type(msg);
 666         bool result;
 667
 668         if (message_type == BT_MESSAGE_TYPE_STREAM_BEGINNING) {
 669                 const struct bt_message_stream *stream_msg = (struct bt_message_stream *) msg;
 670                 const struct bt_clock_class *clock_class = stream_msg->stream->class->default_clock_class;
 671                 bt_uuid clock_class_uuid = NULL;
 672
 673                 if (clock_class) {
 674                         clock_class_uuid = bt_clock_class_get_uuid(clock_class);
 675                 }
 676
 677                 switch (iterator->clock_expectation.type) {
 678                 case CLOCK_EXPECTATION_UNSET:
 679                         /*
 680                          * This is the first time we see a message with a clock
 681                          * snapshot: record the properties of that clock, against
 682                          * which we'll compare the clock properties of the following
 683                          * messages.
 684                          */
 685
 686                         if (!clock_class) {
 687                                 iterator->clock_expectation.type = CLOCK_EXPECTATION_NONE;
 688                         } else if (bt_clock_class_origin_is_unix_epoch(clock_class)) {
 689                                 iterator->clock_expectation.type = CLOCK_EXPECTATION_ORIGIN_UNIX;
 690                         } else if (clock_class_uuid) {
 691                                 iterator->clock_expectation.type = CLOCK_EXPECTATION_ORIGIN_OTHER_UUID;
 692                                 memcpy(iterator->clock_expectation.uuid, clock_class_uuid, BABELTRACE_UUID_LEN);
 693                         } else {
 694                                 iterator->clock_expectation.type = CLOCK_EXPECTATION_ORIGIN_OTHER_NO_UUID;
 695                         }
 696                         break;
 697
 698                 case CLOCK_EXPECTATION_NONE:
 699                         if (clock_class) {
 700                                 BT_ASSERT_POST_MSG("Expecting no clock class, got one: %![cc-]+K",
 701                                         clock_class);
 702                                 result = false;
 703                                 goto end;
 704                         }
 705
 706                         break;
 707
 708                 case CLOCK_EXPECTATION_ORIGIN_UNIX:
 709                         if (!clock_class) {
 710                                 BT_ASSERT_POST_MSG("Expecting a clock class, got none.");
 711                                 result = false;
 712                                 goto end;
 713                         }
 714
 715                         if (!bt_clock_class_origin_is_unix_epoch(clock_class)) {
 716                                 BT_ASSERT_POST_MSG("Expecting a clock class with Unix epoch origin: %![cc-]+K",
 717                                         clock_class);
 718                                 result = false;
 719                                 goto end;
 720                         }
 721                         break;
 722
 723                 case CLOCK_EXPECTATION_ORIGIN_OTHER_UUID:
 724                         if (!clock_class) {
 725                                 BT_ASSERT_POST_MSG("Expecting a clock class, got none.");
 726                                 result = false;
 727                                 goto end;
 728                         }
 729
 730                         if (bt_clock_class_origin_is_unix_epoch(clock_class)) {
 731                                 BT_ASSERT_POST_MSG("Expecting a clock class without Unix epoch origin: %![cc-]+K",
 732                                         clock_class);
 733                                 result = false;
 734                                 goto end;
 735                         }
 736
 737                         if (!clock_class_uuid) {
 738                                 BT_ASSERT_POST_MSG("Expecting a clock class with UUID: %![cc-]+K",
 739                                         clock_class);
 740                                 result = false;
 741                                 goto end;
 742                         }
 743
 744                         if (bt_uuid_compare(iterator->clock_expectation.uuid, clock_class_uuid)) {
 745                                 BT_ASSERT_POST_MSG("Expecting a clock class with UUID, got one "
 746                                         "with a different UUID: %![cc-]+K, expected-uuid=%!u",
 747                                         clock_class, iterator->clock_expectation.uuid);
 748                                 result = false;
 749                                 goto end;
 750                         }
 751                         break;
 752
 753                 case CLOCK_EXPECTATION_ORIGIN_OTHER_NO_UUID:
 754                         if (!clock_class) {
 755                                 BT_ASSERT_POST_MSG("Expecting a clock class, got none.");
 756                                 result = false;
 757                                 goto end;
 758                         }
 759
 760                         if (bt_clock_class_origin_is_unix_epoch(clock_class)) {
 761                                 BT_ASSERT_POST_MSG("Expecting a clock class without Unix epoch origin: %![cc-]+K",
 762                                         clock_class);
 763                                 result = false;
 764                                 goto end;
 765                         }
 766
 767                         if (clock_class_uuid) {
 768                                 BT_ASSERT_POST_MSG("Expecting a clock class without UUID: %![cc-]+K",
 769                                         clock_class);
 770                                 result = false;
 771                                 goto end;
 772                         }
 773                         break;
 774                 }
 775         }
 776
 777         result = true;
 778
 779 end:
 780         return result;
 781 }
 782
 783 BT_ASSERT_POST_FUNC
 784 static
 785 bool clock_classes_are_compatible(
 786                 struct bt_self_component_port_input_message_iterator *iterator,
 787                 bt_message_array_const msgs, uint64_t msg_count)
 788 {
 789         uint64_t i;
 790         bool result;
 791
 792         for (i = 0; i < msg_count; i++) {
 793                 if (!clock_classes_are_compatible_one(iterator, msgs[i])) {
 794                         result = false;
 795                         goto end;
 796                 }
 797         }
 798
 799         result = true;
 800
 801 end:
 802         return result;
 803 }
 804
 805 /*
 806  * Call the `next` method of the iterator.  Do some validation on the returned
 807  * messages.
 808  */
 809
 810 static
 811 enum bt_component_class_message_iterator_next_method_status
 812 call_iterator_next_method(
 813                 struct bt_self_component_port_input_message_iterator *iterator,
 814                 bt_message_array_const msgs, uint64_t capacity, uint64_t *user_count)
 815 {
 816         enum bt_component_class_message_iterator_next_method_status status;
 817
 818         BT_ASSERT(iterator->methods.next);
 819         BT_LOGD_STR("Calling user's \"next\" method.");
 820         status = iterator->methods.next(iterator, msgs, capacity, user_count);
 821         BT_LOGD("User method returned: status=%s, msg-count=%" PRIu64,
 822                 bt_common_func_status_string(status), *user_count);
 823
 824         if (status == BT_FUNC_STATUS_OK) {
 825                 BT_ASSERT_POST(clock_classes_are_compatible(iterator, msgs, *user_count),
 826                         "Clocks are not compatible");
 827                 BT_ASSERT_POST(clock_snapshots_are_monotonic(iterator, msgs, *user_count),
 828                         "Clock snapshots are not monotonic");
 829         }
 830
 831         return status;
 832 }
 833
 834 enum bt_message_iterator_next_status
 835 bt_self_component_port_input_message_iterator_next(
 836                 struct bt_self_component_port_input_message_iterator *iterator,
 837                 bt_message_array_const *msgs, uint64_t *user_count)
 838 {
 839         enum bt_message_iterator_next_status status = BT_FUNC_STATUS_OK;
 840
 841         BT_ASSERT_PRE_NON_NULL(iterator, "Message iterator");
 842         BT_ASSERT_PRE_NON_NULL(msgs, "Message array (output)");
 843         BT_ASSERT_PRE_NON_NULL(user_count, "Message count (output)");
 844         BT_ASSERT_PRE(iterator->state ==
 845                 BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_ACTIVE,
 846                 "Message iterator's \"next\" called, but "
 847                 "message iterator is in the wrong state: %!+i", iterator);
 848         BT_ASSERT(iterator->upstream_component);
 849         BT_ASSERT(iterator->upstream_component->class);
 850         BT_ASSERT_PRE(
 851                 bt_component_borrow_graph(iterator->upstream_component)->config_state !=
 852                         BT_GRAPH_CONFIGURATION_STATE_CONFIGURING,
 853                 "Graph is not configured: %!+g",
 854                 bt_component_borrow_graph(iterator->upstream_component));
 855         BT_LIB_LOGD("Getting next self component input port "
 856                 "message iterator's messages: %!+i, batch-size=%u",
 857                 iterator, MSG_BATCH_SIZE);
 858
 859         /*
 860          * Call the user's "next" method to get the next messages
 861          * and status.
 862          */
 863         *user_count = 0;
 864         status = (int) call_iterator_next_method(iterator,
 865                 (void *) iterator->base.msgs->pdata, MSG_BATCH_SIZE,
 866                 user_count);
 867         BT_LOGD("User method returned: status=%s, msg-count=%" PRIu64,
 868                 bt_common_func_status_string(status), *user_count);
 869         if (status < 0) {
 870                 BT_LIB_LOGW_APPEND_CAUSE(
 871                         "Component input port message iterator's \"next\" method failed: "
 872                         "%![iter-]+i, status=%s",
 873                         iterator, bt_common_func_status_string(status));
 874                 goto end;
 875         }
 876
 877         /*
 878          * There is no way that this iterator could have been finalized
 879          * during its "next" method, as the only way to do this is to
 880          * put the last iterator's reference, and this can only be done
 881          * by its downstream owner.
 882          *
 883          * For the same reason, there is no way that this iterator could
 884          * have seeked (cannot seek a self message iterator).
 885          */
 886         BT_ASSERT(iterator->state ==
 887                 BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_ACTIVE);
 888
 889         switch (status) {
 890         case BT_FUNC_STATUS_OK:
 891                 BT_ASSERT_POST(*user_count <= MSG_BATCH_SIZE,
 892                         "Invalid returned message count: greater than "
 893                         "batch size: count=%" PRIu64 ", batch-size=%u",
 894                         *user_count, MSG_BATCH_SIZE);
 895                 *msgs = (void *) iterator->base.msgs->pdata;
 896                 break;
 897         case BT_FUNC_STATUS_AGAIN:
 898                 goto end;
 899         case BT_FUNC_STATUS_END:
 900                 set_self_comp_port_input_msg_iterator_state(iterator,
 901                         BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_ENDED);
 902                 goto end;
 903         default:
 904                 /* Unknown non-error status */
 905                 abort();
 906         }
 907
 908 end:
 909         return status;
 910 }
 911
 912 enum bt_message_iterator_next_status bt_port_output_message_iterator_next(
 913                 struct bt_port_output_message_iterator *iterator,
 914                 bt_message_array_const *msgs_to_user,
 915                 uint64_t *count_to_user)
 916 {
 917         enum bt_message_iterator_next_status status;
 918         int graph_status;
 919
 920         BT_ASSERT_PRE_NON_NULL(iterator, "Message iterator");
 921         BT_ASSERT_PRE_NON_NULL(msgs_to_user, "Message array (output)");
 922         BT_ASSERT_PRE_NON_NULL(count_to_user, "Message count (output)");
 923         BT_LIB_LOGD("Getting next output port message iterator's messages: "
 924                 "%!+i", iterator);
 925         graph_status = bt_graph_consume_sink_no_check(iterator->graph,
 926                 iterator->colander);
 927         switch (graph_status) {
 928         case BT_FUNC_STATUS_CANCELED:
 929         case BT_FUNC_STATUS_AGAIN:
 930         case BT_FUNC_STATUS_END:
 931         case BT_FUNC_STATUS_MEMORY_ERROR:
 932                 status = (int) graph_status;
 933                 break;
 934         case BT_FUNC_STATUS_OK:
 935                 status = BT_FUNC_STATUS_OK;
 936
 937                 /*
 938                  * On success, the colander sink moves the messages
 939                  * to this iterator's array and sets this iterator's
 940                  * message count: move them to the user.
 941                  */
 942                 *msgs_to_user = (void *) iterator->base.msgs->pdata;
 943                 *count_to_user = iterator->count;
 944                 break;
 945         default:
 946                 /* Other errors */
 947                 status = BT_FUNC_STATUS_ERROR;
 948         }
 949
 950         return status;
 951 }
 952
 953 struct bt_component *
 954 bt_self_component_port_input_message_iterator_borrow_component(
 955                 struct bt_self_component_port_input_message_iterator *iterator)
 956 {
 957         BT_ASSERT_PRE_NON_NULL(iterator, "Message iterator");
 958         return iterator->upstream_component;
 959 }
 960
 961 const struct bt_component *
 962 bt_self_component_port_input_message_iterator_borrow_component_const(
 963                 const struct bt_self_component_port_input_message_iterator *iterator)
 964 {
 965         BT_ASSERT_PRE_NON_NULL(iterator, "Message iterator");
 966         return iterator->upstream_component;
 967 }
 968
 969 struct bt_self_component *bt_self_message_iterator_borrow_component(
 970                 struct bt_self_message_iterator *self_iterator)
 971 {
 972         struct bt_self_component_port_input_message_iterator *iterator =
 973                 (void *) self_iterator;
 974
 975         BT_ASSERT_PRE_NON_NULL(iterator, "Message iterator");
 976         return (void *) iterator->upstream_component;
 977 }
 978
 979 struct bt_self_port_output *bt_self_message_iterator_borrow_port(
 980                 struct bt_self_message_iterator *self_iterator)
 981 {
 982         struct bt_self_component_port_input_message_iterator *iterator =
 983                 (void *) self_iterator;
 984
 985         BT_ASSERT_PRE_NON_NULL(iterator, "Message iterator");
 986         return (void *) iterator->upstream_port;
 987 }
 988
 989 static
 990 void bt_port_output_message_iterator_destroy(struct bt_object *obj)
 991 {
 992         struct bt_port_output_message_iterator *iterator = (void *) obj;
 993
 994         BT_LIB_LOGI("Destroying output port message iterator object: %!+i",
 995                 iterator);
 996         BT_LOGD_STR("Putting graph.");
 997         BT_OBJECT_PUT_REF_AND_RESET(iterator->graph);
 998         BT_LOGD_STR("Putting colander sink component.");
 999         BT_OBJECT_PUT_REF_AND_RESET(iterator->colander);
1000         destroy_base_message_iterator(obj);
1001 }
1002
1003 struct bt_port_output_message_iterator *
1004 bt_port_output_message_iterator_create(struct bt_graph *graph,
1005                 const struct bt_port_output *output_port)
1006 {
1007         struct bt_port_output_message_iterator *iterator = NULL;
1008         struct bt_component_class_sink *colander_comp_cls = NULL;
1009         struct bt_component *output_port_comp = NULL;
1010         struct bt_component_sink *colander_comp;
1011         int graph_status;
1012         struct bt_port_input *colander_in_port = NULL;
1013         struct bt_component_class_sink_colander_data colander_data;
1014         int ret;
1015
1016         BT_ASSERT_PRE_NON_NULL(graph, "Graph");
1017         BT_ASSERT_PRE_NON_NULL(output_port, "Output port");
1018         output_port_comp = bt_port_borrow_component_inline(
1019                 (const void *) output_port);
1020         BT_ASSERT_PRE(output_port_comp,
1021                 "Output port has no component: %!+p", output_port);
1022         BT_ASSERT_PRE(bt_component_borrow_graph(output_port_comp) ==
1023                 (void *) graph,
1024                 "Output port is not part of graph: %![graph-]+g, %![port-]+p",
1025                 graph, output_port);
1026         BT_ASSERT_PRE(!graph->has_sink,
1027                 "Graph already has a sink component: %![graph-]+g");
1028
1029         /* Create message iterator */
1030         BT_LIB_LOGI("Creating message iterator on output port: "
1031                 "%![port-]+p, %![comp-]+c", output_port, output_port_comp);
1032         iterator = g_new0(struct bt_port_output_message_iterator, 1);
1033         if (!iterator) {
1034                 BT_LIB_LOGE_APPEND_CAUSE(
1035                         "Failed to allocate one output port message iterator.");
1036                 goto error;
1037         }
1038
1039         ret = init_message_iterator((void *) iterator,
1040                 BT_MESSAGE_ITERATOR_TYPE_PORT_OUTPUT,
1041                 bt_port_output_message_iterator_destroy);
1042         if (ret) {
1043                 /* init_message_iterator() logs errors */
1044                 BT_OBJECT_PUT_REF_AND_RESET(iterator);
1045                 goto end;
1046         }
1047
1048         /* Create colander component */
1049         colander_comp_cls = bt_component_class_sink_colander_get();
1050         if (!colander_comp_cls) {
1051                 /* bt_component_class_sink_colander_get() logs errors */
1052                 BT_LIB_LOGE_APPEND_CAUSE(
1053                         "Cannot get colander sink component class.");
1054                 goto error;
1055         }
1056
1057         iterator->graph = graph;
1058         bt_object_get_no_null_check(iterator->graph);
1059         colander_data.msgs = (void *) iterator->base.msgs->pdata;
1060         colander_data.count_addr = &iterator->count;
1061
1062         /*
1063          * Hope that nobody uses this very unique name.
1064          *
1065          * We pass `BT_LOGGING_LEVEL_NONE` but the colander component
1066          * class module does not use this level anyway since it belongs
1067          * to the library.
1068          */
1069         graph_status = bt_graph_add_sink_component_with_init_method_data(
1070                         (void *) graph, colander_comp_cls,
1071                         "colander-36ac3409-b1a8-4d60-ab1f-4fdf341a8fb1",
1072                         NULL, &colander_data, BT_LOGGING_LEVEL_NONE,
1073                         (void *) &iterator->colander);
1074         if (graph_status != BT_FUNC_STATUS_OK) {
1075                 BT_LIB_LOGE_APPEND_CAUSE(
1076                         "Cannot add colander sink component to graph: "
1077                         "%![graph-]+g, status=%s", graph,
1078                         bt_common_func_status_string(graph_status));
1079                 goto error;
1080         }
1081
1082         /*
1083          * Connect provided output port to the colander component's
1084          * input port.
1085          */
1086         colander_in_port =
1087                 (void *) bt_component_sink_borrow_input_port_by_index_const(
1088                         (void *) iterator->colander, 0);
1089         BT_ASSERT(colander_in_port);
1090         graph_status = bt_graph_connect_ports(graph,
1091                 output_port, colander_in_port, NULL);
1092         if (graph_status != BT_FUNC_STATUS_OK) {
1093                 BT_LIB_LOGW_APPEND_CAUSE(
1094                         "Cannot connect colander sink's port: "
1095                         "%![graph-]+g, %![comp-]+c, status=%s", graph,
1096                         iterator->colander,
1097                         bt_common_func_status_string(graph_status));
1098                 goto error;
1099         }
1100
1101         /*
1102          * At this point everything went fine. Make the graph
1103          * nonconsumable forever so that only this message iterator
1104          * can consume (thanks to bt_graph_consume_sink_no_check()).
1105          * This avoids leaking the message created by the colander
1106          * sink and moved to the message iterator's message
1107          * member.
1108          */
1109         bt_graph_set_can_consume(iterator->graph, false);
1110
1111         /* Also set the graph as being configured. */
1112         graph_status = bt_graph_configure(graph);
1113         if (graph_status != BT_FUNC_STATUS_OK) {
1114                 BT_LIB_LOGW_APPEND_CAUSE(
1115                         "Cannot configure graph after having "
1116                         "added and connected colander sink: "
1117                         "%![graph-]+g, %![comp-]+c, status=%s", graph,
1118                         iterator->colander,
1119                         bt_common_func_status_string(graph_status));
1120                 goto error;
1121         }
1122         goto end;
1123
1124 error:
1125         if (iterator && iterator->graph && iterator->colander) {
1126                 int ret;
1127
1128                 /* Remove created colander component from graph if any */
1129                 colander_comp = iterator->colander;
1130                 BT_OBJECT_PUT_REF_AND_RESET(iterator->colander);
1131
1132                 /*
1133                  * At this point the colander component's reference
1134                  * count is 0 because iterator->colander was the only
1135                  * owner. We also know that it is not connected because
1136                  * this is the last operation before this function
1137                  * succeeds.
1138                  *
1139                  * Since we honor the preconditions here,
1140                  * bt_graph_remove_unconnected_component() always
1141                  * succeeds.
1142                  */
1143                 ret = bt_graph_remove_unconnected_component(iterator->graph,
1144                         (void *) colander_comp);
1145                 BT_ASSERT(ret == 0);
1146         }
1147
1148         BT_OBJECT_PUT_REF_AND_RESET(iterator);
1149
1150 end:
1151         bt_object_put_ref(colander_comp_cls);
1152         return (void *) iterator;
1153 }
1154
1155 bt_bool bt_self_component_port_input_message_iterator_can_seek_ns_from_origin(
1156                 struct bt_self_component_port_input_message_iterator *iterator,
1157                 int64_t ns_from_origin)
1158 {
1159         bt_bool can = BT_FALSE;
1160
1161         BT_ASSERT_PRE_NON_NULL(iterator, "Message iterator");
1162         BT_ASSERT_PRE_ITER_HAS_STATE_TO_SEEK(iterator);
1163         BT_ASSERT_PRE(
1164                 bt_component_borrow_graph(iterator->upstream_component)->config_state !=
1165                         BT_GRAPH_CONFIGURATION_STATE_CONFIGURING,
1166                 "Graph is not configured: %!+g",
1167                 bt_component_borrow_graph(iterator->upstream_component));
1168
1169         if (iterator->methods.can_seek_ns_from_origin) {
1170                 can = iterator->methods.can_seek_ns_from_origin(iterator,
1171                         ns_from_origin);
1172                 goto end;
1173         }
1174
1175         /*
1176          * Automatic seeking fall back: if we can seek to the beginning,
1177          * then we can automatically seek to any message.
1178          */
1179         if (iterator->methods.can_seek_beginning) {
1180                 can = iterator->methods.can_seek_beginning(iterator);
1181         }
1182
1183 end:
1184         return can;
1185 }
1186
1187 bt_bool bt_self_component_port_input_message_iterator_can_seek_beginning(
1188                 struct bt_self_component_port_input_message_iterator *iterator)
1189 {
1190         bt_bool can = BT_FALSE;
1191
1192         BT_ASSERT_PRE_NON_NULL(iterator, "Message iterator");
1193         BT_ASSERT_PRE_ITER_HAS_STATE_TO_SEEK(iterator);
1194         BT_ASSERT_PRE(
1195                 bt_component_borrow_graph(iterator->upstream_component)->config_state !=
1196                         BT_GRAPH_CONFIGURATION_STATE_CONFIGURING,
1197                 "Graph is not configured: %!+g",
1198                 bt_component_borrow_graph(iterator->upstream_component));
1199
1200         if (iterator->methods.can_seek_beginning) {
1201                 can = iterator->methods.can_seek_beginning(iterator);
1202         }
1203
1204         return can;
1205 }
1206
1207 static inline
1208 void set_iterator_state_after_seeking(
1209                 struct bt_self_component_port_input_message_iterator *iterator,
1210                 int status)
1211 {
1212         enum bt_self_component_port_input_message_iterator_state new_state = 0;
1213
1214         /* Set iterator's state depending on seeking status */
1215         switch (status) {
1216         case BT_FUNC_STATUS_OK:
1217                 new_state = BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_ACTIVE;
1218                 break;
1219         case BT_FUNC_STATUS_AGAIN:
1220                 new_state = BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_LAST_SEEKING_RETURNED_AGAIN;
1221                 break;
1222         case BT_FUNC_STATUS_ERROR:
1223         case BT_FUNC_STATUS_MEMORY_ERROR:
1224                 new_state = BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_LAST_SEEKING_RETURNED_ERROR;
1225                 break;
1226         case BT_FUNC_STATUS_END:
1227                 new_state = BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_ENDED;
1228                 break;
1229         default:
1230                 abort();
1231         }
1232
1233         set_self_comp_port_input_msg_iterator_state(iterator, new_state);
1234 }
1235
1236 static
1237 void reset_iterator_expectations(
1238                 struct bt_self_component_port_input_message_iterator *iterator)
1239 {
1240         iterator->last_ns_from_origin = INT64_MIN;
1241         iterator->clock_expectation.type = CLOCK_EXPECTATION_UNSET;
1242 }
1243
1244 enum bt_message_iterator_seek_beginning_status
1245 bt_self_component_port_input_message_iterator_seek_beginning(
1246                 struct bt_self_component_port_input_message_iterator *iterator)
1247 {
1248         int status;
1249
1250         BT_ASSERT_PRE_NON_NULL(iterator, "Message iterator");
1251         BT_ASSERT_PRE_ITER_HAS_STATE_TO_SEEK(iterator);
1252         BT_ASSERT_PRE(
1253                 bt_component_borrow_graph(iterator->upstream_component)->config_state !=
1254                         BT_GRAPH_CONFIGURATION_STATE_CONFIGURING,
1255                 "Graph is not configured: %!+g",
1256                 bt_component_borrow_graph(iterator->upstream_component));
1257         BT_ASSERT_PRE(
1258                 bt_self_component_port_input_message_iterator_can_seek_beginning(
1259                         iterator),
1260                 "Message iterator cannot seek beginning: %!+i", iterator);
1261
1262         /*
1263          * We are seeking, reset our expectations about how the following
1264          * messages should look like.
1265          */
1266         reset_iterator_expectations(iterator);
1267
1268         BT_LIB_LOGD("Calling user's \"seek beginning\" method: %!+i", iterator);
1269         set_self_comp_port_input_msg_iterator_state(iterator,
1270                 BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_SEEKING);
1271         status = iterator->methods.seek_beginning(iterator);
1272         BT_LOGD("User method returned: status=%s",
1273                 bt_common_func_status_string(status));
1274         BT_ASSERT_POST(status == BT_FUNC_STATUS_OK ||
1275                 status == BT_FUNC_STATUS_ERROR ||
1276                 status == BT_FUNC_STATUS_MEMORY_ERROR ||
1277                 status == BT_FUNC_STATUS_AGAIN,
1278                 "Unexpected status: %![iter-]+i, status=%s",
1279                 iterator, bt_common_func_status_string(status));
1280         if (status < 0) {
1281                 BT_LIB_LOGW_APPEND_CAUSE(
1282                         "Component input port message iterator's \"seek beginning\" method failed: "
1283                         "%![iter-]+i, status=%s",
1284                         iterator, bt_common_func_status_string(status));
1285         }
1286
1287         set_iterator_state_after_seeking(iterator, status);
1288         return status;
1289 }
1290
1291 /*
1292  * Structure used to record the state of a given stream during the fast-forward
1293  * phase of an auto-seek.
1294  */
1295 struct auto_seek_stream_state {
1296         /*
1297          * Value representing which step of this timeline we are at.
1298          *
1299          *      time --->
1300          *   [SB]  1  [SAB]  2  [PB]  3  [PE]  2  [SAE]  1  [SE]
1301          *
1302          * At each point in the timeline, the messages we need to replicate are:
1303          *
1304          *   1: Stream beginning
1305          *   2: Stream beginning, stream activity beginning
1306          *   3: Stream beginning, stream activity beginning, packet beginning
1307          *
1308          * Before "Stream beginning" and after "Stream end", we don't need to
1309          * replicate anything as the stream doesn't exist.
1310          */
1311         enum {
1312                 AUTO_SEEK_STREAM_STATE_STREAM_BEGAN,
1313                 AUTO_SEEK_STREAM_STATE_STREAM_ACTIVITY_BEGAN,
1314                 AUTO_SEEK_STREAM_STATE_PACKET_BEGAN,
1315         } state;
1316
1317         /*
1318          * If `state` is AUTO_SEEK_STREAM_STATE_PACKET_BEGAN, the packet we are
1319          * in.  This is a weak reference, since the packet will always be
1320          * alive by the time we use it.
1321          */
1322         struct bt_packet *packet;
1323 };
1324
1325 static
1326 struct auto_seek_stream_state *create_auto_seek_stream_state(void)
1327 {
1328         return g_new0(struct auto_seek_stream_state, 1);
1329 }
1330
1331 static
1332 void destroy_auto_seek_stream_state(void *ptr)
1333 {
1334         g_free(ptr);
1335 }
1336
1337 static
1338 GHashTable *create_auto_seek_stream_states(void)
1339 {
1340         return g_hash_table_new_full(g_direct_hash, g_direct_equal, NULL,
1341                 destroy_auto_seek_stream_state);
1342 }
1343
1344 static
1345 void destroy_auto_seek_stream_states(GHashTable *stream_states)
1346 {
1347         g_hash_table_destroy(stream_states);
1348 }
1349
1350 /*
1351  * Handle one message while we are in the fast-forward phase of an auto-seek.
1352  *
1353  * Sets `*got_first` to true if the message's timestamp is greater or equal to
1354  * `ns_from_origin`.  In other words, if this is the first message after our
1355  * seek point.
1356  *
1357  * `stream_states` is an hash table of `bt_stream *` (weak reference) to
1358  * `struct auto_seek_stream_state` used to keep the state of each stream
1359  * during the fast-forward.
1360  */
1361
1362 static inline
1363 int auto_seek_handle_message(
1364                 struct bt_self_component_port_input_message_iterator *iterator,
1365                 int64_t ns_from_origin, const struct bt_message *msg,
1366                 bool *got_first, GHashTable *stream_states)
1367 {
1368         int status = BT_FUNC_STATUS_OK;
1369         int64_t msg_ns_from_origin;
1370         const struct bt_clock_snapshot *clk_snapshot = NULL;
1371         int ret;
1372
1373         BT_ASSERT(msg);
1374         BT_ASSERT(got_first);
1375
1376         switch (msg->type) {
1377         case BT_MESSAGE_TYPE_EVENT:
1378         {
1379                 const struct bt_message_event *event_msg =
1380                         (const void *) msg;
1381
1382                 clk_snapshot = event_msg->default_cs;
1383                 BT_ASSERT_POST(clk_snapshot,
1384                         "Event message has no default clock snapshot: %!+n",
1385                         event_msg);
1386                 break;
1387         }
1388         case BT_MESSAGE_TYPE_MESSAGE_ITERATOR_INACTIVITY:
1389         {
1390                 const struct bt_message_message_iterator_inactivity *inactivity_msg =
1391                         (const void *) msg;
1392
1393                 clk_snapshot = inactivity_msg->default_cs;
1394                 BT_ASSERT(clk_snapshot);
1395                 break;
1396         }
1397         case BT_MESSAGE_TYPE_PACKET_BEGINNING:
1398         case BT_MESSAGE_TYPE_PACKET_END:
1399         {
1400                 const struct bt_message_packet *packet_msg =
1401                         (const void *) msg;
1402
1403                 clk_snapshot = packet_msg->default_cs;
1404                 BT_ASSERT_POST(clk_snapshot,
1405                         "Packet message has no default clock snapshot: %!+n",
1406                         packet_msg);
1407                 break;
1408         }
1409         case BT_MESSAGE_TYPE_DISCARDED_EVENTS:
1410         case BT_MESSAGE_TYPE_DISCARDED_PACKETS:
1411         {
1412                 struct bt_message_discarded_items *msg_disc_items =
1413                         (void *) msg;
1414
1415                 BT_ASSERT_POST(msg_disc_items->default_begin_cs &&
1416                         msg_disc_items->default_end_cs,
1417                         "Discarded events/packets message has no default clock snapshots: %!+n",
1418                         msg_disc_items);
1419                 ret = bt_clock_snapshot_get_ns_from_origin(
1420                         msg_disc_items->default_begin_cs,
1421                         &msg_ns_from_origin);
1422                 if (ret) {
1423                         status = BT_FUNC_STATUS_ERROR;
1424                         goto end;
1425                 }
1426
1427                 if (msg_ns_from_origin >= ns_from_origin) {
1428                         *got_first = true;
1429                         goto push_msg;
1430                 }
1431
1432                 ret = bt_clock_snapshot_get_ns_from_origin(
1433                         msg_disc_items->default_end_cs,
1434                         &msg_ns_from_origin);
1435                 if (ret) {
1436                         status = BT_FUNC_STATUS_ERROR;
1437                         goto end;
1438                 }
1439
1440                 if (msg_ns_from_origin >= ns_from_origin) {
1441                         /*
1442                          * The discarded items message's beginning time
1443                          * is before the requested seeking time, but its
1444                          * end time is after. Modify the message so as
1445                          * to set its beginning time to the requested
1446                          * seeking time, and make its item count unknown
1447                          * as we don't know if items were really
1448                          * discarded within the new time range.
1449                          */
1450                         uint64_t new_begin_raw_value;
1451
1452                         ret = bt_clock_class_clock_value_from_ns_from_origin(
1453                                 msg_disc_items->default_end_cs->clock_class,
1454                                 ns_from_origin, &new_begin_raw_value);
1455                         if (ret) {
1456                                 status = BT_FUNC_STATUS_ERROR;
1457                                 goto end;
1458                         }
1459
1460                         bt_clock_snapshot_set_raw_value(
1461                                 msg_disc_items->default_begin_cs,
1462                                 new_begin_raw_value);
1463                         msg_disc_items->count.base.avail =
1464                                 BT_PROPERTY_AVAILABILITY_NOT_AVAILABLE;
1465
1466                         /*
1467                          * It is safe to push it because its beginning
1468                          * time is exactly the requested seeking time.
1469                          */
1470                         goto push_msg;
1471                 } else {
1472                         goto skip_msg;
1473                 }
1474         }
1475         case BT_MESSAGE_TYPE_STREAM_ACTIVITY_BEGINNING:
1476         {
1477                 const struct bt_message_stream_activity *stream_act_msg =
1478                         (const void *) msg;
1479
1480                 switch (stream_act_msg->default_cs_state) {
1481                 case BT_MESSAGE_STREAM_ACTIVITY_CLOCK_SNAPSHOT_STATE_UNKNOWN:
1482                 case BT_MESSAGE_STREAM_ACTIVITY_CLOCK_SNAPSHOT_STATE_INFINITE:
1483                         /*
1484                          * -inf is always less than any requested time,
1485                          * and we can't assume any specific time for an
1486                          * unknown clock snapshot, so skip this.
1487                          */
1488                         goto skip_msg;
1489                 case BT_MESSAGE_STREAM_ACTIVITY_CLOCK_SNAPSHOT_STATE_KNOWN:
1490                         clk_snapshot = stream_act_msg->default_cs;
1491                         BT_ASSERT(clk_snapshot);
1492                         break;
1493                 default:
1494                         abort();
1495                 }
1496
1497                 break;
1498         }
1499         case BT_MESSAGE_TYPE_STREAM_ACTIVITY_END:
1500         {
1501                 const struct bt_message_stream_activity *stream_act_msg =
1502                         (const void *) msg;
1503
1504                 switch (stream_act_msg->default_cs_state) {
1505                 case BT_MESSAGE_STREAM_ACTIVITY_CLOCK_SNAPSHOT_STATE_UNKNOWN:
1506                         /*
1507                          * We can't assume any specific time for an
1508                          * unknown clock snapshot, so skip this.
1509                          */
1510                         goto skip_msg;
1511                 case BT_MESSAGE_STREAM_ACTIVITY_CLOCK_SNAPSHOT_STATE_INFINITE:
1512                         /*
1513                          * +inf is always greater than any requested
1514                          * time.
1515                          */
1516                         *got_first = true;
1517                         goto push_msg;
1518                 case BT_MESSAGE_STREAM_ACTIVITY_CLOCK_SNAPSHOT_STATE_KNOWN:
1519                         clk_snapshot = stream_act_msg->default_cs;
1520                         BT_ASSERT(clk_snapshot);
1521                         break;
1522                 default:
1523                         abort();
1524                 }
1525
1526                 break;
1527         }
1528         case BT_MESSAGE_TYPE_STREAM_BEGINNING:
1529         case BT_MESSAGE_TYPE_STREAM_END:
1530                 /* Ignore */
1531                 goto skip_msg;
1532         default:
1533                 abort();
1534         }
1535
1536         BT_ASSERT(clk_snapshot);
1537         ret = bt_clock_snapshot_get_ns_from_origin(clk_snapshot,
1538                 &msg_ns_from_origin);
1539         if (ret) {
1540                 status = BT_FUNC_STATUS_ERROR;
1541                 goto end;
1542         }
1543
1544         if (msg_ns_from_origin >= ns_from_origin) {
1545                 *got_first = true;
1546                 goto push_msg;
1547         }
1548
1549 skip_msg:
1550         /* This message won't be sent downstream. */
1551         switch (msg->type) {
1552         case BT_MESSAGE_TYPE_STREAM_BEGINNING:
1553         {
1554                 const struct bt_message_stream *stream_msg = (const void *) msg;
1555                 struct auto_seek_stream_state *stream_state;
1556
1557                 /* Update stream's state: stream began. */
1558                 stream_state = create_auto_seek_stream_state();
1559                 if (!stream_state) {
1560                         status = BT_FUNC_STATUS_MEMORY_ERROR;
1561                         goto end;
1562                 }
1563
1564                 stream_state->state = AUTO_SEEK_STREAM_STATE_STREAM_BEGAN;
1565
1566                 BT_ASSERT(!bt_g_hash_table_contains(stream_states, stream_msg->stream));
1567                 g_hash_table_insert(stream_states, stream_msg->stream, stream_state);
1568                 break;
1569         }
1570         case BT_MESSAGE_TYPE_STREAM_ACTIVITY_BEGINNING:
1571         {
1572                 const struct bt_message_stream_activity *stream_act_msg =
1573                         (const void *) msg;
1574                 struct auto_seek_stream_state *stream_state;
1575
1576                 /* Update stream's state: stream activity began. */
1577                 stream_state = g_hash_table_lookup(stream_states, stream_act_msg->stream);
1578                 BT_ASSERT(stream_state);
1579
1580                 BT_ASSERT(stream_state->state == AUTO_SEEK_STREAM_STATE_STREAM_BEGAN);
1581                 stream_state->state = AUTO_SEEK_STREAM_STATE_STREAM_ACTIVITY_BEGAN;
1582                 BT_ASSERT(!stream_state->packet);
1583                 break;
1584         }
1585         case BT_MESSAGE_TYPE_PACKET_BEGINNING:
1586         {
1587                 const struct bt_message_packet *packet_msg =
1588                         (const void *) msg;
1589                 struct auto_seek_stream_state *stream_state;
1590
1591                 /* Update stream's state: packet began. */
1592                 stream_state = g_hash_table_lookup(stream_states, packet_msg->packet->stream);
1593                 BT_ASSERT(stream_state);
1594
1595                 BT_ASSERT(stream_state->state == AUTO_SEEK_STREAM_STATE_STREAM_ACTIVITY_BEGAN);
1596                 stream_state->state = AUTO_SEEK_STREAM_STATE_PACKET_BEGAN;
1597                 BT_ASSERT(!stream_state->packet);
1598                 stream_state->packet = packet_msg->packet;
1599                 break;
1600         }
1601         case BT_MESSAGE_TYPE_PACKET_END:
1602         {
1603                 const struct bt_message_packet *packet_msg =
1604                         (const void *) msg;
1605                 struct auto_seek_stream_state *stream_state;
1606
1607                 /* Update stream's state: packet ended. */
1608                 stream_state = g_hash_table_lookup(stream_states, packet_msg->packet->stream);
1609                 BT_ASSERT(stream_state);
1610
1611                 BT_ASSERT(stream_state->state == AUTO_SEEK_STREAM_STATE_PACKET_BEGAN);
1612                 stream_state->state = AUTO_SEEK_STREAM_STATE_STREAM_ACTIVITY_BEGAN;
1613                 BT_ASSERT(stream_state->packet);
1614                 stream_state->packet = NULL;
1615                 break;
1616         }
1617         case BT_MESSAGE_TYPE_STREAM_ACTIVITY_END:
1618         {
1619                 const struct bt_message_stream_activity *stream_act_msg =
1620                         (const void *) msg;
1621                 struct auto_seek_stream_state *stream_state;
1622
1623                 /* Update stream's state: stream activity ended. */
1624                 stream_state = g_hash_table_lookup(stream_states, stream_act_msg->stream);
1625                 BT_ASSERT(stream_state);
1626
1627                 BT_ASSERT(stream_state->state == AUTO_SEEK_STREAM_STATE_STREAM_ACTIVITY_BEGAN);
1628                 stream_state->state = AUTO_SEEK_STREAM_STATE_STREAM_BEGAN;
1629                 BT_ASSERT(!stream_state->packet);
1630                 break;
1631         }
1632         case BT_MESSAGE_TYPE_STREAM_END:
1633         {
1634                 const struct bt_message_stream *stream_msg = (const void *) msg;
1635                 struct auto_seek_stream_state *stream_state;
1636
1637                 stream_state = g_hash_table_lookup(stream_states, stream_msg->stream);
1638                 BT_ASSERT(stream_state);
1639                 BT_ASSERT(stream_state->state == AUTO_SEEK_STREAM_STATE_STREAM_BEGAN);
1640                 BT_ASSERT(!stream_state->packet);
1641
1642                 /* Update stream's state: this stream doesn't exist anymore. */
1643                 g_hash_table_remove(stream_states, stream_msg->stream);
1644                 break;
1645         }
1646         default:
1647                 break;
1648         }
1649
1650         bt_object_put_no_null_check(msg);
1651         msg = NULL;
1652         goto end;
1653
1654 push_msg:
1655         g_queue_push_tail(iterator->auto_seek.msgs, (void *) msg);
1656         msg = NULL;
1657
1658 end:
1659         BT_ASSERT(!msg || status != BT_FUNC_STATUS_OK);
1660         return status;
1661 }
1662
1663 static
1664 int find_message_ge_ns_from_origin(
1665                 struct bt_self_component_port_input_message_iterator *iterator,
1666                 int64_t ns_from_origin, GHashTable *stream_states)
1667 {
1668         int status;
1669         enum bt_self_component_port_input_message_iterator_state init_state =
1670                 iterator->state;
1671         const struct bt_message *messages[MSG_BATCH_SIZE];
1672         uint64_t user_count = 0;
1673         uint64_t i;
1674         bool got_first = false;
1675
1676         BT_ASSERT(iterator);
1677         memset(&messages[0], 0, sizeof(messages[0]) * MSG_BATCH_SIZE);
1678
1679         /*
1680          * Make this iterator temporarily active (not seeking) to call
1681          * the "next" method.
1682          */
1683         set_self_comp_port_input_msg_iterator_state(iterator,
1684                 BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_ACTIVE);
1685
1686         BT_ASSERT(iterator->methods.next);
1687
1688         while (!got_first) {
1689                 /*
1690                  * Call the user's "next" method to get the next
1691                  * messages and status.
1692                  */
1693                 status = call_iterator_next_method(iterator,
1694                         &messages[0], MSG_BATCH_SIZE, &user_count);
1695                 BT_LOGD("User method returned: status=%s",
1696                         bt_common_func_status_string(status));
1697                 if (status < 0) {
1698                         BT_LIB_LOGW_APPEND_CAUSE(
1699                                 "Component input port message iterator's \"next\" method failed: "
1700                                 "%![iter-]+i, status=%s",
1701                                 iterator, bt_common_func_status_string(status));
1702                 }
1703
1704                 /*
1705                  * The user's "next" method must not do any action which
1706                  * would change the iterator's state.
1707                  */
1708                 BT_ASSERT(iterator->state ==
1709                         BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_ACTIVE);
1710
1711                 switch (status) {
1712                 case BT_FUNC_STATUS_OK:
1713                         BT_ASSERT_POST(user_count <= MSG_BATCH_SIZE,
1714                                 "Invalid returned message count: greater than "
1715                                 "batch size: count=%" PRIu64 ", batch-size=%u",
1716                                 user_count, MSG_BATCH_SIZE);
1717                         break;
1718                 case BT_FUNC_STATUS_AGAIN:
1719                 case BT_FUNC_STATUS_ERROR:
1720                 case BT_FUNC_STATUS_MEMORY_ERROR:
1721                 case BT_FUNC_STATUS_END:
1722                         goto end;
1723                 default:
1724                         abort();
1725                 }
1726
1727                 for (i = 0; i < user_count; i++) {
1728                         if (got_first) {
1729                                 g_queue_push_tail(iterator->auto_seek.msgs,
1730                                         (void *) messages[i]);
1731                                 messages[i] = NULL;
1732                                 continue;
1733                         }
1734
1735                         status = auto_seek_handle_message(iterator,
1736                                 ns_from_origin, messages[i], &got_first,
1737                                 stream_states);
1738                         if (status == BT_FUNC_STATUS_OK) {
1739                                 /* Message was either pushed or moved */
1740                                 messages[i] = NULL;
1741                         } else {
1742                                 goto end;
1743                         }
1744                 }
1745         }
1746
1747 end:
1748         for (i = 0; i < user_count; i++) {
1749                 if (messages[i]) {
1750                         bt_object_put_no_null_check(messages[i]);
1751                 }
1752         }
1753
1754         set_self_comp_port_input_msg_iterator_state(iterator, init_state);
1755         return status;
1756 }
1757
1758 /*
1759  * This function is installed as the iterator's next callback after we have
1760  * auto-seeked (seeked to the beginning and fast-forwarded) to send the
1761  * messages saved in iterator->auto_seek.msgs.  Once this is done, the original
1762  * next callback is put back.
1763  */
1764
1765 static
1766 enum bt_component_class_message_iterator_next_method_status post_auto_seek_next(
1767                 struct bt_self_component_port_input_message_iterator *iterator,
1768                 bt_message_array_const msgs, uint64_t capacity,
1769                 uint64_t *count)
1770 {
1771         BT_ASSERT(!g_queue_is_empty(iterator->auto_seek.msgs));
1772         *count = 0;
1773
1774         /*
1775          * Move auto-seek messages to the output array (which is this
1776          * iterator's base message array).
1777          */
1778         while (capacity > 0 && !g_queue_is_empty(iterator->auto_seek.msgs)) {
1779                 msgs[*count] = g_queue_pop_head(iterator->auto_seek.msgs);
1780                 capacity--;
1781                 (*count)++;
1782         }
1783
1784         BT_ASSERT(*count > 0);
1785
1786         if (g_queue_is_empty(iterator->auto_seek.msgs)) {
1787                 /* No more auto-seek messages, restore user's next callback. */
1788                 BT_ASSERT(iterator->auto_seek.original_next_callback);
1789                 iterator->methods.next = iterator->auto_seek.original_next_callback;
1790                 iterator->auto_seek.original_next_callback = NULL;
1791         }
1792
1793         return BT_FUNC_STATUS_OK;
1794 }
1795
1796 static inline
1797 int clock_raw_value_from_ns_from_origin(const bt_clock_class *clock_class,
1798                 int64_t ns_from_origin, uint64_t *raw_value)
1799 {
1800
1801         int64_t cc_offset_s = clock_class->offset_seconds;
1802         uint64_t cc_offset_cycles = clock_class->offset_cycles;
1803         uint64_t cc_freq = clock_class->frequency;
1804
1805         return bt_common_clock_value_from_ns_from_origin(cc_offset_s,
1806                 cc_offset_cycles, cc_freq, ns_from_origin, raw_value);
1807 }
1808
1809
1810 enum bt_message_iterator_seek_ns_from_origin_status
1811 bt_self_component_port_input_message_iterator_seek_ns_from_origin(
1812                 struct bt_self_component_port_input_message_iterator *iterator,
1813                 int64_t ns_from_origin)
1814 {
1815         int status;
1816         GHashTable *stream_states = NULL;
1817
1818         BT_ASSERT_PRE_NON_NULL(iterator, "Message iterator");
1819         BT_ASSERT_PRE_ITER_HAS_STATE_TO_SEEK(iterator);
1820         BT_ASSERT_PRE(
1821                 bt_component_borrow_graph(iterator->upstream_component)->config_state !=
1822                         BT_GRAPH_CONFIGURATION_STATE_CONFIGURING,
1823                 "Graph is not configured: %!+g",
1824                 bt_component_borrow_graph(iterator->upstream_component));
1825         BT_ASSERT_PRE(
1826                 bt_self_component_port_input_message_iterator_can_seek_ns_from_origin(
1827                         iterator, ns_from_origin),
1828                 "Message iterator cannot seek nanoseconds from origin: %!+i, "
1829                 "ns-from-origin=%" PRId64, iterator, ns_from_origin);
1830         set_self_comp_port_input_msg_iterator_state(iterator,
1831                 BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_SEEKING);
1832
1833         /*
1834          * We are seeking, reset our expectations about how the following
1835          * messages should look like.
1836          */
1837         reset_iterator_expectations(iterator);
1838
1839         if (iterator->methods.seek_ns_from_origin) {
1840                 /* The iterator knows how to seek to a particular time, let it handle this. */
1841                 BT_LIB_LOGD("Calling user's \"seek nanoseconds from origin\" method: "
1842                         "%![iter-]+i, ns=%" PRId64, iterator, ns_from_origin);
1843                 status = iterator->methods.seek_ns_from_origin(iterator,
1844                         ns_from_origin);
1845                 BT_LOGD("User method returned: status=%s",
1846                         bt_common_func_status_string(status));
1847                 BT_ASSERT_POST(status == BT_FUNC_STATUS_OK ||
1848                         status == BT_FUNC_STATUS_ERROR ||
1849                         status == BT_FUNC_STATUS_MEMORY_ERROR ||
1850                         status == BT_FUNC_STATUS_AGAIN,
1851                         "Unexpected status: %![iter-]+i, status=%s",
1852                         iterator, bt_common_func_status_string(status));
1853                 if (status < 0) {
1854                         BT_LIB_LOGW_APPEND_CAUSE(
1855                                 "Component input port message iterator's \"seek nanoseconds from origin\" method failed: "
1856                                 "%![iter-]+i, status=%s",
1857                                 iterator, bt_common_func_status_string(status));
1858                 }
1859         } else {
1860                 /*
1861                  * The iterator doesn't know how to seek to a particular time.  We will
1862                  * seek to the beginning and fast forward to the right place.
1863                  */
1864                 BT_ASSERT(iterator->methods.can_seek_beginning(iterator));
1865                 BT_ASSERT(iterator->methods.seek_beginning);
1866                 BT_LIB_LOGD("Calling user's \"seek beginning\" method: %!+i",
1867                         iterator);
1868                 status = iterator->methods.seek_beginning(iterator);
1869                 BT_LOGD("User method returned: status=%s",
1870                         bt_common_func_status_string(status));
1871                 BT_ASSERT_POST(status == BT_FUNC_STATUS_OK ||
1872                         status == BT_FUNC_STATUS_ERROR ||
1873                         status == BT_FUNC_STATUS_MEMORY_ERROR ||
1874                         status == BT_FUNC_STATUS_AGAIN,
1875                         "Unexpected status: %![iter-]+i, status=%s",
1876                         iterator, bt_common_func_status_string(status));
1877                 if (status < 0) {
1878                         BT_LIB_LOGW_APPEND_CAUSE(
1879                                 "Component input port message iterator's \"seek beginning\" method failed: "
1880                                 "%![iter-]+i, status=%s",
1881                                 iterator, bt_common_func_status_string(status));
1882                 }
1883
1884                 switch (status) {
1885                 case BT_FUNC_STATUS_OK:
1886                         break;
1887                 case BT_FUNC_STATUS_ERROR:
1888                 case BT_FUNC_STATUS_MEMORY_ERROR:
1889                 case BT_FUNC_STATUS_AGAIN:
1890                         goto end;
1891                 default:
1892                         abort();
1893                 }
1894
1895                 /*
1896                  * Find the first message which has a default clock
1897                  * snapshot greater than or equal to the requested
1898                  * seeking time, and move the received messages from
1899                  * this point in the batch to this iterator's auto-seek
1900                  * message queue.
1901                  */
1902                 while (!g_queue_is_empty(iterator->auto_seek.msgs)) {
1903                         bt_object_put_no_null_check(
1904                                 g_queue_pop_tail(iterator->auto_seek.msgs));
1905                 }
1906
1907                 stream_states = create_auto_seek_stream_states();
1908                 if (!stream_states) {
1909                         BT_LIB_LOGE_APPEND_CAUSE(
1910                                 "Failed to allocate one GHashTable.");
1911                         status = BT_FUNC_STATUS_MEMORY_ERROR;
1912                         goto end;
1913                 }
1914
1915                 status = find_message_ge_ns_from_origin(iterator,
1916                         ns_from_origin, stream_states);
1917                 switch (status) {
1918                 case BT_FUNC_STATUS_OK:
1919                 case BT_FUNC_STATUS_END:
1920                 {
1921                         GHashTableIter iter;
1922                         gpointer key, value;
1923
1924                         /*
1925                          * If some streams exist at the seek time, prepend the
1926                          * required messages to put those streams in the right
1927                          * state.
1928                          */
1929                         g_hash_table_iter_init(&iter, stream_states);
1930                         while (g_hash_table_iter_next (&iter, &key, &value)) {
1931                                 const bt_stream *stream = key;
1932                                 struct auto_seek_stream_state *stream_state =
1933                                         (struct auto_seek_stream_state *) value;
1934                                 bt_message *msg;
1935                                 const bt_clock_class *clock_class = bt_stream_class_borrow_default_clock_class_const(
1936                                         bt_stream_borrow_class_const(stream));
1937                                 uint64_t raw_value;
1938
1939                                 if (clock_raw_value_from_ns_from_origin(clock_class, ns_from_origin, &raw_value) != 0) {
1940                                         BT_LIB_LOGW_APPEND_CAUSE(
1941                                                 "Could not convert nanoseconds from origin to clock value: "
1942                                                 "ns-from-origin=%" PRId64 ", %![cc-]+K",
1943                                                 ns_from_origin, clock_class);
1944                                         status = BT_FUNC_STATUS_ERROR;
1945                                         goto end;
1946                                 }
1947
1948                                 switch (stream_state->state) {
1949                                 case AUTO_SEEK_STREAM_STATE_PACKET_BEGAN:
1950                                         BT_ASSERT(stream_state->packet);
1951                                         BT_LIB_LOGD("Creating packet message: %![packet-]+a", stream_state->packet);
1952                                         msg = bt_message_packet_beginning_create_with_default_clock_snapshot(
1953                                                 (bt_self_message_iterator *) iterator, stream_state->packet, raw_value);
1954                                         if (!msg) {
1955                                                 status = BT_FUNC_STATUS_MEMORY_ERROR;
1956                                                 goto end;
1957                                         }
1958
1959                                         g_queue_push_head(iterator->auto_seek.msgs, msg);
1960                                         msg = NULL;
1961                                         /* fall-thru */
1962                                 case AUTO_SEEK_STREAM_STATE_STREAM_ACTIVITY_BEGAN:
1963                                         msg = bt_message_stream_activity_beginning_create(
1964                                                 (bt_self_message_iterator *) iterator, stream);
1965                                         if (!msg) {
1966                                                 status = BT_FUNC_STATUS_MEMORY_ERROR;
1967                                                 goto end;
1968                                         }
1969
1970                                         bt_message_stream_activity_beginning_set_default_clock_snapshot(msg, raw_value);
1971
1972                                         g_queue_push_head(iterator->auto_seek.msgs, msg);
1973                                         msg = NULL;
1974                                         /* fall-thru */
1975                                 case AUTO_SEEK_STREAM_STATE_STREAM_BEGAN:
1976                                         msg = bt_message_stream_beginning_create(
1977                                                 (bt_self_message_iterator *) iterator, stream);
1978                                         if (!msg) {
1979                                                 status = BT_FUNC_STATUS_MEMORY_ERROR;
1980                                                 goto end;
1981                                         }
1982
1983                                         g_queue_push_head(iterator->auto_seek.msgs, msg);
1984                                         msg = NULL;
1985                                         break;
1986                                 }
1987                         }
1988
1989                         /*
1990                          * If there are messages in the auto-seek
1991                          * message queue, replace the user's "next"
1992                          * method with a custom, temporary "next" method
1993                          * which returns them.
1994                          */
1995                         if (!g_queue_is_empty(iterator->auto_seek.msgs)) {
1996                                 BT_ASSERT(!iterator->auto_seek.original_next_callback);
1997                                 iterator->auto_seek.original_next_callback = iterator->methods.next;
1998
1999                                 iterator->methods.next =
2000                                         (bt_self_component_port_input_message_iterator_next_method)
2001                                                 post_auto_seek_next;
2002                         }
2003
2004                         /*
2005                          * `BT_FUNC_STATUS_END` becomes
2006                          * `BT_FUNC_STATUS_OK`: the next
2007                          * time this iterator's "next" method is called,
2008                          * it will return
2009                          * `BT_FUNC_STATUS_END`.
2010                          */
2011                         status = BT_FUNC_STATUS_OK;
2012                         break;
2013                 }
2014                 case BT_FUNC_STATUS_ERROR:
2015                 case BT_FUNC_STATUS_MEMORY_ERROR:
2016                 case BT_FUNC_STATUS_AGAIN:
2017                         goto end;
2018                 default:
2019                         abort();
2020                 }
2021         }
2022
2023         /*
2024          * The following messages returned by the next method (including
2025          * post_auto_seek_next) must be after (or at) `ns_from_origin`.
2026          */
2027         iterator->last_ns_from_origin = ns_from_origin;
2028
2029 end:
2030         if (stream_states) {
2031                 destroy_auto_seek_stream_states(stream_states);
2032                 stream_states = NULL;
2033         }
2034
2035         set_iterator_state_after_seeking(iterator, status);
2036         return status;
2037 }
2038
2039 static inline
2040 bt_self_component_port_input_message_iterator *
2041 borrow_output_port_message_iterator_upstream_iterator(
2042                 struct bt_port_output_message_iterator *iterator)
2043 {
2044         struct bt_component_class_sink_colander_priv_data *colander_data;
2045
2046         BT_ASSERT(iterator);
2047         colander_data = (void *) iterator->colander->parent.user_data;
2048         BT_ASSERT(colander_data);
2049         BT_ASSERT(colander_data->msg_iter);
2050         return colander_data->msg_iter;
2051 }
2052
2053 bt_bool bt_port_output_message_iterator_can_seek_ns_from_origin(
2054                 struct bt_port_output_message_iterator *iterator,
2055                 int64_t ns_from_origin)
2056 {
2057         BT_ASSERT_PRE_NON_NULL(iterator, "Message iterator");
2058         return bt_self_component_port_input_message_iterator_can_seek_ns_from_origin(
2059                 borrow_output_port_message_iterator_upstream_iterator(
2060                         iterator), ns_from_origin);
2061 }
2062
2063 bt_bool bt_port_output_message_iterator_can_seek_beginning(
2064                 struct bt_port_output_message_iterator *iterator)
2065 {
2066         BT_ASSERT_PRE_NON_NULL(iterator, "Message iterator");
2067         return bt_self_component_port_input_message_iterator_can_seek_beginning(
2068                 borrow_output_port_message_iterator_upstream_iterator(
2069                         iterator));
2070 }
2071
2072 enum bt_message_iterator_seek_ns_from_origin_status
2073 bt_port_output_message_iterator_seek_ns_from_origin(
2074                 struct bt_port_output_message_iterator *iterator,
2075                 int64_t ns_from_origin)
2076 {
2077         BT_ASSERT_PRE_NON_NULL(iterator, "Message iterator");
2078         return bt_self_component_port_input_message_iterator_seek_ns_from_origin(
2079                 borrow_output_port_message_iterator_upstream_iterator(iterator),
2080                 ns_from_origin);
2081 }
2082
2083 enum bt_message_iterator_seek_beginning_status
2084 bt_port_output_message_iterator_seek_beginning(
2085                 struct bt_port_output_message_iterator *iterator)
2086 {
2087         BT_ASSERT_PRE_NON_NULL(iterator, "Message iterator");
2088         return bt_self_component_port_input_message_iterator_seek_beginning(
2089                 borrow_output_port_message_iterator_upstream_iterator(
2090                         iterator));
2091 }
2092
2093 void bt_port_output_message_iterator_get_ref(
2094                 const struct bt_port_output_message_iterator *iterator)
2095 {
2096         bt_object_get_ref(iterator);
2097 }
2098
2099 void bt_port_output_message_iterator_put_ref(
2100                 const struct bt_port_output_message_iterator *iterator)
2101 {
2102         bt_object_put_ref(iterator);
2103 }
2104
2105 void bt_self_component_port_input_message_iterator_get_ref(
2106                 const struct bt_self_component_port_input_message_iterator *iterator)
2107 {
2108         bt_object_get_ref(iterator);
2109 }
2110
2111 void bt_self_component_port_input_message_iterator_put_ref(
2112                 const struct bt_self_component_port_input_message_iterator *iterator)
2113 {
2114         bt_object_put_ref(iterator);
2115 }