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