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