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