lib/graph/iterator.c

   1 /*
   2  * Copyright 2017-2018 Philippe Proulx <pproulx@efficios.com>
   3  * Copyright 2015 Jérémie Galarneau <jeremie.galarneau@efficios.com>
   4  *
   5  * Permission is hereby granted, free of charge, to any person obtaining a copy
   6  * of this software and associated documentation files (the "Software"), to deal
   7  * in the Software without restriction, including without limitation the rights
   8  * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
   9  * copies of the Software, and to permit persons to whom the Software is
  10  * furnished to do so, subject to the following conditions:
  11  *
  12  * The above copyright notice and this permission notice shall be included in
  13  * all copies or substantial portions of the Software.
  14  *
  15  * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
  16  * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
  17  * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
  18  * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
  19  * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
  20  * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
  21  * SOFTWARE.
  22  */
  23
  24 #define BT_LOG_TAG "MSG-ITER"
  25 #include <babeltrace/lib-logging-internal.h>
  26
  27 #include <babeltrace/compiler-internal.h>
  28 #include <babeltrace/trace-ir/field.h>
  29 #include <babeltrace/trace-ir/event-const.h>
  30 #include <babeltrace/trace-ir/event-internal.h>
  31 #include <babeltrace/trace-ir/packet-const.h>
  32 #include <babeltrace/trace-ir/packet-internal.h>
  33 #include <babeltrace/trace-ir/stream-internal.h>
  34 #include <babeltrace/graph/connection-const.h>
  35 #include <babeltrace/graph/connection-internal.h>
  36 #include <babeltrace/graph/component-const.h>
  37 #include <babeltrace/graph/component-internal.h>
  38 #include <babeltrace/graph/component-source-internal.h>
  39 #include <babeltrace/graph/component-class-internal.h>
  40 #include <babeltrace/graph/component-class-sink-colander-internal.h>
  41 #include <babeltrace/graph/component-sink-const.h>
  42 #include <babeltrace/graph/component-sink-internal.h>
  43 #include <babeltrace/graph/message-const.h>
  44 #include <babeltrace/graph/message-iterator-const.h>
  45 #include <babeltrace/graph/message-iterator-internal.h>
  46 #include <babeltrace/graph/self-component-port-input-message-iterator.h>
  47 #include <babeltrace/graph/port-output-message-iterator.h>
  48 #include <babeltrace/graph/message-internal.h>
  49 #include <babeltrace/graph/message-event-const.h>
  50 #include <babeltrace/graph/message-event-internal.h>
  51 #include <babeltrace/graph/message-packet-beginning-const.h>
  52 #include <babeltrace/graph/message-packet-end-const.h>
  53 #include <babeltrace/graph/message-packet-internal.h>
  54 #include <babeltrace/graph/message-stream-beginning-const.h>
  55 #include <babeltrace/graph/message-stream-end-const.h>
  56 #include <babeltrace/graph/message-stream-internal.h>
  57 #include <babeltrace/graph/message-inactivity-internal.h>
  58 #include <babeltrace/graph/port-const.h>
  59 #include <babeltrace/graph/graph.h>
  60 #include <babeltrace/graph/graph-const.h>
  61 #include <babeltrace/graph/graph-internal.h>
  62 #include <babeltrace/types.h>
  63 #include <babeltrace/assert-internal.h>
  64 #include <babeltrace/assert-pre-internal.h>
  65 #include <stdint.h>
  66 #include <inttypes.h>
  67 #include <stdlib.h>
  68
  69 /*
  70  * TODO: Use graph's state (number of active iterators, etc.) and
  71  * possibly system specifications to make a better guess than this.
  72  */
  73 #define MSG_BATCH_SIZE  15
  74
  75 #define BT_ASSERT_PRE_ITER_HAS_STATE_TO_SEEK(_iter)                     \
  76         BT_ASSERT_PRE((_iter)->state == BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_ACTIVE || \
  77                 (_iter)->state == BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_ENDED || \
  78                 (_iter)->state == BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_LAST_SEEKING_RETURNED_AGAIN || \
  79                 (_iter)->state == BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_LAST_SEEKING_RETURNED_ERROR, \
  80                 "Message iterator is in the wrong state: %!+i", _iter)
  81
  82 static inline
  83 void _set_self_comp_port_input_msg_iterator_state(
  84                 struct bt_self_component_port_input_message_iterator *iterator,
  85                 enum bt_self_component_port_input_message_iterator_state state)
  86 {
  87         BT_ASSERT(iterator);
  88         BT_LIB_LOGD("Updating message iterator's state: new-state=%s",
  89                 bt_self_component_port_input_message_iterator_state_string(state));
  90         iterator->state = state;
  91 }
  92
  93 #ifdef BT_DEV_MODE
  94 # define set_self_comp_port_input_msg_iterator_state _set_self_comp_port_input_msg_iterator_state
  95 #else
  96 # define set_self_comp_port_input_msg_iterator_state(_a, _b)
  97 #endif
  98
  99 static
 100 void destroy_base_message_iterator(struct bt_object *obj)
 101 {
 102         struct bt_message_iterator *iterator = (void *) obj;
 103
 104         BT_ASSERT(iterator);
 105
 106         if (iterator->msgs) {
 107                 g_ptr_array_free(iterator->msgs, TRUE);
 108                 iterator->msgs = NULL;
 109         }
 110
 111         g_free(iterator);
 112 }
 113
 114 static
 115 void bt_self_component_port_input_message_iterator_destroy(struct bt_object *obj)
 116 {
 117         struct bt_self_component_port_input_message_iterator *iterator;
 118
 119         BT_ASSERT(obj);
 120
 121         /*
 122          * The message iterator's reference count is 0 if we're
 123          * here. Increment it to avoid a double-destroy (possibly
 124          * infinitely recursive). This could happen for example if the
 125          * message iterator's finalization function does
 126          * bt_object_get_ref() (or anything that causes
 127          * bt_object_get_ref() to be called) on itself (ref. count goes
 128          * from 0 to 1), and then bt_object_put_ref(): the reference
 129          * count would go from 1 to 0 again and this function would be
 130          * called again.
 131          */
 132         obj->ref_count++;
 133         iterator = (void *) obj;
 134         BT_LIB_LOGD("Destroying self component input port message iterator object: "
 135                 "%!+i", iterator);
 136         bt_self_component_port_input_message_iterator_try_finalize(iterator);
 137
 138         if (iterator->connection) {
 139                 /*
 140                  * Remove ourself from the originating connection so
 141                  * that it does not try to finalize a dangling pointer
 142                  * later.
 143                  */
 144                 bt_connection_remove_iterator(iterator->connection, iterator);
 145                 iterator->connection = NULL;
 146         }
 147
 148         if (iterator->auto_seek_msgs) {
 149                 uint64_t i;
 150
 151                 /* Put any remaining message in the auto-seek array */
 152                 for (i = 0; i < iterator->auto_seek_msgs->len; i++) {
 153                         if (iterator->auto_seek_msgs->pdata[i]) {
 154                                 bt_object_put_no_null_check(
 155                                         iterator->auto_seek_msgs->pdata[i]);
 156                         }
 157                 }
 158
 159                 g_ptr_array_free(iterator->auto_seek_msgs, TRUE);
 160                 iterator->auto_seek_msgs = NULL;
 161         }
 162
 163         destroy_base_message_iterator(obj);
 164 }
 165
 166 BT_HIDDEN
 167 void bt_self_component_port_input_message_iterator_try_finalize(
 168                 struct bt_self_component_port_input_message_iterator *iterator)
 169 {
 170         typedef void (*method_t)(void *);
 171
 172         struct bt_component_class *comp_class = NULL;
 173         method_t method = NULL;
 174
 175         BT_ASSERT(iterator);
 176
 177         switch (iterator->state) {
 178         case BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_NON_INITIALIZED:
 179                 /* Skip user finalization if user initialization failed */
 180                 BT_LIB_LOGD("Not finalizing non-initialized message iterator: "
 181                         "%!+i", iterator);
 182                 goto end;
 183         case BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_FINALIZED:
 184                 /* Already finalized */
 185                 BT_LIB_LOGD("Not finalizing message iterator: already finalized: "
 186                         "%!+i", iterator);
 187                 goto end;
 188         case BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_FINALIZING:
 189                 /* Already finalized */
 190                 BT_LIB_LOGF("Message iterator is already being finalized: "
 191                         "%!+i", iterator);
 192                 abort();
 193         default:
 194                 break;
 195         }
 196
 197         BT_LIB_LOGD("Finalizing message iterator: %!+i", iterator);
 198         set_self_comp_port_input_msg_iterator_state(iterator,
 199                 BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_FINALIZING);
 200         BT_ASSERT(iterator->upstream_component);
 201         comp_class = iterator->upstream_component->class;
 202
 203         /* Call user-defined destroy method */
 204         switch (comp_class->type) {
 205         case BT_COMPONENT_CLASS_TYPE_SOURCE:
 206         {
 207                 struct bt_component_class_source *src_comp_cls =
 208                         (void *) comp_class;
 209
 210                 method = (method_t) src_comp_cls->methods.msg_iter_finalize;
 211                 break;
 212         }
 213         case BT_COMPONENT_CLASS_TYPE_FILTER:
 214         {
 215                 struct bt_component_class_filter *flt_comp_cls =
 216                         (void *) comp_class;
 217
 218                 method = (method_t) flt_comp_cls->methods.msg_iter_finalize;
 219                 break;
 220         }
 221         default:
 222                 /* Unreachable */
 223                 abort();
 224         }
 225
 226         if (method) {
 227                 BT_LIB_LOGD("Calling user's finalization method: %!+i",
 228                         iterator);
 229                 method(iterator);
 230         }
 231
 232         iterator->upstream_component = NULL;
 233         iterator->upstream_port = NULL;
 234         set_self_comp_port_input_msg_iterator_state(iterator,
 235                 BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_FINALIZED);
 236         BT_LIB_LOGD("Finalized message iterator: %!+i", iterator);
 237
 238 end:
 239         return;
 240 }
 241
 242 BT_HIDDEN
 243 void bt_self_component_port_input_message_iterator_set_connection(
 244                 struct bt_self_component_port_input_message_iterator *iterator,
 245                 struct bt_connection *connection)
 246 {
 247         BT_ASSERT(iterator);
 248         iterator->connection = connection;
 249         BT_LIB_LOGV("Set message iterator's connection: "
 250                 "%![iter-]+i, %![conn-]+x", iterator, connection);
 251 }
 252
 253 static
 254 int init_message_iterator(struct bt_message_iterator *iterator,
 255                 enum bt_message_iterator_type type,
 256                 bt_object_release_func destroy)
 257 {
 258         int ret = 0;
 259
 260         bt_object_init_shared(&iterator->base, destroy);
 261         iterator->type = type;
 262         iterator->msgs = g_ptr_array_new();
 263         if (!iterator->msgs) {
 264                 BT_LOGE_STR("Failed to allocate a GPtrArray.");
 265                 ret = -1;
 266                 goto end;
 267         }
 268
 269         g_ptr_array_set_size(iterator->msgs, MSG_BATCH_SIZE);
 270
 271 end:
 272         return ret;
 273 }
 274
 275 static
 276 bt_bool can_seek_ns_from_origin_true(
 277                 struct bt_self_component_port_input_message_iterator *iterator,
 278                 int64_t ns_from_origin)
 279 {
 280         return BT_TRUE;
 281 }
 282
 283 static
 284 bt_bool can_seek_beginning_true(
 285                 struct bt_self_component_port_input_message_iterator *iterator)
 286 {
 287         return BT_TRUE;
 288 }
 289
 290 static
 291 struct bt_self_component_port_input_message_iterator *
 292 bt_self_component_port_input_message_iterator_create_initial(
 293                 struct bt_component *upstream_comp,
 294                 struct bt_port *upstream_port)
 295 {
 296         int ret;
 297         struct bt_self_component_port_input_message_iterator *iterator = NULL;
 298
 299         BT_ASSERT(upstream_comp);
 300         BT_ASSERT(upstream_port);
 301         BT_ASSERT(bt_port_is_connected(upstream_port));
 302         BT_LIB_LOGD("Creating initial message iterator on self component input port: "
 303                 "%![up-comp-]+c, %![up-port-]+p", upstream_comp, upstream_port);
 304         BT_ASSERT(bt_component_get_class_type(upstream_comp) ==
 305                 BT_COMPONENT_CLASS_TYPE_SOURCE ||
 306                 bt_component_get_class_type(upstream_comp) ==
 307                 BT_COMPONENT_CLASS_TYPE_FILTER);
 308         iterator = g_new0(
 309                 struct bt_self_component_port_input_message_iterator, 1);
 310         if (!iterator) {
 311                 BT_LOGE_STR("Failed to allocate one self component input port "
 312                         "message iterator.");
 313                 goto end;
 314         }
 315
 316         ret = init_message_iterator((void *) iterator,
 317                 BT_MESSAGE_ITERATOR_TYPE_SELF_COMPONENT_PORT_INPUT,
 318                 bt_self_component_port_input_message_iterator_destroy);
 319         if (ret) {
 320                 /* init_message_iterator() logs errors */
 321                 BT_OBJECT_PUT_REF_AND_RESET(iterator);
 322                 goto end;
 323         }
 324
 325         iterator->auto_seek_msgs = g_ptr_array_new();
 326         if (!iterator->auto_seek_msgs) {
 327                 BT_LOGE_STR("Failed to allocate a GPtrArray.");
 328                 ret = -1;
 329                 goto end;
 330         }
 331
 332         g_ptr_array_set_size(iterator->auto_seek_msgs, MSG_BATCH_SIZE);
 333         iterator->upstream_component = upstream_comp;
 334         iterator->upstream_port = upstream_port;
 335         iterator->connection = iterator->upstream_port->connection;
 336         iterator->graph = bt_component_borrow_graph(upstream_comp);
 337         set_self_comp_port_input_msg_iterator_state(iterator,
 338                 BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_NON_INITIALIZED);
 339
 340         switch (iterator->upstream_component->class->type) {
 341         case BT_COMPONENT_CLASS_TYPE_SOURCE:
 342         {
 343                 struct bt_component_class_source *src_comp_cls =
 344                         (void *) iterator->upstream_component->class;
 345
 346                 iterator->methods.next =
 347                         (bt_self_component_port_input_message_iterator_next_method)
 348                                 src_comp_cls->methods.msg_iter_next;
 349                 iterator->methods.seek_ns_from_origin =
 350                         (bt_self_component_port_input_message_iterator_seek_ns_from_origin_method)
 351                                 src_comp_cls->methods.msg_iter_seek_ns_from_origin;
 352                 iterator->methods.seek_beginning =
 353                         (bt_self_component_port_input_message_iterator_seek_beginning_method)
 354                                 src_comp_cls->methods.msg_iter_seek_beginning;
 355                 iterator->methods.can_seek_ns_from_origin =
 356                         (bt_self_component_port_input_message_iterator_can_seek_ns_from_origin_method)
 357                                 src_comp_cls->methods.msg_iter_can_seek_ns_from_origin;
 358                 iterator->methods.can_seek_beginning =
 359                         (bt_self_component_port_input_message_iterator_can_seek_beginning_method)
 360                                 src_comp_cls->methods.msg_iter_can_seek_beginning;
 361                 break;
 362         }
 363         case BT_COMPONENT_CLASS_TYPE_FILTER:
 364         {
 365                 struct bt_component_class_filter *flt_comp_cls =
 366                         (void *) iterator->upstream_component->class;
 367
 368                 iterator->methods.next =
 369                         (bt_self_component_port_input_message_iterator_next_method)
 370                                 flt_comp_cls->methods.msg_iter_next;
 371                 iterator->methods.seek_ns_from_origin =
 372                         (bt_self_component_port_input_message_iterator_seek_ns_from_origin_method)
 373                                 flt_comp_cls->methods.msg_iter_seek_ns_from_origin;
 374                 iterator->methods.seek_beginning =
 375                         (bt_self_component_port_input_message_iterator_seek_beginning_method)
 376                                 flt_comp_cls->methods.msg_iter_seek_beginning;
 377                 iterator->methods.can_seek_ns_from_origin =
 378                         (bt_self_component_port_input_message_iterator_can_seek_ns_from_origin_method)
 379                                 flt_comp_cls->methods.msg_iter_can_seek_ns_from_origin;
 380                 iterator->methods.can_seek_beginning =
 381                         (bt_self_component_port_input_message_iterator_can_seek_beginning_method)
 382                                 flt_comp_cls->methods.msg_iter_can_seek_beginning;
 383                 break;
 384         }
 385         default:
 386                 abort();
 387         }
 388
 389         if (iterator->methods.seek_ns_from_origin &&
 390                         !iterator->methods.can_seek_ns_from_origin) {
 391                 iterator->methods.can_seek_ns_from_origin =
 392                         (bt_self_component_port_input_message_iterator_can_seek_ns_from_origin_method)
 393                                 can_seek_ns_from_origin_true;
 394         }
 395
 396         if (iterator->methods.seek_beginning &&
 397                         !iterator->methods.can_seek_beginning) {
 398                 iterator->methods.can_seek_beginning =
 399                         (bt_self_component_port_input_message_iterator_seek_beginning_method)
 400                                 can_seek_beginning_true;
 401         }
 402
 403         BT_LIB_LOGD("Created initial message iterator on self component input port: "
 404                 "%![up-port-]+p, %![up-comp-]+c, %![iter-]+i",
 405                 upstream_port, upstream_comp, iterator);
 406
 407 end:
 408         return iterator;
 409 }
 410
 411 struct bt_self_component_port_input_message_iterator *
 412 bt_self_component_port_input_message_iterator_create(
 413                 struct bt_self_component_port_input *self_port)
 414 {
 415         typedef enum bt_self_message_iterator_status (*init_method_t)(
 416                         void *, void *, void *);
 417
 418         init_method_t init_method = NULL;
 419         struct bt_self_component_port_input_message_iterator *iterator =
 420                 NULL;
 421         struct bt_port *port = (void *) self_port;
 422         struct bt_port *upstream_port;
 423         struct bt_component *comp;
 424         struct bt_component *upstream_comp;
 425         struct bt_component_class *upstream_comp_cls;
 426
 427         BT_ASSERT_PRE_NON_NULL(port, "Port");
 428         comp = bt_port_borrow_component_inline(port);
 429         BT_ASSERT_PRE(bt_port_is_connected(port),
 430                 "Port is not connected: %![port-]+p", port);
 431         BT_ASSERT_PRE(comp, "Port is not part of a component: %![port-]+p",
 432                 port);
 433         BT_ASSERT_PRE(!bt_component_graph_is_canceled(comp),
 434                 "Port's component's graph is canceled: "
 435                 "%![port-]+p, %![comp-]+c", port, comp);
 436         BT_ASSERT(port->connection);
 437         upstream_port = port->connection->upstream_port;
 438         BT_ASSERT(upstream_port);
 439         upstream_comp = bt_port_borrow_component_inline(upstream_port);
 440         BT_ASSERT(upstream_comp);
 441         upstream_comp_cls = upstream_comp->class;
 442         BT_ASSERT(upstream_comp->class->type ==
 443                 BT_COMPONENT_CLASS_TYPE_SOURCE ||
 444                 upstream_comp->class->type ==
 445                 BT_COMPONENT_CLASS_TYPE_FILTER);
 446         iterator = bt_self_component_port_input_message_iterator_create_initial(
 447                 upstream_comp, upstream_port);
 448         if (!iterator) {
 449                 BT_LOGW_STR("Cannot create self component input port "
 450                         "message iterator.");
 451                 goto end;
 452         }
 453
 454         switch (upstream_comp_cls->type) {
 455         case BT_COMPONENT_CLASS_TYPE_SOURCE:
 456         {
 457                 struct bt_component_class_source *src_comp_cls =
 458                         (void *) upstream_comp_cls;
 459
 460                 init_method =
 461                         (init_method_t) src_comp_cls->methods.msg_iter_init;
 462                 break;
 463         }
 464         case BT_COMPONENT_CLASS_TYPE_FILTER:
 465         {
 466                 struct bt_component_class_filter *flt_comp_cls =
 467                         (void *) upstream_comp_cls;
 468
 469                 init_method =
 470                         (init_method_t) flt_comp_cls->methods.msg_iter_init;
 471                 break;
 472         }
 473         default:
 474                 /* Unreachable */
 475                 abort();
 476         }
 477
 478         if (init_method) {
 479                 int iter_status;
 480
 481                 BT_LIB_LOGD("Calling user's initialization method: %!+i", iterator);
 482                 iter_status = init_method(iterator, upstream_comp,
 483                         upstream_port);
 484                 BT_LOGD("User method returned: status=%s",
 485                         bt_message_iterator_status_string(iter_status));
 486                 if (iter_status != BT_MESSAGE_ITERATOR_STATUS_OK) {
 487                         BT_LOGW_STR("Initialization method failed.");
 488                         BT_OBJECT_PUT_REF_AND_RESET(iterator);
 489                         goto end;
 490                 }
 491         }
 492
 493         set_self_comp_port_input_msg_iterator_state(iterator,
 494                 BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_ACTIVE);
 495         g_ptr_array_add(port->connection->iterators, iterator);
 496         BT_LIB_LOGD("Created message iterator on self component input port: "
 497                 "%![up-port-]+p, %![up-comp-]+c, %![iter-]+i",
 498                 upstream_port, upstream_comp, iterator);
 499
 500 end:
 501         return iterator;
 502 }
 503
 504 void *bt_self_message_iterator_get_data(
 505                 const struct bt_self_message_iterator *self_iterator)
 506 {
 507         struct bt_self_component_port_input_message_iterator *iterator =
 508                 (void *) self_iterator;
 509
 510         BT_ASSERT_PRE_NON_NULL(iterator, "Message iterator");
 511         return iterator->user_data;
 512 }
 513
 514 void bt_self_message_iterator_set_data(
 515                 struct bt_self_message_iterator *self_iterator, void *data)
 516 {
 517         struct bt_self_component_port_input_message_iterator *iterator =
 518                 (void *) self_iterator;
 519
 520         BT_ASSERT_PRE_NON_NULL(iterator, "Message iterator");
 521         iterator->user_data = data;
 522         BT_LIB_LOGV("Set message iterator's user data: "
 523                 "%!+i, user-data-addr=%p", iterator, data);
 524 }
 525
 526 BT_ASSERT_PRE_FUNC
 527 static inline
 528 void bt_message_borrow_packet_stream(const struct bt_message *msg,
 529                 const struct bt_stream **stream,
 530                 const struct bt_packet **packet)
 531 {
 532         BT_ASSERT(msg);
 533
 534         switch (msg->type) {
 535         case BT_MESSAGE_TYPE_EVENT:
 536                 *packet = bt_event_borrow_packet_const(
 537                         bt_message_event_borrow_event_const(msg));
 538                 *stream = bt_packet_borrow_stream_const(*packet);
 539                 break;
 540         case BT_MESSAGE_TYPE_STREAM_BEGINNING:
 541                 *stream = bt_message_stream_beginning_borrow_stream_const(msg);
 542                 break;
 543         case BT_MESSAGE_TYPE_STREAM_END:
 544                 *stream = bt_message_stream_end_borrow_stream_const(msg);
 545                 break;
 546         case BT_MESSAGE_TYPE_PACKET_BEGINNING:
 547                 *packet = bt_message_packet_beginning_borrow_packet_const(msg);
 548                 *stream = bt_packet_borrow_stream_const(*packet);
 549                 break;
 550         case BT_MESSAGE_TYPE_PACKET_END:
 551                 *packet = bt_message_packet_end_borrow_packet_const(msg);
 552                 *stream = bt_packet_borrow_stream_const(*packet);
 553                 break;
 554         default:
 555                 break;
 556         }
 557 }
 558
 559 enum bt_message_iterator_status
 560 bt_self_component_port_input_message_iterator_next(
 561                 struct bt_self_component_port_input_message_iterator *iterator,
 562                 bt_message_array_const *msgs, uint64_t *user_count)
 563 {
 564         int status = BT_MESSAGE_ITERATOR_STATUS_OK;
 565
 566         BT_ASSERT_PRE_NON_NULL(iterator, "Message iterator");
 567         BT_ASSERT_PRE_NON_NULL(msgs, "Message array (output)");
 568         BT_ASSERT_PRE_NON_NULL(user_count, "Message count (output)");
 569         BT_ASSERT_PRE(iterator->state ==
 570                 BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_ACTIVE,
 571                 "Message iterator's \"next\" called, but "
 572                 "message iterator is in the wrong state: %!+i", iterator);
 573         BT_ASSERT(iterator->upstream_component);
 574         BT_ASSERT(iterator->upstream_component->class);
 575         BT_ASSERT_PRE(bt_component_borrow_graph(iterator->upstream_component)->is_configured,
 576                 "Graph is not configured: %!+g",
 577                 bt_component_borrow_graph(iterator->upstream_component));
 578         BT_LIB_LOGD("Getting next self component input port "
 579                 "message iterator's messages: %!+i", iterator);
 580
 581         /*
 582          * Call the user's "next" method to get the next messages
 583          * and status.
 584          */
 585         BT_ASSERT(iterator->methods.next);
 586         BT_LOGD_STR("Calling user's \"next\" method.");
 587         status = iterator->methods.next(iterator,
 588                 (void *) iterator->base.msgs->pdata, MSG_BATCH_SIZE,
 589                 user_count);
 590         BT_LOGD("User method returned: status=%s",
 591                 bt_message_iterator_status_string(status));
 592         if (status < 0) {
 593                 BT_LOGW_STR("User method failed.");
 594                 goto end;
 595         }
 596
 597 #ifdef BT_DEV_MODE
 598         /*
 599          * There is no way that this iterator could have been finalized
 600          * during its "next" method, as the only way to do this is to
 601          * put the last iterator's reference, and this can only be done
 602          * by its downstream owner.
 603          *
 604          * For the same reason, there is no way that this iterator could
 605          * have seeked (cannot seek a self message iterator).
 606          */
 607         BT_ASSERT(iterator->state ==
 608                 BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_ACTIVE);
 609 #endif
 610
 611         switch (status) {
 612         case BT_MESSAGE_ITERATOR_STATUS_OK:
 613                 BT_ASSERT_PRE(*user_count <= MSG_BATCH_SIZE,
 614                         "Invalid returned message count: greater than "
 615                         "batch size: count=%" PRIu64 ", batch-size=%u",
 616                         *user_count, MSG_BATCH_SIZE);
 617                 *msgs = (void *) iterator->base.msgs->pdata;
 618                 break;
 619         case BT_MESSAGE_ITERATOR_STATUS_AGAIN:
 620                 goto end;
 621         case BT_MESSAGE_ITERATOR_STATUS_END:
 622                 set_self_comp_port_input_msg_iterator_state(iterator,
 623                         BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_ENDED);
 624                 goto end;
 625         default:
 626                 /* Unknown non-error status */
 627                 abort();
 628         }
 629
 630 end:
 631         return status;
 632 }
 633
 634 enum bt_message_iterator_status bt_port_output_message_iterator_next(
 635                 struct bt_port_output_message_iterator *iterator,
 636                 bt_message_array_const *msgs_to_user,
 637                 uint64_t *count_to_user)
 638 {
 639         enum bt_message_iterator_status status;
 640         enum bt_graph_status graph_status;
 641
 642         BT_ASSERT_PRE_NON_NULL(iterator, "Message iterator");
 643         BT_ASSERT_PRE_NON_NULL(msgs_to_user, "Message array (output)");
 644         BT_ASSERT_PRE_NON_NULL(count_to_user, "Message count (output)");
 645         BT_LIB_LOGD("Getting next output port message iterator's messages: "
 646                 "%!+i", iterator);
 647
 648         /*
 649          * As soon as the user calls this function, we mark the graph as
 650          * being definitely configured.
 651          */
 652         bt_graph_set_is_configured(iterator->graph, true);
 653
 654         graph_status = bt_graph_consume_sink_no_check(iterator->graph,
 655                 iterator->colander);
 656         switch (graph_status) {
 657         case BT_GRAPH_STATUS_CANCELED:
 658         case BT_GRAPH_STATUS_AGAIN:
 659         case BT_GRAPH_STATUS_END:
 660         case BT_GRAPH_STATUS_NOMEM:
 661                 status = (int) graph_status;
 662                 break;
 663         case BT_GRAPH_STATUS_OK:
 664                 status = BT_MESSAGE_ITERATOR_STATUS_OK;
 665
 666                 /*
 667                  * On success, the colander sink moves the messages
 668                  * to this iterator's array and sets this iterator's
 669                  * message count: move them to the user.
 670                  */
 671                 *msgs_to_user = (void *) iterator->base.msgs->pdata;
 672                 *count_to_user = iterator->count;
 673                 break;
 674         default:
 675                 /* Other errors */
 676                 status = BT_MESSAGE_ITERATOR_STATUS_ERROR;
 677         }
 678
 679         return status;
 680 }
 681
 682 struct bt_component *
 683 bt_self_component_port_input_message_iterator_borrow_component(
 684                 struct bt_self_component_port_input_message_iterator *iterator)
 685 {
 686         BT_ASSERT_PRE_NON_NULL(iterator, "Message iterator");
 687         return iterator->upstream_component;
 688 }
 689
 690 const struct bt_component *
 691 bt_self_component_port_input_message_iterator_borrow_component_const(
 692                 const struct bt_self_component_port_input_message_iterator *iterator)
 693 {
 694         BT_ASSERT_PRE_NON_NULL(iterator, "Message iterator");
 695         return iterator->upstream_component;
 696 }
 697
 698 struct bt_self_component *bt_self_message_iterator_borrow_component(
 699                 struct bt_self_message_iterator *self_iterator)
 700 {
 701         struct bt_self_component_port_input_message_iterator *iterator =
 702                 (void *) self_iterator;
 703
 704         BT_ASSERT_PRE_NON_NULL(iterator, "Message iterator");
 705         return (void *) iterator->upstream_component;
 706 }
 707
 708 struct bt_self_port_output *bt_self_message_iterator_borrow_port(
 709                 struct bt_self_message_iterator *self_iterator)
 710 {
 711         struct bt_self_component_port_input_message_iterator *iterator =
 712                 (void *) self_iterator;
 713
 714         BT_ASSERT_PRE_NON_NULL(iterator, "Message iterator");
 715         return (void *) iterator->upstream_port;
 716 }
 717
 718 static
 719 void bt_port_output_message_iterator_destroy(struct bt_object *obj)
 720 {
 721         struct bt_port_output_message_iterator *iterator = (void *) obj;
 722
 723         BT_LIB_LOGD("Destroying output port message iterator object: %!+i",
 724                 iterator);
 725         BT_LOGD_STR("Putting graph.");
 726         BT_OBJECT_PUT_REF_AND_RESET(iterator->graph);
 727         BT_LOGD_STR("Putting colander sink component.");
 728         BT_OBJECT_PUT_REF_AND_RESET(iterator->colander);
 729         destroy_base_message_iterator(obj);
 730 }
 731
 732 struct bt_port_output_message_iterator *
 733 bt_port_output_message_iterator_create(struct bt_graph *graph,
 734                 const struct bt_port_output *output_port)
 735 {
 736         struct bt_port_output_message_iterator *iterator = NULL;
 737         struct bt_component_class_sink *colander_comp_cls = NULL;
 738         struct bt_component *output_port_comp = NULL;
 739         struct bt_component_sink *colander_comp;
 740         enum bt_graph_status graph_status;
 741         struct bt_port_input *colander_in_port = NULL;
 742         struct bt_component_class_sink_colander_data colander_data;
 743         int ret;
 744
 745         BT_ASSERT_PRE_NON_NULL(graph, "Graph");
 746         BT_ASSERT_PRE_NON_NULL(output_port, "Output port");
 747         output_port_comp = bt_port_borrow_component_inline(
 748                 (const void *) output_port);
 749         BT_ASSERT_PRE(output_port_comp,
 750                 "Output port has no component: %!+p", output_port);
 751         BT_ASSERT_PRE(bt_component_borrow_graph(output_port_comp) ==
 752                 (void *) graph,
 753                 "Output port is not part of graph: %![graph-]+g, %![port-]+p",
 754                 graph, output_port);
 755
 756         /* Create message iterator */
 757         BT_LIB_LOGD("Creating message iterator on output port: "
 758                 "%![port-]+p, %![comp-]+c", output_port, output_port_comp);
 759         iterator = g_new0(struct bt_port_output_message_iterator, 1);
 760         if (!iterator) {
 761                 BT_LOGE_STR("Failed to allocate one output port message iterator.");
 762                 goto error;
 763         }
 764
 765         ret = init_message_iterator((void *) iterator,
 766                 BT_MESSAGE_ITERATOR_TYPE_PORT_OUTPUT,
 767                 bt_port_output_message_iterator_destroy);
 768         if (ret) {
 769                 /* init_message_iterator() logs errors */
 770                 BT_OBJECT_PUT_REF_AND_RESET(iterator);
 771                 goto end;
 772         }
 773
 774         /* Create colander component */
 775         colander_comp_cls = bt_component_class_sink_colander_get();
 776         if (!colander_comp_cls) {
 777                 BT_LOGW("Cannot get colander sink component class.");
 778                 goto error;
 779         }
 780
 781         iterator->graph = graph;
 782         bt_object_get_no_null_check(iterator->graph);
 783         colander_data.msgs = (void *) iterator->base.msgs->pdata;
 784         colander_data.count_addr = &iterator->count;
 785
 786         /* Hope that nobody uses this very unique name */
 787         graph_status =
 788                 bt_graph_add_sink_component_with_init_method_data(
 789                         (void *) graph, colander_comp_cls,
 790                         "colander-36ac3409-b1a8-4d60-ab1f-4fdf341a8fb1",
 791                         NULL, &colander_data, (void *) &iterator->colander);
 792         if (graph_status != BT_GRAPH_STATUS_OK) {
 793                 BT_LIB_LOGW("Cannot add colander sink component to graph: "
 794                         "%1[graph-]+g, status=%s", graph,
 795                         bt_graph_status_string(graph_status));
 796                 goto error;
 797         }
 798
 799         /*
 800          * Connect provided output port to the colander component's
 801          * input port.
 802          */
 803         colander_in_port =
 804                 (void *) bt_component_sink_borrow_input_port_by_index_const(
 805                         (void *) iterator->colander, 0);
 806         BT_ASSERT(colander_in_port);
 807         graph_status = bt_graph_connect_ports(graph,
 808                 output_port, colander_in_port, NULL);
 809         if (graph_status != BT_GRAPH_STATUS_OK) {
 810                 BT_LIB_LOGW("Cannot add colander sink component to graph: "
 811                         "%![graph-]+g, %![comp-]+c, status=%s", graph,
 812                         iterator->colander,
 813                         bt_graph_status_string(graph_status));
 814                 goto error;
 815         }
 816
 817         /*
 818          * At this point everything went fine. Make the graph
 819          * nonconsumable forever so that only this message iterator
 820          * can consume (thanks to bt_graph_consume_sink_no_check()).
 821          * This avoids leaking the message created by the colander
 822          * sink and moved to the message iterator's message
 823          * member.
 824          */
 825         bt_graph_set_can_consume(iterator->graph, false);
 826         goto end;
 827
 828 error:
 829         if (iterator && iterator->graph && iterator->colander) {
 830                 int ret;
 831
 832                 /* Remove created colander component from graph if any */
 833                 colander_comp = iterator->colander;
 834                 BT_OBJECT_PUT_REF_AND_RESET(iterator->colander);
 835
 836                 /*
 837                  * At this point the colander component's reference
 838                  * count is 0 because iterator->colander was the only
 839                  * owner. We also know that it is not connected because
 840                  * this is the last operation before this function
 841                  * succeeds.
 842                  *
 843                  * Since we honor the preconditions here,
 844                  * bt_graph_remove_unconnected_component() always
 845                  * succeeds.
 846                  */
 847                 ret = bt_graph_remove_unconnected_component(iterator->graph,
 848                         (void *) colander_comp);
 849                 BT_ASSERT(ret == 0);
 850         }
 851
 852         BT_OBJECT_PUT_REF_AND_RESET(iterator);
 853
 854 end:
 855         bt_object_put_ref(colander_comp_cls);
 856         return (void *) iterator;
 857 }
 858
 859 bt_bool bt_self_component_port_input_message_iterator_can_seek_ns_from_origin(
 860                 struct bt_self_component_port_input_message_iterator *iterator,
 861                 int64_t ns_from_origin)
 862 {
 863         bt_bool can = BT_FALSE;
 864
 865         BT_ASSERT_PRE_NON_NULL(iterator, "Message iterator");
 866         BT_ASSERT_PRE_ITER_HAS_STATE_TO_SEEK(iterator);
 867         BT_ASSERT_PRE(bt_component_borrow_graph(iterator->upstream_component)->is_configured,
 868                 "Graph is not configured: %!+g",
 869                 bt_component_borrow_graph(iterator->upstream_component));
 870
 871         if (iterator->methods.can_seek_ns_from_origin) {
 872                 can = iterator->methods.can_seek_ns_from_origin(iterator,
 873                         ns_from_origin);
 874                 goto end;
 875         }
 876
 877         /*
 878          * Automatic seeking fall back: if we can seek to the beginning,
 879          * then we can automatically seek to any message.
 880          */
 881         if (iterator->methods.can_seek_beginning) {
 882                 can = iterator->methods.can_seek_beginning(iterator);
 883         }
 884
 885 end:
 886         return can;
 887 }
 888
 889 bt_bool bt_self_component_port_input_message_iterator_can_seek_beginning(
 890                 struct bt_self_component_port_input_message_iterator *iterator)
 891 {
 892         bt_bool can = BT_FALSE;
 893
 894         BT_ASSERT_PRE_NON_NULL(iterator, "Message iterator");
 895         BT_ASSERT_PRE_ITER_HAS_STATE_TO_SEEK(iterator);
 896         BT_ASSERT_PRE(bt_component_borrow_graph(iterator->upstream_component)->is_configured,
 897                 "Graph is not configured: %!+g",
 898                 bt_component_borrow_graph(iterator->upstream_component));
 899
 900         if (iterator->methods.can_seek_beginning) {
 901                 can = iterator->methods.can_seek_beginning(iterator);
 902         }
 903
 904         return can;
 905 }
 906
 907 static inline
 908 void _set_iterator_state_after_seeking(
 909                 struct bt_self_component_port_input_message_iterator *iterator,
 910                 enum bt_message_iterator_status status)
 911 {
 912         enum bt_self_component_port_input_message_iterator_state new_state = 0;
 913
 914         /* Set iterator's state depending on seeking status */
 915         switch (status) {
 916         case BT_MESSAGE_ITERATOR_STATUS_OK:
 917                 new_state = BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_ACTIVE;
 918                 break;
 919         case BT_MESSAGE_ITERATOR_STATUS_AGAIN:
 920                 new_state = BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_LAST_SEEKING_RETURNED_AGAIN;
 921                 break;
 922         case BT_MESSAGE_ITERATOR_STATUS_ERROR:
 923         case BT_MESSAGE_ITERATOR_STATUS_NOMEM:
 924                 new_state = BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_LAST_SEEKING_RETURNED_ERROR;
 925                 break;
 926         case BT_MESSAGE_ITERATOR_STATUS_END:
 927                 new_state = BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_ENDED;
 928                 break;
 929         default:
 930                 abort();
 931         }
 932
 933         set_self_comp_port_input_msg_iterator_state(iterator, new_state);
 934 }
 935
 936 #ifdef BT_DEV_MODE
 937 # define set_iterator_state_after_seeking       _set_iterator_state_after_seeking
 938 #else
 939 # define set_iterator_state_after_seeking(_iter, _status)
 940 #endif
 941
 942 enum bt_message_iterator_status
 943 bt_self_component_port_input_message_iterator_seek_beginning(
 944                 struct bt_self_component_port_input_message_iterator *iterator)
 945 {
 946         int status;
 947
 948         BT_ASSERT_PRE_NON_NULL(iterator, "Message iterator");
 949         BT_ASSERT_PRE_ITER_HAS_STATE_TO_SEEK(iterator);
 950         BT_ASSERT_PRE(bt_component_borrow_graph(iterator->upstream_component)->is_configured,
 951                 "Graph is not configured: %!+g",
 952                 bt_component_borrow_graph(iterator->upstream_component));
 953         BT_ASSERT_PRE(
 954                 bt_self_component_port_input_message_iterator_can_seek_beginning(
 955                         iterator),
 956                 "Message iterator cannot seek beginning: %!+i", iterator);
 957         BT_LIB_LOGD("Calling user's \"seek beginning\" method: %!+i", iterator);
 958         set_self_comp_port_input_msg_iterator_state(iterator,
 959                 BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_SEEKING);
 960         status = iterator->methods.seek_beginning(iterator);
 961         BT_LOGD("User method returned: status=%s",
 962                 bt_message_iterator_status_string(status));
 963         BT_ASSERT_PRE(status == BT_MESSAGE_ITERATOR_STATUS_OK ||
 964                 status == BT_MESSAGE_ITERATOR_STATUS_ERROR ||
 965                 status == BT_MESSAGE_ITERATOR_STATUS_NOMEM ||
 966                 status == BT_MESSAGE_ITERATOR_STATUS_AGAIN,
 967                 "Unexpected status: %![iter-]+i, status=%s",
 968                 iterator, bt_self_message_iterator_status_string(status));
 969         set_iterator_state_after_seeking(iterator, status);
 970         return status;
 971 }
 972
 973 static inline
 974 int get_message_ns_from_origin(const struct bt_message *msg,
 975                 int64_t *ns_from_origin, bool *ignore)
 976 {
 977         const struct bt_clock_snapshot *clk_snapshot = NULL;
 978         int ret = 0;
 979
 980         switch (msg->type) {
 981         case BT_MESSAGE_TYPE_EVENT:
 982         {
 983                 const struct bt_message_event *event_msg =
 984                         (const void *) msg;
 985
 986                 clk_snapshot = event_msg->event->default_cs;
 987                 BT_ASSERT_PRE(clk_snapshot,
 988                         "Event has no default clock snapshot: %!+e",
 989                         event_msg->event);
 990                 break;
 991         }
 992         case BT_MESSAGE_TYPE_INACTIVITY:
 993         {
 994                 const struct bt_message_inactivity *inactivity_msg =
 995                         (const void *) msg;
 996
 997                 BT_ASSERT(inactivity_msg->default_cs);
 998                 clk_snapshot = inactivity_msg->default_cs;
 999                 break;
1000         }
1001         case BT_MESSAGE_TYPE_STREAM_BEGINNING:
1002         case BT_MESSAGE_TYPE_STREAM_END:
1003                 /* Ignore */
1004                 goto end;
1005         case BT_MESSAGE_TYPE_PACKET_BEGINNING:
1006         {
1007                 const struct bt_message_packet *pkt_msg =
1008                         (const void *) msg;
1009
1010                 clk_snapshot = pkt_msg->packet->default_beginning_cs;
1011                 break;
1012         }
1013         case BT_MESSAGE_TYPE_PACKET_END:
1014         {
1015                 const struct bt_message_packet *pkt_msg =
1016                         (const void *) msg;
1017
1018                 clk_snapshot = pkt_msg->packet->default_end_cs;
1019                 break;
1020         }
1021         default:
1022                 abort();
1023         }
1024
1025         if (!clk_snapshot) {
1026                 *ignore = true;
1027                 goto end;
1028         }
1029
1030         ret = bt_clock_snapshot_get_ns_from_origin(clk_snapshot,
1031                 ns_from_origin);
1032
1033 end:
1034         return ret;
1035 }
1036
1037 static
1038 enum bt_message_iterator_status find_message_ge_ns_from_origin(
1039                 struct bt_self_component_port_input_message_iterator *iterator,
1040                 int64_t ns_from_origin)
1041 {
1042         int status;
1043         enum bt_self_component_port_input_message_iterator_state init_state =
1044                 iterator->state;
1045         const struct bt_message *messages[MSG_BATCH_SIZE];
1046         uint64_t user_count = 0;
1047         uint64_t i;
1048
1049         BT_ASSERT(iterator);
1050         memset(&messages[0], 0, sizeof(messages[0]) * MSG_BATCH_SIZE);
1051
1052         /*
1053          * Make this iterator temporarily active (not seeking) to call
1054          * the "next" method.
1055          */
1056         set_self_comp_port_input_msg_iterator_state(iterator,
1057                 BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_ACTIVE);
1058
1059         BT_ASSERT(iterator->methods.next);
1060
1061         while (true) {
1062                 /*
1063                  * Call the user's "next" method to get the next
1064                  * messages and status.
1065                  */
1066                 BT_LOGD_STR("Calling user's \"next\" method.");
1067                 status = iterator->methods.next(iterator,
1068                         &messages[0], MSG_BATCH_SIZE, &user_count);
1069                 BT_LOGD("User method returned: status=%s",
1070                         bt_message_iterator_status_string(status));
1071
1072 #ifdef BT_DEV_MODE
1073                 /*
1074                  * The user's "next" method must not do any action which
1075                  * would change the iterator's state.
1076                  */
1077                 BT_ASSERT(iterator->state ==
1078                         BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_ACTIVE);
1079 #endif
1080
1081                 switch (status) {
1082                 case BT_MESSAGE_ITERATOR_STATUS_OK:
1083                         BT_ASSERT_PRE(user_count <= MSG_BATCH_SIZE,
1084                                 "Invalid returned message count: greater than "
1085                                 "batch size: count=%" PRIu64 ", batch-size=%u",
1086                                 user_count, MSG_BATCH_SIZE);
1087                         break;
1088                 case BT_MESSAGE_ITERATOR_STATUS_AGAIN:
1089                 case BT_MESSAGE_ITERATOR_STATUS_ERROR:
1090                 case BT_MESSAGE_ITERATOR_STATUS_NOMEM:
1091                 case BT_MESSAGE_ITERATOR_STATUS_END:
1092                         goto end;
1093                 default:
1094                         abort();
1095                 }
1096
1097                 /*
1098                  * Find first message which has a default clock snapshot
1099                  * that is greater than or equal to the requested value.
1100                  *
1101                  * For event and inactivity messages, compare with the
1102                  * default clock snapshot.
1103                  *
1104                  * For packet beginning messages, compare with the
1105                  * default beginning clock snapshot, if any.
1106                  *
1107                  * For packet end messages, compare with the default end
1108                  * clock snapshot, if any.
1109                  *
1110                  * For stream beginning, stream end, ignore.
1111                  */
1112                 for (i = 0; i < user_count; i++) {
1113                         const struct bt_message *msg = messages[i];
1114                         int64_t msg_ns_from_origin;
1115                         bool ignore = false;
1116                         int ret;
1117
1118                         BT_ASSERT(msg);
1119                         ret = get_message_ns_from_origin(msg, &msg_ns_from_origin,
1120                                 &ignore);
1121                         if (ret) {
1122                                 status = BT_MESSAGE_ITERATOR_STATUS_ERROR;
1123                                 goto end;
1124                         }
1125
1126                         if (ignore) {
1127                                 /* Skip message without a clock snapshot */
1128                                 continue;
1129                         }
1130
1131                         if (msg_ns_from_origin >= ns_from_origin) {
1132                                 /*
1133                                  * We found it: move this message and
1134                                  * the following ones to the iterator's
1135                                  * auto-seek message array.
1136                                  */
1137                                 uint64_t j;
1138
1139                                 for (j = i; j < user_count; j++) {
1140                                         iterator->auto_seek_msgs->pdata[j - i] =
1141                                                 (void *) messages[j];
1142                                         messages[j] = NULL;
1143                                 }
1144
1145                                 iterator->auto_seek_msg_count = user_count - i;
1146                                 goto end;
1147                         }
1148
1149                         bt_object_put_no_null_check(msg);
1150                         messages[i] = NULL;
1151                 }
1152         }
1153
1154 end:
1155         for (i = 0; i < user_count; i++) {
1156                 if (messages[i]) {
1157                         bt_object_put_no_null_check(messages[i]);
1158                 }
1159         }
1160
1161         set_self_comp_port_input_msg_iterator_state(iterator, init_state);
1162         return status;
1163 }
1164
1165 static
1166 enum bt_self_message_iterator_status post_auto_seek_next(
1167                 struct bt_self_component_port_input_message_iterator *iterator,
1168                 bt_message_array_const msgs, uint64_t capacity,
1169                 uint64_t *count)
1170 {
1171         BT_ASSERT(iterator->auto_seek_msg_count <= capacity);
1172         BT_ASSERT(iterator->auto_seek_msg_count > 0);
1173
1174         /*
1175          * Move auto-seek messages to the output array (which is this
1176          * iterator's base message array.
1177          */
1178         memcpy(&msgs[0], &iterator->auto_seek_msgs->pdata[0],
1179                 sizeof(msgs[0]) * iterator->auto_seek_msg_count);
1180         memset(&iterator->auto_seek_msgs->pdata[0], 0,
1181                 sizeof(iterator->auto_seek_msgs->pdata[0]) *
1182                 iterator->auto_seek_msg_count);
1183         *count = iterator->auto_seek_msg_count;
1184
1185         /* Restore real user's "next" method */
1186         switch (iterator->upstream_component->class->type) {
1187         case BT_COMPONENT_CLASS_TYPE_SOURCE:
1188         {
1189                 struct bt_component_class_source *src_comp_cls =
1190                         (void *) iterator->upstream_component->class;
1191
1192                 iterator->methods.next =
1193                         (bt_self_component_port_input_message_iterator_next_method)
1194                                 src_comp_cls->methods.msg_iter_next;
1195                 break;
1196         }
1197         case BT_COMPONENT_CLASS_TYPE_FILTER:
1198         {
1199                 struct bt_component_class_filter *flt_comp_cls =
1200                         (void *) iterator->upstream_component->class;
1201
1202                 iterator->methods.next =
1203                         (bt_self_component_port_input_message_iterator_next_method)
1204                                 flt_comp_cls->methods.msg_iter_next;
1205                 break;
1206         }
1207         default:
1208                 abort();
1209         }
1210
1211         return BT_SELF_MESSAGE_ITERATOR_STATUS_OK;
1212 }
1213
1214 enum bt_message_iterator_status
1215 bt_self_component_port_input_message_iterator_seek_ns_from_origin(
1216                 struct bt_self_component_port_input_message_iterator *iterator,
1217                 int64_t ns_from_origin)
1218 {
1219         int status;
1220
1221         BT_ASSERT_PRE_NON_NULL(iterator, "Message iterator");
1222         BT_ASSERT_PRE_ITER_HAS_STATE_TO_SEEK(iterator);
1223         BT_ASSERT_PRE(bt_component_borrow_graph(iterator->upstream_component)->is_configured,
1224                 "Graph is not configured: %!+g",
1225                 bt_component_borrow_graph(iterator->upstream_component));
1226         BT_ASSERT_PRE(
1227                 bt_self_component_port_input_message_iterator_can_seek_ns_from_origin(
1228                         iterator, ns_from_origin),
1229                 "Message iterator cannot seek nanoseconds from origin: %!+i, "
1230                 "ns-from-origin=%" PRId64, iterator, ns_from_origin);
1231         set_self_comp_port_input_msg_iterator_state(iterator,
1232                 BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_SEEKING);
1233
1234         if (iterator->methods.seek_ns_from_origin) {
1235                 BT_LIB_LOGD("Calling user's \"seek nanoseconds from origin\" method: "
1236                         "%![iter-]+i, ns=%" PRId64, iterator, ns_from_origin);
1237                 status = iterator->methods.seek_ns_from_origin(iterator,
1238                         ns_from_origin);
1239                 BT_LOGD("User method returned: status=%s",
1240                         bt_message_iterator_status_string(status));
1241                 BT_ASSERT_PRE(status == BT_MESSAGE_ITERATOR_STATUS_OK ||
1242                         status == BT_MESSAGE_ITERATOR_STATUS_ERROR ||
1243                         status == BT_MESSAGE_ITERATOR_STATUS_NOMEM ||
1244                         status == BT_MESSAGE_ITERATOR_STATUS_AGAIN,
1245                         "Unexpected status: %![iter-]+i, status=%s",
1246                         iterator,
1247                         bt_self_message_iterator_status_string(status));
1248         } else {
1249                 /* Start automatic seeking: seek beginning first */
1250                 BT_ASSERT(iterator->methods.can_seek_beginning(iterator));
1251                 BT_ASSERT(iterator->methods.seek_beginning);
1252                 BT_LIB_LOGD("Calling user's \"seek beginning\" method: %!+i",
1253                         iterator);
1254                 status = iterator->methods.seek_beginning(iterator);
1255                 BT_LOGD("User method returned: status=%s",
1256                         bt_message_iterator_status_string(status));
1257                 BT_ASSERT_PRE(status == BT_MESSAGE_ITERATOR_STATUS_OK ||
1258                         status == BT_MESSAGE_ITERATOR_STATUS_ERROR ||
1259                         status == BT_MESSAGE_ITERATOR_STATUS_NOMEM ||
1260                         status == BT_MESSAGE_ITERATOR_STATUS_AGAIN,
1261                         "Unexpected status: %![iter-]+i, status=%s",
1262                         iterator,
1263                         bt_self_message_iterator_status_string(status));
1264                 switch (status) {
1265                 case BT_MESSAGE_ITERATOR_STATUS_OK:
1266                         break;
1267                 case BT_MESSAGE_ITERATOR_STATUS_ERROR:
1268                 case BT_MESSAGE_ITERATOR_STATUS_NOMEM:
1269                 case BT_MESSAGE_ITERATOR_STATUS_AGAIN:
1270                         goto end;
1271                 default:
1272                         abort();
1273                 }
1274
1275                 /*
1276                  * Find the first message which has a default clock
1277                  * snapshot greater than or equal to the requested
1278                  * nanoseconds from origin, and move the received
1279                  * messages from this point in the batch to this
1280                  * iterator's auto-seek message array.
1281                  */
1282                 status = find_message_ge_ns_from_origin(iterator,
1283                         ns_from_origin);
1284                 switch (status) {
1285                 case BT_MESSAGE_ITERATOR_STATUS_OK:
1286                         /*
1287                          * Replace the user's "next" method with a
1288                          * custom, temporary "next" method which returns
1289                          * the messages in the iterator's message array.
1290                          */
1291                         iterator->methods.next =
1292                                 (bt_self_component_port_input_message_iterator_next_method)
1293                                         post_auto_seek_next;
1294                         break;
1295                 case BT_MESSAGE_ITERATOR_STATUS_ERROR:
1296                 case BT_MESSAGE_ITERATOR_STATUS_NOMEM:
1297                 case BT_MESSAGE_ITERATOR_STATUS_AGAIN:
1298                         goto end;
1299                 case BT_MESSAGE_ITERATOR_STATUS_END:
1300                         /*
1301                          * The iterator reached the end: just return
1302                          * `BT_MESSAGE_ITERATOR_STATUS_OK` here, as if
1303                          * the seeking operation occured: the next
1304                          * "next" method will return
1305                          * `BT_MESSAGE_ITERATOR_STATUS_END` itself.
1306                          */
1307                         break;
1308                 default:
1309                         abort();
1310                 }
1311         }
1312
1313 end:
1314         set_iterator_state_after_seeking(iterator, status);
1315
1316         if (status == BT_MESSAGE_ITERATOR_STATUS_END) {
1317                 status = BT_MESSAGE_ITERATOR_STATUS_OK;
1318         }
1319
1320         return status;
1321 }
1322
1323 static inline
1324 bt_self_component_port_input_message_iterator *
1325 borrow_output_port_message_iterator_upstream_iterator(
1326                 struct bt_port_output_message_iterator *iterator)
1327 {
1328         struct bt_component_class_sink_colander_priv_data *colander_data;
1329
1330         BT_ASSERT(iterator);
1331         colander_data = (void *) iterator->colander->parent.user_data;
1332         BT_ASSERT(colander_data);
1333         BT_ASSERT(colander_data->msg_iter);
1334         return colander_data->msg_iter;
1335 }
1336
1337 bt_bool bt_port_output_message_iterator_can_seek_ns_from_origin(
1338                 struct bt_port_output_message_iterator *iterator,
1339                 int64_t ns_from_origin)
1340 {
1341         BT_ASSERT_PRE_NON_NULL(iterator, "Message iterator");
1342         return bt_self_component_port_input_message_iterator_can_seek_ns_from_origin(
1343                 borrow_output_port_message_iterator_upstream_iterator(
1344                         iterator), ns_from_origin);
1345 }
1346
1347 bt_bool bt_port_output_message_iterator_can_seek_beginning(
1348                 struct bt_port_output_message_iterator *iterator)
1349 {
1350         BT_ASSERT_PRE_NON_NULL(iterator, "Message iterator");
1351         return bt_self_component_port_input_message_iterator_can_seek_beginning(
1352                 borrow_output_port_message_iterator_upstream_iterator(
1353                         iterator));
1354 }
1355
1356 enum bt_message_iterator_status bt_port_output_message_iterator_seek_ns_from_origin(
1357                 struct bt_port_output_message_iterator *iterator,
1358                 int64_t ns_from_origin)
1359 {
1360         BT_ASSERT_PRE_NON_NULL(iterator, "Message iterator");
1361         return bt_self_component_port_input_message_iterator_seek_ns_from_origin(
1362                 borrow_output_port_message_iterator_upstream_iterator(iterator),
1363                 ns_from_origin);
1364 }
1365
1366 enum bt_message_iterator_status bt_port_output_message_iterator_seek_beginning(
1367                 struct bt_port_output_message_iterator *iterator)
1368 {
1369         BT_ASSERT_PRE_NON_NULL(iterator, "Message iterator");
1370         return bt_self_component_port_input_message_iterator_seek_beginning(
1371                 borrow_output_port_message_iterator_upstream_iterator(
1372                         iterator));
1373 }
1374
1375 void bt_port_output_message_iterator_get_ref(
1376                 const struct bt_port_output_message_iterator *iterator)
1377 {
1378         bt_object_get_ref(iterator);
1379 }
1380
1381 void bt_port_output_message_iterator_put_ref(
1382                 const struct bt_port_output_message_iterator *iterator)
1383 {
1384         bt_object_put_ref(iterator);
1385 }
1386
1387 void bt_self_component_port_input_message_iterator_get_ref(
1388                 const struct bt_self_component_port_input_message_iterator *iterator)
1389 {
1390         bt_object_get_ref(iterator);
1391 }
1392
1393 void bt_self_component_port_input_message_iterator_put_ref(
1394                 const struct bt_self_component_port_input_message_iterator *iterator)
1395 {
1396         bt_object_put_ref(iterator);
1397 }