src/lib/graph/iterator.c

   1 /*
   2  * SPDX-License-Identifier: MIT
   3  *
   4  * Copyright 2017-2018 Philippe Proulx <pproulx@efficios.com>
   5  * Copyright 2015 Jérémie Galarneau <jeremie.galarneau@efficios.com>
   6  */
   7
   8 #define BT_LOG_TAG "LIB/MSG-ITER"
   9 #include "lib/logging.h"
  10
  11 #include "compat/compiler.h"
  12 #include "compat/glib.h"
  13 #include "lib/trace-ir/clock-class.h"
  14 #include "lib/trace-ir/clock-snapshot.h"
  15 #include <babeltrace2/trace-ir/field.h>
  16 #include <babeltrace2/trace-ir/event.h>
  17 #include "lib/trace-ir/event.h"
  18 #include <babeltrace2/trace-ir/packet.h>
  19 #include "lib/trace-ir/packet.h"
  20 #include "lib/trace-ir/stream.h"
  21 #include "lib/trace-ir/stream-class.h"
  22 #include <babeltrace2/trace-ir/clock-class.h>
  23 #include <babeltrace2/trace-ir/stream-class.h>
  24 #include <babeltrace2/trace-ir/stream.h>
  25 #include <babeltrace2/graph/connection.h>
  26 #include <babeltrace2/graph/component.h>
  27 #include <babeltrace2/graph/message.h>
  28 #include <babeltrace2/graph/self-component.h>
  29 #include <babeltrace2/graph/port.h>
  30 #include <babeltrace2/graph/graph.h>
  31 #include <babeltrace2/graph/message-iterator.h>
  32 #include <babeltrace2/types.h>
  33 #include "common/assert.h"
  34 #include "lib/assert-cond.h"
  35 #include <stdint.h>
  36 #include <inttypes.h>
  37 #include <stdbool.h>
  38 #include <stdlib.h>
  39
  40 #include "component-class.h"
  41 #include "component.h"
  42 #include "connection.h"
  43 #include "graph.h"
  44 #include "iterator.h"
  45 #include "message-iterator-class.h"
  46 #include "message/discarded-items.h"
  47 #include "message/event.h"
  48 #include "message/message.h"
  49 #include "message/message-iterator-inactivity.h"
  50 #include "message/stream.h"
  51 #include "message/packet.h"
  52 #include "lib/func-status.h"
  53
  54 /*
  55  * TODO: Use graph's state (number of active iterators, etc.) and
  56  * possibly system specifications to make a better guess than this.
  57  */
  58 #define MSG_BATCH_SIZE  15
  59
  60 #define BT_ASSERT_PRE_ITER_HAS_STATE_TO_SEEK(_iter)                     \
  61         BT_ASSERT_PRE("has-state-to-seek",                              \
  62                 (_iter)->state == BT_MESSAGE_ITERATOR_STATE_ACTIVE ||   \
  63                 (_iter)->state == BT_MESSAGE_ITERATOR_STATE_ENDED ||    \
  64                 (_iter)->state == BT_MESSAGE_ITERATOR_STATE_LAST_SEEKING_RETURNED_AGAIN || \
  65                 (_iter)->state == BT_MESSAGE_ITERATOR_STATE_LAST_SEEKING_RETURNED_ERROR, \
  66                 "Message iterator is in the wrong state: %!+i", (_iter))
  67
  68 #ifdef BT_DEV_MODE
  69 struct per_stream_state
  70 {
  71         bt_packet *cur_packet;
  72
  73         /* Bit mask of expected message types. */
  74         guint expected_msg_types;
  75 };
  76 #endif
  77
  78 static void
  79 clear_per_stream_state (struct bt_message_iterator *iterator)
  80 {
  81 #ifdef BT_DEV_MODE
  82         g_hash_table_remove_all(iterator->per_stream_state);
  83 #else
  84         BT_USE_EXPR(iterator);
  85 #endif
  86 }
  87
  88 static inline
  89 void set_msg_iterator_state(struct bt_message_iterator *iterator,
  90                 enum bt_message_iterator_state state)
  91 {
  92         BT_ASSERT_DBG(iterator);
  93         BT_LIB_LOGD("Updating message iterator's state: new-state=%s",
  94                 bt_message_iterator_state_string(state));
  95         iterator->state = state;
  96 }
  97
  98 static
  99 void bt_message_iterator_destroy(struct bt_object *obj)
 100 {
 101         struct bt_message_iterator *iterator;
 102
 103         BT_ASSERT(obj);
 104
 105         /*
 106          * The message iterator's reference count is 0 if we're
 107          * here. Increment it to avoid a double-destroy (possibly
 108          * infinitely recursive). This could happen for example if the
 109          * message iterator's finalization function does
 110          * bt_object_get_ref() (or anything that causes
 111          * bt_object_get_ref() to be called) on itself (ref. count goes
 112          * from 0 to 1), and then bt_object_put_ref(): the reference
 113          * count would go from 1 to 0 again and this function would be
 114          * called again.
 115          */
 116         obj->ref_count++;
 117         iterator = (void *) obj;
 118         BT_LIB_LOGI("Destroying self component input port message iterator object: "
 119                 "%!+i", iterator);
 120         bt_message_iterator_try_finalize(iterator);
 121
 122         if (iterator->clock_expectation.type ==
 123                         CLOCK_EXPECTATION_ORIGIN_OTHER_NO_UUID) {
 124                 BT_CLOCK_CLASS_PUT_REF_AND_RESET(
 125                         iterator->clock_expectation.clock_class);
 126         }
 127
 128         if (iterator->connection) {
 129                 /*
 130                  * Remove ourself from the originating connection so
 131                  * that it does not try to finalize a dangling pointer
 132                  * later.
 133                  */
 134                 bt_connection_remove_iterator(iterator->connection, iterator);
 135                 iterator->connection = NULL;
 136         }
 137
 138         if (iterator->auto_seek.msgs) {
 139                 while (!g_queue_is_empty(iterator->auto_seek.msgs)) {
 140                         bt_object_put_ref_no_null_check(
 141                                 g_queue_pop_tail(iterator->auto_seek.msgs));
 142                 }
 143
 144                 g_queue_free(iterator->auto_seek.msgs);
 145                 iterator->auto_seek.msgs = NULL;
 146         }
 147
 148         if (iterator->upstream_msg_iters) {
 149                 /*
 150                  * At this point the message iterator is finalized, so
 151                  * it's detached from any upstream message iterator.
 152                  */
 153                 BT_ASSERT(iterator->upstream_msg_iters->len == 0);
 154                 g_ptr_array_free(iterator->upstream_msg_iters, TRUE);
 155                 iterator->upstream_msg_iters = NULL;
 156         }
 157
 158         if (iterator->msgs) {
 159                 g_ptr_array_free(iterator->msgs, TRUE);
 160                 iterator->msgs = NULL;
 161         }
 162
 163 #ifdef BT_DEV_MODE
 164         g_hash_table_destroy(iterator->per_stream_state);
 165 #endif
 166
 167         g_free(iterator);
 168 }
 169
 170 void bt_message_iterator_try_finalize(
 171                 struct bt_message_iterator *iterator)
 172 {
 173         uint64_t i;
 174         bool call_user_finalize = true;
 175
 176         BT_ASSERT(iterator);
 177
 178         switch (iterator->state) {
 179         case BT_MESSAGE_ITERATOR_STATE_NON_INITIALIZED:
 180                 /*
 181                  * If this function is called while the iterator is in the
 182                  * NON_INITIALIZED state, it means the user initialization
 183                  * method has either not been called, or has failed.  We
 184                  * therefore don't want to call the user finalization method.
 185                  * However, the initialization method might have created some
 186                  * upstream message iterators before failing, so we want to
 187                  * execute the rest of this function, which unlinks the related
 188                  * iterators.
 189                  */
 190                 call_user_finalize = false;
 191                 break;
 192         case BT_MESSAGE_ITERATOR_STATE_FINALIZED:
 193                 /* Already finalized */
 194                 BT_LIB_LOGD("Not finalizing message iterator: already finalized: "
 195                         "%!+i", iterator);
 196                 goto end;
 197         case BT_MESSAGE_ITERATOR_STATE_FINALIZING:
 198                 /* Finalizing */
 199                 BT_LIB_LOGF("Message iterator is already being finalized: "
 200                         "%!+i", iterator);
 201                 bt_common_abort();
 202         default:
 203                 break;
 204         }
 205
 206         BT_LIB_LOGD("Finalizing message iterator: %!+i", iterator);
 207         set_msg_iterator_state(iterator,
 208                 BT_MESSAGE_ITERATOR_STATE_FINALIZING);
 209         BT_ASSERT(iterator->upstream_component);
 210
 211         /* Call user-defined destroy method */
 212         if (call_user_finalize) {
 213                 typedef void (*method_t)(void *);
 214                 method_t method;
 215                 struct bt_component_class *comp_class =
 216                         iterator->upstream_component->class;
 217                 struct bt_component_class_with_iterator_class *class_with_iter_class;
 218
 219                 BT_ASSERT(bt_component_class_has_message_iterator_class(comp_class));
 220                 class_with_iter_class = container_of(comp_class,
 221                         struct bt_component_class_with_iterator_class, parent);
 222                 method = (method_t) class_with_iter_class->msg_iter_cls->methods.finalize;
 223
 224                 if (method) {
 225                         const bt_error *saved_error;
 226
 227                         saved_error = bt_current_thread_take_error();
 228
 229                         BT_LIB_LOGD("Calling user's finalization method: %!+i",
 230                                 iterator);
 231                         method(iterator);
 232                         BT_ASSERT_POST_NO_ERROR("bt_message_iterator_class_finalize_method");
 233
 234                         if (saved_error) {
 235                                 BT_CURRENT_THREAD_MOVE_ERROR_AND_RESET(saved_error);
 236                         }
 237                 }
 238         }
 239
 240         /* Detach upstream message iterators */
 241         for (i = 0; i < iterator->upstream_msg_iters->len; i++) {
 242                 struct bt_message_iterator *upstream_msg_iter =
 243                         iterator->upstream_msg_iters->pdata[i];
 244
 245                 upstream_msg_iter->downstream_msg_iter = NULL;
 246         }
 247
 248         g_ptr_array_set_size(iterator->upstream_msg_iters, 0);
 249
 250         /* Detach downstream message iterator */
 251         if (iterator->downstream_msg_iter) {
 252                 gboolean existed;
 253
 254                 BT_ASSERT(iterator->downstream_msg_iter->upstream_msg_iters);
 255                 existed = g_ptr_array_remove_fast(
 256                         iterator->downstream_msg_iter->upstream_msg_iters,
 257                         iterator);
 258                 BT_ASSERT(existed);
 259         }
 260
 261         iterator->upstream_component = NULL;
 262         iterator->upstream_port = NULL;
 263         set_msg_iterator_state(iterator,
 264                 BT_MESSAGE_ITERATOR_STATE_FINALIZED);
 265         BT_LIB_LOGD("Finalized message iterator: %!+i", iterator);
 266
 267 end:
 268         return;
 269 }
 270
 271 void bt_message_iterator_set_connection(
 272                 struct bt_message_iterator *iterator,
 273                 struct bt_connection *connection)
 274 {
 275         BT_ASSERT(iterator);
 276         iterator->connection = connection;
 277         BT_LIB_LOGI("Set message iterator's connection: "
 278                 "%![iter-]+i, %![conn-]+x", iterator, connection);
 279 }
 280
 281 static
 282 enum bt_message_iterator_can_seek_beginning_status can_seek_ns_from_origin_true(
 283                 struct bt_message_iterator *iterator __attribute__((unused)),
 284                 int64_t ns_from_origin __attribute__((unused)),
 285                 bt_bool *can_seek)
 286 {
 287         *can_seek = BT_TRUE;
 288
 289         return BT_FUNC_STATUS_OK;
 290 }
 291
 292 static
 293 enum bt_message_iterator_can_seek_beginning_status can_seek_beginning_true(
 294                 struct bt_message_iterator *iterator __attribute__((unused)),
 295                 bt_bool *can_seek)
 296 {
 297         *can_seek = BT_TRUE;
 298
 299         return BT_FUNC_STATUS_OK;
 300 }
 301
 302 static
 303 int create_self_component_input_port_message_iterator(
 304                 struct bt_self_message_iterator *self_downstream_msg_iter,
 305                 struct bt_self_component_port_input *self_port,
 306                 struct bt_message_iterator **message_iterator,
 307                 const char *api_func)
 308 {
 309         bt_message_iterator_class_initialize_method init_method = NULL;
 310         struct bt_message_iterator *iterator =
 311                 NULL;
 312         struct bt_message_iterator *downstream_msg_iter =
 313                 (void *) self_downstream_msg_iter;
 314         struct bt_port *port = (void *) self_port;
 315         struct bt_port *upstream_port;
 316         struct bt_component *comp;
 317         struct bt_component *upstream_comp;
 318         struct bt_component_class *upstream_comp_cls;
 319         struct bt_component_class_with_iterator_class *upstream_comp_cls_with_iter_cls;
 320         int status;
 321
 322         BT_ASSERT_PRE_NON_NULL_FROM_FUNC(api_func, "message-iterator-output",
 323                 message_iterator, "Created message iterator (output)");
 324         BT_ASSERT_PRE_NON_NULL_FROM_FUNC(api_func, "input-port", port,
 325                 "Input port");
 326         comp = bt_port_borrow_component_inline(port);
 327         BT_ASSERT_PRE_FROM_FUNC(api_func, "input-port-is-connected",
 328                 bt_port_is_connected(port),
 329                 "Input port is not connected: %![port-]+p", port);
 330         BT_ASSERT_PRE_FROM_FUNC(api_func, "input-port-has-component",
 331                 comp, "Input port is not part of a component: %![port-]+p",
 332                 port);
 333         BT_ASSERT(port->connection);
 334         upstream_port = port->connection->upstream_port;
 335         BT_ASSERT(upstream_port);
 336         upstream_comp = bt_port_borrow_component_inline(upstream_port);
 337         BT_ASSERT(upstream_comp);
 338         BT_ASSERT_PRE_FROM_FUNC(api_func, "graph-is-configured",
 339                 bt_component_borrow_graph(upstream_comp)->config_state ==
 340                         BT_GRAPH_CONFIGURATION_STATE_PARTIALLY_CONFIGURED ||
 341                 bt_component_borrow_graph(upstream_comp)->config_state ==
 342                         BT_GRAPH_CONFIGURATION_STATE_CONFIGURED,
 343                 "Graph is not configured: %!+g",
 344                 bt_component_borrow_graph(upstream_comp));
 345         upstream_comp_cls = upstream_comp->class;
 346         BT_ASSERT(upstream_comp->class->type ==
 347                 BT_COMPONENT_CLASS_TYPE_SOURCE ||
 348                 upstream_comp->class->type ==
 349                 BT_COMPONENT_CLASS_TYPE_FILTER);
 350         BT_LIB_LOGI("Creating message iterator on self component input port: "
 351                 "%![up-comp-]+c, %![up-port-]+p", upstream_comp, upstream_port);
 352         iterator = g_new0(
 353                 struct bt_message_iterator, 1);
 354         if (!iterator) {
 355                 BT_LIB_LOGE_APPEND_CAUSE(
 356                         "Failed to allocate one self component input port "
 357                         "message iterator.");
 358                 status = BT_FUNC_STATUS_MEMORY_ERROR;
 359                 goto error;
 360         }
 361
 362         bt_object_init_shared(&iterator->base,
 363                 bt_message_iterator_destroy);
 364         iterator->msgs = g_ptr_array_new();
 365         if (!iterator->msgs) {
 366                 BT_LIB_LOGE_APPEND_CAUSE("Failed to allocate a GPtrArray.");
 367                 status = BT_FUNC_STATUS_MEMORY_ERROR;
 368                 goto error;
 369         }
 370
 371         g_ptr_array_set_size(iterator->msgs, MSG_BATCH_SIZE);
 372         iterator->last_ns_from_origin = INT64_MIN;
 373
 374 #ifdef BT_DEV_MODE
 375         /* The per-stream state is only used for dev assertions right now. */
 376         iterator->per_stream_state = g_hash_table_new_full(
 377                 g_direct_hash,
 378                 g_direct_equal,
 379                 NULL,
 380                 g_free);
 381 #endif
 382
 383         iterator->auto_seek.msgs = g_queue_new();
 384         if (!iterator->auto_seek.msgs) {
 385                 BT_LIB_LOGE_APPEND_CAUSE("Failed to allocate a GQueue.");
 386                 status = BT_FUNC_STATUS_MEMORY_ERROR;
 387                 goto error;
 388         }
 389
 390         iterator->upstream_msg_iters = g_ptr_array_new();
 391         if (!iterator->upstream_msg_iters) {
 392                 BT_LIB_LOGE_APPEND_CAUSE("Failed to allocate a GPtrArray.");
 393                 status = BT_FUNC_STATUS_MEMORY_ERROR;
 394                 goto error;
 395         }
 396
 397         iterator->upstream_component = upstream_comp;
 398         iterator->upstream_port = upstream_port;
 399         iterator->connection = iterator->upstream_port->connection;
 400         iterator->graph = bt_component_borrow_graph(upstream_comp);
 401         set_msg_iterator_state(iterator,
 402                 BT_MESSAGE_ITERATOR_STATE_NON_INITIALIZED);
 403
 404         /* Copy methods from the message iterator class to the message iterator. */
 405         BT_ASSERT(bt_component_class_has_message_iterator_class(upstream_comp_cls));
 406         upstream_comp_cls_with_iter_cls = container_of(upstream_comp_cls,
 407                 struct bt_component_class_with_iterator_class, parent);
 408
 409         iterator->methods.next =
 410                 (bt_message_iterator_next_method)
 411                         upstream_comp_cls_with_iter_cls->msg_iter_cls->methods.next;
 412         iterator->methods.seek_ns_from_origin =
 413                 (bt_message_iterator_seek_ns_from_origin_method)
 414                         upstream_comp_cls_with_iter_cls->msg_iter_cls->methods.seek_ns_from_origin;
 415         iterator->methods.seek_beginning =
 416                 (bt_message_iterator_seek_beginning_method)
 417                         upstream_comp_cls_with_iter_cls->msg_iter_cls->methods.seek_beginning;
 418         iterator->methods.can_seek_ns_from_origin =
 419                 (bt_message_iterator_can_seek_ns_from_origin_method)
 420                         upstream_comp_cls_with_iter_cls->msg_iter_cls->methods.can_seek_ns_from_origin;
 421         iterator->methods.can_seek_beginning =
 422                 (bt_message_iterator_can_seek_beginning_method)
 423                         upstream_comp_cls_with_iter_cls->msg_iter_cls->methods.can_seek_beginning;
 424
 425         if (iterator->methods.seek_ns_from_origin &&
 426                         !iterator->methods.can_seek_ns_from_origin) {
 427                 iterator->methods.can_seek_ns_from_origin =
 428                         (bt_message_iterator_can_seek_ns_from_origin_method)
 429                                 can_seek_ns_from_origin_true;
 430         }
 431
 432         if (iterator->methods.seek_beginning &&
 433                         !iterator->methods.can_seek_beginning) {
 434                 iterator->methods.can_seek_beginning =
 435                         (bt_message_iterator_can_seek_beginning_method)
 436                                 can_seek_beginning_true;
 437         }
 438
 439         /* Call iterator's init method. */
 440         init_method = upstream_comp_cls_with_iter_cls->msg_iter_cls->methods.initialize;
 441
 442         if (init_method) {
 443                 enum bt_message_iterator_class_initialize_method_status iter_status;
 444
 445                 BT_LIB_LOGD("Calling user's initialization method: %!+i", iterator);
 446                 iter_status = init_method(
 447                         (struct bt_self_message_iterator *) iterator,
 448                         &iterator->config,
 449                         (struct bt_self_component_port_output *) upstream_port);
 450                 BT_LOGD("User method returned: status=%s",
 451                         bt_common_func_status_string(iter_status));
 452                 BT_ASSERT_POST_NO_ERROR_IF_NO_ERROR_STATUS(
 453                         "bt_message_iterator_class_initialize_method",
 454                         iter_status);
 455                 if (iter_status != BT_FUNC_STATUS_OK) {
 456                         BT_LIB_LOGW_APPEND_CAUSE(
 457                                 "Component input port message iterator initialization method failed: "
 458                                 "%![iter-]+i, status=%s",
 459                                 iterator,
 460                                 bt_common_func_status_string(iter_status));
 461                         status = iter_status;
 462                         goto error;
 463                 }
 464
 465                 iterator->config.frozen = true;
 466         }
 467
 468         if (downstream_msg_iter) {
 469                 /* Set this message iterator's downstream message iterator */
 470                 iterator->downstream_msg_iter = downstream_msg_iter;
 471
 472                 /*
 473                  * Add this message iterator to the downstream message
 474                  * iterator's array of upstream message iterators.
 475                  */
 476                 g_ptr_array_add(downstream_msg_iter->upstream_msg_iters,
 477                         iterator);
 478         }
 479
 480         set_msg_iterator_state(iterator,
 481                 BT_MESSAGE_ITERATOR_STATE_ACTIVE);
 482         g_ptr_array_add(port->connection->iterators, iterator);
 483         BT_LIB_LOGI("Created message iterator on self component input port: "
 484                 "%![up-port-]+p, %![up-comp-]+c, %![iter-]+i",
 485                 upstream_port, upstream_comp, iterator);
 486
 487         *message_iterator = iterator;
 488         status = BT_FUNC_STATUS_OK;
 489         goto end;
 490
 491 error:
 492         BT_OBJECT_PUT_REF_AND_RESET(iterator);
 493
 494 end:
 495         return status;
 496 }
 497
 498 BT_EXPORT
 499 bt_message_iterator_create_from_message_iterator_status
 500 bt_message_iterator_create_from_message_iterator(
 501                 struct bt_self_message_iterator *self_msg_iter,
 502                 struct bt_self_component_port_input *input_port,
 503                 struct bt_message_iterator **message_iterator)
 504 {
 505         BT_ASSERT_PRE_NO_ERROR();
 506         BT_ASSERT_PRE_MSG_ITER_NON_NULL(self_msg_iter);
 507         return create_self_component_input_port_message_iterator(self_msg_iter,
 508                 input_port, message_iterator, __func__);
 509 }
 510
 511 BT_EXPORT
 512 bt_message_iterator_create_from_sink_component_status
 513 bt_message_iterator_create_from_sink_component(
 514                 struct bt_self_component_sink *self_comp,
 515                 struct bt_self_component_port_input *input_port,
 516                 struct bt_message_iterator **message_iterator)
 517 {
 518         BT_ASSERT_PRE_NO_ERROR();
 519         BT_ASSERT_PRE_NON_NULL("sink-component", self_comp, "Sink component");
 520         return create_self_component_input_port_message_iterator(NULL,
 521                 input_port, message_iterator, __func__);
 522 }
 523
 524 BT_EXPORT
 525 void *bt_self_message_iterator_get_data(
 526                 const struct bt_self_message_iterator *self_iterator)
 527 {
 528         struct bt_message_iterator *iterator =
 529                 (void *) self_iterator;
 530
 531         BT_ASSERT_PRE_DEV_MSG_ITER_NON_NULL(iterator);
 532         return iterator->user_data;
 533 }
 534
 535 BT_EXPORT
 536 void bt_self_message_iterator_set_data(
 537                 struct bt_self_message_iterator *self_iterator, void *data)
 538 {
 539         struct bt_message_iterator *iterator =
 540                 (void *) self_iterator;
 541
 542         BT_ASSERT_PRE_DEV_MSG_ITER_NON_NULL(iterator);
 543         iterator->user_data = data;
 544         BT_LIB_LOGD("Set message iterator's user data: "
 545                 "%!+i, user-data-addr=%p", iterator, data);
 546 }
 547
 548 BT_EXPORT
 549 void bt_self_message_iterator_configuration_set_can_seek_forward(
 550                 bt_self_message_iterator_configuration *config,
 551                 bt_bool can_seek_forward)
 552 {
 553         BT_ASSERT_PRE_NON_NULL("message-iterator-configuration", config,
 554                 "Message iterator configuration");
 555         BT_ASSERT_PRE_DEV_HOT("message-iterator-configuration", config,
 556                 "Message iterator configuration", "");
 557
 558         config->can_seek_forward = can_seek_forward;
 559 }
 560
 561 /*
 562  * Validate that the default clock snapshot in `msg` doesn't make us go back in
 563  * time.
 564  */
 565
 566 BT_ASSERT_COND_DEV_FUNC
 567 static
 568 bool clock_snapshots_are_monotonic_one(
 569                 struct bt_message_iterator *iterator,
 570                 const bt_message *msg)
 571 {
 572         const struct bt_clock_snapshot *clock_snapshot = NULL;
 573         bt_message_type message_type = bt_message_get_type(msg);
 574         int64_t ns_from_origin;
 575         enum bt_clock_snapshot_get_ns_from_origin_status clock_snapshot_status;
 576
 577         /*
 578          * The default is true: if we can't figure out the clock snapshot
 579          * (or there is none), assume it is fine.
 580          */
 581         bool result = true;
 582
 583         switch (message_type) {
 584         case BT_MESSAGE_TYPE_EVENT:
 585         {
 586                 struct bt_message_event *event_msg = (struct bt_message_event *) msg;
 587                 clock_snapshot = event_msg->default_cs;
 588                 break;
 589         }
 590         case BT_MESSAGE_TYPE_MESSAGE_ITERATOR_INACTIVITY:
 591         {
 592                 struct bt_message_message_iterator_inactivity *inactivity_msg =
 593                         (struct bt_message_message_iterator_inactivity *) msg;
 594                 clock_snapshot = inactivity_msg->cs;
 595                 break;
 596         }
 597         case BT_MESSAGE_TYPE_PACKET_BEGINNING:
 598         case BT_MESSAGE_TYPE_PACKET_END:
 599         {
 600                 struct bt_message_packet *packet_msg = (struct bt_message_packet *) msg;
 601                 clock_snapshot = packet_msg->default_cs;
 602                 break;
 603         }
 604         case BT_MESSAGE_TYPE_STREAM_BEGINNING:
 605         case BT_MESSAGE_TYPE_STREAM_END:
 606         {
 607                 struct bt_message_stream *stream_msg = (struct bt_message_stream *) msg;
 608                 if (stream_msg->default_cs_state != BT_MESSAGE_STREAM_CLOCK_SNAPSHOT_STATE_KNOWN) {
 609                         goto end;
 610                 }
 611
 612                 clock_snapshot = stream_msg->default_cs;
 613                 break;
 614         }
 615         case BT_MESSAGE_TYPE_DISCARDED_EVENTS:
 616         case BT_MESSAGE_TYPE_DISCARDED_PACKETS:
 617         {
 618                 struct bt_message_discarded_items *discarded_msg =
 619                         (struct bt_message_discarded_items *) msg;
 620
 621                 clock_snapshot = discarded_msg->default_begin_cs;
 622                 break;
 623         }
 624         }
 625
 626         if (!clock_snapshot) {
 627                 goto end;
 628         }
 629
 630         clock_snapshot_status = bt_clock_snapshot_get_ns_from_origin(
 631                 clock_snapshot, &ns_from_origin);
 632         if (clock_snapshot_status != BT_FUNC_STATUS_OK) {
 633                 /*
 634                  * bt_clock_snapshot_get_ns_from_origin can return
 635                  * OVERFLOW_ERROR.  We don't really want to report an error to
 636                  * our caller, so just clear it.
 637                  */
 638                 bt_current_thread_clear_error();
 639                 goto end;
 640         }
 641
 642         result = ns_from_origin >= iterator->last_ns_from_origin;
 643         iterator->last_ns_from_origin = ns_from_origin;
 644 end:
 645         return result;
 646 }
 647
 648 BT_ASSERT_COND_DEV_FUNC
 649 static
 650 bool clock_snapshots_are_monotonic(
 651                 struct bt_message_iterator *iterator,
 652                 bt_message_array_const msgs, uint64_t msg_count)
 653 {
 654         uint64_t i;
 655         bool result;
 656
 657         for (i = 0; i < msg_count; i++) {
 658                 if (!clock_snapshots_are_monotonic_one(iterator, msgs[i])) {
 659                         result = false;
 660                         goto end;
 661                 }
 662         }
 663
 664         result = true;
 665
 666 end:
 667         return result;
 668 }
 669
 670 #define NEXT_METHOD_NAME        "bt_message_iterator_class_next_method"
 671
 672 #ifdef BT_DEV_MODE
 673
 674 /*
 675  * When a new stream begins, verify that the clock class tied to this
 676  * stream is compatible with what we've seen before.
 677  */
 678
 679 static
 680 void assert_post_dev_clock_classes_are_compatible_one(
 681                 struct bt_message_iterator *iterator,
 682                 const struct bt_message *msg)
 683 {
 684         enum bt_message_type message_type = bt_message_get_type(msg);
 685
 686         if (message_type == BT_MESSAGE_TYPE_STREAM_BEGINNING) {
 687                 const struct bt_message_stream *stream_msg = (struct bt_message_stream *) msg;
 688                 const struct bt_clock_class *clock_class = stream_msg->stream->class->default_clock_class;
 689                 bt_uuid clock_class_uuid = NULL;
 690
 691                 if (clock_class) {
 692                         clock_class_uuid = bt_clock_class_get_uuid(clock_class);
 693                 }
 694
 695                 switch (iterator->clock_expectation.type) {
 696                 case CLOCK_EXPECTATION_UNSET:
 697                         /*
 698                          * This is the first time we see a message with a clock
 699                          * snapshot: record the properties of that clock, against
 700                          * which we'll compare the clock properties of the following
 701                          * messages.
 702                          */
 703
 704                         if (!clock_class) {
 705                                 iterator->clock_expectation.type = CLOCK_EXPECTATION_NONE;
 706                         } else if (bt_clock_class_origin_is_unix_epoch(clock_class)) {
 707                                 iterator->clock_expectation.type = CLOCK_EXPECTATION_ORIGIN_UNIX;
 708                         } else if (clock_class_uuid) {
 709                                 iterator->clock_expectation.type = CLOCK_EXPECTATION_ORIGIN_OTHER_UUID;
 710                                 bt_uuid_copy(iterator->clock_expectation.uuid, clock_class_uuid);
 711                         } else {
 712                                 iterator->clock_expectation.type = CLOCK_EXPECTATION_ORIGIN_OTHER_NO_UUID;
 713                                 iterator->clock_expectation.clock_class = clock_class;
 714                                 bt_clock_class_get_ref(iterator->clock_expectation.clock_class);
 715                         }
 716                         break;
 717
 718                 case CLOCK_EXPECTATION_NONE:
 719                         BT_ASSERT_POST_DEV(NEXT_METHOD_NAME,
 720                                 "stream-class-has-no-clock-class", !clock_class,
 721                                 "Expecting no clock class, got one: %![cc-]+K",
 722                                 clock_class);
 723                         break;
 724
 725                 case CLOCK_EXPECTATION_ORIGIN_UNIX:
 726                         BT_ASSERT_POST_DEV(NEXT_METHOD_NAME,
 727                                 "stream-class-has-clock-class-with-unix-epoch-origin", clock_class,
 728                                 "Expecting a clock class with Unix epoch origin, got none.");
 729
 730                         BT_ASSERT_POST_DEV(NEXT_METHOD_NAME,
 731                                 "clock-class-has-unix-epoch-origin",
 732                                 bt_clock_class_origin_is_unix_epoch(clock_class),
 733                                 "Expecting a clock class with Unix epoch origin: %![cc-]+K",
 734                                 clock_class);
 735                         break;
 736
 737                 case CLOCK_EXPECTATION_ORIGIN_OTHER_UUID:
 738                         BT_ASSERT_POST_DEV(NEXT_METHOD_NAME,
 739                                 "stream-class-has-clock-class-with-uuid", clock_class,
 740                                 "Expecting a clock class with UUID, got none.");
 741
 742                         BT_ASSERT_POST_DEV(NEXT_METHOD_NAME,
 743                                 "clock-class-has-non-unix-epoch-origin",
 744                                 !bt_clock_class_origin_is_unix_epoch(clock_class),
 745                                 "Expecting a clock class without Unix epoch origin: %![cc-]+K",
 746                                 clock_class);
 747
 748                         BT_ASSERT_POST_DEV(NEXT_METHOD_NAME,
 749                                 "clock-class-has-uuid",
 750                                 clock_class_uuid,
 751                                 "Expecting a clock class with UUID, got one without UUID: %![cc-]+K",
 752                                 clock_class);
 753
 754                         BT_ASSERT_POST_DEV(NEXT_METHOD_NAME,
 755                                 "clock-class-has-expected-uuid",
 756                                 !bt_uuid_compare(iterator->clock_expectation.uuid, clock_class_uuid),
 757                                 "Expecting a clock class with UUID, got one "
 758                                         "with a different UUID: %![cc-]+K, expected-uuid=%!u",
 759                                         clock_class, iterator->clock_expectation.uuid);
 760                         break;
 761
 762                 case CLOCK_EXPECTATION_ORIGIN_OTHER_NO_UUID:
 763                         BT_ASSERT_POST_DEV(NEXT_METHOD_NAME,
 764                                 "stream-class-has-clock-class", clock_class,
 765                                 "Expecting a clock class, got none.");
 766
 767                         BT_ASSERT_POST_DEV(NEXT_METHOD_NAME,
 768                                 "clock-class-is-expected",
 769                                 clock_class == iterator->clock_expectation.clock_class,
 770                                 "Expecting clock class %![cc-]+K, got %![cc-]+K.",
 771                                 iterator->clock_expectation.clock_class,
 772                                 clock_class);
 773                         break;
 774                 }
 775         }
 776 }
 777
 778 static
 779 void assert_post_dev_clock_classes_are_compatible(
 780                 struct bt_message_iterator *iterator,
 781                 bt_message_array_const msgs, uint64_t msg_count)
 782 {
 783         uint64_t i;
 784
 785         for (i = 0; i < msg_count; i++) {
 786                 assert_post_dev_clock_classes_are_compatible_one(iterator, msgs[i]);
 787         }
 788 }
 789
 790 static
 791 const bt_stream *get_stream_from_msg(const struct bt_message *msg)
 792 {
 793         struct bt_stream *stream;
 794
 795         switch (msg->type) {
 796         case BT_MESSAGE_TYPE_STREAM_BEGINNING:
 797         case BT_MESSAGE_TYPE_STREAM_END:
 798         {
 799                 struct bt_message_stream *msg_stream =
 800                         (struct bt_message_stream *) msg;
 801                 stream = msg_stream->stream;
 802                 break;
 803         }
 804         case BT_MESSAGE_TYPE_EVENT:
 805         {
 806                 struct bt_message_event *msg_event =
 807                         (struct bt_message_event *) msg;
 808                 stream = msg_event->event->stream;
 809                 break;
 810         }
 811         case BT_MESSAGE_TYPE_PACKET_BEGINNING:
 812         case BT_MESSAGE_TYPE_PACKET_END:
 813         {
 814                 struct bt_message_packet *msg_packet =
 815                         (struct bt_message_packet *) msg;
 816                 stream = msg_packet->packet->stream;
 817                 break;
 818         }
 819         case BT_MESSAGE_TYPE_DISCARDED_EVENTS:
 820         case BT_MESSAGE_TYPE_DISCARDED_PACKETS:
 821         {
 822                 struct bt_message_discarded_items *msg_discarded =
 823                         (struct bt_message_discarded_items *) msg;
 824                 stream = msg_discarded->stream;
 825                 break;
 826         }
 827         case BT_MESSAGE_TYPE_MESSAGE_ITERATOR_INACTIVITY:
 828                 stream = NULL;
 829                 break;
 830         default:
 831                 bt_common_abort();
 832         }
 833
 834         return stream;
 835 }
 836
 837 static
 838 GString *message_types_to_string(guint msg_types)
 839 {
 840         GString *str = g_string_new("");
 841
 842         for (int msg_type = 1; msg_type <= BT_MESSAGE_TYPE_MESSAGE_ITERATOR_INACTIVITY;
 843                         msg_type <<= 1) {
 844                 if (msg_type & msg_types) {
 845                         if (str->len > 0) {
 846                                 g_string_append_c(str, '|');
 847                         }
 848
 849                         g_string_append(str,
 850                                 bt_common_message_type_string(msg_type));
 851                 }
 852         }
 853
 854         return str;
 855 }
 856
 857 static
 858 void update_expected_msg_type(const struct bt_stream *stream,
 859                 struct per_stream_state *state,
 860                 const struct bt_message *msg)
 861 {
 862         switch (msg->type) {
 863         case BT_MESSAGE_TYPE_STREAM_BEGINNING:
 864                 state->expected_msg_types = BT_MESSAGE_TYPE_STREAM_END;
 865
 866                 if (stream->class->supports_packets) {
 867                         state->expected_msg_types |=
 868                                 BT_MESSAGE_TYPE_PACKET_BEGINNING;
 869
 870                         if (stream->class->supports_discarded_packets) {
 871                                 state->expected_msg_types |=
 872                                         BT_MESSAGE_TYPE_DISCARDED_PACKETS;
 873                         }
 874                 } else {
 875                         state->expected_msg_types |= BT_MESSAGE_TYPE_EVENT;
 876                 }
 877
 878                 if (stream->class->supports_discarded_events) {
 879                         state->expected_msg_types |=
 880                                 BT_MESSAGE_TYPE_DISCARDED_EVENTS;
 881                 }
 882
 883                 break;
 884         case BT_MESSAGE_TYPE_STREAM_END:
 885                 state->expected_msg_types = 0;
 886                 break;
 887         case BT_MESSAGE_TYPE_EVENT:
 888         {
 889                 state->expected_msg_types = BT_MESSAGE_TYPE_EVENT;
 890
 891                 if (stream->class->supports_packets) {
 892                         state->expected_msg_types |= BT_MESSAGE_TYPE_PACKET_END;
 893                 } else {
 894                         state->expected_msg_types |= BT_MESSAGE_TYPE_STREAM_END;
 895                 }
 896
 897                 if (stream->class->supports_discarded_events) {
 898                         state->expected_msg_types |=
 899                                 BT_MESSAGE_TYPE_DISCARDED_EVENTS;
 900                 }
 901
 902                 break;
 903         }
 904         case BT_MESSAGE_TYPE_PACKET_BEGINNING:
 905         {
 906                 state->expected_msg_types = BT_MESSAGE_TYPE_EVENT |
 907                         BT_MESSAGE_TYPE_PACKET_END;
 908
 909                 if (stream->class->supports_discarded_events) {
 910                         state->expected_msg_types |=
 911                                 BT_MESSAGE_TYPE_DISCARDED_EVENTS;
 912                 }
 913
 914                 break;
 915         }
 916         case BT_MESSAGE_TYPE_PACKET_END:
 917         {
 918                 state->expected_msg_types = BT_MESSAGE_TYPE_PACKET_BEGINNING |
 919                         BT_MESSAGE_TYPE_STREAM_END;
 920
 921                 if (stream->class->supports_discarded_events) {
 922                         state->expected_msg_types |=
 923                                 BT_MESSAGE_TYPE_DISCARDED_EVENTS;
 924                 }
 925
 926                 if (stream->class->supports_discarded_packets) {
 927                         state->expected_msg_types |=
 928                                 BT_MESSAGE_TYPE_DISCARDED_PACKETS;
 929                 }
 930
 931                 break;
 932         }
 933         case BT_MESSAGE_TYPE_DISCARDED_EVENTS:
 934                 state->expected_msg_types = BT_MESSAGE_TYPE_DISCARDED_EVENTS;
 935
 936                 if (state->cur_packet) {
 937                         state->expected_msg_types |= BT_MESSAGE_TYPE_EVENT |
 938                                 BT_MESSAGE_TYPE_PACKET_END;
 939                 } else {
 940                         state->expected_msg_types |= BT_MESSAGE_TYPE_STREAM_END;
 941
 942                         if (stream->class->supports_packets) {
 943                                 state->expected_msg_types |=
 944                                         BT_MESSAGE_TYPE_PACKET_BEGINNING;
 945
 946                                 if (stream->class->supports_discarded_packets) {
 947                                         state->expected_msg_types |=
 948                                                 BT_MESSAGE_TYPE_DISCARDED_PACKETS;
 949                                 }
 950                         } else {
 951                                 state->expected_msg_types |=
 952                                         BT_MESSAGE_TYPE_EVENT;
 953                         }
 954                 }
 955
 956                 break;
 957         case BT_MESSAGE_TYPE_DISCARDED_PACKETS:
 958                 state->expected_msg_types = BT_MESSAGE_TYPE_DISCARDED_PACKETS |
 959                         BT_MESSAGE_TYPE_PACKET_BEGINNING |
 960                         BT_MESSAGE_TYPE_STREAM_END;
 961
 962                 if (stream->class->supports_discarded_events) {
 963                         state->expected_msg_types |=
 964                                 BT_MESSAGE_TYPE_DISCARDED_EVENTS;
 965                 }
 966                 break;
 967         default:
 968                 /*
 969                  * Other message types are not associated to a stream, so we
 970                  * should not get them here.
 971                  */
 972                 bt_common_abort();
 973         }
 974 }
 975
 976 static
 977 struct per_stream_state *get_per_stream_state(
 978                 struct bt_message_iterator *iterator,
 979                 const struct bt_stream *stream)
 980 {
 981         struct per_stream_state *state = g_hash_table_lookup(
 982                 iterator->per_stream_state, stream);
 983
 984         if (!state) {
 985                 state = g_new0(struct per_stream_state, 1);
 986                 state->expected_msg_types = BT_MESSAGE_TYPE_STREAM_BEGINNING;
 987                 g_hash_table_insert(iterator->per_stream_state,
 988                         (gpointer) stream, state);
 989         }
 990
 991         return state;
 992 }
 993
 994 static
 995 void assert_post_dev_expected_sequence(struct bt_message_iterator *iterator,
 996                 const struct bt_message *msg)
 997 {
 998         const bt_stream *stream = get_stream_from_msg(msg);
 999         struct per_stream_state *state;
1000
1001         if (!stream) {
1002                 goto end;
1003         }
1004
1005         state = get_per_stream_state(iterator, stream);
1006
1007         /*
1008          * We don't free the return value of message_types_to_string(), but
1009          * that's because we know the program is going to abort anyway, and
1010          * we don't want to call it if the assertion holds.
1011          */
1012         BT_ASSERT_POST_DEV(NEXT_METHOD_NAME,
1013                 "message-type-is-expected",
1014                 msg->type & state->expected_msg_types,
1015                 "Unexpected message type: %![stream-]s, %![iterator-]i, "
1016                 "%![message-]n, expected-msg-types=%s",
1017                 stream, iterator, msg,
1018                 message_types_to_string(state->expected_msg_types)->str);
1019
1020         update_expected_msg_type(stream, state, msg);
1021
1022 end:
1023         return;
1024 }
1025
1026 static
1027 void assert_post_dev_expected_packet(struct bt_message_iterator *iterator,
1028                 const struct bt_message *msg)
1029 {
1030         const bt_stream *stream = get_stream_from_msg(msg);
1031         struct per_stream_state *state;
1032         const bt_packet *actual_packet = NULL;
1033         const bt_packet *expected_packet = NULL;
1034
1035         if (!stream) {
1036                 goto end;
1037         }
1038
1039         state = get_per_stream_state(iterator, stream);
1040
1041         switch (msg->type) {
1042         case BT_MESSAGE_TYPE_EVENT:
1043         {
1044                 const struct bt_message_event *msg_event =
1045                         (const struct bt_message_event *) msg;
1046
1047                 actual_packet = msg_event->event->packet;
1048                 expected_packet = state->cur_packet;
1049                 break;
1050         }
1051         case BT_MESSAGE_TYPE_PACKET_BEGINNING:
1052         {
1053                 const struct bt_message_packet *msg_packet =
1054                         (const struct bt_message_packet *) msg;
1055
1056                 BT_ASSERT(!state->cur_packet);
1057                 state->cur_packet = msg_packet->packet;
1058                 break;
1059         }
1060         case BT_MESSAGE_TYPE_PACKET_END:
1061         {
1062                 const struct bt_message_packet *msg_packet =
1063                         (const struct bt_message_packet *) msg;
1064
1065                 actual_packet = msg_packet->packet;
1066                 expected_packet = state->cur_packet;
1067                 BT_ASSERT(state->cur_packet);
1068                 state->cur_packet = NULL;
1069                 break;
1070         }
1071         default:
1072                 break;
1073         }
1074
1075         BT_ASSERT_POST_DEV(NEXT_METHOD_NAME,
1076                 "message-packet-is-expected",
1077                 actual_packet == expected_packet,
1078                 "Message's packet is not expected: %![stream-]s, %![iterator-]i, "
1079                 "%![message-]n, %![received-packet-]a, %![expected-packet-]a",
1080                 stream, iterator, msg, actual_packet, expected_packet);
1081
1082 end:
1083         return;
1084 }
1085
1086 static
1087 void assert_post_dev_next(
1088                 struct bt_message_iterator *iterator,
1089                 bt_message_iterator_class_next_method_status status,
1090                 bt_message_array_const msgs, uint64_t msg_count)
1091 {
1092         if (status == BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_OK) {
1093                 uint64_t i;
1094
1095                 for (i = 0; i < msg_count; i++) {
1096                         assert_post_dev_expected_sequence(iterator, msgs[i]);
1097                         assert_post_dev_expected_packet(iterator, msgs[i]);
1098                 }
1099         } else if (status == BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_END) {
1100                 GHashTableIter iter;
1101
1102                 gpointer stream_v, stream_state_v;
1103
1104                 g_hash_table_iter_init(&iter, iterator->per_stream_state);
1105                 while (g_hash_table_iter_next(&iter, &stream_v,
1106                                 &stream_state_v)) {
1107                         struct bt_stream *stream = stream_v;
1108                         struct per_stream_state *stream_state = stream_state_v;
1109
1110                         BT_ASSERT_POST_DEV(NEXT_METHOD_NAME,
1111                                 "stream-is-ended",
1112                                 stream_state->expected_msg_types == 0,
1113                                 "Stream is not ended: %![stream-]s, "
1114                                 "%![iterator-]i, expected-msg-types=%s",
1115                                 stream, iterator,
1116                                 message_types_to_string(
1117                                         stream_state->expected_msg_types)->str);
1118                 }
1119
1120         }
1121 }
1122 #endif
1123
1124 /*
1125  * Call the `next` method of the iterator.  Do some validation on the returned
1126  * messages.
1127  */
1128
1129 static
1130 enum bt_message_iterator_class_next_method_status
1131 call_iterator_next_method(
1132                 struct bt_message_iterator *iterator,
1133                 bt_message_array_const msgs, uint64_t capacity, uint64_t *user_count)
1134 {
1135         enum bt_message_iterator_class_next_method_status status;
1136
1137         BT_ASSERT_DBG(iterator->methods.next);
1138         BT_LOGD_STR("Calling user's \"next\" method.");
1139         status = iterator->methods.next(iterator, msgs, capacity, user_count);
1140         BT_LOGD("User method returned: status=%s, msg-count=%" PRIu64,
1141                 bt_common_func_status_string(status), *user_count);
1142
1143         if (status == BT_FUNC_STATUS_OK) {
1144                 BT_IF_DEV_MODE(assert_post_dev_clock_classes_are_compatible(
1145                         iterator, msgs, *user_count));
1146
1147                 BT_ASSERT_POST_DEV(NEXT_METHOD_NAME,
1148                         "message-clock-snapshots-are-monotonic",
1149                         clock_snapshots_are_monotonic(iterator, msgs,
1150                                 *user_count),
1151                         "Clock snapshots are not monotonic");
1152         }
1153
1154 #ifdef BT_DEV_MODE
1155         assert_post_dev_next(iterator, status, msgs, *user_count);
1156 #endif
1157
1158         BT_ASSERT_POST_DEV_NO_ERROR_IF_NO_ERROR_STATUS(NEXT_METHOD_NAME,
1159                 status);
1160
1161         return status;
1162 }
1163
1164 BT_EXPORT
1165 enum bt_message_iterator_next_status
1166 bt_message_iterator_next(
1167                 struct bt_message_iterator *iterator,
1168                 bt_message_array_const *msgs, uint64_t *user_count)
1169 {
1170         enum bt_message_iterator_next_status status = BT_FUNC_STATUS_OK;
1171
1172         BT_ASSERT_PRE_DEV_NO_ERROR();
1173         BT_ASSERT_PRE_DEV_MSG_ITER_NON_NULL(iterator);
1174         BT_ASSERT_PRE_DEV_NON_NULL("message-array-output", msgs,
1175                 "Message array (output)");
1176         BT_ASSERT_PRE_DEV_NON_NULL("user-count-output", user_count,
1177                 "Message count (output)");
1178         BT_ASSERT_PRE_DEV("message-iterator-is-active",
1179                 iterator->state == BT_MESSAGE_ITERATOR_STATE_ACTIVE,
1180                 "Message iterator's \"next\" called, but "
1181                 "message iterator is in the wrong state: %!+i", iterator);
1182         BT_ASSERT_DBG(iterator->upstream_component);
1183         BT_ASSERT_DBG(iterator->upstream_component->class);
1184         BT_ASSERT_PRE_DEV("graph-is-configured",
1185                 bt_component_borrow_graph(iterator->upstream_component)->config_state !=
1186                         BT_GRAPH_CONFIGURATION_STATE_CONFIGURING,
1187                 "Graph is not configured: %!+g",
1188                 bt_component_borrow_graph(iterator->upstream_component));
1189         BT_LIB_LOGD("Getting next self component input port "
1190                 "message iterator's messages: %!+i, batch-size=%u",
1191                 iterator, MSG_BATCH_SIZE);
1192
1193         /*
1194          * Call the user's "next" method to get the next messages
1195          * and status.
1196          */
1197         *user_count = 0;
1198         status = (int) call_iterator_next_method(iterator,
1199                 (void *) iterator->msgs->pdata, MSG_BATCH_SIZE,
1200                 user_count);
1201         BT_LOGD("User method returned: status=%s, msg-count=%" PRIu64,
1202                 bt_common_func_status_string(status), *user_count);
1203         if (status < 0) {
1204                 BT_LIB_LOGW_APPEND_CAUSE(
1205                         "Component input port message iterator's \"next\" method failed: "
1206                         "%![iter-]+i, status=%s",
1207                         iterator, bt_common_func_status_string(status));
1208                 goto end;
1209         }
1210
1211         /*
1212          * There is no way that this iterator could have been finalized
1213          * during its "next" method, as the only way to do this is to
1214          * put the last iterator's reference, and this can only be done
1215          * by its downstream owner.
1216          *
1217          * For the same reason, there is no way that this iterator could
1218          * have sought (cannot seek a self message iterator).
1219          */
1220         BT_ASSERT_DBG(iterator->state ==
1221                 BT_MESSAGE_ITERATOR_STATE_ACTIVE);
1222
1223         switch (status) {
1224         case BT_FUNC_STATUS_OK:
1225                 BT_ASSERT_POST_DEV(NEXT_METHOD_NAME, "count-lteq-capacity",
1226                         *user_count <= MSG_BATCH_SIZE,
1227                         "Invalid returned message count: greater than "
1228                         "batch size: count=%" PRIu64 ", batch-size=%u",
1229                         *user_count, MSG_BATCH_SIZE);
1230                 *msgs = (void *) iterator->msgs->pdata;
1231                 break;
1232         case BT_FUNC_STATUS_AGAIN:
1233                 goto end;
1234         case BT_FUNC_STATUS_END:
1235                 set_msg_iterator_state(iterator,
1236                         BT_MESSAGE_ITERATOR_STATE_ENDED);
1237                 goto end;
1238         default:
1239                 /* Unknown non-error status */
1240                 bt_common_abort();
1241         }
1242
1243 end:
1244         return status;
1245 }
1246
1247 BT_EXPORT
1248 struct bt_component *
1249 bt_message_iterator_borrow_component(
1250                 struct bt_message_iterator *iterator)
1251 {
1252         BT_ASSERT_PRE_DEV_MSG_ITER_NON_NULL(iterator);
1253         return iterator->upstream_component;
1254 }
1255
1256 BT_EXPORT
1257 struct bt_self_component *bt_self_message_iterator_borrow_component(
1258                 struct bt_self_message_iterator *self_iterator)
1259 {
1260         struct bt_message_iterator *iterator =
1261                 (void *) self_iterator;
1262
1263         BT_ASSERT_PRE_DEV_MSG_ITER_NON_NULL(iterator);
1264         return (void *) iterator->upstream_component;
1265 }
1266
1267 BT_EXPORT
1268 struct bt_self_component_port_output *bt_self_message_iterator_borrow_port(
1269                 struct bt_self_message_iterator *self_iterator)
1270 {
1271         struct bt_message_iterator *iterator =
1272                 (void *) self_iterator;
1273
1274         BT_ASSERT_PRE_DEV_MSG_ITER_NON_NULL(iterator);
1275         return (void *) iterator->upstream_port;
1276 }
1277
1278 #define CAN_SEEK_NS_FROM_ORIGIN_METHOD_NAME                             \
1279         "bt_message_iterator_class_can_seek_ns_from_origin_method"
1280
1281 BT_EXPORT
1282 enum bt_message_iterator_can_seek_ns_from_origin_status
1283 bt_message_iterator_can_seek_ns_from_origin(
1284                 struct bt_message_iterator *iterator,
1285                 int64_t ns_from_origin, bt_bool *can_seek)
1286 {
1287         enum bt_message_iterator_can_seek_ns_from_origin_status status;
1288
1289         BT_ASSERT_PRE_NO_ERROR();
1290         BT_ASSERT_PRE_MSG_ITER_NON_NULL(iterator);
1291         BT_ASSERT_PRE_RES_OUT_NON_NULL(can_seek);
1292         BT_ASSERT_PRE_ITER_HAS_STATE_TO_SEEK(iterator);
1293         BT_ASSERT_PRE("graph-is-configured",
1294                 bt_component_borrow_graph(iterator->upstream_component)->config_state !=
1295                         BT_GRAPH_CONFIGURATION_STATE_CONFIGURING,
1296                 "Graph is not configured: %!+g",
1297                 bt_component_borrow_graph(iterator->upstream_component));
1298
1299         if (iterator->methods.can_seek_ns_from_origin) {
1300                 /*
1301                  * Initialize to an invalid value, so we can post-assert that
1302                  * the method returned a valid value.
1303                  */
1304                 *can_seek = -1;
1305
1306                 BT_LIB_LOGD("Calling user's \"can seek nanoseconds from origin\" method: %!+i",
1307                         iterator);
1308
1309                 status = (int) iterator->methods.can_seek_ns_from_origin(iterator,
1310                         ns_from_origin, can_seek);
1311
1312                 BT_ASSERT_POST_NO_ERROR_IF_NO_ERROR_STATUS(
1313                         CAN_SEEK_NS_FROM_ORIGIN_METHOD_NAME, status);
1314
1315                 if (status != BT_FUNC_STATUS_OK) {
1316                         BT_LIB_LOGW_APPEND_CAUSE(
1317                                 "Component input port message iterator's \"can seek nanoseconds from origin\" method failed: "
1318                                 "%![iter-]+i, status=%s",
1319                                 iterator, bt_common_func_status_string(status));
1320                         goto end;
1321                 }
1322
1323                 BT_ASSERT_POST(CAN_SEEK_NS_FROM_ORIGIN_METHOD_NAME,
1324                         "valid-return-value",
1325                         *can_seek == BT_TRUE || *can_seek == BT_FALSE,
1326                         "Unexpected boolean value returned from user's \"can seek ns from origin\" method: val=%d, %![iter-]+i",
1327                         *can_seek, iterator);
1328
1329                 BT_LIB_LOGD(
1330                         "User's \"can seek nanoseconds from origin\" returned successfully: "
1331                         "%![iter-]+i, can-seek=%d",
1332                         iterator, *can_seek);
1333
1334                 if (*can_seek) {
1335                         goto end;
1336                 }
1337         }
1338
1339         /*
1340          * Automatic seeking fall back: if we can seek to the beginning and the
1341          * iterator supports forward seeking then we can automatically seek to
1342          * any timestamp.
1343          */
1344         status = (int) bt_message_iterator_can_seek_beginning(
1345                 iterator, can_seek);
1346         if (status != BT_FUNC_STATUS_OK) {
1347                 goto end;
1348         }
1349
1350         *can_seek = *can_seek && iterator->config.can_seek_forward;
1351
1352 end:
1353         return status;
1354 }
1355
1356 #define CAN_SEEK_BEGINNING_METHOD_NAME                                  \
1357         "bt_message_iterator_class_can_seek_beginning"
1358
1359 BT_EXPORT
1360 enum bt_message_iterator_can_seek_beginning_status
1361 bt_message_iterator_can_seek_beginning(
1362                 struct bt_message_iterator *iterator,
1363                 bt_bool *can_seek)
1364 {
1365         enum bt_message_iterator_can_seek_beginning_status status;
1366
1367         BT_ASSERT_PRE_NO_ERROR();
1368         BT_ASSERT_PRE_MSG_ITER_NON_NULL(iterator);
1369         BT_ASSERT_PRE_RES_OUT_NON_NULL(can_seek);
1370         BT_ASSERT_PRE_ITER_HAS_STATE_TO_SEEK(iterator);
1371         BT_ASSERT_PRE("graph-is-configured",
1372                 bt_component_borrow_graph(iterator->upstream_component)->config_state !=
1373                         BT_GRAPH_CONFIGURATION_STATE_CONFIGURING,
1374                 "Graph is not configured: %!+g",
1375                 bt_component_borrow_graph(iterator->upstream_component));
1376
1377         if (iterator->methods.can_seek_beginning) {
1378                 /*
1379                  * Initialize to an invalid value, so we can post-assert that
1380                  * the method returned a valid value.
1381                  */
1382                 *can_seek = -1;
1383
1384                 status = (int) iterator->methods.can_seek_beginning(iterator, can_seek);
1385
1386                 BT_ASSERT_POST(CAN_SEEK_BEGINNING_METHOD_NAME,
1387                         "valid-return-value",
1388                         status != BT_FUNC_STATUS_OK ||
1389                                 *can_seek == BT_TRUE ||
1390                                 *can_seek == BT_FALSE,
1391                         "Unexpected boolean value returned from user's \"can seek beginning\" method: val=%d, %![iter-]+i",
1392                         *can_seek, iterator);
1393                 BT_ASSERT_POST_NO_ERROR_IF_NO_ERROR_STATUS(
1394                         CAN_SEEK_BEGINNING_METHOD_NAME, status);
1395         } else {
1396                 *can_seek = BT_FALSE;
1397                 status = BT_FUNC_STATUS_OK;
1398         }
1399
1400         return status;
1401 }
1402
1403 static inline
1404 void set_iterator_state_after_seeking(
1405                 struct bt_message_iterator *iterator,
1406                 int status)
1407 {
1408         enum bt_message_iterator_state new_state = 0;
1409
1410         /* Set iterator's state depending on seeking status */
1411         switch (status) {
1412         case BT_FUNC_STATUS_OK:
1413                 new_state = BT_MESSAGE_ITERATOR_STATE_ACTIVE;
1414                 break;
1415         case BT_FUNC_STATUS_AGAIN:
1416                 new_state = BT_MESSAGE_ITERATOR_STATE_LAST_SEEKING_RETURNED_AGAIN;
1417                 break;
1418         case BT_FUNC_STATUS_ERROR:
1419         case BT_FUNC_STATUS_MEMORY_ERROR:
1420                 new_state = BT_MESSAGE_ITERATOR_STATE_LAST_SEEKING_RETURNED_ERROR;
1421                 break;
1422         case BT_FUNC_STATUS_END:
1423                 new_state = BT_MESSAGE_ITERATOR_STATE_ENDED;
1424                 break;
1425         default:
1426                 bt_common_abort();
1427         }
1428
1429         set_msg_iterator_state(iterator, new_state);
1430 }
1431
1432 static
1433 void reset_iterator_expectations(
1434                 struct bt_message_iterator *iterator)
1435 {
1436         iterator->last_ns_from_origin = INT64_MIN;
1437         iterator->clock_expectation.type = CLOCK_EXPECTATION_UNSET;
1438 }
1439
1440 static
1441 bool message_iterator_can_seek_beginning(
1442                 struct bt_message_iterator *iterator)
1443 {
1444         enum bt_message_iterator_can_seek_beginning_status status;
1445         bt_bool can_seek;
1446
1447         status = bt_message_iterator_can_seek_beginning(
1448                 iterator, &can_seek);
1449         if (status != BT_FUNC_STATUS_OK) {
1450                 can_seek = BT_FALSE;
1451         }
1452
1453         return can_seek;
1454 }
1455
1456 #define SEEK_BEGINNING_METHOD_NAME                                      \
1457         "bt_message_iterator_class_seek_beginning_method"
1458
1459 BT_EXPORT
1460 enum bt_message_iterator_seek_beginning_status
1461 bt_message_iterator_seek_beginning(struct bt_message_iterator *iterator)
1462 {
1463         int status;
1464
1465         BT_ASSERT_PRE_NO_ERROR();
1466         BT_ASSERT_PRE_MSG_ITER_NON_NULL(iterator);
1467         BT_ASSERT_PRE_ITER_HAS_STATE_TO_SEEK(iterator);
1468         BT_ASSERT_PRE("graph-is-configured",
1469                 bt_component_borrow_graph(iterator->upstream_component)->config_state !=
1470                         BT_GRAPH_CONFIGURATION_STATE_CONFIGURING,
1471                 "Graph is not configured: %!+g",
1472                 bt_component_borrow_graph(iterator->upstream_component));
1473         BT_ASSERT_PRE("can-seek-beginning",
1474                 message_iterator_can_seek_beginning(iterator),
1475                 "Message iterator cannot seek beginning: %!+i", iterator);
1476
1477         /*
1478          * We are seeking, reset our expectations about how the following
1479          * messages should look like.
1480          */
1481         reset_iterator_expectations(iterator);
1482
1483         BT_LIB_LOGD("Calling user's \"seek beginning\" method: %!+i", iterator);
1484         set_msg_iterator_state(iterator,
1485                 BT_MESSAGE_ITERATOR_STATE_SEEKING);
1486         status = iterator->methods.seek_beginning(iterator);
1487         BT_LOGD("User method returned: status=%s",
1488                 bt_common_func_status_string(status));
1489         BT_ASSERT_POST(SEEK_BEGINNING_METHOD_NAME, "valid-status",
1490                 status == BT_FUNC_STATUS_OK ||
1491                 status == BT_FUNC_STATUS_ERROR ||
1492                 status == BT_FUNC_STATUS_MEMORY_ERROR ||
1493                 status == BT_FUNC_STATUS_AGAIN,
1494                 "Unexpected status: %![iter-]+i, status=%s",
1495                 iterator, bt_common_func_status_string(status));
1496         BT_ASSERT_POST_NO_ERROR_IF_NO_ERROR_STATUS(SEEK_BEGINNING_METHOD_NAME,
1497                 status);
1498         if (status < 0) {
1499                 BT_LIB_LOGW_APPEND_CAUSE(
1500                         "Component input port message iterator's \"seek beginning\" method failed: "
1501                         "%![iter-]+i, status=%s",
1502                         iterator, bt_common_func_status_string(status));
1503         }
1504
1505         clear_per_stream_state(iterator);
1506
1507         set_iterator_state_after_seeking(iterator, status);
1508         return status;
1509 }
1510
1511 BT_EXPORT
1512 bt_bool
1513 bt_message_iterator_can_seek_forward(
1514                 bt_message_iterator *iterator)
1515 {
1516         BT_ASSERT_PRE_MSG_ITER_NON_NULL(iterator);
1517
1518         return iterator->config.can_seek_forward;
1519 }
1520
1521 /*
1522  * Structure used to record the state of a given stream during the fast-forward
1523  * phase of an auto-seek.
1524  */
1525 struct auto_seek_stream_state {
1526         /*
1527          * Value representing which step of this timeline we are at.
1528          *
1529          *      time --->
1530          *   [SB]  1  [PB]  2  [PE]  1  [SE]
1531          *
1532          * At each point in the timeline, the messages we need to replicate are:
1533          *
1534          *   1: Stream beginning
1535          *   2: Stream beginning, packet beginning
1536          *
1537          * Before "Stream beginning" and after "Stream end", we don't need to
1538          * replicate anything as the stream doesn't exist.
1539          */
1540         enum {
1541                 AUTO_SEEK_STREAM_STATE_STREAM_BEGAN,
1542                 AUTO_SEEK_STREAM_STATE_PACKET_BEGAN,
1543         } state;
1544
1545         /*
1546          * If `state` is AUTO_SEEK_STREAM_STATE_PACKET_BEGAN, the packet we are
1547          * in.  This is a weak reference, since the packet will always be
1548          * alive by the time we use it.
1549          */
1550         struct bt_packet *packet;
1551
1552         /* Have we see a message with a clock snapshot yet? */
1553         bool seen_clock_snapshot;
1554 };
1555
1556 static
1557 struct auto_seek_stream_state *create_auto_seek_stream_state(void)
1558 {
1559         return g_new0(struct auto_seek_stream_state, 1);
1560 }
1561
1562 static
1563 void destroy_auto_seek_stream_state(void *ptr)
1564 {
1565         g_free(ptr);
1566 }
1567
1568 static
1569 GHashTable *create_auto_seek_stream_states(void)
1570 {
1571         return g_hash_table_new_full(g_direct_hash, g_direct_equal, NULL,
1572                 destroy_auto_seek_stream_state);
1573 }
1574
1575 static
1576 void destroy_auto_seek_stream_states(GHashTable *stream_states)
1577 {
1578         g_hash_table_destroy(stream_states);
1579 }
1580
1581 /*
1582  * Handle one message while we are in the fast-forward phase of an auto-seek.
1583  *
1584  * Sets `*got_first` to true if the message's timestamp is greater or equal to
1585  * `ns_from_origin`.  In other words, if this is the first message after our
1586  * seek point.
1587  *
1588  * `stream_states` is an hash table of `bt_stream *` (weak reference) to
1589  * `struct auto_seek_stream_state` used to keep the state of each stream
1590  * during the fast-forward.
1591  */
1592
1593 static inline
1594 int auto_seek_handle_message(
1595                 struct bt_message_iterator *iterator,
1596                 int64_t ns_from_origin, const struct bt_message *msg,
1597                 bool *got_first, GHashTable *stream_states)
1598 {
1599         int status = BT_FUNC_STATUS_OK;
1600         int64_t msg_ns_from_origin;
1601         const struct bt_clock_snapshot *clk_snapshot = NULL;
1602         int ret;
1603
1604         BT_ASSERT_DBG(msg);
1605         BT_ASSERT_DBG(got_first);
1606
1607         switch (msg->type) {
1608         case BT_MESSAGE_TYPE_EVENT:
1609         {
1610                 const struct bt_message_event *event_msg =
1611                         (const void *) msg;
1612
1613                 clk_snapshot = event_msg->default_cs;
1614                 BT_ASSERT_POST_DEV(NEXT_METHOD_NAME,
1615                         "event-message-has-default-clock-snapshot",
1616                         clk_snapshot,
1617                         "Event message has no default clock snapshot: %!+n",
1618                         event_msg);
1619                 break;
1620         }
1621         case BT_MESSAGE_TYPE_MESSAGE_ITERATOR_INACTIVITY:
1622         {
1623                 const struct bt_message_message_iterator_inactivity *inactivity_msg =
1624                         (const void *) msg;
1625
1626                 clk_snapshot = inactivity_msg->cs;
1627                 BT_ASSERT_DBG(clk_snapshot);
1628                 break;
1629         }
1630         case BT_MESSAGE_TYPE_PACKET_BEGINNING:
1631         case BT_MESSAGE_TYPE_PACKET_END:
1632         {
1633                 const struct bt_message_packet *packet_msg =
1634                         (const void *) msg;
1635
1636                 if (msg->type == BT_MESSAGE_TYPE_PACKET_BEGINNING
1637                                 && !packet_msg->packet->stream->class->packets_have_beginning_default_clock_snapshot) {
1638                         goto skip_msg;
1639                 }
1640
1641                 if (msg->type == BT_MESSAGE_TYPE_PACKET_END
1642                                 && !packet_msg->packet->stream->class->packets_have_end_default_clock_snapshot) {
1643                         goto skip_msg;
1644                 }
1645
1646                 clk_snapshot = packet_msg->default_cs;
1647                 BT_ASSERT_POST_DEV(NEXT_METHOD_NAME,
1648                         "packet-message-has-default-clock-snapshot",
1649                         clk_snapshot,
1650                         "Packet message has no default clock snapshot: %!+n",
1651                         packet_msg);
1652                 break;
1653         }
1654         case BT_MESSAGE_TYPE_DISCARDED_EVENTS:
1655         case BT_MESSAGE_TYPE_DISCARDED_PACKETS:
1656         {
1657                 struct bt_message_discarded_items *msg_disc_items =
1658                         (void *) msg;
1659
1660                 if (msg->type == BT_MESSAGE_TYPE_DISCARDED_EVENTS &&
1661                                 !msg_disc_items->stream->class->discarded_events_have_default_clock_snapshots) {
1662                         goto skip_msg;
1663                 }
1664
1665                 if (msg->type == BT_MESSAGE_TYPE_DISCARDED_PACKETS &&
1666                                 !msg_disc_items->stream->class->discarded_packets_have_default_clock_snapshots) {
1667                         goto skip_msg;
1668                 }
1669
1670                 BT_ASSERT_POST_DEV(NEXT_METHOD_NAME,
1671                         "discarded-events-packets-message-has-default-clock-snapshot",
1672                         msg_disc_items->default_begin_cs &&
1673                                 msg_disc_items->default_end_cs,
1674                         "Discarded events/packets message has no default clock snapshots: %!+n",
1675                         msg_disc_items);
1676                 ret = bt_clock_snapshot_get_ns_from_origin(
1677                         msg_disc_items->default_begin_cs,
1678                         &msg_ns_from_origin);
1679                 if (ret) {
1680                         status = BT_FUNC_STATUS_ERROR;
1681                         goto end;
1682                 }
1683
1684                 if (msg_ns_from_origin >= ns_from_origin) {
1685                         *got_first = true;
1686                         goto push_msg;
1687                 }
1688
1689                 ret = bt_clock_snapshot_get_ns_from_origin(
1690                         msg_disc_items->default_end_cs,
1691                         &msg_ns_from_origin);
1692                 if (ret) {
1693                         status = BT_FUNC_STATUS_ERROR;
1694                         goto end;
1695                 }
1696
1697                 if (msg_ns_from_origin >= ns_from_origin) {
1698                         /*
1699                          * The discarded items message's beginning time
1700                          * is before the requested seeking time, but its
1701                          * end time is after. Modify the message so as
1702                          * to set its beginning time to the requested
1703                          * seeking time, and make its item count unknown
1704                          * as we don't know if items were really
1705                          * discarded within the new time range.
1706                          */
1707                         uint64_t new_begin_raw_value = 0;
1708
1709                         ret = bt_clock_class_clock_value_from_ns_from_origin(
1710                                 msg_disc_items->default_end_cs->clock_class,
1711                                 ns_from_origin, &new_begin_raw_value);
1712                         if (ret) {
1713                                 status = BT_FUNC_STATUS_ERROR;
1714                                 goto end;
1715                         }
1716
1717                         bt_clock_snapshot_set_raw_value(
1718                                 msg_disc_items->default_begin_cs,
1719                                 new_begin_raw_value);
1720                         msg_disc_items->count.base.avail =
1721                                 BT_PROPERTY_AVAILABILITY_NOT_AVAILABLE;
1722
1723                         /*
1724                          * It is safe to push it because its beginning
1725                          * time is exactly the requested seeking time.
1726                          */
1727                         goto push_msg;
1728                 } else {
1729                         goto skip_msg;
1730                 }
1731         }
1732         case BT_MESSAGE_TYPE_STREAM_BEGINNING:
1733         case BT_MESSAGE_TYPE_STREAM_END:
1734         {
1735                 struct bt_message_stream *stream_msg =
1736                         (struct bt_message_stream *) msg;
1737
1738                 if (stream_msg->default_cs_state != BT_MESSAGE_STREAM_CLOCK_SNAPSHOT_STATE_KNOWN) {
1739                         /* Ignore */
1740                         goto skip_msg;
1741                 }
1742
1743                 clk_snapshot = stream_msg->default_cs;
1744                 break;
1745         }
1746         default:
1747                 bt_common_abort();
1748         }
1749
1750         BT_ASSERT_DBG(clk_snapshot);
1751         ret = bt_clock_snapshot_get_ns_from_origin(clk_snapshot,
1752                 &msg_ns_from_origin);
1753         if (ret) {
1754                 status = BT_FUNC_STATUS_ERROR;
1755                 goto end;
1756         }
1757
1758         if (msg_ns_from_origin >= ns_from_origin) {
1759                 *got_first = true;
1760                 goto push_msg;
1761         }
1762
1763 skip_msg:
1764         /* This message won't be sent downstream. */
1765         switch (msg->type) {
1766         case BT_MESSAGE_TYPE_STREAM_BEGINNING:
1767         {
1768                 const struct bt_message_stream *stream_msg = (const void *) msg;
1769                 struct auto_seek_stream_state *stream_state;
1770
1771                 /* Update stream's state: stream began. */
1772                 stream_state = create_auto_seek_stream_state();
1773                 if (!stream_state) {
1774                         status = BT_FUNC_STATUS_MEMORY_ERROR;
1775                         goto end;
1776                 }
1777
1778                 stream_state->state = AUTO_SEEK_STREAM_STATE_STREAM_BEGAN;
1779
1780                 if (stream_msg->default_cs_state == BT_MESSAGE_STREAM_CLOCK_SNAPSHOT_STATE_KNOWN) {
1781                         stream_state->seen_clock_snapshot = true;
1782                 }
1783
1784                 BT_ASSERT_DBG(!bt_g_hash_table_contains(stream_states, stream_msg->stream));
1785                 g_hash_table_insert(stream_states, stream_msg->stream, stream_state);
1786                 break;
1787         }
1788         case BT_MESSAGE_TYPE_PACKET_BEGINNING:
1789         {
1790                 const struct bt_message_packet *packet_msg =
1791                         (const void *) msg;
1792                 struct auto_seek_stream_state *stream_state;
1793
1794                 /* Update stream's state: packet began. */
1795                 stream_state = g_hash_table_lookup(stream_states, packet_msg->packet->stream);
1796                 BT_ASSERT_DBG(stream_state);
1797                 BT_ASSERT_DBG(stream_state->state == AUTO_SEEK_STREAM_STATE_STREAM_BEGAN);
1798                 stream_state->state = AUTO_SEEK_STREAM_STATE_PACKET_BEGAN;
1799                 BT_ASSERT_DBG(!stream_state->packet);
1800                 stream_state->packet = packet_msg->packet;
1801
1802                 if (packet_msg->packet->stream->class->packets_have_beginning_default_clock_snapshot) {
1803                         stream_state->seen_clock_snapshot = true;
1804                 }
1805
1806                 break;
1807         }
1808         case BT_MESSAGE_TYPE_EVENT:
1809         {
1810                 const struct bt_message_event *event_msg = (const void *) msg;
1811                 struct auto_seek_stream_state *stream_state;
1812
1813                 stream_state = g_hash_table_lookup(stream_states,
1814                         event_msg->event->stream);
1815                 BT_ASSERT_DBG(stream_state);
1816
1817                 // HELPME: are we sure that event messages have clock snapshots at this point?
1818                 stream_state->seen_clock_snapshot = true;
1819
1820                 break;
1821         }
1822         case BT_MESSAGE_TYPE_PACKET_END:
1823         {
1824                 const struct bt_message_packet *packet_msg =
1825                         (const void *) msg;
1826                 struct auto_seek_stream_state *stream_state;
1827
1828                 /* Update stream's state: packet ended. */
1829                 stream_state = g_hash_table_lookup(stream_states, packet_msg->packet->stream);
1830                 BT_ASSERT_DBG(stream_state);
1831                 BT_ASSERT_DBG(stream_state->state == AUTO_SEEK_STREAM_STATE_PACKET_BEGAN);
1832                 stream_state->state = AUTO_SEEK_STREAM_STATE_STREAM_BEGAN;
1833                 BT_ASSERT_DBG(stream_state->packet);
1834                 stream_state->packet = NULL;
1835
1836                 if (packet_msg->packet->stream->class->packets_have_end_default_clock_snapshot) {
1837                         stream_state->seen_clock_snapshot = true;
1838                 }
1839
1840                 break;
1841         }
1842         case BT_MESSAGE_TYPE_STREAM_END:
1843         {
1844                 const struct bt_message_stream *stream_msg = (const void *) msg;
1845                 struct auto_seek_stream_state *stream_state;
1846
1847                 stream_state = g_hash_table_lookup(stream_states, stream_msg->stream);
1848                 BT_ASSERT_DBG(stream_state);
1849                 BT_ASSERT_DBG(stream_state->state == AUTO_SEEK_STREAM_STATE_STREAM_BEGAN);
1850                 BT_ASSERT_DBG(!stream_state->packet);
1851
1852                 /* Update stream's state: this stream doesn't exist anymore. */
1853                 g_hash_table_remove(stream_states, stream_msg->stream);
1854                 break;
1855         }
1856         case BT_MESSAGE_TYPE_DISCARDED_EVENTS:
1857         case BT_MESSAGE_TYPE_DISCARDED_PACKETS:
1858         {
1859                 const struct bt_message_discarded_items *discarded_msg =
1860                         (const void *) msg;
1861                 struct auto_seek_stream_state *stream_state;
1862
1863                 stream_state = g_hash_table_lookup(stream_states, discarded_msg->stream);
1864                 BT_ASSERT_DBG(stream_state);
1865
1866                 if ((msg->type == BT_MESSAGE_TYPE_DISCARDED_EVENTS && discarded_msg->stream->class->discarded_events_have_default_clock_snapshots) ||
1867                         (msg->type == BT_MESSAGE_TYPE_DISCARDED_PACKETS && discarded_msg->stream->class->discarded_packets_have_default_clock_snapshots)) {
1868                         stream_state->seen_clock_snapshot = true;
1869                 }
1870
1871                 break;
1872         }
1873         default:
1874                 break;
1875         }
1876
1877         bt_object_put_ref_no_null_check(msg);
1878         msg = NULL;
1879         goto end;
1880
1881 push_msg:
1882         g_queue_push_tail(iterator->auto_seek.msgs, (void *) msg);
1883         msg = NULL;
1884
1885 end:
1886         BT_ASSERT_DBG(!msg || status != BT_FUNC_STATUS_OK);
1887         return status;
1888 }
1889
1890 static
1891 int find_message_ge_ns_from_origin(
1892                 struct bt_message_iterator *iterator,
1893                 int64_t ns_from_origin, GHashTable *stream_states)
1894 {
1895         int status = BT_FUNC_STATUS_OK;
1896         enum bt_message_iterator_state init_state =
1897                 iterator->state;
1898         const struct bt_message *messages[MSG_BATCH_SIZE];
1899         uint64_t user_count = 0;
1900         uint64_t i;
1901         bool got_first = false;
1902
1903         BT_ASSERT_DBG(iterator);
1904         memset(&messages[0], 0, sizeof(messages[0]) * MSG_BATCH_SIZE);
1905
1906         /*
1907          * Make this iterator temporarily active (not seeking) to call
1908          * the "next" method.
1909          */
1910         set_msg_iterator_state(iterator,
1911                 BT_MESSAGE_ITERATOR_STATE_ACTIVE);
1912
1913         BT_ASSERT_DBG(iterator->methods.next);
1914
1915         while (!got_first) {
1916                 /*
1917                  * Call the user's "next" method to get the next
1918                  * messages and status.
1919                  */
1920                 status = call_iterator_next_method(iterator,
1921                         &messages[0], MSG_BATCH_SIZE, &user_count);
1922                 BT_LOGD("User method returned: status=%s",
1923                         bt_common_func_status_string(status));
1924                 if (status < 0) {
1925                         BT_LIB_LOGW_APPEND_CAUSE(
1926                                 "Component input port message iterator's \"next\" method failed: "
1927                                 "%![iter-]+i, status=%s",
1928                                 iterator, bt_common_func_status_string(status));
1929                 }
1930
1931                 /*
1932                  * The user's "next" method must not do any action which
1933                  * would change the iterator's state.
1934                  */
1935                 BT_ASSERT_DBG(iterator->state ==
1936                         BT_MESSAGE_ITERATOR_STATE_ACTIVE);
1937
1938                 switch (status) {
1939                 case BT_FUNC_STATUS_OK:
1940                         BT_ASSERT_POST_DEV(NEXT_METHOD_NAME,
1941                                 "count-lteq-capacity",
1942                                 user_count <= MSG_BATCH_SIZE,
1943                                 "Invalid returned message count: greater than "
1944                                 "batch size: count=%" PRIu64 ", batch-size=%u",
1945                                 user_count, MSG_BATCH_SIZE);
1946                         break;
1947                 case BT_FUNC_STATUS_AGAIN:
1948                 case BT_FUNC_STATUS_ERROR:
1949                 case BT_FUNC_STATUS_MEMORY_ERROR:
1950                 case BT_FUNC_STATUS_END:
1951                         goto end;
1952                 default:
1953                         bt_common_abort();
1954                 }
1955
1956                 for (i = 0; i < user_count; i++) {
1957                         if (got_first) {
1958                                 g_queue_push_tail(iterator->auto_seek.msgs,
1959                                         (void *) messages[i]);
1960                                 messages[i] = NULL;
1961                                 continue;
1962                         }
1963
1964                         status = auto_seek_handle_message(iterator,
1965                                 ns_from_origin, messages[i], &got_first,
1966                                 stream_states);
1967                         if (status == BT_FUNC_STATUS_OK) {
1968                                 /* Message was either pushed or moved */
1969                                 messages[i] = NULL;
1970                         } else {
1971                                 goto end;
1972                         }
1973                 }
1974         }
1975
1976 end:
1977         for (i = 0; i < user_count; i++) {
1978                 if (messages[i]) {
1979                         bt_object_put_ref_no_null_check(messages[i]);
1980                 }
1981         }
1982
1983         set_msg_iterator_state(iterator, init_state);
1984         return status;
1985 }
1986
1987 /*
1988  * This function is installed as the iterator's next callback after we have
1989  * auto-sought (sought to the beginning and fast-forwarded) to send the
1990  * messages saved in iterator->auto_seek.msgs.  Once this is done, the original
1991  * next callback is put back.
1992  */
1993
1994 static
1995 enum bt_message_iterator_class_next_method_status post_auto_seek_next(
1996                 struct bt_message_iterator *iterator,
1997                 bt_message_array_const msgs, uint64_t capacity,
1998                 uint64_t *count)
1999 {
2000         BT_ASSERT(!g_queue_is_empty(iterator->auto_seek.msgs));
2001         *count = 0;
2002
2003         /*
2004          * Move auto-seek messages to the output array (which is this
2005          * iterator's base message array).
2006          */
2007         while (capacity > 0 && !g_queue_is_empty(iterator->auto_seek.msgs)) {
2008                 msgs[*count] = g_queue_pop_head(iterator->auto_seek.msgs);
2009                 capacity--;
2010                 (*count)++;
2011         }
2012
2013         BT_ASSERT(*count > 0);
2014
2015         if (g_queue_is_empty(iterator->auto_seek.msgs)) {
2016                 /* No more auto-seek messages, restore user's next callback. */
2017                 BT_ASSERT(iterator->auto_seek.original_next_callback);
2018                 iterator->methods.next = iterator->auto_seek.original_next_callback;
2019                 iterator->auto_seek.original_next_callback = NULL;
2020         }
2021
2022         return BT_FUNC_STATUS_OK;
2023 }
2024
2025 static inline
2026 int clock_raw_value_from_ns_from_origin(const bt_clock_class *clock_class,
2027                 int64_t ns_from_origin, uint64_t *raw_value)
2028 {
2029
2030         int64_t cc_offset_s = clock_class->offset_seconds;
2031         uint64_t cc_offset_cycles = clock_class->offset_cycles;
2032         uint64_t cc_freq = clock_class->frequency;
2033
2034         return bt_common_clock_value_from_ns_from_origin(cc_offset_s,
2035                 cc_offset_cycles, cc_freq, ns_from_origin, raw_value);
2036 }
2037
2038 static
2039 bool message_iterator_can_seek_ns_from_origin(
2040                 struct bt_message_iterator *iterator,
2041                 int64_t ns_from_origin)
2042 {
2043         enum bt_message_iterator_can_seek_ns_from_origin_status status;
2044         bt_bool can_seek;
2045
2046         status = bt_message_iterator_can_seek_ns_from_origin(
2047                 iterator, ns_from_origin, &can_seek);
2048         if (status != BT_FUNC_STATUS_OK) {
2049                 can_seek = BT_FALSE;
2050         }
2051
2052         return can_seek;
2053 }
2054
2055 #define SEEK_NS_FROM_ORIGIN_METHOD_NAME                                 \
2056         "bt_message_iterator_class_seek_ns_from_origin_method"
2057
2058
2059 BT_EXPORT
2060 enum bt_message_iterator_seek_ns_from_origin_status
2061 bt_message_iterator_seek_ns_from_origin(
2062                 struct bt_message_iterator *iterator,
2063                 int64_t ns_from_origin)
2064 {
2065         int status;
2066         GHashTable *stream_states = NULL;
2067         bt_bool can_seek_by_itself;
2068
2069         BT_ASSERT_PRE_NO_ERROR();
2070         BT_ASSERT_PRE_MSG_ITER_NON_NULL(iterator);
2071         BT_ASSERT_PRE_ITER_HAS_STATE_TO_SEEK(iterator);
2072         BT_ASSERT_PRE("graph-is-configured",
2073                 bt_component_borrow_graph(iterator->upstream_component)->config_state !=
2074                         BT_GRAPH_CONFIGURATION_STATE_CONFIGURING,
2075                 "Graph is not configured: %!+g",
2076                 bt_component_borrow_graph(iterator->upstream_component));
2077         /* The iterator must be able to seek ns from origin one way or another. */
2078         BT_ASSERT_PRE("can-seek-ns-from-origin",
2079                 message_iterator_can_seek_ns_from_origin(iterator, ns_from_origin),
2080                 "Message iterator cannot seek nanoseconds from origin: %!+i, "
2081                 "ns-from-origin=%" PRId64, iterator, ns_from_origin);
2082         set_msg_iterator_state(iterator,
2083                 BT_MESSAGE_ITERATOR_STATE_SEEKING);
2084
2085         /*
2086          * We are seeking, reset our expectations about how the following
2087          * messages should look like.
2088          */
2089         reset_iterator_expectations(iterator);
2090
2091         /* Check if the iterator can seek by itself.  If not we'll use autoseek. */
2092         if (iterator->methods.can_seek_ns_from_origin) {
2093                 bt_message_iterator_class_can_seek_ns_from_origin_method_status
2094                         can_seek_status;
2095
2096                 can_seek_status =
2097                         iterator->methods.can_seek_ns_from_origin(
2098                                 iterator, ns_from_origin, &can_seek_by_itself);
2099                 if (can_seek_status != BT_FUNC_STATUS_OK) {
2100                         status = can_seek_status;
2101                         goto end;
2102                 }
2103         } else {
2104                 can_seek_by_itself = false;
2105         }
2106
2107         if (can_seek_by_itself) {
2108                 /* The iterator knows how to seek to a particular time, let it handle this. */
2109                 BT_ASSERT(iterator->methods.seek_ns_from_origin);
2110                 BT_LIB_LOGD("Calling user's \"seek nanoseconds from origin\" method: "
2111                         "%![iter-]+i, ns=%" PRId64, iterator, ns_from_origin);
2112                 status = iterator->methods.seek_ns_from_origin(iterator,
2113                         ns_from_origin);
2114                 BT_LOGD("User method returned: status=%s",
2115                         bt_common_func_status_string(status));
2116                 BT_ASSERT_POST(SEEK_NS_FROM_ORIGIN_METHOD_NAME, "valid-status",
2117                         status == BT_FUNC_STATUS_OK ||
2118                         status == BT_FUNC_STATUS_ERROR ||
2119                         status == BT_FUNC_STATUS_MEMORY_ERROR ||
2120                         status == BT_FUNC_STATUS_AGAIN,
2121                         "Unexpected status: %![iter-]+i, status=%s",
2122                         iterator, bt_common_func_status_string(status));
2123                 BT_ASSERT_POST_NO_ERROR_IF_NO_ERROR_STATUS(
2124                         SEEK_NS_FROM_ORIGIN_METHOD_NAME, status);
2125                 if (status < 0) {
2126                         BT_LIB_LOGW_APPEND_CAUSE(
2127                                 "Component input port message iterator's \"seek nanoseconds from origin\" method failed: "
2128                                 "%![iter-]+i, status=%s",
2129                                 iterator, bt_common_func_status_string(status));
2130                 }
2131         } else {
2132                 /*
2133                  * The iterator doesn't know how to seek by itself to a
2134                  * particular time.  We will seek to the beginning and fast
2135                  * forward to the right place.
2136                  */
2137                 enum bt_message_iterator_class_can_seek_beginning_method_status can_seek_status;
2138                 bt_bool can_seek_beginning;
2139
2140                 can_seek_status = iterator->methods.can_seek_beginning(iterator,
2141                         &can_seek_beginning);
2142                 BT_ASSERT(can_seek_status == BT_FUNC_STATUS_OK);
2143                 BT_ASSERT(can_seek_beginning);
2144                 BT_ASSERT(iterator->methods.seek_beginning);
2145                 BT_LIB_LOGD("Calling user's \"seek beginning\" method: %!+i",
2146                         iterator);
2147                 status = iterator->methods.seek_beginning(iterator);
2148                 BT_LOGD("User method returned: status=%s",
2149                         bt_common_func_status_string(status));
2150                 BT_ASSERT_POST(SEEK_BEGINNING_METHOD_NAME, "valid-status",
2151                         status == BT_FUNC_STATUS_OK ||
2152                         status == BT_FUNC_STATUS_ERROR ||
2153                         status == BT_FUNC_STATUS_MEMORY_ERROR ||
2154                         status == BT_FUNC_STATUS_AGAIN,
2155                         "Unexpected status: %![iter-]+i, status=%s",
2156                         iterator, bt_common_func_status_string(status));
2157                 if (status < 0) {
2158                         BT_LIB_LOGW_APPEND_CAUSE(
2159                                 "Component input port message iterator's \"seek beginning\" method failed: "
2160                                 "%![iter-]+i, status=%s",
2161                                 iterator, bt_common_func_status_string(status));
2162                 }
2163
2164                 clear_per_stream_state(iterator);
2165
2166                 switch (status) {
2167                 case BT_FUNC_STATUS_OK:
2168                         break;
2169                 case BT_FUNC_STATUS_ERROR:
2170                 case BT_FUNC_STATUS_MEMORY_ERROR:
2171                 case BT_FUNC_STATUS_AGAIN:
2172                         goto end;
2173                 default:
2174                         bt_common_abort();
2175                 }
2176
2177                 /*
2178                  * Find the first message which has a default clock
2179                  * snapshot greater than or equal to the requested
2180                  * seeking time, and move the received messages from
2181                  * this point in the batch to this iterator's auto-seek
2182                  * message queue.
2183                  */
2184                 while (!g_queue_is_empty(iterator->auto_seek.msgs)) {
2185                         bt_object_put_ref_no_null_check(
2186                                 g_queue_pop_tail(iterator->auto_seek.msgs));
2187                 }
2188
2189                 stream_states = create_auto_seek_stream_states();
2190                 if (!stream_states) {
2191                         BT_LIB_LOGE_APPEND_CAUSE(
2192                                 "Failed to allocate one GHashTable.");
2193                         status = BT_FUNC_STATUS_MEMORY_ERROR;
2194                         goto end;
2195                 }
2196
2197                 status = find_message_ge_ns_from_origin(iterator,
2198                         ns_from_origin, stream_states);
2199                 switch (status) {
2200                 case BT_FUNC_STATUS_OK:
2201                 case BT_FUNC_STATUS_END:
2202                 {
2203                         GHashTableIter iter;
2204                         gpointer key, value;
2205
2206                         /*
2207                          * If some streams exist at the seek time, prepend the
2208                          * required messages to put those streams in the right
2209                          * state.
2210                          */
2211                         g_hash_table_iter_init(&iter, stream_states);
2212                         while (g_hash_table_iter_next (&iter, &key, &value)) {
2213                                 const bt_stream *stream = key;
2214                                 struct auto_seek_stream_state *stream_state =
2215                                         (struct auto_seek_stream_state *) value;
2216                                 bt_message *msg;
2217                                 const bt_clock_class *clock_class = bt_stream_class_borrow_default_clock_class_const(
2218                                         bt_stream_borrow_class_const(stream));
2219                                 /* Initialize to silence maybe-uninitialized warning. */
2220                                 uint64_t raw_value = 0;
2221
2222                                 /*
2223                                  * If we haven't seen a message with a clock snapshot, we don't know if our seek time is within
2224                                  * the clock's range, so it wouldn't be safe to try to convert ns_from_origin to a clock value.
2225                                  *
2226                                  * Also, it would be a bit of a lie to generate a stream begin message with the seek time as its
2227                                  * clock snapshot, because we don't really know if the stream existed at that time.  If we have
2228                                  * seen a message with a clock snapshot in our seeking, then we are sure that the
2229                                  * seek time is not below the clock range, and we know the stream was active at that
2230                                  * time (and that we cut it short).
2231                                  */
2232                                 if (stream_state->seen_clock_snapshot) {
2233                                         if (clock_raw_value_from_ns_from_origin(clock_class, ns_from_origin, &raw_value) != 0) {
2234                                                 BT_LIB_LOGW("Could not convert nanoseconds from origin to clock value: ns-from-origin=%" PRId64 ", %![cc-]+K",
2235                                                         ns_from_origin, clock_class);
2236                                                 status = BT_FUNC_STATUS_ERROR;
2237                                                 goto end;
2238                                         }
2239                                 }
2240
2241                                 switch (stream_state->state) {
2242                                 case AUTO_SEEK_STREAM_STATE_PACKET_BEGAN:
2243                                         BT_ASSERT(stream_state->packet);
2244                                         BT_LIB_LOGD("Creating packet message: %![packet-]+a", stream_state->packet);
2245
2246                                         if (stream->class->packets_have_beginning_default_clock_snapshot) {
2247                                                 /*
2248                                                  * If we are in the PACKET_BEGAN state, it means we have seen a "packet beginning"
2249                                                  * message.  If "packet beginning" packets have clock snapshots, then we must have
2250                                                  * seen a clock snapshot.
2251                                                  */
2252                                                 BT_ASSERT(stream_state->seen_clock_snapshot);
2253
2254                                                 msg = bt_message_packet_beginning_create_with_default_clock_snapshot(
2255                                                         (bt_self_message_iterator *) iterator, stream_state->packet, raw_value);
2256                                         } else {
2257                                                 msg = bt_message_packet_beginning_create((bt_self_message_iterator *) iterator,
2258                                                                 stream_state->packet);
2259                                         }
2260
2261                                         if (!msg) {
2262                                                 status = BT_FUNC_STATUS_MEMORY_ERROR;
2263                                                 goto end;
2264                                         }
2265
2266                                         g_queue_push_head(iterator->auto_seek.msgs, msg);
2267                                         msg = NULL;
2268                                         /* fall-thru */
2269
2270                                 case AUTO_SEEK_STREAM_STATE_STREAM_BEGAN:
2271                                         msg = bt_message_stream_beginning_create(
2272                                                 (bt_self_message_iterator *) iterator, stream);
2273                                         if (!msg) {
2274                                                 status = BT_FUNC_STATUS_MEMORY_ERROR;
2275                                                 goto end;
2276                                         }
2277
2278                                         if (stream_state->seen_clock_snapshot) {
2279                                                 bt_message_stream_beginning_set_default_clock_snapshot(msg, raw_value);
2280                                         }
2281
2282                                         g_queue_push_head(iterator->auto_seek.msgs, msg);
2283                                         msg = NULL;
2284                                         break;
2285                                 }
2286                         }
2287
2288                         /*
2289                          * If there are messages in the auto-seek
2290                          * message queue, replace the user's "next"
2291                          * method with a custom, temporary "next" method
2292                          * which returns them.
2293                          */
2294                         if (!g_queue_is_empty(iterator->auto_seek.msgs)) {
2295                                 BT_ASSERT(!iterator->auto_seek.original_next_callback);
2296                                 iterator->auto_seek.original_next_callback = iterator->methods.next;
2297
2298                                 iterator->methods.next =
2299                                         (bt_message_iterator_next_method)
2300                                                 post_auto_seek_next;
2301                         }
2302
2303                         /*
2304                          * `BT_FUNC_STATUS_END` becomes
2305                          * `BT_FUNC_STATUS_OK`: the next
2306                          * time this iterator's "next" method is called,
2307                          * it will return
2308                          * `BT_FUNC_STATUS_END`.
2309                          */
2310                         status = BT_FUNC_STATUS_OK;
2311                         break;
2312                 }
2313                 case BT_FUNC_STATUS_ERROR:
2314                 case BT_FUNC_STATUS_MEMORY_ERROR:
2315                 case BT_FUNC_STATUS_AGAIN:
2316                         goto end;
2317                 default:
2318                         bt_common_abort();
2319                 }
2320         }
2321
2322         clear_per_stream_state(iterator);
2323
2324         /*
2325          * The following messages returned by the next method (including
2326          * post_auto_seek_next) must be after (or at) `ns_from_origin`.
2327          */
2328         iterator->last_ns_from_origin = ns_from_origin;
2329
2330 end:
2331         if (stream_states) {
2332                 destroy_auto_seek_stream_states(stream_states);
2333                 stream_states = NULL;
2334         }
2335
2336         set_iterator_state_after_seeking(iterator, status);
2337         return status;
2338 }
2339
2340 BT_EXPORT
2341 bt_bool bt_self_message_iterator_is_interrupted(
2342                 const struct bt_self_message_iterator *self_msg_iter)
2343 {
2344         const struct bt_message_iterator *iterator =
2345                 (const void *) self_msg_iter;
2346
2347         BT_ASSERT_PRE_MSG_ITER_NON_NULL(iterator);
2348         return (bt_bool) bt_graph_is_interrupted(iterator->graph);
2349 }
2350
2351 BT_EXPORT
2352 void bt_message_iterator_get_ref(
2353                 const struct bt_message_iterator *iterator)
2354 {
2355         bt_object_get_ref(iterator);
2356 }
2357
2358 BT_EXPORT
2359 void bt_message_iterator_put_ref(
2360                 const struct bt_message_iterator *iterator)
2361 {
2362         bt_object_put_ref(iterator);
2363 }