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                 BT_ASSERT_POST_NO_ERROR_IF_NO_ERROR_STATUS(iter_status);
 481                 if (iter_status != BT_FUNC_STATUS_OK) {
 482                         BT_LIB_LOGW_APPEND_CAUSE(
 483                                 "Component input port message iterator initialization method failed: "
 484                                 "%![iter-]+i, status=%s",
 485                                 iterator,
 486                                 bt_common_func_status_string(iter_status));
 487                         status = iter_status;
 488                         goto error;
 489                 }
 490
 491                 iterator->config.frozen = true;
 492         }
 493
 494         if (downstream_msg_iter) {
 495                 /* Set this message iterator's downstream message iterator */
 496                 iterator->downstream_msg_iter = downstream_msg_iter;
 497
 498                 /*
 499                  * Add this message iterator to the downstream message
 500                  * iterator's array of upstream message iterators.
 501                  */
 502                 g_ptr_array_add(downstream_msg_iter->upstream_msg_iters,
 503                         iterator);
 504         }
 505
 506         set_self_comp_port_input_msg_iterator_state(iterator,
 507                 BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_ACTIVE);
 508         g_ptr_array_add(port->connection->iterators, iterator);
 509         BT_LIB_LOGI("Created message iterator on self component input port: "
 510                 "%![up-port-]+p, %![up-comp-]+c, %![iter-]+i",
 511                 upstream_port, upstream_comp, iterator);
 512
 513         *message_iterator = iterator;
 514         status = BT_FUNC_STATUS_OK;
 515         goto end;
 516
 517 error:
 518         BT_OBJECT_PUT_REF_AND_RESET(iterator);
 519
 520 end:
 521         return status;
 522 }
 523
 524 bt_self_component_port_input_message_iterator_create_from_message_iterator_status
 525 bt_self_component_port_input_message_iterator_create_from_message_iterator(
 526                 struct bt_self_message_iterator *self_msg_iter,
 527                 struct bt_self_component_port_input *input_port,
 528                 struct bt_self_component_port_input_message_iterator **message_iterator)
 529 {
 530         BT_ASSERT_PRE_NON_NULL(self_msg_iter, "Message iterator");
 531         return create_self_component_input_port_message_iterator(self_msg_iter,
 532                 input_port, message_iterator);
 533 }
 534
 535 bt_self_component_port_input_message_iterator_create_from_sink_component_status
 536 bt_self_component_port_input_message_iterator_create_from_sink_component(
 537                 struct bt_self_component_sink *self_comp,
 538                 struct bt_self_component_port_input *input_port,
 539                 struct bt_self_component_port_input_message_iterator **message_iterator)
 540 {
 541         BT_ASSERT_PRE_NON_NULL(self_comp, "Sink component");
 542         return create_self_component_input_port_message_iterator(NULL,
 543                 input_port, message_iterator);
 544 }
 545
 546 void *bt_self_message_iterator_get_data(
 547                 const struct bt_self_message_iterator *self_iterator)
 548 {
 549         struct bt_self_component_port_input_message_iterator *iterator =
 550                 (void *) self_iterator;
 551
 552         BT_ASSERT_PRE_DEV_NON_NULL(iterator, "Message iterator");
 553         return iterator->user_data;
 554 }
 555
 556 void bt_self_message_iterator_set_data(
 557                 struct bt_self_message_iterator *self_iterator, void *data)
 558 {
 559         struct bt_self_component_port_input_message_iterator *iterator =
 560                 (void *) self_iterator;
 561
 562         BT_ASSERT_PRE_DEV_NON_NULL(iterator, "Message iterator");
 563         iterator->user_data = data;
 564         BT_LIB_LOGD("Set message iterator's user data: "
 565                 "%!+i, user-data-addr=%p", iterator, data);
 566 }
 567
 568 void bt_self_message_iterator_configuration_set_can_seek_forward(
 569                 bt_self_message_iterator_configuration *config,
 570                 bt_bool can_seek_forward)
 571 {
 572         BT_ASSERT_PRE_NON_NULL(config, "Message iterator configuration");
 573         BT_ASSERT_PRE_DEV_HOT(config, "Message iterator configuration", "");
 574
 575         config->can_seek_forward = can_seek_forward;
 576 }
 577
 578 /*
 579  * Validate that the default clock snapshot in `msg` doesn't make us go back in
 580  * time.
 581  */
 582
 583 BT_ASSERT_POST_DEV_FUNC
 584 static
 585 bool clock_snapshots_are_monotonic_one(
 586                 struct bt_self_component_port_input_message_iterator *iterator,
 587                 const bt_message *msg)
 588 {
 589         const struct bt_clock_snapshot *clock_snapshot = NULL;
 590         bt_message_type message_type = bt_message_get_type(msg);
 591         int64_t ns_from_origin;
 592         enum bt_clock_snapshot_get_ns_from_origin_status clock_snapshot_status;
 593
 594         /*
 595          * The default is true: if we can't figure out the clock snapshot
 596          * (or there is none), assume it is fine.
 597          */
 598         bool result = true;
 599
 600         switch (message_type) {
 601         case BT_MESSAGE_TYPE_EVENT:
 602         {
 603                 struct bt_message_event *event_msg = (struct bt_message_event *) msg;
 604                 clock_snapshot = event_msg->default_cs;
 605                 break;
 606         }
 607         case BT_MESSAGE_TYPE_MESSAGE_ITERATOR_INACTIVITY:
 608         {
 609                 struct bt_message_message_iterator_inactivity *inactivity_msg =
 610                         (struct bt_message_message_iterator_inactivity *) msg;
 611                 clock_snapshot = inactivity_msg->default_cs;
 612                 break;
 613         }
 614         case BT_MESSAGE_TYPE_PACKET_BEGINNING:
 615         case BT_MESSAGE_TYPE_PACKET_END:
 616         {
 617                 struct bt_message_packet *packet_msg = (struct bt_message_packet *) msg;
 618                 clock_snapshot = packet_msg->default_cs;
 619                 break;
 620         }
 621         case BT_MESSAGE_TYPE_STREAM_BEGINNING:
 622         case BT_MESSAGE_TYPE_STREAM_END:
 623         {
 624                 struct bt_message_stream *stream_msg = (struct bt_message_stream *) msg;
 625                 if (stream_msg->default_cs_state != BT_MESSAGE_STREAM_CLOCK_SNAPSHOT_STATE_KNOWN) {
 626                         goto end;
 627                 }
 628
 629                 clock_snapshot = stream_msg->default_cs;
 630                 break;
 631         }
 632         case BT_MESSAGE_TYPE_DISCARDED_EVENTS:
 633         case BT_MESSAGE_TYPE_DISCARDED_PACKETS:
 634         {
 635                 struct bt_message_discarded_items *discarded_msg =
 636                         (struct bt_message_discarded_items *) msg;
 637
 638                 clock_snapshot = discarded_msg->default_begin_cs;
 639                 break;
 640         }
 641         }
 642
 643         if (!clock_snapshot) {
 644                 goto end;
 645         }
 646
 647         clock_snapshot_status = bt_clock_snapshot_get_ns_from_origin(
 648                 clock_snapshot, &ns_from_origin);
 649         if (clock_snapshot_status != BT_FUNC_STATUS_OK) {
 650                 /*
 651                  * bt_clock_snapshot_get_ns_from_origin can return
 652                  * OVERFLOW_ERROR.  We don't really want to report an error to
 653                  * our caller, so just clear it.
 654                  */
 655                 bt_current_thread_clear_error();
 656                 goto end;
 657         }
 658
 659         result = ns_from_origin >= iterator->last_ns_from_origin;
 660         iterator->last_ns_from_origin = ns_from_origin;
 661 end:
 662         return result;
 663 }
 664
 665 BT_ASSERT_POST_DEV_FUNC
 666 static
 667 bool clock_snapshots_are_monotonic(
 668                 struct bt_self_component_port_input_message_iterator *iterator,
 669                 bt_message_array_const msgs, uint64_t msg_count)
 670 {
 671         uint64_t i;
 672         bool result;
 673
 674         for (i = 0; i < msg_count; i++) {
 675                 if (!clock_snapshots_are_monotonic_one(iterator, msgs[i])) {
 676                         result = false;
 677                         goto end;
 678                 }
 679         }
 680
 681         result = true;
 682
 683 end:
 684         return result;
 685 }
 686
 687 /*
 688  * When a new stream begins, verify that the clock class tied to this
 689  * stream is compatible with what we've seen before.
 690  */
 691
 692 BT_ASSERT_POST_DEV_FUNC
 693 static
 694 bool clock_classes_are_compatible_one(struct bt_self_component_port_input_message_iterator *iterator,
 695                 const struct bt_message *msg)
 696 {
 697         enum bt_message_type message_type = bt_message_get_type(msg);
 698         bool result;
 699
 700         if (message_type == BT_MESSAGE_TYPE_STREAM_BEGINNING) {
 701                 const struct bt_message_stream *stream_msg = (struct bt_message_stream *) msg;
 702                 const struct bt_clock_class *clock_class = stream_msg->stream->class->default_clock_class;
 703                 bt_uuid clock_class_uuid = NULL;
 704
 705                 if (clock_class) {
 706                         clock_class_uuid = bt_clock_class_get_uuid(clock_class);
 707                 }
 708
 709                 switch (iterator->clock_expectation.type) {
 710                 case CLOCK_EXPECTATION_UNSET:
 711                         /*
 712                          * This is the first time we see a message with a clock
 713                          * snapshot: record the properties of that clock, against
 714                          * which we'll compare the clock properties of the following
 715                          * messages.
 716                          */
 717
 718                         if (!clock_class) {
 719                                 iterator->clock_expectation.type = CLOCK_EXPECTATION_NONE;
 720                         } else if (bt_clock_class_origin_is_unix_epoch(clock_class)) {
 721                                 iterator->clock_expectation.type = CLOCK_EXPECTATION_ORIGIN_UNIX;
 722                         } else if (clock_class_uuid) {
 723                                 iterator->clock_expectation.type = CLOCK_EXPECTATION_ORIGIN_OTHER_UUID;
 724                                 bt_uuid_copy(iterator->clock_expectation.uuid, clock_class_uuid);
 725                         } else {
 726                                 iterator->clock_expectation.type = CLOCK_EXPECTATION_ORIGIN_OTHER_NO_UUID;
 727                         }
 728                         break;
 729
 730                 case CLOCK_EXPECTATION_NONE:
 731                         if (clock_class) {
 732                                 BT_ASSERT_POST_DEV_MSG(
 733                                         "Expecting no clock class, got one: %![cc-]+K",
 734                                         clock_class);
 735                                 result = false;
 736                                 goto end;
 737                         }
 738
 739                         break;
 740
 741                 case CLOCK_EXPECTATION_ORIGIN_UNIX:
 742                         if (!clock_class) {
 743                                 BT_ASSERT_POST_DEV_MSG(
 744                                         "Expecting a clock class, got none.");
 745                                 result = false;
 746                                 goto end;
 747                         }
 748
 749                         if (!bt_clock_class_origin_is_unix_epoch(clock_class)) {
 750                                 BT_ASSERT_POST_DEV_MSG(
 751                                         "Expecting a clock class with Unix epoch origin: %![cc-]+K",
 752                                         clock_class);
 753                                 result = false;
 754                                 goto end;
 755                         }
 756                         break;
 757
 758                 case CLOCK_EXPECTATION_ORIGIN_OTHER_UUID:
 759                         if (!clock_class) {
 760                                 BT_ASSERT_POST_DEV_MSG(
 761                                         "Expecting a clock class, got none.");
 762                                 result = false;
 763                                 goto end;
 764                         }
 765
 766                         if (bt_clock_class_origin_is_unix_epoch(clock_class)) {
 767                                 BT_ASSERT_POST_DEV_MSG(
 768                                         "Expecting a clock class without Unix epoch origin: %![cc-]+K",
 769                                         clock_class);
 770                                 result = false;
 771                                 goto end;
 772                         }
 773
 774                         if (!clock_class_uuid) {
 775                                 BT_ASSERT_POST_DEV_MSG(
 776                                         "Expecting a clock class with UUID: %![cc-]+K",
 777                                         clock_class);
 778                                 result = false;
 779                                 goto end;
 780                         }
 781
 782                         if (bt_uuid_compare(iterator->clock_expectation.uuid, clock_class_uuid)) {
 783                                 BT_ASSERT_POST_DEV_MSG(
 784                                         "Expecting a clock class with UUID, got one "
 785                                         "with a different UUID: %![cc-]+K, expected-uuid=%!u",
 786                                         clock_class, iterator->clock_expectation.uuid);
 787                                 result = false;
 788                                 goto end;
 789                         }
 790                         break;
 791
 792                 case CLOCK_EXPECTATION_ORIGIN_OTHER_NO_UUID:
 793                         if (!clock_class) {
 794                                 BT_ASSERT_POST_DEV_MSG(
 795                                         "Expecting a clock class, got none.");
 796                                 result = false;
 797                                 goto end;
 798                         }
 799
 800                         if (bt_clock_class_origin_is_unix_epoch(clock_class)) {
 801                                 BT_ASSERT_POST_DEV_MSG(
 802                                         "Expecting a clock class without Unix epoch origin: %![cc-]+K",
 803                                         clock_class);
 804                                 result = false;
 805                                 goto end;
 806                         }
 807
 808                         if (clock_class_uuid) {
 809                                 BT_ASSERT_POST_DEV_MSG(
 810                                         "Expecting a clock class without UUID: %![cc-]+K",
 811                                         clock_class);
 812                                 result = false;
 813                                 goto end;
 814                         }
 815                         break;
 816                 }
 817         }
 818
 819         result = true;
 820
 821 end:
 822         return result;
 823 }
 824
 825 BT_ASSERT_POST_DEV_FUNC
 826 static
 827 bool clock_classes_are_compatible(
 828                 struct bt_self_component_port_input_message_iterator *iterator,
 829                 bt_message_array_const msgs, uint64_t msg_count)
 830 {
 831         uint64_t i;
 832         bool result;
 833
 834         for (i = 0; i < msg_count; i++) {
 835                 if (!clock_classes_are_compatible_one(iterator, msgs[i])) {
 836                         result = false;
 837                         goto end;
 838                 }
 839         }
 840
 841         result = true;
 842
 843 end:
 844         return result;
 845 }
 846
 847 /*
 848  * Call the `next` method of the iterator.  Do some validation on the returned
 849  * messages.
 850  */
 851
 852 static
 853 enum bt_component_class_message_iterator_next_method_status
 854 call_iterator_next_method(
 855                 struct bt_self_component_port_input_message_iterator *iterator,
 856                 bt_message_array_const msgs, uint64_t capacity, uint64_t *user_count)
 857 {
 858         enum bt_component_class_message_iterator_next_method_status status;
 859
 860         BT_ASSERT_DBG(iterator->methods.next);
 861         BT_LOGD_STR("Calling user's \"next\" method.");
 862         status = iterator->methods.next(iterator, msgs, capacity, user_count);
 863         BT_LOGD("User method returned: status=%s, msg-count=%" PRIu64,
 864                 bt_common_func_status_string(status), *user_count);
 865
 866         if (status == BT_FUNC_STATUS_OK) {
 867                 BT_ASSERT_POST_DEV(clock_classes_are_compatible(iterator, msgs, *user_count),
 868                         "Clocks are not compatible");
 869                 BT_ASSERT_POST_DEV(clock_snapshots_are_monotonic(iterator, msgs, *user_count),
 870                         "Clock snapshots are not monotonic");
 871         }
 872
 873         BT_ASSERT_POST_DEV_NO_ERROR_IF_NO_ERROR_STATUS(status);
 874
 875         return status;
 876 }
 877
 878 enum bt_message_iterator_next_status
 879 bt_self_component_port_input_message_iterator_next(
 880                 struct bt_self_component_port_input_message_iterator *iterator,
 881                 bt_message_array_const *msgs, uint64_t *user_count)
 882 {
 883         enum bt_message_iterator_next_status status = BT_FUNC_STATUS_OK;
 884
 885         BT_ASSERT_PRE_DEV_NON_NULL(iterator, "Message iterator");
 886         BT_ASSERT_PRE_DEV_NON_NULL(msgs, "Message array (output)");
 887         BT_ASSERT_PRE_DEV_NON_NULL(user_count, "Message count (output)");
 888         BT_ASSERT_PRE_DEV(iterator->state ==
 889                 BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_ACTIVE,
 890                 "Message iterator's \"next\" called, but "
 891                 "message iterator is in the wrong state: %!+i", iterator);
 892         BT_ASSERT_DBG(iterator->upstream_component);
 893         BT_ASSERT_DBG(iterator->upstream_component->class);
 894         BT_ASSERT_PRE_DEV(
 895                 bt_component_borrow_graph(iterator->upstream_component)->config_state !=
 896                         BT_GRAPH_CONFIGURATION_STATE_CONFIGURING,
 897                 "Graph is not configured: %!+g",
 898                 bt_component_borrow_graph(iterator->upstream_component));
 899         BT_LIB_LOGD("Getting next self component input port "
 900                 "message iterator's messages: %!+i, batch-size=%u",
 901                 iterator, MSG_BATCH_SIZE);
 902
 903         /*
 904          * Call the user's "next" method to get the next messages
 905          * and status.
 906          */
 907         *user_count = 0;
 908         status = (int) call_iterator_next_method(iterator,
 909                 (void *) iterator->msgs->pdata, MSG_BATCH_SIZE,
 910                 user_count);
 911         BT_LOGD("User method returned: status=%s, msg-count=%" PRIu64,
 912                 bt_common_func_status_string(status), *user_count);
 913         if (status < 0) {
 914                 BT_LIB_LOGW_APPEND_CAUSE(
 915                         "Component input port message iterator's \"next\" method failed: "
 916                         "%![iter-]+i, status=%s",
 917                         iterator, bt_common_func_status_string(status));
 918                 goto end;
 919         }
 920
 921         /*
 922          * There is no way that this iterator could have been finalized
 923          * during its "next" method, as the only way to do this is to
 924          * put the last iterator's reference, and this can only be done
 925          * by its downstream owner.
 926          *
 927          * For the same reason, there is no way that this iterator could
 928          * have seeked (cannot seek a self message iterator).
 929          */
 930         BT_ASSERT_DBG(iterator->state ==
 931                 BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_ACTIVE);
 932
 933         switch (status) {
 934         case BT_FUNC_STATUS_OK:
 935                 BT_ASSERT_POST_DEV(*user_count <= MSG_BATCH_SIZE,
 936                         "Invalid returned message count: greater than "
 937                         "batch size: count=%" PRIu64 ", batch-size=%u",
 938                         *user_count, MSG_BATCH_SIZE);
 939                 *msgs = (void *) iterator->msgs->pdata;
 940                 break;
 941         case BT_FUNC_STATUS_AGAIN:
 942                 goto end;
 943         case BT_FUNC_STATUS_END:
 944                 set_self_comp_port_input_msg_iterator_state(iterator,
 945                         BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_ENDED);
 946                 goto end;
 947         default:
 948                 /* Unknown non-error status */
 949                 bt_common_abort();
 950         }
 951
 952 end:
 953         return status;
 954 }
 955
 956 struct bt_component *
 957 bt_self_component_port_input_message_iterator_borrow_component(
 958                 struct bt_self_component_port_input_message_iterator *iterator)
 959 {
 960         BT_ASSERT_PRE_DEV_NON_NULL(iterator, "Message iterator");
 961         return iterator->upstream_component;
 962 }
 963
 964 struct bt_self_component *bt_self_message_iterator_borrow_component(
 965                 struct bt_self_message_iterator *self_iterator)
 966 {
 967         struct bt_self_component_port_input_message_iterator *iterator =
 968                 (void *) self_iterator;
 969
 970         BT_ASSERT_PRE_DEV_NON_NULL(iterator, "Message iterator");
 971         return (void *) iterator->upstream_component;
 972 }
 973
 974 struct bt_self_component_port_output *bt_self_message_iterator_borrow_port(
 975                 struct bt_self_message_iterator *self_iterator)
 976 {
 977         struct bt_self_component_port_input_message_iterator *iterator =
 978                 (void *) self_iterator;
 979
 980         BT_ASSERT_PRE_DEV_NON_NULL(iterator, "Message iterator");
 981         return (void *) iterator->upstream_port;
 982 }
 983
 984 enum bt_message_iterator_can_seek_ns_from_origin_status
 985 bt_self_component_port_input_message_iterator_can_seek_ns_from_origin(
 986                 struct bt_self_component_port_input_message_iterator *iterator,
 987                 int64_t ns_from_origin, bt_bool *can_seek)
 988 {
 989         enum bt_message_iterator_can_seek_ns_from_origin_status status;
 990
 991         BT_ASSERT_PRE_NON_NULL(iterator, "Message iterator");
 992         BT_ASSERT_PRE_NON_NULL(can_seek, "Result (output)");
 993         BT_ASSERT_PRE_ITER_HAS_STATE_TO_SEEK(iterator);
 994         BT_ASSERT_PRE(
 995                 bt_component_borrow_graph(iterator->upstream_component)->config_state !=
 996                         BT_GRAPH_CONFIGURATION_STATE_CONFIGURING,
 997                 "Graph is not configured: %!+g",
 998                 bt_component_borrow_graph(iterator->upstream_component));
 999
1000         if (iterator->methods.can_seek_ns_from_origin) {
1001                 /*
1002                  * Initialize to an invalid value, so we can post-assert that
1003                  * the method returned a valid value.
1004                  */
1005                 *can_seek = -1;
1006
1007                 BT_LIB_LOGD("Calling user's \"can seek nanoseconds from origin\" method: %!+i",
1008                         iterator);
1009
1010                 status = (int) iterator->methods.can_seek_ns_from_origin(iterator,
1011                         ns_from_origin, can_seek);
1012
1013                 BT_ASSERT_POST_NO_ERROR_IF_NO_ERROR_STATUS(status);
1014
1015                 if (status != BT_FUNC_STATUS_OK) {
1016                         BT_LIB_LOGW_APPEND_CAUSE(
1017                                 "Component input port message iterator's \"can seek nanoseconds from origin\" method failed: "
1018                                 "%![iter-]+i, status=%s",
1019                                 iterator, bt_common_func_status_string(status));
1020                         goto end;
1021                 }
1022
1023                 BT_ASSERT_POST(*can_seek == BT_TRUE || *can_seek == BT_FALSE,
1024                         "Unexpected boolean value returned from user's \"can seek ns from origin\" method: val=%d, %![iter-]+i",
1025                         *can_seek, iterator);
1026
1027                 BT_LIB_LOGD(
1028                         "User's \"can seek nanoseconds from origin\" returned successfully: "
1029                         "%![iter-]+i, can-seek=%d",
1030                         iterator, *can_seek);
1031
1032                 if (*can_seek) {
1033                         goto end;
1034                 }
1035         }
1036
1037         /*
1038          * Automatic seeking fall back: if we can seek to the beginning and the
1039          * iterator supports forward seeking then we can automatically seek to
1040          * any timestamp.
1041          */
1042         status = (int) bt_self_component_port_input_message_iterator_can_seek_beginning(
1043                 iterator, can_seek);
1044         if (status != BT_FUNC_STATUS_OK) {
1045                 goto end;
1046         }
1047
1048         *can_seek = *can_seek && iterator->config.can_seek_forward;
1049
1050 end:
1051         return status;
1052 }
1053
1054 enum bt_message_iterator_can_seek_beginning_status
1055 bt_self_component_port_input_message_iterator_can_seek_beginning(
1056                 struct bt_self_component_port_input_message_iterator *iterator,
1057                 bt_bool *can_seek)
1058 {
1059         enum bt_message_iterator_can_seek_beginning_status status;
1060
1061         BT_ASSERT_PRE_NON_NULL(iterator, "Message iterator");
1062         BT_ASSERT_PRE_NON_NULL(can_seek, "Result (output)");
1063         BT_ASSERT_PRE_ITER_HAS_STATE_TO_SEEK(iterator);
1064         BT_ASSERT_PRE(
1065                 bt_component_borrow_graph(iterator->upstream_component)->config_state !=
1066                         BT_GRAPH_CONFIGURATION_STATE_CONFIGURING,
1067                 "Graph is not configured: %!+g",
1068                 bt_component_borrow_graph(iterator->upstream_component));
1069
1070         if (iterator->methods.can_seek_beginning) {
1071                 /*
1072                  * Initialize to an invalid value, so we can post-assert that
1073                  * the method returned a valid value.
1074                  */
1075                 *can_seek = -1;
1076
1077                 status = (int) iterator->methods.can_seek_beginning(iterator, can_seek);
1078
1079                 BT_ASSERT_POST(
1080                         status != BT_FUNC_STATUS_OK ||
1081                                 *can_seek == BT_TRUE ||
1082                                 *can_seek == BT_FALSE,
1083                         "Unexpected boolean value returned from user's \"can seek beginning\" method: val=%d, %![iter-]+i",
1084                         *can_seek, iterator);
1085                 BT_ASSERT_POST_NO_ERROR_IF_NO_ERROR_STATUS(status);
1086         } else {
1087                 *can_seek = BT_FALSE;
1088                 status = BT_FUNC_STATUS_OK;
1089         }
1090
1091         return status;
1092 }
1093
1094 static inline
1095 void set_iterator_state_after_seeking(
1096                 struct bt_self_component_port_input_message_iterator *iterator,
1097                 int status)
1098 {
1099         enum bt_self_component_port_input_message_iterator_state new_state = 0;
1100
1101         /* Set iterator's state depending on seeking status */
1102         switch (status) {
1103         case BT_FUNC_STATUS_OK:
1104                 new_state = BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_ACTIVE;
1105                 break;
1106         case BT_FUNC_STATUS_AGAIN:
1107                 new_state = BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_LAST_SEEKING_RETURNED_AGAIN;
1108                 break;
1109         case BT_FUNC_STATUS_ERROR:
1110         case BT_FUNC_STATUS_MEMORY_ERROR:
1111                 new_state = BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_LAST_SEEKING_RETURNED_ERROR;
1112                 break;
1113         case BT_FUNC_STATUS_END:
1114                 new_state = BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_ENDED;
1115                 break;
1116         default:
1117                 bt_common_abort();
1118         }
1119
1120         set_self_comp_port_input_msg_iterator_state(iterator, new_state);
1121 }
1122
1123 static
1124 void reset_iterator_expectations(
1125                 struct bt_self_component_port_input_message_iterator *iterator)
1126 {
1127         iterator->last_ns_from_origin = INT64_MIN;
1128         iterator->clock_expectation.type = CLOCK_EXPECTATION_UNSET;
1129 }
1130
1131 static
1132 bool message_iterator_can_seek_beginning(
1133                 struct bt_self_component_port_input_message_iterator *iterator)
1134 {
1135         enum bt_message_iterator_can_seek_beginning_status status;
1136         bt_bool can_seek;
1137
1138         status = bt_self_component_port_input_message_iterator_can_seek_beginning(
1139                 iterator, &can_seek);
1140         if (status != BT_FUNC_STATUS_OK) {
1141                 can_seek = BT_FALSE;
1142         }
1143
1144         return can_seek;
1145 }
1146
1147 enum bt_message_iterator_seek_beginning_status
1148 bt_self_component_port_input_message_iterator_seek_beginning(
1149                 struct bt_self_component_port_input_message_iterator *iterator)
1150 {
1151         int status;
1152
1153         BT_ASSERT_PRE_NON_NULL(iterator, "Message iterator");
1154         BT_ASSERT_PRE_ITER_HAS_STATE_TO_SEEK(iterator);
1155         BT_ASSERT_PRE(
1156                 bt_component_borrow_graph(iterator->upstream_component)->config_state !=
1157                         BT_GRAPH_CONFIGURATION_STATE_CONFIGURING,
1158                 "Graph is not configured: %!+g",
1159                 bt_component_borrow_graph(iterator->upstream_component));
1160         BT_ASSERT_PRE(message_iterator_can_seek_beginning(iterator),
1161                 "Message iterator cannot seek beginning: %!+i", iterator);
1162
1163         /*
1164          * We are seeking, reset our expectations about how the following
1165          * messages should look like.
1166          */
1167         reset_iterator_expectations(iterator);
1168
1169         BT_LIB_LOGD("Calling user's \"seek beginning\" method: %!+i", iterator);
1170         set_self_comp_port_input_msg_iterator_state(iterator,
1171                 BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_SEEKING);
1172         status = iterator->methods.seek_beginning(iterator);
1173         BT_LOGD("User method returned: status=%s",
1174                 bt_common_func_status_string(status));
1175         BT_ASSERT_POST(status == BT_FUNC_STATUS_OK ||
1176                 status == BT_FUNC_STATUS_ERROR ||
1177                 status == BT_FUNC_STATUS_MEMORY_ERROR ||
1178                 status == BT_FUNC_STATUS_AGAIN,
1179                 "Unexpected status: %![iter-]+i, status=%s",
1180                 iterator, bt_common_func_status_string(status));
1181         BT_ASSERT_POST_NO_ERROR_IF_NO_ERROR_STATUS(status);
1182         if (status < 0) {
1183                 BT_LIB_LOGW_APPEND_CAUSE(
1184                         "Component input port message iterator's \"seek beginning\" method failed: "
1185                         "%![iter-]+i, status=%s",
1186                         iterator, bt_common_func_status_string(status));
1187         }
1188
1189         set_iterator_state_after_seeking(iterator, status);
1190         return status;
1191 }
1192
1193 bt_bool
1194 bt_self_component_port_input_message_iterator_can_seek_forward(
1195                 bt_self_component_port_input_message_iterator *iterator)
1196 {
1197         BT_ASSERT_PRE_NON_NULL(iterator, "Message iterator");
1198
1199         return iterator->config.can_seek_forward;
1200 }
1201
1202 /*
1203  * Structure used to record the state of a given stream during the fast-forward
1204  * phase of an auto-seek.
1205  */
1206 struct auto_seek_stream_state {
1207         /*
1208          * Value representing which step of this timeline we are at.
1209          *
1210          *      time --->
1211          *   [SB]  1  [PB]  2  [PE]  1  [SE]
1212          *
1213          * At each point in the timeline, the messages we need to replicate are:
1214          *
1215          *   1: Stream beginning
1216          *   2: Stream beginning, packet beginning
1217          *
1218          * Before "Stream beginning" and after "Stream end", we don't need to
1219          * replicate anything as the stream doesn't exist.
1220          */
1221         enum {
1222                 AUTO_SEEK_STREAM_STATE_STREAM_BEGAN,
1223                 AUTO_SEEK_STREAM_STATE_PACKET_BEGAN,
1224         } state;
1225
1226         /*
1227          * If `state` is AUTO_SEEK_STREAM_STATE_PACKET_BEGAN, the packet we are
1228          * in.  This is a weak reference, since the packet will always be
1229          * alive by the time we use it.
1230          */
1231         struct bt_packet *packet;
1232
1233         /* Have we see a message with a clock snapshot yet? */
1234         bool seen_clock_snapshot;
1235 };
1236
1237 static
1238 struct auto_seek_stream_state *create_auto_seek_stream_state(void)
1239 {
1240         return g_new0(struct auto_seek_stream_state, 1);
1241 }
1242
1243 static
1244 void destroy_auto_seek_stream_state(void *ptr)
1245 {
1246         g_free(ptr);
1247 }
1248
1249 static
1250 GHashTable *create_auto_seek_stream_states(void)
1251 {
1252         return g_hash_table_new_full(g_direct_hash, g_direct_equal, NULL,
1253                 destroy_auto_seek_stream_state);
1254 }
1255
1256 static
1257 void destroy_auto_seek_stream_states(GHashTable *stream_states)
1258 {
1259         g_hash_table_destroy(stream_states);
1260 }
1261
1262 /*
1263  * Handle one message while we are in the fast-forward phase of an auto-seek.
1264  *
1265  * Sets `*got_first` to true if the message's timestamp is greater or equal to
1266  * `ns_from_origin`.  In other words, if this is the first message after our
1267  * seek point.
1268  *
1269  * `stream_states` is an hash table of `bt_stream *` (weak reference) to
1270  * `struct auto_seek_stream_state` used to keep the state of each stream
1271  * during the fast-forward.
1272  */
1273
1274 static inline
1275 int auto_seek_handle_message(
1276                 struct bt_self_component_port_input_message_iterator *iterator,
1277                 int64_t ns_from_origin, const struct bt_message *msg,
1278                 bool *got_first, GHashTable *stream_states)
1279 {
1280         int status = BT_FUNC_STATUS_OK;
1281         int64_t msg_ns_from_origin;
1282         const struct bt_clock_snapshot *clk_snapshot = NULL;
1283         int ret;
1284
1285         BT_ASSERT_DBG(msg);
1286         BT_ASSERT_DBG(got_first);
1287
1288         switch (msg->type) {
1289         case BT_MESSAGE_TYPE_EVENT:
1290         {
1291                 const struct bt_message_event *event_msg =
1292                         (const void *) msg;
1293
1294                 clk_snapshot = event_msg->default_cs;
1295                 BT_ASSERT_POST_DEV(clk_snapshot,
1296                         "Event message has no default clock snapshot: %!+n",
1297                         event_msg);
1298                 break;
1299         }
1300         case BT_MESSAGE_TYPE_MESSAGE_ITERATOR_INACTIVITY:
1301         {
1302                 const struct bt_message_message_iterator_inactivity *inactivity_msg =
1303                         (const void *) msg;
1304
1305                 clk_snapshot = inactivity_msg->default_cs;
1306                 BT_ASSERT_DBG(clk_snapshot);
1307                 break;
1308         }
1309         case BT_MESSAGE_TYPE_PACKET_BEGINNING:
1310         case BT_MESSAGE_TYPE_PACKET_END:
1311         {
1312                 const struct bt_message_packet *packet_msg =
1313                         (const void *) msg;
1314
1315                 clk_snapshot = packet_msg->default_cs;
1316                 BT_ASSERT_POST_DEV(clk_snapshot,
1317                         "Packet message has no default clock snapshot: %!+n",
1318                         packet_msg);
1319                 break;
1320         }
1321         case BT_MESSAGE_TYPE_DISCARDED_EVENTS:
1322         case BT_MESSAGE_TYPE_DISCARDED_PACKETS:
1323         {
1324                 struct bt_message_discarded_items *msg_disc_items =
1325                         (void *) msg;
1326
1327                 BT_ASSERT_POST_DEV(msg_disc_items->default_begin_cs &&
1328                         msg_disc_items->default_end_cs,
1329                         "Discarded events/packets message has no default clock snapshots: %!+n",
1330                         msg_disc_items);
1331                 ret = bt_clock_snapshot_get_ns_from_origin(
1332                         msg_disc_items->default_begin_cs,
1333                         &msg_ns_from_origin);
1334                 if (ret) {
1335                         status = BT_FUNC_STATUS_ERROR;
1336                         goto end;
1337                 }
1338
1339                 if (msg_ns_from_origin >= ns_from_origin) {
1340                         *got_first = true;
1341                         goto push_msg;
1342                 }
1343
1344                 ret = bt_clock_snapshot_get_ns_from_origin(
1345                         msg_disc_items->default_end_cs,
1346                         &msg_ns_from_origin);
1347                 if (ret) {
1348                         status = BT_FUNC_STATUS_ERROR;
1349                         goto end;
1350                 }
1351
1352                 if (msg_ns_from_origin >= ns_from_origin) {
1353                         /*
1354                          * The discarded items message's beginning time
1355                          * is before the requested seeking time, but its
1356                          * end time is after. Modify the message so as
1357                          * to set its beginning time to the requested
1358                          * seeking time, and make its item count unknown
1359                          * as we don't know if items were really
1360                          * discarded within the new time range.
1361                          */
1362                         uint64_t new_begin_raw_value = 0;
1363
1364                         ret = bt_clock_class_clock_value_from_ns_from_origin(
1365                                 msg_disc_items->default_end_cs->clock_class,
1366                                 ns_from_origin, &new_begin_raw_value);
1367                         if (ret) {
1368                                 status = BT_FUNC_STATUS_ERROR;
1369                                 goto end;
1370                         }
1371
1372                         bt_clock_snapshot_set_raw_value(
1373                                 msg_disc_items->default_begin_cs,
1374                                 new_begin_raw_value);
1375                         msg_disc_items->count.base.avail =
1376                                 BT_PROPERTY_AVAILABILITY_NOT_AVAILABLE;
1377
1378                         /*
1379                          * It is safe to push it because its beginning
1380                          * time is exactly the requested seeking time.
1381                          */
1382                         goto push_msg;
1383                 } else {
1384                         goto skip_msg;
1385                 }
1386         }
1387         case BT_MESSAGE_TYPE_STREAM_BEGINNING:
1388         case BT_MESSAGE_TYPE_STREAM_END:
1389         {
1390                 struct bt_message_stream *stream_msg =
1391                         (struct bt_message_stream *) msg;
1392
1393                 if (stream_msg->default_cs_state != BT_MESSAGE_STREAM_CLOCK_SNAPSHOT_STATE_KNOWN) {
1394                         /* Ignore */
1395                         goto skip_msg;
1396                 }
1397
1398                 clk_snapshot = stream_msg->default_cs;
1399                 break;
1400         }
1401         default:
1402                 bt_common_abort();
1403         }
1404
1405         BT_ASSERT_DBG(clk_snapshot);
1406         ret = bt_clock_snapshot_get_ns_from_origin(clk_snapshot,
1407                 &msg_ns_from_origin);
1408         if (ret) {
1409                 status = BT_FUNC_STATUS_ERROR;
1410                 goto end;
1411         }
1412
1413         if (msg_ns_from_origin >= ns_from_origin) {
1414                 *got_first = true;
1415                 goto push_msg;
1416         }
1417
1418 skip_msg:
1419         /* This message won't be sent downstream. */
1420         switch (msg->type) {
1421         case BT_MESSAGE_TYPE_STREAM_BEGINNING:
1422         {
1423                 const struct bt_message_stream *stream_msg = (const void *) msg;
1424                 struct auto_seek_stream_state *stream_state;
1425
1426                 /* Update stream's state: stream began. */
1427                 stream_state = create_auto_seek_stream_state();
1428                 if (!stream_state) {
1429                         status = BT_FUNC_STATUS_MEMORY_ERROR;
1430                         goto end;
1431                 }
1432
1433                 stream_state->state = AUTO_SEEK_STREAM_STATE_STREAM_BEGAN;
1434
1435                 if (stream_msg->default_cs_state == BT_MESSAGE_STREAM_CLOCK_SNAPSHOT_STATE_KNOWN) {
1436                         stream_state->seen_clock_snapshot = true;
1437                 }
1438
1439                 BT_ASSERT_DBG(!bt_g_hash_table_contains(stream_states, stream_msg->stream));
1440                 g_hash_table_insert(stream_states, stream_msg->stream, stream_state);
1441                 break;
1442         }
1443         case BT_MESSAGE_TYPE_PACKET_BEGINNING:
1444         {
1445                 const struct bt_message_packet *packet_msg =
1446                         (const void *) msg;
1447                 struct auto_seek_stream_state *stream_state;
1448
1449                 /* Update stream's state: packet began. */
1450                 stream_state = g_hash_table_lookup(stream_states, packet_msg->packet->stream);
1451                 BT_ASSERT_DBG(stream_state);
1452                 BT_ASSERT_DBG(stream_state->state == AUTO_SEEK_STREAM_STATE_STREAM_BEGAN);
1453                 stream_state->state = AUTO_SEEK_STREAM_STATE_PACKET_BEGAN;
1454                 BT_ASSERT_DBG(!stream_state->packet);
1455                 stream_state->packet = packet_msg->packet;
1456
1457                 if (packet_msg->packet->stream->class->packets_have_beginning_default_clock_snapshot) {
1458                         stream_state->seen_clock_snapshot = true;
1459                 }
1460
1461                 break;
1462         }
1463         case BT_MESSAGE_TYPE_EVENT:
1464         {
1465                 const struct bt_message_event *event_msg = (const void *) msg;
1466                 struct auto_seek_stream_state *stream_state;
1467
1468                 stream_state = g_hash_table_lookup(stream_states,
1469                         event_msg->event->packet->stream);
1470                 BT_ASSERT_DBG(stream_state);
1471
1472                 // HELPME: are we sure that event messages have clock snapshots at this point?
1473                 stream_state->seen_clock_snapshot = true;
1474
1475                 break;
1476         }
1477         case BT_MESSAGE_TYPE_PACKET_END:
1478         {
1479                 const struct bt_message_packet *packet_msg =
1480                         (const void *) msg;
1481                 struct auto_seek_stream_state *stream_state;
1482
1483                 /* Update stream's state: packet ended. */
1484                 stream_state = g_hash_table_lookup(stream_states, packet_msg->packet->stream);
1485                 BT_ASSERT_DBG(stream_state);
1486                 BT_ASSERT_DBG(stream_state->state == AUTO_SEEK_STREAM_STATE_PACKET_BEGAN);
1487                 stream_state->state = AUTO_SEEK_STREAM_STATE_STREAM_BEGAN;
1488                 BT_ASSERT_DBG(stream_state->packet);
1489                 stream_state->packet = NULL;
1490
1491                 if (packet_msg->packet->stream->class->packets_have_end_default_clock_snapshot) {
1492                         stream_state->seen_clock_snapshot = true;
1493                 }
1494
1495                 break;
1496         }
1497         case BT_MESSAGE_TYPE_STREAM_END:
1498         {
1499                 const struct bt_message_stream *stream_msg = (const void *) msg;
1500                 struct auto_seek_stream_state *stream_state;
1501
1502                 stream_state = g_hash_table_lookup(stream_states, stream_msg->stream);
1503                 BT_ASSERT_DBG(stream_state);
1504                 BT_ASSERT_DBG(stream_state->state == AUTO_SEEK_STREAM_STATE_STREAM_BEGAN);
1505                 BT_ASSERT_DBG(!stream_state->packet);
1506
1507                 /* Update stream's state: this stream doesn't exist anymore. */
1508                 g_hash_table_remove(stream_states, stream_msg->stream);
1509                 break;
1510         }
1511         case BT_MESSAGE_TYPE_DISCARDED_EVENTS:
1512         case BT_MESSAGE_TYPE_DISCARDED_PACKETS:
1513         {
1514                 const struct bt_message_discarded_items *discarded_msg =
1515                         (const void *) msg;
1516                 struct auto_seek_stream_state *stream_state;
1517
1518                 stream_state = g_hash_table_lookup(stream_states, discarded_msg->stream);
1519                 BT_ASSERT_DBG(stream_state);
1520
1521                 if ((msg->type == BT_MESSAGE_TYPE_DISCARDED_EVENTS && discarded_msg->stream->class->discarded_events_have_default_clock_snapshots) ||
1522                         (msg->type == BT_MESSAGE_TYPE_DISCARDED_PACKETS && discarded_msg->stream->class->discarded_packets_have_default_clock_snapshots)) {
1523                         stream_state->seen_clock_snapshot = true;
1524                 }
1525
1526                 break;
1527         }
1528         default:
1529                 break;
1530         }
1531
1532         bt_object_put_ref_no_null_check(msg);
1533         msg = NULL;
1534         goto end;
1535
1536 push_msg:
1537         g_queue_push_tail(iterator->auto_seek.msgs, (void *) msg);
1538         msg = NULL;
1539
1540 end:
1541         BT_ASSERT_DBG(!msg || status != BT_FUNC_STATUS_OK);
1542         return status;
1543 }
1544
1545 static
1546 int find_message_ge_ns_from_origin(
1547                 struct bt_self_component_port_input_message_iterator *iterator,
1548                 int64_t ns_from_origin, GHashTable *stream_states)
1549 {
1550         int status = BT_FUNC_STATUS_OK;
1551         enum bt_self_component_port_input_message_iterator_state init_state =
1552                 iterator->state;
1553         const struct bt_message *messages[MSG_BATCH_SIZE];
1554         uint64_t user_count = 0;
1555         uint64_t i;
1556         bool got_first = false;
1557
1558         BT_ASSERT_DBG(iterator);
1559         memset(&messages[0], 0, sizeof(messages[0]) * MSG_BATCH_SIZE);
1560
1561         /*
1562          * Make this iterator temporarily active (not seeking) to call
1563          * the "next" method.
1564          */
1565         set_self_comp_port_input_msg_iterator_state(iterator,
1566                 BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_ACTIVE);
1567
1568         BT_ASSERT_DBG(iterator->methods.next);
1569
1570         while (!got_first) {
1571                 /*
1572                  * Call the user's "next" method to get the next
1573                  * messages and status.
1574                  */
1575                 status = call_iterator_next_method(iterator,
1576                         &messages[0], MSG_BATCH_SIZE, &user_count);
1577                 BT_LOGD("User method returned: status=%s",
1578                         bt_common_func_status_string(status));
1579                 if (status < 0) {
1580                         BT_LIB_LOGW_APPEND_CAUSE(
1581                                 "Component input port message iterator's \"next\" method failed: "
1582                                 "%![iter-]+i, status=%s",
1583                                 iterator, bt_common_func_status_string(status));
1584                 }
1585
1586                 /*
1587                  * The user's "next" method must not do any action which
1588                  * would change the iterator's state.
1589                  */
1590                 BT_ASSERT_DBG(iterator->state ==
1591                         BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_ACTIVE);
1592
1593                 switch (status) {
1594                 case BT_FUNC_STATUS_OK:
1595                         BT_ASSERT_POST_DEV(user_count <= MSG_BATCH_SIZE,
1596                                 "Invalid returned message count: greater than "
1597                                 "batch size: count=%" PRIu64 ", batch-size=%u",
1598                                 user_count, MSG_BATCH_SIZE);
1599                         break;
1600                 case BT_FUNC_STATUS_AGAIN:
1601                 case BT_FUNC_STATUS_ERROR:
1602                 case BT_FUNC_STATUS_MEMORY_ERROR:
1603                 case BT_FUNC_STATUS_END:
1604                         goto end;
1605                 default:
1606                         bt_common_abort();
1607                 }
1608
1609                 for (i = 0; i < user_count; i++) {
1610                         if (got_first) {
1611                                 g_queue_push_tail(iterator->auto_seek.msgs,
1612                                         (void *) messages[i]);
1613                                 messages[i] = NULL;
1614                                 continue;
1615                         }
1616
1617                         status = auto_seek_handle_message(iterator,
1618                                 ns_from_origin, messages[i], &got_first,
1619                                 stream_states);
1620                         if (status == BT_FUNC_STATUS_OK) {
1621                                 /* Message was either pushed or moved */
1622                                 messages[i] = NULL;
1623                         } else {
1624                                 goto end;
1625                         }
1626                 }
1627         }
1628
1629 end:
1630         for (i = 0; i < user_count; i++) {
1631                 if (messages[i]) {
1632                         bt_object_put_ref_no_null_check(messages[i]);
1633                 }
1634         }
1635
1636         set_self_comp_port_input_msg_iterator_state(iterator, init_state);
1637         return status;
1638 }
1639
1640 /*
1641  * This function is installed as the iterator's next callback after we have
1642  * auto-seeked (seeked to the beginning and fast-forwarded) to send the
1643  * messages saved in iterator->auto_seek.msgs.  Once this is done, the original
1644  * next callback is put back.
1645  */
1646
1647 static
1648 enum bt_component_class_message_iterator_next_method_status post_auto_seek_next(
1649                 struct bt_self_component_port_input_message_iterator *iterator,
1650                 bt_message_array_const msgs, uint64_t capacity,
1651                 uint64_t *count)
1652 {
1653         BT_ASSERT(!g_queue_is_empty(iterator->auto_seek.msgs));
1654         *count = 0;
1655
1656         /*
1657          * Move auto-seek messages to the output array (which is this
1658          * iterator's base message array).
1659          */
1660         while (capacity > 0 && !g_queue_is_empty(iterator->auto_seek.msgs)) {
1661                 msgs[*count] = g_queue_pop_head(iterator->auto_seek.msgs);
1662                 capacity--;
1663                 (*count)++;
1664         }
1665
1666         BT_ASSERT(*count > 0);
1667
1668         if (g_queue_is_empty(iterator->auto_seek.msgs)) {
1669                 /* No more auto-seek messages, restore user's next callback. */
1670                 BT_ASSERT(iterator->auto_seek.original_next_callback);
1671                 iterator->methods.next = iterator->auto_seek.original_next_callback;
1672                 iterator->auto_seek.original_next_callback = NULL;
1673         }
1674
1675         return BT_FUNC_STATUS_OK;
1676 }
1677
1678 static inline
1679 int clock_raw_value_from_ns_from_origin(const bt_clock_class *clock_class,
1680                 int64_t ns_from_origin, uint64_t *raw_value)
1681 {
1682
1683         int64_t cc_offset_s = clock_class->offset_seconds;
1684         uint64_t cc_offset_cycles = clock_class->offset_cycles;
1685         uint64_t cc_freq = clock_class->frequency;
1686
1687         return bt_common_clock_value_from_ns_from_origin(cc_offset_s,
1688                 cc_offset_cycles, cc_freq, ns_from_origin, raw_value);
1689 }
1690
1691 static
1692 bool message_iterator_can_seek_ns_from_origin(
1693                 struct bt_self_component_port_input_message_iterator *iterator,
1694                 int64_t ns_from_origin)
1695 {
1696         enum bt_message_iterator_can_seek_ns_from_origin_status status;
1697         bt_bool can_seek;
1698
1699         status = bt_self_component_port_input_message_iterator_can_seek_ns_from_origin(
1700                 iterator, ns_from_origin, &can_seek);
1701         if (status != BT_FUNC_STATUS_OK) {
1702                 can_seek = BT_FALSE;
1703         }
1704
1705         return can_seek;
1706 }
1707
1708 enum bt_message_iterator_seek_ns_from_origin_status
1709 bt_self_component_port_input_message_iterator_seek_ns_from_origin(
1710                 struct bt_self_component_port_input_message_iterator *iterator,
1711                 int64_t ns_from_origin)
1712 {
1713         int status;
1714         GHashTable *stream_states = NULL;
1715         bt_bool can_seek_by_itself;
1716
1717         BT_ASSERT_PRE_NON_NULL(iterator, "Message iterator");
1718         BT_ASSERT_PRE_ITER_HAS_STATE_TO_SEEK(iterator);
1719         BT_ASSERT_PRE(
1720                 bt_component_borrow_graph(iterator->upstream_component)->config_state !=
1721                         BT_GRAPH_CONFIGURATION_STATE_CONFIGURING,
1722                 "Graph is not configured: %!+g",
1723                 bt_component_borrow_graph(iterator->upstream_component));
1724         /* The iterator must be able to seek ns from origin one way or another. */
1725         BT_ASSERT_PRE(
1726                 message_iterator_can_seek_ns_from_origin(iterator, ns_from_origin),
1727                 "Message iterator cannot seek nanoseconds from origin: %!+i, "
1728                 "ns-from-origin=%" PRId64, iterator, ns_from_origin);
1729         set_self_comp_port_input_msg_iterator_state(iterator,
1730                 BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_SEEKING);
1731
1732         /*
1733          * We are seeking, reset our expectations about how the following
1734          * messages should look like.
1735          */
1736         reset_iterator_expectations(iterator);
1737
1738         /* Check if the iterator can seek by itself.  If not we'll use autoseek. */
1739         if (iterator->methods.can_seek_ns_from_origin) {
1740                 bt_component_class_message_iterator_can_seek_ns_from_origin_method_status
1741                         can_seek_status;
1742
1743                 can_seek_status =
1744                         iterator->methods.can_seek_ns_from_origin(
1745                                 iterator, ns_from_origin, &can_seek_by_itself);
1746                 if (can_seek_status != BT_FUNC_STATUS_OK) {
1747                         status = can_seek_status;
1748                         goto end;
1749                 }
1750         } else {
1751                 can_seek_by_itself = false;
1752         }
1753
1754         if (can_seek_by_itself) {
1755                 /* The iterator knows how to seek to a particular time, let it handle this. */
1756                 BT_ASSERT(iterator->methods.seek_ns_from_origin);
1757                 BT_LIB_LOGD("Calling user's \"seek nanoseconds from origin\" method: "
1758                         "%![iter-]+i, ns=%" PRId64, iterator, ns_from_origin);
1759                 status = iterator->methods.seek_ns_from_origin(iterator,
1760                         ns_from_origin);
1761                 BT_LOGD("User method returned: status=%s",
1762                         bt_common_func_status_string(status));
1763                 BT_ASSERT_POST(status == BT_FUNC_STATUS_OK ||
1764                         status == BT_FUNC_STATUS_ERROR ||
1765                         status == BT_FUNC_STATUS_MEMORY_ERROR ||
1766                         status == BT_FUNC_STATUS_AGAIN,
1767                         "Unexpected status: %![iter-]+i, status=%s",
1768                         iterator, bt_common_func_status_string(status));
1769                 BT_ASSERT_POST_NO_ERROR_IF_NO_ERROR_STATUS(status);
1770                 if (status < 0) {
1771                         BT_LIB_LOGW_APPEND_CAUSE(
1772                                 "Component input port message iterator's \"seek nanoseconds from origin\" method failed: "
1773                                 "%![iter-]+i, status=%s",
1774                                 iterator, bt_common_func_status_string(status));
1775                 }
1776         } else {
1777                 /*
1778                  * The iterator doesn't know how to seek by itself to a
1779                  * particular time.  We will seek to the beginning and fast
1780                  * forward to the right place.
1781                  */
1782                 enum bt_component_class_message_iterator_can_seek_beginning_method_status
1783                         can_seek_status;
1784                 bt_bool can_seek_beginning;
1785
1786                 can_seek_status = iterator->methods.can_seek_beginning(iterator,
1787                         &can_seek_beginning);
1788                 BT_ASSERT(can_seek_status == BT_FUNC_STATUS_OK);
1789                 BT_ASSERT(can_seek_beginning);
1790                 BT_ASSERT(iterator->methods.seek_beginning);
1791                 BT_LIB_LOGD("Calling user's \"seek beginning\" method: %!+i",
1792                         iterator);
1793                 status = iterator->methods.seek_beginning(iterator);
1794                 BT_LOGD("User method returned: status=%s",
1795                         bt_common_func_status_string(status));
1796                 BT_ASSERT_POST(status == BT_FUNC_STATUS_OK ||
1797                         status == BT_FUNC_STATUS_ERROR ||
1798                         status == BT_FUNC_STATUS_MEMORY_ERROR ||
1799                         status == BT_FUNC_STATUS_AGAIN,
1800                         "Unexpected status: %![iter-]+i, status=%s",
1801                         iterator, bt_common_func_status_string(status));
1802                 if (status < 0) {
1803                         BT_LIB_LOGW_APPEND_CAUSE(
1804                                 "Component input port message iterator's \"seek beginning\" method failed: "
1805                                 "%![iter-]+i, status=%s",
1806                                 iterator, bt_common_func_status_string(status));
1807                 }
1808
1809                 switch (status) {
1810                 case BT_FUNC_STATUS_OK:
1811                         break;
1812                 case BT_FUNC_STATUS_ERROR:
1813                 case BT_FUNC_STATUS_MEMORY_ERROR:
1814                 case BT_FUNC_STATUS_AGAIN:
1815                         goto end;
1816                 default:
1817                         bt_common_abort();
1818                 }
1819
1820                 /*
1821                  * Find the first message which has a default clock
1822                  * snapshot greater than or equal to the requested
1823                  * seeking time, and move the received messages from
1824                  * this point in the batch to this iterator's auto-seek
1825                  * message queue.
1826                  */
1827                 while (!g_queue_is_empty(iterator->auto_seek.msgs)) {
1828                         bt_object_put_ref_no_null_check(
1829                                 g_queue_pop_tail(iterator->auto_seek.msgs));
1830                 }
1831
1832                 stream_states = create_auto_seek_stream_states();
1833                 if (!stream_states) {
1834                         BT_LIB_LOGE_APPEND_CAUSE(
1835                                 "Failed to allocate one GHashTable.");
1836                         status = BT_FUNC_STATUS_MEMORY_ERROR;
1837                         goto end;
1838                 }
1839
1840                 status = find_message_ge_ns_from_origin(iterator,
1841                         ns_from_origin, stream_states);
1842                 switch (status) {
1843                 case BT_FUNC_STATUS_OK:
1844                 case BT_FUNC_STATUS_END:
1845                 {
1846                         GHashTableIter iter;
1847                         gpointer key, value;
1848
1849                         /*
1850                          * If some streams exist at the seek time, prepend the
1851                          * required messages to put those streams in the right
1852                          * state.
1853                          */
1854                         g_hash_table_iter_init(&iter, stream_states);
1855                         while (g_hash_table_iter_next (&iter, &key, &value)) {
1856                                 const bt_stream *stream = key;
1857                                 struct auto_seek_stream_state *stream_state =
1858                                         (struct auto_seek_stream_state *) value;
1859                                 bt_message *msg;
1860                                 const bt_clock_class *clock_class = bt_stream_class_borrow_default_clock_class_const(
1861                                         bt_stream_borrow_class_const(stream));
1862                                 /* Initialize to silence maybe-uninitialized warning. */
1863                                 uint64_t raw_value = 0;
1864
1865                                 /*
1866                                  * If we haven't seen a message with a clock snapshot, we don't know if our seek time is within
1867                                  * the clock's range, so it wouldn't be safe to try to convert ns_from_origin to a clock value.
1868                                  *
1869                                  * Also, it would be a bit of a lie to generate a stream begin message with the seek time as its
1870                                  * clock snapshot, because we don't really know if the stream existed at that time.  If we have
1871                                  * seen a message with a clock snapshot in our seeking, then we are sure that the
1872                                  * seek time is not below the clock range, and we know the stream was active at that
1873                                  * time (and that we cut it short).
1874                                  */
1875                                 if (stream_state->seen_clock_snapshot) {
1876                                         if (clock_raw_value_from_ns_from_origin(clock_class, ns_from_origin, &raw_value) != 0) {
1877                                                 BT_LIB_LOGW("Could not convert nanoseconds from origin to clock value: ns-from-origin=%" PRId64 ", %![cc-]+K",
1878                                                         ns_from_origin, clock_class);
1879                                                 status = BT_FUNC_STATUS_ERROR;
1880                                                 goto end;
1881                                         }
1882                                 }
1883
1884                                 switch (stream_state->state) {
1885                                 case AUTO_SEEK_STREAM_STATE_PACKET_BEGAN:
1886                                         BT_ASSERT(stream_state->packet);
1887                                         BT_LIB_LOGD("Creating packet message: %![packet-]+a", stream_state->packet);
1888
1889                                         if (stream->class->packets_have_beginning_default_clock_snapshot) {
1890                                                 /*
1891                                                  * If we are in the PACKET_BEGAN state, it means we have seen a "packet beginning"
1892                                                  * message.  If "packet beginning" packets have clock snapshots, then we must have
1893                                                  * seen a clock snapshot.
1894                                                  */
1895                                                 BT_ASSERT(stream_state->seen_clock_snapshot);
1896
1897                                                 msg = bt_message_packet_beginning_create_with_default_clock_snapshot(
1898                                                         (bt_self_message_iterator *) iterator, stream_state->packet, raw_value);
1899                                         } else {
1900                                                 msg = bt_message_packet_beginning_create((bt_self_message_iterator *) iterator,
1901                                                                 stream_state->packet);
1902                                         }
1903
1904                                         if (!msg) {
1905                                                 status = BT_FUNC_STATUS_MEMORY_ERROR;
1906                                                 goto end;
1907                                         }
1908
1909                                         g_queue_push_head(iterator->auto_seek.msgs, msg);
1910                                         msg = NULL;
1911                                         /* fall-thru */
1912
1913                                 case AUTO_SEEK_STREAM_STATE_STREAM_BEGAN:
1914                                         msg = bt_message_stream_beginning_create(
1915                                                 (bt_self_message_iterator *) iterator, stream);
1916                                         if (!msg) {
1917                                                 status = BT_FUNC_STATUS_MEMORY_ERROR;
1918                                                 goto end;
1919                                         }
1920
1921                                         if (stream_state->seen_clock_snapshot) {
1922                                                 bt_message_stream_beginning_set_default_clock_snapshot(msg, raw_value);
1923                                         }
1924
1925                                         g_queue_push_head(iterator->auto_seek.msgs, msg);
1926                                         msg = NULL;
1927                                         break;
1928                                 }
1929                         }
1930
1931                         /*
1932                          * If there are messages in the auto-seek
1933                          * message queue, replace the user's "next"
1934                          * method with a custom, temporary "next" method
1935                          * which returns them.
1936                          */
1937                         if (!g_queue_is_empty(iterator->auto_seek.msgs)) {
1938                                 BT_ASSERT(!iterator->auto_seek.original_next_callback);
1939                                 iterator->auto_seek.original_next_callback = iterator->methods.next;
1940
1941                                 iterator->methods.next =
1942                                         (bt_self_component_port_input_message_iterator_next_method)
1943                                                 post_auto_seek_next;
1944                         }
1945
1946                         /*
1947                          * `BT_FUNC_STATUS_END` becomes
1948                          * `BT_FUNC_STATUS_OK`: the next
1949                          * time this iterator's "next" method is called,
1950                          * it will return
1951                          * `BT_FUNC_STATUS_END`.
1952                          */
1953                         status = BT_FUNC_STATUS_OK;
1954                         break;
1955                 }
1956                 case BT_FUNC_STATUS_ERROR:
1957                 case BT_FUNC_STATUS_MEMORY_ERROR:
1958                 case BT_FUNC_STATUS_AGAIN:
1959                         goto end;
1960                 default:
1961                         bt_common_abort();
1962                 }
1963         }
1964
1965         /*
1966          * The following messages returned by the next method (including
1967          * post_auto_seek_next) must be after (or at) `ns_from_origin`.
1968          */
1969         iterator->last_ns_from_origin = ns_from_origin;
1970
1971 end:
1972         if (stream_states) {
1973                 destroy_auto_seek_stream_states(stream_states);
1974                 stream_states = NULL;
1975         }
1976
1977         set_iterator_state_after_seeking(iterator, status);
1978         return status;
1979 }
1980
1981 bt_bool bt_self_message_iterator_is_interrupted(
1982                 const struct bt_self_message_iterator *self_msg_iter)
1983 {
1984         const struct bt_self_component_port_input_message_iterator *iterator =
1985                 (const void *) self_msg_iter;
1986
1987         BT_ASSERT_PRE_NON_NULL(iterator, "Message iterator");
1988         return (bt_bool) bt_graph_is_interrupted(iterator->graph);
1989 }
1990
1991 void bt_self_component_port_input_message_iterator_get_ref(
1992                 const struct bt_self_component_port_input_message_iterator *iterator)
1993 {
1994         bt_object_get_ref(iterator);
1995 }
1996
1997 void bt_self_component_port_input_message_iterator_put_ref(
1998                 const struct bt_self_component_port_input_message_iterator *iterator)
1999 {
2000         bt_object_put_ref(iterator);
2001 }