lib/graph/iterator.c

   1 /*
   2  * Copyright 2015 Jérémie Galarneau <jeremie.galarneau@efficios.com>
   3  * Copyright 2017 Philippe Proulx <pproulx@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 "NOTIF-ITER"
  25 #include <babeltrace/lib-logging-internal.h>
  26
  27 #include <babeltrace/compiler-internal.h>
  28 #include <babeltrace/object.h>
  29 #include <babeltrace/trace-ir/fields.h>
  30 #include <babeltrace/trace-ir/event-const.h>
  31 #include <babeltrace/trace-ir/event-internal.h>
  32 #include <babeltrace/trace-ir/packet-const.h>
  33 #include <babeltrace/trace-ir/packet-internal.h>
  34 #include <babeltrace/trace-ir/stream-internal.h>
  35 #include <babeltrace/graph/connection.h>
  36 #include <babeltrace/graph/connection-internal.h>
  37 #include <babeltrace/graph/component.h>
  38 #include <babeltrace/graph/component-internal.h>
  39 #include <babeltrace/graph/component-source-internal.h>
  40 #include <babeltrace/graph/component-class-internal.h>
  41 #include <babeltrace/graph/component-class-sink-colander-internal.h>
  42 #include <babeltrace/graph/component-sink.h>
  43 #include <babeltrace/graph/notification.h>
  44 #include <babeltrace/graph/notification-iterator.h>
  45 #include <babeltrace/graph/notification-iterator-internal.h>
  46 #include <babeltrace/graph/self-component-port-input-notification-iterator.h>
  47 #include <babeltrace/graph/port-output-notification-iterator.h>
  48 #include <babeltrace/graph/notification-internal.h>
  49 #include <babeltrace/graph/notification-event.h>
  50 #include <babeltrace/graph/notification-event-internal.h>
  51 #include <babeltrace/graph/notification-packet.h>
  52 #include <babeltrace/graph/notification-packet-internal.h>
  53 #include <babeltrace/graph/notification-stream.h>
  54 #include <babeltrace/graph/notification-stream-internal.h>
  55 #include <babeltrace/graph/port.h>
  56 #include <babeltrace/graph/private-graph.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 NOTIF_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_notif_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_notification_iterator(struct bt_object *obj)
 120 {
 121         struct bt_notification_iterator *iterator = (void *) obj;
 122
 123         BT_ASSERT(iterator);
 124
 125         if (iterator->notifs) {
 126                 g_ptr_array_free(iterator->notifs, TRUE);
 127                 iterator->notifs = NULL;
 128         }
 129
 130         g_free(iterator);
 131 }
 132
 133 static
 134 void bt_self_component_port_input_notification_iterator_destroy(struct bt_object *obj)
 135 {
 136         struct bt_self_component_port_input_notification_iterator *iterator;
 137
 138         BT_ASSERT(obj);
 139
 140         /*
 141          * The notification 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          * notification 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 notification iterator object: "
 154                 "%!+i", iterator);
 155         bt_self_component_port_input_notification_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                  * notification 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_notification_iterator(obj);
 179 }
 180
 181 BT_HIDDEN
 182 void bt_self_component_port_input_notification_iterator_finalize(
 183                 struct bt_self_component_port_input_notification_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_NOTIFICATION_ITERATOR_STATE_NON_INITIALIZED:
 194                 /* Skip user finalization if user initialization failed */
 195                 BT_LIB_LOGD("Not finalizing non-initialized notification iterator: "
 196                         "%!+i", iterator);
 197                 return;
 198         case BT_SELF_COMPONENT_PORT_INPUT_NOTIFICATION_ITERATOR_STATE_FINALIZED:
 199         case BT_SELF_COMPONENT_PORT_INPUT_NOTIFICATION_ITERATOR_STATE_FINALIZED_AND_ENDED:
 200                 /* Already finalized */
 201                 BT_LIB_LOGD("Not finalizing notification iterator: already finalized: "
 202                         "%!+i", iterator);
 203                 return;
 204         default:
 205                 break;
 206         }
 207
 208         BT_LIB_LOGD("Finalizing notification iterator: %!+i", iterator);
 209
 210         if (iterator->state == BT_SELF_COMPONENT_PORT_INPUT_NOTIFICATION_ITERATOR_STATE_ENDED) {
 211                 BT_LIB_LOGD("Updating notification iterator's state: "
 212                         "new-state=BT_SELF_COMPONENT_PORT_INPUT_NOTIFICATION_ITERATOR_STATE_FINALIZED_AND_ENDED");
 213                 iterator->state = BT_SELF_COMPONENT_PORT_INPUT_NOTIFICATION_ITERATOR_STATE_FINALIZED_AND_ENDED;
 214         } else {
 215                 BT_LIB_LOGD("Updating notification iterator's state: "
 216                         "new-state=BT_SELF_COMPONENT_PORT_INPUT_NOTIFICATION_ITERATOR_STATE_FINALIZED");
 217                 iterator->state = BT_SELF_COMPONENT_PORT_INPUT_NOTIFICATION_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.notif_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.notif_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 notification iterator: %!+i", iterator);
 255 }
 256
 257 BT_HIDDEN
 258 void bt_self_component_port_input_notification_iterator_set_connection(
 259                 struct bt_self_component_port_input_notification_iterator *iterator,
 260                 struct bt_connection *connection)
 261 {
 262         BT_ASSERT(iterator);
 263         iterator->connection = connection;
 264         BT_LIB_LOGV("Set notification iterator's connection: "
 265                 "%![iter-]+i, %![conn-]+x", iterator, connection);
 266 }
 267
 268 static
 269 int init_notification_iterator(struct bt_notification_iterator *iterator,
 270                 enum bt_notification_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->notifs = g_ptr_array_new();
 278         if (!iterator->notifs) {
 279                 BT_LOGE_STR("Failed to allocate a GPtrArray.");
 280                 ret = -1;
 281                 goto end;
 282         }
 283
 284         g_ptr_array_set_size(iterator->notifs, NOTIF_BATCH_SIZE);
 285
 286 end:
 287         return ret;
 288 }
 289
 290 static
 291 struct bt_self_component_port_input_notification_iterator *
 292 bt_self_component_port_input_notification_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_notification_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 notification 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_notification_iterator, 1);
 310         if (!iterator) {
 311                 BT_LOGE_STR("Failed to allocate one self component input port "
 312                         "notification iterator.");
 313                 goto end;
 314         }
 315
 316         ret = init_notification_iterator((void *) iterator,
 317                 BT_NOTIFICATION_ITERATOR_TYPE_SELF_COMPONENT_PORT_INPUT,
 318                 bt_self_component_port_input_notification_iterator_destroy);
 319         if (ret) {
 320                 /* init_notification_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_NOTIFICATION_ITERATOR_STATE_NON_INITIALIZED;
 338         BT_LIB_LOGD("Created initial notification 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_notification_iterator *
 347 bt_self_component_port_input_notification_iterator_create(
 348                 struct bt_self_component_port_input *self_port)
 349 {
 350         typedef enum bt_self_notification_iterator_status (*init_method_t)(
 351                         void *, void *, void *);
 352
 353         init_method_t init_method = NULL;
 354         struct bt_self_component_port_input_notification_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(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(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_notification_iterator_create_initial(
 382                 upstream_comp, upstream_port);
 383         if (!iterator) {
 384                 BT_LOGW_STR("Cannot create self component input port "
 385                         "notification 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.notif_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.notif_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_notification_iterator_status_string(iter_status));
 421                 if (iter_status != BT_NOTIFICATION_ITERATOR_STATUS_OK) {
 422                         BT_LOGW_STR("Initialization method failed.");
 423                         goto end;
 424                 }
 425         }
 426
 427         iterator->state = BT_SELF_COMPONENT_PORT_INPUT_NOTIFICATION_ITERATOR_STATE_ACTIVE;
 428         g_ptr_array_add(port->connection->iterators, iterator);
 429         BT_LIB_LOGD("Created notification 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_notification_iterator_get_data(
 438                 struct bt_self_notification_iterator *self_iterator)
 439 {
 440         struct bt_self_component_port_input_notification_iterator *iterator =
 441                 (void *) self_iterator;
 442
 443         BT_ASSERT_PRE_NON_NULL(iterator, "Notification iterator");
 444         return iterator->user_data;
 445 }
 446
 447 void bt_self_notification_iterator_set_data(
 448                 struct bt_self_notification_iterator *self_iterator, void *data)
 449 {
 450         struct bt_self_component_port_input_notification_iterator *iterator =
 451                 (void *) self_iterator;
 452
 453         BT_ASSERT_PRE_NON_NULL(iterator, "Notification iterator");
 454         iterator->user_data = data;
 455         BT_LIB_LOGV("Set notification 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_notification_borrow_packet_stream(struct bt_notification *notif,
 462                 const struct bt_stream **stream,
 463                 const struct bt_packet **packet)
 464 {
 465         BT_ASSERT(notif);
 466
 467         switch (notif->type) {
 468         case BT_NOTIFICATION_TYPE_EVENT:
 469                 *packet = bt_event_borrow_packet_const(
 470                         bt_notification_event_borrow_event(notif));
 471                 *stream = bt_packet_borrow_stream_const(*packet);
 472                 break;
 473         case BT_NOTIFICATION_TYPE_STREAM_BEGIN:
 474                 *stream = bt_notification_stream_begin_borrow_stream(notif);
 475                 break;
 476         case BT_NOTIFICATION_TYPE_STREAM_END:
 477                 *stream = bt_notification_stream_end_borrow_stream(notif);
 478                 break;
 479         case BT_NOTIFICATION_TYPE_PACKET_BEGIN:
 480                 *packet = bt_notification_packet_begin_borrow_packet(notif);
 481                 *stream = bt_packet_borrow_stream_const(*packet);
 482                 break;
 483         case BT_NOTIFICATION_TYPE_PACKET_END:
 484                 *packet = bt_notification_packet_end_borrow_packet(notif);
 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_notification(
 495                 struct bt_self_component_port_input_notification_iterator *iterator,
 496                 struct bt_notification *notif)
 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
 503         BT_ASSERT(notif);
 504         bt_notification_borrow_packet_stream(notif, &stream, &packet);
 505
 506         if (!stream) {
 507                 /* we don't care about notifications not attached to streams */
 508                 goto end;
 509         }
 510
 511         stream_state = g_hash_table_lookup(iterator->stream_states, stream);
 512         if (!stream_state) {
 513                 /*
 514                  * No stream state for this stream: this notification
 515                  * MUST be a BT_NOTIFICATION_TYPE_STREAM_BEGIN notification
 516                  * and its sequence number must be 0.
 517                  */
 518                 if (notif->type != BT_NOTIFICATION_TYPE_STREAM_BEGIN) {
 519                         BT_ASSERT_PRE_MSG("Unexpected notification: missing a "
 520                                 "BT_NOTIFICATION_TYPE_STREAM_BEGIN "
 521                                 "notification prior to this notification: "
 522                                 "%![stream-]+s", stream);
 523                         is_valid = false;
 524                         goto end;
 525                 }
 526
 527                 if (notif->seq_num == -1ULL) {
 528                         notif->seq_num = 0;
 529                 }
 530
 531                 if (notif->seq_num != 0) {
 532                         BT_ASSERT_PRE_MSG("Unexpected notification sequence "
 533                                 "number for this notification iterator: "
 534                                 "this is the first notification for this "
 535                                 "stream, expecting sequence number 0: "
 536                                 "seq-num=%" PRIu64 ", %![stream-]+s",
 537                                 notif->seq_num, stream);
 538                         is_valid = false;
 539                         goto end;
 540                 }
 541
 542                 stream_state = create_stream_state(stream);
 543                 if (!stream_state) {
 544                         abort();
 545                 }
 546
 547                 g_hash_table_insert(iterator->stream_states,
 548                         (void *) stream, stream_state);
 549                 stream_state->expected_notif_seq_num++;
 550                 goto end;
 551         }
 552
 553         if (stream_state->is_ended) {
 554                 /*
 555                  * There's a new notification which has a reference to a
 556                  * stream which, from this iterator's point of view, is
 557                  * ended ("end of stream" notification was returned).
 558                  * This is bad: the API guarantees that it can never
 559                  * happen.
 560                  */
 561                 BT_ASSERT_PRE_MSG("Stream is already ended: %![stream-]+s",
 562                         stream);
 563                 is_valid = false;
 564                 goto end;
 565         }
 566
 567         if (notif->seq_num == -1ULL) {
 568                 notif->seq_num = stream_state->expected_notif_seq_num;
 569         }
 570
 571         if (notif->seq_num != -1ULL &&
 572                         notif->seq_num != stream_state->expected_notif_seq_num) {
 573                 BT_ASSERT_PRE_MSG("Unexpected notification sequence number: "
 574                         "seq-num=%" PRIu64 ", "
 575                         "expected-seq-num=%" PRIu64 ", %![stream-]+s",
 576                         notif->seq_num, stream_state->expected_notif_seq_num,
 577                         stream);
 578                 is_valid = false;
 579                 goto end;
 580         }
 581
 582         switch (notif->type) {
 583         case BT_NOTIFICATION_TYPE_STREAM_BEGIN:
 584                 BT_ASSERT_PRE_MSG("Unexpected BT_NOTIFICATION_TYPE_STREAM_BEGIN "
 585                         "notification at this point: notif-seq-num=%" PRIu64 ", "
 586                         "%![stream-]+s", notif->seq_num, stream);
 587                 is_valid = false;
 588                 goto end;
 589         case BT_NOTIFICATION_TYPE_STREAM_END:
 590                 if (stream_state->cur_packet) {
 591                         BT_ASSERT_PRE_MSG("Unexpected BT_NOTIFICATION_TYPE_STREAM_END "
 592                                 "notification: missing a "
 593                                 "BT_NOTIFICATION_TYPE_PACKET_END notification "
 594                                 "prior to this notification: "
 595                                 "notif-seq-num=%" PRIu64 ", "
 596                                 "%![stream-]+s", notif->seq_num, stream);
 597                         is_valid = false;
 598                         goto end;
 599                 }
 600                 stream_state->expected_notif_seq_num++;
 601                 stream_state->is_ended = true;
 602                 goto end;
 603         case BT_NOTIFICATION_TYPE_PACKET_BEGIN:
 604                 if (stream_state->cur_packet) {
 605                         BT_ASSERT_PRE_MSG("Unexpected BT_NOTIFICATION_TYPE_PACKET_BEGIN "
 606                                 "notification at this point: missing a "
 607                                 "BT_NOTIFICATION_TYPE_PACKET_END notification "
 608                                 "prior to this notification: "
 609                                 "notif-seq-num=%" PRIu64 ", %![stream-]+s, "
 610                                 "%![packet-]+a", notif->seq_num, stream,
 611                                 packet);
 612                         is_valid = false;
 613                         goto end;
 614                 }
 615                 stream_state->expected_notif_seq_num++;
 616                 stream_state->cur_packet = packet;
 617                 bt_object_get_no_null_check(stream_state->cur_packet);
 618                 goto end;
 619         case BT_NOTIFICATION_TYPE_PACKET_END:
 620                 if (!stream_state->cur_packet) {
 621                         BT_ASSERT_PRE_MSG("Unexpected BT_NOTIFICATION_TYPE_PACKET_END "
 622                                 "notification at this point: missing a "
 623                                 "BT_NOTIFICATION_TYPE_PACKET_BEGIN notification "
 624                                 "prior to this notification: "
 625                                 "notif-seq-num=%" PRIu64 ", %![stream-]+s, "
 626                                 "%![packet-]+a", notif->seq_num, stream,
 627                                 packet);
 628                         is_valid = false;
 629                         goto end;
 630                 }
 631                 stream_state->expected_notif_seq_num++;
 632                 BT_OBJECT_PUT_REF_AND_RESET(stream_state->cur_packet);
 633                 goto end;
 634         case BT_NOTIFICATION_TYPE_EVENT:
 635                 if (packet != stream_state->cur_packet) {
 636                         BT_ASSERT_PRE_MSG("Unexpected packet for "
 637                                 "BT_NOTIFICATION_TYPE_EVENT notification: "
 638                                 "notif-seq-num=%" PRIu64 ", %![stream-]+s, "
 639                                 "%![notif-packet-]+a, %![expected-packet-]+a",
 640                                 notif->seq_num, stream,
 641                                 stream_state->cur_packet, packet);
 642                         is_valid = false;
 643                         goto end;
 644                 }
 645                 stream_state->expected_notif_seq_num++;
 646                 goto end;
 647         default:
 648                 break;
 649         }
 650
 651 end:
 652         return is_valid;
 653 }
 654
 655 BT_ASSERT_PRE_FUNC
 656 static inline
 657 bool validate_notifications(
 658                 struct bt_self_component_port_input_notification_iterator *iterator,
 659                 uint64_t count)
 660 {
 661         bool ret = true;
 662         bt_notification_array notifs = (void *) iterator->base.notifs->pdata;
 663         uint64_t i;
 664
 665         for (i = 0; i < count; i++) {
 666                 ret = validate_notification(iterator, notifs[i]);
 667                 if (!ret) {
 668                         break;
 669                 }
 670         }
 671
 672         return ret;
 673 }
 674
 675 BT_ASSERT_PRE_FUNC
 676 static inline bool priv_conn_notif_iter_can_end(
 677                 struct bt_self_component_port_input_notification_iterator *iterator)
 678 {
 679         GHashTableIter iter;
 680         gpointer stream_key, state_value;
 681         bool ret = true;
 682
 683         /*
 684          * Verify that this iterator received a
 685          * BT_NOTIFICATION_TYPE_STREAM_END notification for each stream
 686          * which has a state.
 687          */
 688
 689         g_hash_table_iter_init(&iter, iterator->stream_states);
 690
 691         while (g_hash_table_iter_next(&iter, &stream_key, &state_value)) {
 692                 struct stream_state *stream_state = (void *) state_value;
 693
 694                 BT_ASSERT(stream_state);
 695                 BT_ASSERT(stream_key);
 696
 697                 if (!stream_state->is_ended) {
 698                         BT_ASSERT_PRE_MSG("Ending notification iterator, "
 699                                 "but stream is not ended: "
 700                                 "%![stream-]s", stream_key);
 701                         ret = false;
 702                         goto end;
 703                 }
 704         }
 705
 706 end:
 707         return ret;
 708 }
 709
 710 enum bt_notification_iterator_status
 711 bt_self_component_port_input_notification_iterator_next(
 712                 struct bt_self_component_port_input_notification_iterator *iterator,
 713                 bt_notification_array *notifs, uint64_t *user_count)
 714 {
 715         typedef enum bt_self_notification_iterator_status (*method_t)(
 716                         void *, bt_notification_array, uint64_t, uint64_t *);
 717
 718         method_t method = NULL;
 719         struct bt_component_class *comp_cls;
 720         int status = BT_NOTIFICATION_ITERATOR_STATUS_OK;
 721
 722         BT_ASSERT_PRE_NON_NULL(iterator, "Notification iterator");
 723         BT_ASSERT_PRE_NON_NULL(notifs, "Notification array (output)");
 724         BT_ASSERT_PRE_NON_NULL(user_count, "Notification count (output)");
 725         BT_ASSERT_PRE(iterator->state ==
 726                 BT_SELF_COMPONENT_PORT_INPUT_NOTIFICATION_ITERATOR_STATE_ACTIVE,
 727                 "Notification iterator's \"next\" called, but "
 728                 "iterator is in the wrong state: %!+i", iterator);
 729         BT_ASSERT(iterator->upstream_component);
 730         BT_ASSERT(iterator->upstream_component->class);
 731         BT_LIB_LOGD("Getting next self component input port "
 732                 "notification iterator's notifications: %!+i", iterator);
 733         comp_cls = iterator->upstream_component->class;
 734
 735         /* Pick the appropriate "next" method */
 736         switch (comp_cls->type) {
 737         case BT_COMPONENT_CLASS_TYPE_SOURCE:
 738         {
 739                 struct bt_component_class_source *src_comp_cls =
 740                         (void *) comp_cls;
 741
 742                 method = (method_t) src_comp_cls->methods.notif_iter_next;
 743                 break;
 744         }
 745         case BT_COMPONENT_CLASS_TYPE_FILTER:
 746         {
 747                 struct bt_component_class_filter *flt_comp_cls =
 748                         (void *) comp_cls;
 749
 750                 method = (method_t) flt_comp_cls->methods.notif_iter_next;
 751                 break;
 752         }
 753         default:
 754                 abort();
 755         }
 756
 757         /*
 758          * Call the user's "next" method to get the next notifications
 759          * and status.
 760          */
 761         BT_ASSERT(method);
 762         BT_LOGD_STR("Calling user's \"next\" method.");
 763         status = method(iterator,
 764                 (void *) iterator->base.notifs->pdata,
 765                 NOTIF_BATCH_SIZE, user_count);
 766         BT_LOGD("User method returned: status=%s",
 767                 bt_notification_iterator_status_string(status));
 768         if (status < 0) {
 769                 BT_LOGW_STR("User method failed.");
 770                 goto end;
 771         }
 772
 773         if (iterator->state == BT_SELF_COMPONENT_PORT_INPUT_NOTIFICATION_ITERATOR_STATE_FINALIZED ||
 774                         iterator->state == BT_SELF_COMPONENT_PORT_INPUT_NOTIFICATION_ITERATOR_STATE_FINALIZED_AND_ENDED) {
 775                 /*
 776                  * The user's "next" method, somehow, cancelled its own
 777                  * notification iterator. This can happen, for example,
 778                  * when the user's method removes the port on which
 779                  * there's the connection from which the iterator was
 780                  * created. In this case, said connection is ended, and
 781                  * all its notification iterators are finalized.
 782                  *
 783                  * Only bt_object_put_ref() the returned notification if
 784                  * the status is BT_NOTIFICATION_ITERATOR_STATUS_OK
 785                  * because otherwise this field could be garbage.
 786                  */
 787                 if (status == BT_NOTIFICATION_ITERATOR_STATUS_OK) {
 788                         uint64_t i;
 789                         bt_notification_array notifs =
 790                                 (void *) iterator->base.notifs->pdata;
 791
 792                         for (i = 0; i < *user_count; i++) {
 793                                 bt_object_put_ref(notifs[i]);
 794                         }
 795                 }
 796
 797                 status = BT_NOTIFICATION_ITERATOR_STATUS_CANCELED;
 798                 goto end;
 799         }
 800
 801         switch (status) {
 802         case BT_NOTIFICATION_ITERATOR_STATUS_OK:
 803                 BT_ASSERT_PRE(validate_notifications(iterator, *user_count),
 804                         "Notifications are invalid at this point: "
 805                         "%![notif-iter-]+i, count=%" PRIu64,
 806                         iterator, *user_count);
 807                 *notifs = (void *) iterator->base.notifs->pdata;
 808                 break;
 809         case BT_NOTIFICATION_ITERATOR_STATUS_AGAIN:
 810                 goto end;
 811         case BT_NOTIFICATION_ITERATOR_STATUS_END:
 812                 BT_ASSERT_PRE(priv_conn_notif_iter_can_end(iterator),
 813                         "Notification iterator cannot end at this point: "
 814                         "%!+i", iterator);
 815                 BT_ASSERT(iterator->state ==
 816                         BT_SELF_COMPONENT_PORT_INPUT_NOTIFICATION_ITERATOR_STATE_ACTIVE);
 817                 iterator->state = BT_SELF_COMPONENT_PORT_INPUT_NOTIFICATION_ITERATOR_STATE_ENDED;
 818                 BT_LOGD("Set new status: status=%s",
 819                         bt_notification_iterator_status_string(status));
 820                 goto end;
 821         default:
 822                 /* Unknown non-error status */
 823                 abort();
 824         }
 825
 826 end:
 827         return status;
 828 }
 829
 830 enum bt_notification_iterator_status
 831 bt_port_output_notification_iterator_next(
 832                 struct bt_port_output_notification_iterator *iterator,
 833                 bt_notification_array *notifs_to_user,
 834                 uint64_t *count_to_user)
 835 {
 836         enum bt_notification_iterator_status status;
 837         enum bt_graph_status graph_status;
 838
 839         BT_ASSERT_PRE_NON_NULL(iterator, "Notification iterator");
 840         BT_ASSERT_PRE_NON_NULL(notifs_to_user, "Notification array (output)");
 841         BT_ASSERT_PRE_NON_NULL(count_to_user, "Notification count (output)");
 842         BT_LIB_LOGD("Getting next output port notification iterator's notifications: "
 843                 "%!+i", iterator);
 844
 845         graph_status = bt_graph_consume_sink_no_check(iterator->graph,
 846                 iterator->colander);
 847         switch (graph_status) {
 848         case BT_GRAPH_STATUS_CANCELED:
 849         case BT_GRAPH_STATUS_AGAIN:
 850         case BT_GRAPH_STATUS_END:
 851         case BT_GRAPH_STATUS_NOMEM:
 852                 status = (int) graph_status;
 853                 break;
 854         case BT_GRAPH_STATUS_OK:
 855                 status = BT_NOTIFICATION_ITERATOR_STATUS_OK;
 856
 857                 /*
 858                  * On success, the colander sink moves the notifications
 859                  * to this iterator's array and sets this iterator's
 860                  * notification count: move them to the user.
 861                  */
 862                 *notifs_to_user = (void *) iterator->base.notifs->pdata;
 863                 *count_to_user = iterator->count;
 864                 break;
 865         default:
 866                 /* Other errors */
 867                 status = BT_NOTIFICATION_ITERATOR_STATUS_ERROR;
 868         }
 869
 870         return status;
 871 }
 872
 873 struct bt_component *bt_self_component_port_input_notification_iterator_borrow_component(
 874                 struct bt_self_component_port_input_notification_iterator *iterator)
 875 {
 876         BT_ASSERT_PRE_NON_NULL(iterator, "Notification iterator");
 877         return iterator->upstream_component;
 878 }
 879
 880 struct bt_self_component *bt_self_notification_iterator_borrow_component(
 881                 struct bt_self_notification_iterator *self_iterator)
 882 {
 883         struct bt_self_component_port_input_notification_iterator *iterator =
 884                 (void *) self_iterator;
 885
 886         BT_ASSERT_PRE_NON_NULL(iterator, "Notification iterator");
 887         return (void *) iterator->upstream_component;
 888 }
 889
 890 struct bt_self_port_output *bt_self_notification_iterator_borrow_port(
 891                 struct bt_self_notification_iterator *self_iterator)
 892 {
 893         struct bt_self_component_port_input_notification_iterator *iterator =
 894                 (void *) self_iterator;
 895
 896         BT_ASSERT_PRE_NON_NULL(iterator, "Notification iterator");
 897         return (void *) iterator->upstream_port;
 898 }
 899
 900 static
 901 void bt_port_output_notification_iterator_destroy(struct bt_object *obj)
 902 {
 903         struct bt_port_output_notification_iterator *iterator = (void *) obj;
 904
 905         BT_LIB_LOGD("Destroying output port notification iterator object: %!+i",
 906                 iterator);
 907         BT_LOGD_STR("Putting graph.");
 908         BT_OBJECT_PUT_REF_AND_RESET(iterator->graph);
 909         BT_LOGD_STR("Putting colander sink component.");
 910         BT_OBJECT_PUT_REF_AND_RESET(iterator->colander);
 911         destroy_base_notification_iterator(obj);
 912 }
 913
 914 struct bt_port_output_notification_iterator *
 915 bt_port_output_notification_iterator_create(
 916                 struct bt_private_graph *priv_graph,
 917                 struct bt_port_output *output_port)
 918 {
 919         struct bt_port_output_notification_iterator *iterator = NULL;
 920         struct bt_component_class_sink *colander_comp_cls = NULL;
 921         struct bt_component *output_port_comp = NULL;
 922         struct bt_component_sink *colander_comp;
 923         struct bt_graph *graph = (void *) priv_graph;
 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((void *) output_port);
 932         BT_ASSERT_PRE(output_port_comp,
 933                 "Output port has no component: %!+p", output_port);
 934         BT_ASSERT_PRE(bt_component_borrow_graph(output_port_comp) ==
 935                 (void *) graph,
 936                 "Output port is not part of graph: %![graph-]+g, %![port-]+p",
 937                 graph, output_port);
 938
 939         /* Create notification iterator */
 940         BT_LIB_LOGD("Creating notification iterator on output port: "
 941                 "%![port-]+p, %![comp-]+c", output_port, output_port_comp);
 942         iterator = g_new0(struct bt_port_output_notification_iterator, 1);
 943         if (!iterator) {
 944                 BT_LOGE_STR("Failed to allocate one output port notification iterator.");
 945                 goto error;
 946         }
 947
 948         ret = init_notification_iterator((void *) iterator,
 949                 BT_NOTIFICATION_ITERATOR_TYPE_PORT_OUTPUT,
 950                 bt_port_output_notification_iterator_destroy);
 951         if (ret) {
 952                 /* init_notification_iterator() logs errors */
 953                 BT_OBJECT_PUT_REF_AND_RESET(iterator);
 954                 goto end;
 955         }
 956
 957         /* Create colander component */
 958         colander_comp_cls = bt_component_class_sink_colander_get();
 959         if (!colander_comp_cls) {
 960                 BT_LOGW("Cannot get colander sink component class.");
 961                 goto error;
 962         }
 963
 964         iterator->graph = graph;
 965         bt_object_get_no_null_check(iterator->graph);
 966         colander_data.notifs = (void *) iterator->base.notifs->pdata;
 967         colander_data.count_addr = &iterator->count;
 968
 969         /* Hope that nobody uses this very unique name */
 970         graph_status =
 971                 bt_private_graph_add_sink_component_with_init_method_data(
 972                         (void *) graph, colander_comp_cls,
 973                         "colander-36ac3409-b1a8-4d60-ab1f-4fdf341a8fb1",
 974                         NULL, &colander_data, &iterator->colander);
 975         if (graph_status != BT_GRAPH_STATUS_OK) {
 976                 BT_LIB_LOGW("Cannot add colander sink component to graph: "
 977                         "%1[graph-]+g, status=%s", graph,
 978                         bt_graph_status_string(graph_status));
 979                 goto error;
 980         }
 981
 982         /*
 983          * Connect provided output port to the colander component's
 984          * input port.
 985          */
 986         colander_in_port = bt_component_sink_borrow_input_port_by_index(
 987                 iterator->colander, 0);
 988         BT_ASSERT(colander_in_port);
 989         graph_status = bt_private_graph_connect_ports(priv_graph,
 990                 output_port, colander_in_port, NULL);
 991         if (graph_status != BT_GRAPH_STATUS_OK) {
 992                 BT_LIB_LOGW("Cannot add colander sink component to graph: "
 993                         "%![graph-]+g, %![comp-]+c, status=%s", graph,
 994                         iterator->colander,
 995                         bt_graph_status_string(graph_status));
 996                 goto error;
 997         }
 998
 999         /*
1000          * At this point everything went fine. Make the graph
1001          * nonconsumable forever so that only this notification iterator
1002          * can consume (thanks to bt_graph_consume_sink_no_check()).
1003          * This avoids leaking the notification created by the colander
1004          * sink and moved to the notification iterator's notification
1005          * member.
1006          */
1007         bt_graph_set_can_consume(iterator->graph, false);
1008         goto end;
1009
1010 error:
1011         if (iterator && iterator->graph && iterator->colander) {
1012                 int ret;
1013
1014                 /* Remove created colander component from graph if any */
1015                 colander_comp = iterator->colander;
1016                 BT_OBJECT_PUT_REF_AND_RESET(iterator->colander);
1017
1018                 /*
1019                  * At this point the colander component's reference
1020                  * count is 0 because iterator->colander was the only
1021                  * owner. We also know that it is not connected because
1022                  * this is the last operation before this function
1023                  * succeeds.
1024                  *
1025                  * Since we honor the preconditions here,
1026                  * bt_graph_remove_unconnected_component() always
1027                  * succeeds.
1028                  */
1029                 ret = bt_graph_remove_unconnected_component(iterator->graph,
1030                         (void *) colander_comp);
1031                 BT_ASSERT(ret == 0);
1032         }
1033
1034         BT_OBJECT_PUT_REF_AND_RESET(iterator);
1035
1036 end:
1037         bt_object_put_ref(colander_comp_cls);
1038         return (void *) iterator;
1039 }