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/message-const.h>
  43 #include <babeltrace/graph/message-iterator.h>
  44 #include <babeltrace/graph/message-iterator-internal.h>
  45 #include <babeltrace/graph/self-component-port-input-message-iterator.h>
  46 #include <babeltrace/graph/port-output-message-iterator.h>
  47 #include <babeltrace/graph/message-internal.h>
  48 #include <babeltrace/graph/message-event-const.h>
  49 #include <babeltrace/graph/message-event-internal.h>
  50 #include <babeltrace/graph/message-packet-const.h>
  51 #include <babeltrace/graph/message-packet-internal.h>
  52 #include <babeltrace/graph/message-stream-const.h>
  53 #include <babeltrace/graph/message-stream-internal.h>
  54 #include <babeltrace/graph/port-const.h>
  55 #include <babeltrace/graph/graph.h>
  56 #include <babeltrace/graph/graph-const.h>
  57 #include <babeltrace/graph/graph-internal.h>
  58 #include <babeltrace/types.h>
  59 #include <babeltrace/assert-internal.h>
  60 #include <babeltrace/assert-pre-internal.h>
  61 #include <stdint.h>
  62 #include <inttypes.h>
  63 #include <stdlib.h>
  64
  65 /*
  66  * TODO: Use graph's state (number of active iterators, etc.) and
  67  * possibly system specifications to make a better guess than this.
  68  */
  69 #define MSG_BATCH_SIZE  15
  70
  71 struct stream_state {
  72         const struct bt_stream *stream; /* owned by this */
  73         const struct bt_packet *cur_packet; /* owned by this */
  74         uint64_t expected_msg_seq_num;
  75         bt_bool is_ended;
  76 };
  77
  78 BT_ASSERT_PRE_FUNC
  79 static
  80 void destroy_stream_state(struct stream_state *stream_state)
  81 {
  82         if (!stream_state) {
  83                 return;
  84         }
  85
  86         BT_LOGV("Destroying stream state: stream-state-addr=%p", stream_state);
  87         BT_LOGV_STR("Putting stream state's current packet.");
  88         BT_OBJECT_PUT_REF_AND_RESET(stream_state->cur_packet);
  89         BT_LOGV_STR("Putting stream state's stream.");
  90         BT_OBJECT_PUT_REF_AND_RESET(stream_state->stream);
  91         g_free(stream_state);
  92 }
  93
  94 BT_ASSERT_PRE_FUNC
  95 static
  96 struct stream_state *create_stream_state(const struct bt_stream *stream)
  97 {
  98         struct stream_state *stream_state = g_new0(struct stream_state, 1);
  99
 100         if (!stream_state) {
 101                 BT_LOGE_STR("Failed to allocate one stream state.");
 102                 goto end;
 103         }
 104
 105         /*
 106          * We keep a reference to the stream until we know it's ended.
 107          */
 108         stream_state->stream = stream;
 109         bt_object_get_no_null_check(stream_state->stream);
 110         BT_LIB_LOGV("Created stream state: %![stream-]+s, "
 111                 "stream-state-addr=%p",
 112                 stream, stream_state);
 113
 114 end:
 115         return stream_state;
 116 }
 117
 118 static inline
 119 void _set_self_comp_port_input_msg_iterator_state(
 120                 struct bt_self_component_port_input_message_iterator *iterator,
 121                 enum bt_self_component_port_input_message_iterator_state state)
 122 {
 123         BT_ASSERT(iterator);
 124         BT_LIB_LOGD("Updating message iterator's state: "
 125                 "new-state=%s",
 126                 bt_self_component_port_input_message_iterator_state_string(state));
 127         iterator->state = state;
 128 }
 129
 130 #ifdef BT_DEV_MODE
 131 # define set_self_comp_port_input_msg_iterator_state _set_self_comp_port_input_msg_iterator_state
 132 #else
 133 # define set_self_comp_port_input_msg_iterator_state(_a, _b)
 134 #endif
 135
 136 static
 137 void destroy_base_message_iterator(struct bt_object *obj)
 138 {
 139         struct bt_message_iterator *iterator = (void *) obj;
 140
 141         BT_ASSERT(iterator);
 142
 143         if (iterator->msgs) {
 144                 g_ptr_array_free(iterator->msgs, TRUE);
 145                 iterator->msgs = NULL;
 146         }
 147
 148         g_free(iterator);
 149 }
 150
 151 static
 152 void bt_self_component_port_input_message_iterator_destroy(struct bt_object *obj)
 153 {
 154         struct bt_self_component_port_input_message_iterator *iterator;
 155
 156         BT_ASSERT(obj);
 157
 158         /*
 159          * The message iterator's reference count is 0 if we're
 160          * here. Increment it to avoid a double-destroy (possibly
 161          * infinitely recursive). This could happen for example if the
 162          * message iterator's finalization function does
 163          * bt_object_get_ref() (or anything that causes
 164          * bt_object_get_ref() to be called) on itself (ref. count goes
 165          * from 0 to 1), and then bt_object_put_ref(): the reference
 166          * count would go from 1 to 0 again and this function would be
 167          * called again.
 168          */
 169         obj->ref_count++;
 170         iterator = (void *) obj;
 171         BT_LIB_LOGD("Destroying self component input port message iterator object: "
 172                 "%!+i", iterator);
 173         bt_self_component_port_input_message_iterator_try_finalize(iterator);
 174
 175         if (iterator->stream_states) {
 176                 /*
 177                  * Remove our destroy listener from each stream which
 178                  * has a state in this iterator. Otherwise the destroy
 179                  * listener would be called with an invalid/other
 180                  * message iterator object.
 181                  */
 182                 g_hash_table_destroy(iterator->stream_states);
 183                 iterator->stream_states = NULL;
 184         }
 185
 186         if (iterator->connection) {
 187                 /*
 188                  * Remove ourself from the originating connection so
 189                  * that it does not try to finalize a dangling pointer
 190                  * later.
 191                  */
 192                 bt_connection_remove_iterator(iterator->connection, iterator);
 193                 iterator->connection = NULL;
 194         }
 195
 196         destroy_base_message_iterator(obj);
 197 }
 198
 199 BT_HIDDEN
 200 void bt_self_component_port_input_message_iterator_try_finalize(
 201                 struct bt_self_component_port_input_message_iterator *iterator)
 202 {
 203         typedef void (*method_t)(void *);
 204
 205         struct bt_component_class *comp_class = NULL;
 206         method_t method = NULL;
 207
 208         BT_ASSERT(iterator);
 209
 210         switch (iterator->state) {
 211         case BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_NON_INITIALIZED:
 212                 /* Skip user finalization if user initialization failed */
 213                 BT_LIB_LOGD("Not finalizing non-initialized message iterator: "
 214                         "%!+i", iterator);
 215                 goto end;
 216         case BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_FINALIZED:
 217                 /* Already finalized */
 218                 BT_LIB_LOGD("Not finalizing message iterator: already finalized: "
 219                         "%!+i", iterator);
 220                 goto end;
 221         case BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_FINALIZING:
 222                 /* Already finalized */
 223                 BT_LIB_LOGF("Message iterator is already being finalized: "
 224                         "%!+i", iterator);
 225                 abort();
 226         default:
 227                 break;
 228         }
 229
 230         BT_LIB_LOGD("Finalizing message iterator: %!+i", iterator);
 231         set_self_comp_port_input_msg_iterator_state(iterator,
 232                 BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_FINALIZING);
 233         BT_ASSERT(iterator->upstream_component);
 234         comp_class = iterator->upstream_component->class;
 235
 236         /* Call user-defined destroy method */
 237         switch (comp_class->type) {
 238         case BT_COMPONENT_CLASS_TYPE_SOURCE:
 239         {
 240                 struct bt_component_class_source *src_comp_cls =
 241                         (void *) comp_class;
 242
 243                 method = (method_t) src_comp_cls->methods.msg_iter_finalize;
 244                 break;
 245         }
 246         case BT_COMPONENT_CLASS_TYPE_FILTER:
 247         {
 248                 struct bt_component_class_filter *flt_comp_cls =
 249                         (void *) comp_class;
 250
 251                 method = (method_t) flt_comp_cls->methods.msg_iter_finalize;
 252                 break;
 253         }
 254         default:
 255                 /* Unreachable */
 256                 abort();
 257         }
 258
 259         if (method) {
 260                 BT_LIB_LOGD("Calling user's finalization method: %!+i",
 261                         iterator);
 262                 method(iterator);
 263         }
 264
 265         iterator->upstream_component = NULL;
 266         iterator->upstream_port = NULL;
 267         set_self_comp_port_input_msg_iterator_state(iterator,
 268                 BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_FINALIZED);
 269         BT_LIB_LOGD("Finalized message iterator: %!+i", iterator);
 270
 271 end:
 272         return;
 273 }
 274
 275 BT_HIDDEN
 276 void bt_self_component_port_input_message_iterator_set_connection(
 277                 struct bt_self_component_port_input_message_iterator *iterator,
 278                 struct bt_connection *connection)
 279 {
 280         BT_ASSERT(iterator);
 281         iterator->connection = connection;
 282         BT_LIB_LOGV("Set message iterator's connection: "
 283                 "%![iter-]+i, %![conn-]+x", iterator, connection);
 284 }
 285
 286 static
 287 int init_message_iterator(struct bt_message_iterator *iterator,
 288                 enum bt_message_iterator_type type,
 289                 bt_object_release_func destroy)
 290 {
 291         int ret = 0;
 292
 293         bt_object_init_shared(&iterator->base, destroy);
 294         iterator->type = type;
 295         iterator->msgs = g_ptr_array_new();
 296         if (!iterator->msgs) {
 297                 BT_LOGE_STR("Failed to allocate a GPtrArray.");
 298                 ret = -1;
 299                 goto end;
 300         }
 301
 302         g_ptr_array_set_size(iterator->msgs, MSG_BATCH_SIZE);
 303
 304 end:
 305         return ret;
 306 }
 307
 308 static
 309 struct bt_self_component_port_input_message_iterator *
 310 bt_self_component_port_input_message_iterator_create_initial(
 311                 struct bt_component *upstream_comp,
 312                 struct bt_port *upstream_port)
 313 {
 314         int ret;
 315         struct bt_self_component_port_input_message_iterator *iterator = NULL;
 316
 317         BT_ASSERT(upstream_comp);
 318         BT_ASSERT(upstream_port);
 319         BT_ASSERT(bt_port_is_connected(upstream_port));
 320         BT_LIB_LOGD("Creating initial message iterator on self component input port: "
 321                 "%![up-comp-]+c, %![up-port-]+p", upstream_comp, upstream_port);
 322         BT_ASSERT(bt_component_get_class_type(upstream_comp) ==
 323                 BT_COMPONENT_CLASS_TYPE_SOURCE ||
 324                 bt_component_get_class_type(upstream_comp) ==
 325                 BT_COMPONENT_CLASS_TYPE_FILTER);
 326         iterator = g_new0(
 327                 struct bt_self_component_port_input_message_iterator, 1);
 328         if (!iterator) {
 329                 BT_LOGE_STR("Failed to allocate one self component input port "
 330                         "message iterator.");
 331                 goto end;
 332         }
 333
 334         ret = init_message_iterator((void *) iterator,
 335                 BT_MESSAGE_ITERATOR_TYPE_SELF_COMPONENT_PORT_INPUT,
 336                 bt_self_component_port_input_message_iterator_destroy);
 337         if (ret) {
 338                 /* init_message_iterator() logs errors */
 339                 BT_OBJECT_PUT_REF_AND_RESET(iterator);
 340                 goto end;
 341         }
 342
 343         iterator->stream_states = g_hash_table_new_full(g_direct_hash,
 344                 g_direct_equal, NULL, (GDestroyNotify) destroy_stream_state);
 345         if (!iterator->stream_states) {
 346                 BT_LOGE_STR("Failed to allocate a GHashTable.");
 347                 BT_OBJECT_PUT_REF_AND_RESET(iterator);
 348                 goto end;
 349         }
 350
 351         iterator->upstream_component = upstream_comp;
 352         iterator->upstream_port = upstream_port;
 353         iterator->connection = iterator->upstream_port->connection;
 354         iterator->graph = bt_component_borrow_graph(upstream_comp);
 355         set_self_comp_port_input_msg_iterator_state(iterator,
 356                 BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_NON_INITIALIZED);
 357         BT_LIB_LOGD("Created initial message iterator on self component input port: "
 358                 "%![up-port-]+p, %![up-comp-]+c, %![iter-]+i",
 359                 upstream_port, upstream_comp, iterator);
 360
 361 end:
 362         return iterator;
 363 }
 364
 365 struct bt_self_component_port_input_message_iterator *
 366 bt_self_component_port_input_message_iterator_create(
 367                 struct bt_self_component_port_input *self_port)
 368 {
 369         typedef enum bt_self_message_iterator_status (*init_method_t)(
 370                         void *, void *, void *);
 371
 372         init_method_t init_method = NULL;
 373         struct bt_self_component_port_input_message_iterator *iterator =
 374                 NULL;
 375         struct bt_port *port = (void *) self_port;
 376         struct bt_port *upstream_port;
 377         struct bt_component *comp;
 378         struct bt_component *upstream_comp;
 379         struct bt_component_class *upstream_comp_cls;
 380
 381         BT_ASSERT_PRE_NON_NULL(port, "Port");
 382         comp = bt_port_borrow_component_inline(port);
 383         BT_ASSERT_PRE(bt_port_is_connected(port),
 384                 "Port is not connected: %![port-]+p", port);
 385         BT_ASSERT_PRE(comp, "Port is not part of a component: %![port-]+p",
 386                 port);
 387         BT_ASSERT_PRE(!bt_component_graph_is_canceled(comp),
 388                 "Port's component's graph is canceled: "
 389                 "%![port-]+p, %![comp-]+c", port, comp);
 390         BT_ASSERT(port->connection);
 391         upstream_port = port->connection->upstream_port;
 392         BT_ASSERT(upstream_port);
 393         upstream_comp = bt_port_borrow_component_inline(upstream_port);
 394         BT_ASSERT(upstream_comp);
 395         upstream_comp_cls = upstream_comp->class;
 396         BT_ASSERT(upstream_comp->class->type ==
 397                 BT_COMPONENT_CLASS_TYPE_SOURCE ||
 398                 upstream_comp->class->type ==
 399                 BT_COMPONENT_CLASS_TYPE_FILTER);
 400         iterator = bt_self_component_port_input_message_iterator_create_initial(
 401                 upstream_comp, upstream_port);
 402         if (!iterator) {
 403                 BT_LOGW_STR("Cannot create self component input port "
 404                         "message iterator.");
 405                 goto end;
 406         }
 407
 408         switch (upstream_comp_cls->type) {
 409         case BT_COMPONENT_CLASS_TYPE_SOURCE:
 410         {
 411                 struct bt_component_class_source *src_comp_cls =
 412                         (void *) upstream_comp_cls;
 413
 414                 init_method =
 415                         (init_method_t) src_comp_cls->methods.msg_iter_init;
 416                 break;
 417         }
 418         case BT_COMPONENT_CLASS_TYPE_FILTER:
 419         {
 420                 struct bt_component_class_filter *flt_comp_cls =
 421                         (void *) upstream_comp_cls;
 422
 423                 init_method =
 424                         (init_method_t) flt_comp_cls->methods.msg_iter_init;
 425                 break;
 426         }
 427         default:
 428                 /* Unreachable */
 429                 abort();
 430         }
 431
 432         if (init_method) {
 433                 int iter_status;
 434
 435                 BT_LIB_LOGD("Calling user's initialization method: %!+i", iterator);
 436                 iter_status = init_method(iterator, upstream_comp,
 437                         upstream_port);
 438                 BT_LOGD("User method returned: status=%s",
 439                         bt_message_iterator_status_string(iter_status));
 440                 if (iter_status != BT_MESSAGE_ITERATOR_STATUS_OK) {
 441                         BT_LOGW_STR("Initialization method failed.");
 442                         BT_OBJECT_PUT_REF_AND_RESET(iterator);
 443                         goto end;
 444                 }
 445         }
 446
 447         set_self_comp_port_input_msg_iterator_state(iterator,
 448                 BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_ACTIVE);
 449         g_ptr_array_add(port->connection->iterators, iterator);
 450         BT_LIB_LOGD("Created message iterator on self component input port: "
 451                 "%![up-port-]+p, %![up-comp-]+c, %![iter-]+i",
 452                 upstream_port, upstream_comp, iterator);
 453
 454 end:
 455         return iterator;
 456 }
 457
 458 void *bt_self_message_iterator_get_data(
 459                 const struct bt_self_message_iterator *self_iterator)
 460 {
 461         struct bt_self_component_port_input_message_iterator *iterator =
 462                 (void *) self_iterator;
 463
 464         BT_ASSERT_PRE_NON_NULL(iterator, "Message iterator");
 465         return iterator->user_data;
 466 }
 467
 468 void bt_self_message_iterator_set_data(
 469                 struct bt_self_message_iterator *self_iterator, void *data)
 470 {
 471         struct bt_self_component_port_input_message_iterator *iterator =
 472                 (void *) self_iterator;
 473
 474         BT_ASSERT_PRE_NON_NULL(iterator, "Message iterator");
 475         iterator->user_data = data;
 476         BT_LIB_LOGV("Set message iterator's user data: "
 477                 "%!+i, user-data-addr=%p", iterator, data);
 478 }
 479
 480 BT_ASSERT_PRE_FUNC
 481 static inline
 482 void bt_message_borrow_packet_stream(const struct bt_message *msg,
 483                 const struct bt_stream **stream,
 484                 const struct bt_packet **packet)
 485 {
 486         BT_ASSERT(msg);
 487
 488         switch (msg->type) {
 489         case BT_MESSAGE_TYPE_EVENT:
 490                 *packet = bt_event_borrow_packet_const(
 491                         bt_message_event_borrow_event_const(msg));
 492                 *stream = bt_packet_borrow_stream_const(*packet);
 493                 break;
 494         case BT_MESSAGE_TYPE_STREAM_BEGINNING:
 495                 *stream = bt_message_stream_beginning_borrow_stream_const(msg);
 496                 break;
 497         case BT_MESSAGE_TYPE_STREAM_END:
 498                 *stream = bt_message_stream_end_borrow_stream_const(msg);
 499                 break;
 500         case BT_MESSAGE_TYPE_PACKET_BEGINNING:
 501                 *packet = bt_message_packet_beginning_borrow_packet_const(msg);
 502                 *stream = bt_packet_borrow_stream_const(*packet);
 503                 break;
 504         case BT_MESSAGE_TYPE_PACKET_END:
 505                 *packet = bt_message_packet_end_borrow_packet_const(msg);
 506                 *stream = bt_packet_borrow_stream_const(*packet);
 507                 break;
 508         default:
 509                 break;
 510         }
 511 }
 512
 513 BT_ASSERT_PRE_FUNC
 514 static inline
 515 bool validate_message(
 516                 struct bt_self_component_port_input_message_iterator *iterator,
 517                 const struct bt_message *c_msg)
 518 {
 519         bool is_valid = true;
 520         struct stream_state *stream_state;
 521         const struct bt_stream *stream = NULL;
 522         const struct bt_packet *packet = NULL;
 523         struct bt_message *msg = (void *) c_msg;
 524
 525         BT_ASSERT(msg);
 526         bt_message_borrow_packet_stream(c_msg, &stream, &packet);
 527
 528         if (!stream) {
 529                 /* we don't care about messages not attached to streams */
 530                 goto end;
 531         }
 532
 533         stream_state = g_hash_table_lookup(iterator->stream_states, stream);
 534         if (!stream_state) {
 535                 /*
 536                  * No stream state for this stream: this message
 537                  * MUST be a BT_MESSAGE_TYPE_STREAM_BEGINNING message
 538                  * and its sequence number must be 0.
 539                  */
 540                 if (c_msg->type != BT_MESSAGE_TYPE_STREAM_BEGINNING) {
 541                         BT_ASSERT_PRE_MSG("Unexpected message: missing a "
 542                                 "BT_MESSAGE_TYPE_STREAM_BEGINNING "
 543                                 "message prior to this message: "
 544                                 "%![stream-]+s", stream);
 545                         is_valid = false;
 546                         goto end;
 547                 }
 548
 549                 if (c_msg->seq_num == -1ULL) {
 550                         msg->seq_num = 0;
 551                 }
 552
 553                 if (c_msg->seq_num != 0) {
 554                         BT_ASSERT_PRE_MSG("Unexpected message sequence "
 555                                 "number for this message iterator: "
 556                                 "this is the first message for this "
 557                                 "stream, expecting sequence number 0: "
 558                                 "seq-num=%" PRIu64 ", %![stream-]+s",
 559                                 c_msg->seq_num, stream);
 560                         is_valid = false;
 561                         goto end;
 562                 }
 563
 564                 stream_state = create_stream_state(stream);
 565                 if (!stream_state) {
 566                         abort();
 567                 }
 568
 569                 g_hash_table_insert(iterator->stream_states,
 570                         (void *) stream, stream_state);
 571                 stream_state->expected_msg_seq_num++;
 572                 goto end;
 573         }
 574
 575         if (stream_state->is_ended) {
 576                 /*
 577                  * There's a new message which has a reference to a
 578                  * stream which, from this iterator's point of view, is
 579                  * ended ("end of stream" message was returned).
 580                  * This is bad: the API guarantees that it can never
 581                  * happen.
 582                  */
 583                 BT_ASSERT_PRE_MSG("Stream is already ended: %![stream-]+s",
 584                         stream);
 585                 is_valid = false;
 586                 goto end;
 587         }
 588
 589         if (c_msg->seq_num == -1ULL) {
 590                 msg->seq_num = stream_state->expected_msg_seq_num;
 591         }
 592
 593         if (c_msg->seq_num != -1ULL &&
 594                         c_msg->seq_num != stream_state->expected_msg_seq_num) {
 595                 BT_ASSERT_PRE_MSG("Unexpected message sequence number: "
 596                         "seq-num=%" PRIu64 ", "
 597                         "expected-seq-num=%" PRIu64 ", %![stream-]+s",
 598                         c_msg->seq_num, stream_state->expected_msg_seq_num,
 599                         stream);
 600                 is_valid = false;
 601                 goto end;
 602         }
 603
 604         switch (c_msg->type) {
 605         case BT_MESSAGE_TYPE_STREAM_BEGINNING:
 606                 BT_ASSERT_PRE_MSG("Unexpected BT_MESSAGE_TYPE_STREAM_BEGINNING "
 607                         "message at this point: msg-seq-num=%" PRIu64 ", "
 608                         "%![stream-]+s", c_msg->seq_num, stream);
 609                 is_valid = false;
 610                 goto end;
 611         case BT_MESSAGE_TYPE_STREAM_END:
 612                 if (stream_state->cur_packet) {
 613                         BT_ASSERT_PRE_MSG("Unexpected BT_MESSAGE_TYPE_STREAM_END "
 614                                 "message: missing a "
 615                                 "BT_MESSAGE_TYPE_PACKET_END message "
 616                                 "prior to this message: "
 617                                 "msg-seq-num=%" PRIu64 ", "
 618                                 "%![stream-]+s", c_msg->seq_num, stream);
 619                         is_valid = false;
 620                         goto end;
 621                 }
 622                 stream_state->expected_msg_seq_num++;
 623                 stream_state->is_ended = true;
 624                 goto end;
 625         case BT_MESSAGE_TYPE_PACKET_BEGINNING:
 626                 if (stream_state->cur_packet) {
 627                         BT_ASSERT_PRE_MSG("Unexpected BT_MESSAGE_TYPE_PACKET_BEGINNING "
 628                                 "message at this point: missing a "
 629                                 "BT_MESSAGE_TYPE_PACKET_END message "
 630                                 "prior to this message: "
 631                                 "msg-seq-num=%" PRIu64 ", %![stream-]+s, "
 632                                 "%![packet-]+a", c_msg->seq_num, stream,
 633                                 packet);
 634                         is_valid = false;
 635                         goto end;
 636                 }
 637                 stream_state->expected_msg_seq_num++;
 638                 stream_state->cur_packet = packet;
 639                 bt_object_get_no_null_check(stream_state->cur_packet);
 640                 goto end;
 641         case BT_MESSAGE_TYPE_PACKET_END:
 642                 if (!stream_state->cur_packet) {
 643                         BT_ASSERT_PRE_MSG("Unexpected BT_MESSAGE_TYPE_PACKET_END "
 644                                 "message at this point: missing a "
 645                                 "BT_MESSAGE_TYPE_PACKET_BEGINNING message "
 646                                 "prior to this message: "
 647                                 "msg-seq-num=%" PRIu64 ", %![stream-]+s, "
 648                                 "%![packet-]+a", c_msg->seq_num, stream,
 649                                 packet);
 650                         is_valid = false;
 651                         goto end;
 652                 }
 653                 stream_state->expected_msg_seq_num++;
 654                 BT_OBJECT_PUT_REF_AND_RESET(stream_state->cur_packet);
 655                 goto end;
 656         case BT_MESSAGE_TYPE_EVENT:
 657                 if (packet != stream_state->cur_packet) {
 658                         BT_ASSERT_PRE_MSG("Unexpected packet for "
 659                                 "BT_MESSAGE_TYPE_EVENT message: "
 660                                 "msg-seq-num=%" PRIu64 ", %![stream-]+s, "
 661                                 "%![msg-packet-]+a, %![expected-packet-]+a",
 662                                 c_msg->seq_num, stream,
 663                                 stream_state->cur_packet, packet);
 664                         is_valid = false;
 665                         goto end;
 666                 }
 667                 stream_state->expected_msg_seq_num++;
 668                 goto end;
 669         default:
 670                 break;
 671         }
 672
 673 end:
 674         return is_valid;
 675 }
 676
 677 BT_ASSERT_PRE_FUNC
 678 static inline
 679 bool validate_messages(
 680                 struct bt_self_component_port_input_message_iterator *iterator,
 681                 uint64_t count)
 682 {
 683         bool ret = true;
 684         bt_message_array_const msgs =
 685                 (void *) iterator->base.msgs->pdata;
 686         uint64_t i;
 687
 688         for (i = 0; i < count; i++) {
 689                 ret = validate_message(iterator, msgs[i]);
 690                 if (!ret) {
 691                         break;
 692                 }
 693         }
 694
 695         return ret;
 696 }
 697
 698 BT_ASSERT_PRE_FUNC
 699 static inline bool self_comp_port_input_msg_iter_can_end(
 700                 struct bt_self_component_port_input_message_iterator *iterator)
 701 {
 702         GHashTableIter iter;
 703         gpointer stream_key, state_value;
 704         bool ret = true;
 705
 706         /*
 707          * Verify that this iterator received a
 708          * BT_MESSAGE_TYPE_STREAM_END message for each stream
 709          * which has a state.
 710          */
 711
 712         g_hash_table_iter_init(&iter, iterator->stream_states);
 713
 714         while (g_hash_table_iter_next(&iter, &stream_key, &state_value)) {
 715                 struct stream_state *stream_state = (void *) state_value;
 716
 717                 BT_ASSERT(stream_state);
 718                 BT_ASSERT(stream_key);
 719
 720                 if (!stream_state->is_ended) {
 721                         BT_ASSERT_PRE_MSG("Ending message iterator, "
 722                                 "but stream is not ended: "
 723                                 "%![stream-]s", stream_key);
 724                         ret = false;
 725                         goto end;
 726                 }
 727         }
 728
 729 end:
 730         return ret;
 731 }
 732
 733 enum bt_message_iterator_status
 734 bt_self_component_port_input_message_iterator_next(
 735                 struct bt_self_component_port_input_message_iterator *iterator,
 736                 bt_message_array_const *msgs, uint64_t *user_count)
 737 {
 738         typedef enum bt_self_message_iterator_status (*method_t)(
 739                         void *, bt_message_array_const, uint64_t, uint64_t *);
 740
 741         method_t method = NULL;
 742         struct bt_component_class *comp_cls;
 743         int status = BT_MESSAGE_ITERATOR_STATUS_OK;
 744
 745         BT_ASSERT_PRE_NON_NULL(iterator, "Message iterator");
 746         BT_ASSERT_PRE_NON_NULL(msgs, "Message array (output)");
 747         BT_ASSERT_PRE_NON_NULL(user_count, "Message count (output)");
 748         BT_ASSERT_PRE(iterator->state ==
 749                 BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_ACTIVE,
 750                 "Message iterator's \"next\" called, but "
 751                 "iterator is in the wrong state: %!+i", iterator);
 752         BT_ASSERT(iterator->upstream_component);
 753         BT_ASSERT(iterator->upstream_component->class);
 754         BT_LIB_LOGD("Getting next self component input port "
 755                 "message iterator's messages: %!+i", iterator);
 756         comp_cls = iterator->upstream_component->class;
 757
 758         /* Pick the appropriate "next" method */
 759         switch (comp_cls->type) {
 760         case BT_COMPONENT_CLASS_TYPE_SOURCE:
 761         {
 762                 struct bt_component_class_source *src_comp_cls =
 763                         (void *) comp_cls;
 764
 765                 method = (method_t) src_comp_cls->methods.msg_iter_next;
 766                 break;
 767         }
 768         case BT_COMPONENT_CLASS_TYPE_FILTER:
 769         {
 770                 struct bt_component_class_filter *flt_comp_cls =
 771                         (void *) comp_cls;
 772
 773                 method = (method_t) flt_comp_cls->methods.msg_iter_next;
 774                 break;
 775         }
 776         default:
 777                 abort();
 778         }
 779
 780         /*
 781          * Call the user's "next" method to get the next messages
 782          * and status.
 783          */
 784         BT_ASSERT(method);
 785         BT_LOGD_STR("Calling user's \"next\" method.");
 786         status = method(iterator, (void *) iterator->base.msgs->pdata,
 787                 MSG_BATCH_SIZE, user_count);
 788         BT_LOGD("User method returned: status=%s",
 789                 bt_message_iterator_status_string(status));
 790         if (status < 0) {
 791                 BT_LOGW_STR("User method failed.");
 792                 goto end;
 793         }
 794
 795 #ifdef BT_DEV_MODE
 796         /*
 797          * There is no way that this iterator could have been finalized
 798          * during its "next" method, as the only way to do this is to
 799          * put the last iterator's reference, and this can only be done
 800          * by its downstream owner.
 801          */
 802         BT_ASSERT(iterator->state ==
 803                 BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_ACTIVE);
 804 #endif
 805
 806         switch (status) {
 807         case BT_MESSAGE_ITERATOR_STATUS_OK:
 808                 BT_ASSERT_PRE(validate_messages(iterator, *user_count),
 809                         "Messages are invalid at this point: "
 810                         "%![msg-iter-]+i, count=%" PRIu64,
 811                         iterator, *user_count);
 812                 *msgs = (void *) iterator->base.msgs->pdata;
 813                 break;
 814         case BT_MESSAGE_ITERATOR_STATUS_AGAIN:
 815                 goto end;
 816         case BT_MESSAGE_ITERATOR_STATUS_END:
 817                 BT_ASSERT_PRE(self_comp_port_input_msg_iter_can_end(iterator),
 818                         "Message iterator cannot end at this point: "
 819                         "%!+i", iterator);
 820                 set_self_comp_port_input_msg_iterator_state(iterator,
 821                         BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_ENDED);
 822                 goto end;
 823         default:
 824                 /* Unknown non-error status */
 825                 abort();
 826         }
 827
 828 end:
 829         return status;
 830 }
 831
 832 enum bt_message_iterator_status bt_port_output_message_iterator_next(
 833                 struct bt_port_output_message_iterator *iterator,
 834                 bt_message_array_const *msgs_to_user,
 835                 uint64_t *count_to_user)
 836 {
 837         enum bt_message_iterator_status status;
 838         enum bt_graph_status graph_status;
 839
 840         BT_ASSERT_PRE_NON_NULL(iterator, "Message iterator");
 841         BT_ASSERT_PRE_NON_NULL(msgs_to_user, "Message array (output)");
 842         BT_ASSERT_PRE_NON_NULL(count_to_user, "Message count (output)");
 843         BT_LIB_LOGD("Getting next output port message iterator's messages: "
 844                 "%!+i", iterator);
 845
 846         graph_status = bt_graph_consume_sink_no_check(iterator->graph,
 847                 iterator->colander);
 848         switch (graph_status) {
 849         case BT_GRAPH_STATUS_CANCELED:
 850         case BT_GRAPH_STATUS_AGAIN:
 851         case BT_GRAPH_STATUS_END:
 852         case BT_GRAPH_STATUS_NOMEM:
 853                 status = (int) graph_status;
 854                 break;
 855         case BT_GRAPH_STATUS_OK:
 856                 status = BT_MESSAGE_ITERATOR_STATUS_OK;
 857
 858                 /*
 859                  * On success, the colander sink moves the messages
 860                  * to this iterator's array and sets this iterator's
 861                  * message count: move them to the user.
 862                  */
 863                 *msgs_to_user = (void *) iterator->base.msgs->pdata;
 864                 *count_to_user = iterator->count;
 865                 break;
 866         default:
 867                 /* Other errors */
 868                 status = BT_MESSAGE_ITERATOR_STATUS_ERROR;
 869         }
 870
 871         return status;
 872 }
 873
 874 struct bt_component *bt_self_component_port_input_message_iterator_borrow_component(
 875                 struct bt_self_component_port_input_message_iterator *iterator)
 876 {
 877         BT_ASSERT_PRE_NON_NULL(iterator, "Message iterator");
 878         return iterator->upstream_component;
 879 }
 880
 881 struct bt_self_component *bt_self_message_iterator_borrow_component(
 882                 struct bt_self_message_iterator *self_iterator)
 883 {
 884         struct bt_self_component_port_input_message_iterator *iterator =
 885                 (void *) self_iterator;
 886
 887         BT_ASSERT_PRE_NON_NULL(iterator, "Message iterator");
 888         return (void *) iterator->upstream_component;
 889 }
 890
 891 struct bt_self_port_output *bt_self_message_iterator_borrow_port(
 892                 struct bt_self_message_iterator *self_iterator)
 893 {
 894         struct bt_self_component_port_input_message_iterator *iterator =
 895                 (void *) self_iterator;
 896
 897         BT_ASSERT_PRE_NON_NULL(iterator, "Message iterator");
 898         return (void *) iterator->upstream_port;
 899 }
 900
 901 static
 902 void bt_port_output_message_iterator_destroy(struct bt_object *obj)
 903 {
 904         struct bt_port_output_message_iterator *iterator = (void *) obj;
 905
 906         BT_LIB_LOGD("Destroying output port message iterator object: %!+i",
 907                 iterator);
 908         BT_LOGD_STR("Putting graph.");
 909         BT_OBJECT_PUT_REF_AND_RESET(iterator->graph);
 910         BT_LOGD_STR("Putting colander sink component.");
 911         BT_OBJECT_PUT_REF_AND_RESET(iterator->colander);
 912         destroy_base_message_iterator(obj);
 913 }
 914
 915 struct bt_port_output_message_iterator *
 916 bt_port_output_message_iterator_create(
 917                 struct bt_graph *graph,
 918                 const struct bt_port_output *output_port)
 919 {
 920         struct bt_port_output_message_iterator *iterator = NULL;
 921         struct bt_component_class_sink *colander_comp_cls = NULL;
 922         struct bt_component *output_port_comp = NULL;
 923         struct bt_component_sink *colander_comp;
 924         enum bt_graph_status graph_status;
 925         struct bt_port_input *colander_in_port = NULL;
 926         struct bt_component_class_sink_colander_data colander_data;
 927         int ret;
 928
 929         BT_ASSERT_PRE_NON_NULL(graph, "Graph");
 930         BT_ASSERT_PRE_NON_NULL(output_port, "Output port");
 931         output_port_comp = bt_port_borrow_component_inline(
 932                 (const void *) output_port);
 933         BT_ASSERT_PRE(output_port_comp,
 934                 "Output port has no component: %!+p", output_port);
 935         BT_ASSERT_PRE(bt_component_borrow_graph(output_port_comp) ==
 936                 (void *) graph,
 937                 "Output port is not part of graph: %![graph-]+g, %![port-]+p",
 938                 graph, output_port);
 939
 940         /* Create message iterator */
 941         BT_LIB_LOGD("Creating message iterator on output port: "
 942                 "%![port-]+p, %![comp-]+c", output_port, output_port_comp);
 943         iterator = g_new0(struct bt_port_output_message_iterator, 1);
 944         if (!iterator) {
 945                 BT_LOGE_STR("Failed to allocate one output port message iterator.");
 946                 goto error;
 947         }
 948
 949         ret = init_message_iterator((void *) iterator,
 950                 BT_MESSAGE_ITERATOR_TYPE_PORT_OUTPUT,
 951                 bt_port_output_message_iterator_destroy);
 952         if (ret) {
 953                 /* init_message_iterator() logs errors */
 954                 BT_OBJECT_PUT_REF_AND_RESET(iterator);
 955                 goto end;
 956         }
 957
 958         /* Create colander component */
 959         colander_comp_cls = bt_component_class_sink_colander_get();
 960         if (!colander_comp_cls) {
 961                 BT_LOGW("Cannot get colander sink component class.");
 962                 goto error;
 963         }
 964
 965         iterator->graph = graph;
 966         bt_object_get_no_null_check(iterator->graph);
 967         colander_data.msgs = (void *) iterator->base.msgs->pdata;
 968         colander_data.count_addr = &iterator->count;
 969
 970         /* Hope that nobody uses this very unique name */
 971         graph_status =
 972                 bt_graph_add_sink_component_with_init_method_data(
 973                         (void *) graph, colander_comp_cls,
 974                         "colander-36ac3409-b1a8-4d60-ab1f-4fdf341a8fb1",
 975                         NULL, &colander_data, (void *) &iterator->colander);
 976         if (graph_status != BT_GRAPH_STATUS_OK) {
 977                 BT_LIB_LOGW("Cannot add colander sink component to graph: "
 978                         "%1[graph-]+g, status=%s", graph,
 979                         bt_graph_status_string(graph_status));
 980                 goto error;
 981         }
 982
 983         /*
 984          * Connect provided output port to the colander component's
 985          * input port.
 986          */
 987         colander_in_port =
 988                 (void *) bt_component_sink_borrow_input_port_by_index_const(
 989                         (void *) iterator->colander, 0);
 990         BT_ASSERT(colander_in_port);
 991         graph_status = bt_graph_connect_ports(graph,
 992                 output_port, colander_in_port, NULL);
 993         if (graph_status != BT_GRAPH_STATUS_OK) {
 994                 BT_LIB_LOGW("Cannot add colander sink component to graph: "
 995                         "%![graph-]+g, %![comp-]+c, status=%s", graph,
 996                         iterator->colander,
 997                         bt_graph_status_string(graph_status));
 998                 goto error;
 999         }
1000
1001         /*
1002          * At this point everything went fine. Make the graph
1003          * nonconsumable forever so that only this message iterator
1004          * can consume (thanks to bt_graph_consume_sink_no_check()).
1005          * This avoids leaking the message created by the colander
1006          * sink and moved to the message iterator's message
1007          * member.
1008          */
1009         bt_graph_set_can_consume(iterator->graph, false);
1010         goto end;
1011
1012 error:
1013         if (iterator && iterator->graph && iterator->colander) {
1014                 int ret;
1015
1016                 /* Remove created colander component from graph if any */
1017                 colander_comp = iterator->colander;
1018                 BT_OBJECT_PUT_REF_AND_RESET(iterator->colander);
1019
1020                 /*
1021                  * At this point the colander component's reference
1022                  * count is 0 because iterator->colander was the only
1023                  * owner. We also know that it is not connected because
1024                  * this is the last operation before this function
1025                  * succeeds.
1026                  *
1027                  * Since we honor the preconditions here,
1028                  * bt_graph_remove_unconnected_component() always
1029                  * succeeds.
1030                  */
1031                 ret = bt_graph_remove_unconnected_component(iterator->graph,
1032                         (void *) colander_comp);
1033                 BT_ASSERT(ret == 0);
1034         }
1035
1036         BT_OBJECT_PUT_REF_AND_RESET(iterator);
1037
1038 end:
1039         bt_object_put_ref(colander_comp_cls);
1040         return (void *) iterator;
1041 }
1042
1043 void bt_port_output_message_iterator_get_ref(
1044                 const struct bt_port_output_message_iterator *iterator)
1045 {
1046         bt_object_get_ref(iterator);
1047 }
1048
1049 void bt_port_output_message_iterator_put_ref(
1050                 const struct bt_port_output_message_iterator *iterator)
1051 {
1052         bt_object_put_ref(iterator);
1053 }
1054
1055 void bt_self_component_port_input_message_iterator_get_ref(
1056                 const struct bt_self_component_port_input_message_iterator *iterator)
1057 {
1058         bt_object_get_ref(iterator);
1059 }
1060
1061 void bt_self_component_port_input_message_iterator_put_ref(
1062                 const struct bt_self_component_port_input_message_iterator *iterator)
1063 {
1064         bt_object_put_ref(iterator);
1065 }