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