lib/graph/iterator.c

   1 /*
   2  * iterator.c
   3  *
   4  * Babeltrace Notification Iterator
   5  *
   6  * Copyright 2015 Jérémie Galarneau <jeremie.galarneau@efficios.com>
   7  * Copyright 2017 Philippe Proulx <pproulx@efficios.com>
   8  *
   9  * Permission is hereby granted, free of charge, to any person obtaining a copy
  10  * of this software and associated documentation files (the "Software"), to deal
  11  * in the Software without restriction, including without limitation the rights
  12  * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
  13  * copies of the Software, and to permit persons to whom the Software is
  14  * furnished to do so, subject to the following conditions:
  15  *
  16  * The above copyright notice and this permission notice shall be included in
  17  * all copies or substantial portions of the Software.
  18  *
  19  * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
  20  * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
  21  * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
  22  * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
  23  * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
  24  * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
  25  * SOFTWARE.
  26  */
  27
  28 #define BT_LOG_TAG "NOTIF-ITER"
  29 #include <babeltrace/lib-logging-internal.h>
  30
  31 #include <babeltrace/compiler-internal.h>
  32 #include <babeltrace/ref.h>
  33 #include <babeltrace/ctf-ir/fields.h>
  34 #include <babeltrace/ctf-ir/field-types.h>
  35 #include <babeltrace/ctf-ir/field-types-internal.h>
  36 #include <babeltrace/ctf-ir/event-internal.h>
  37 #include <babeltrace/ctf-ir/packet-internal.h>
  38 #include <babeltrace/ctf-ir/stream-internal.h>
  39 #include <babeltrace/graph/connection.h>
  40 #include <babeltrace/graph/connection-internal.h>
  41 #include <babeltrace/graph/component.h>
  42 #include <babeltrace/graph/component-source-internal.h>
  43 #include <babeltrace/graph/component-class-internal.h>
  44 #include <babeltrace/graph/component-class-sink-colander-internal.h>
  45 #include <babeltrace/graph/component-sink.h>
  46 #include <babeltrace/graph/notification.h>
  47 #include <babeltrace/graph/notification-iterator.h>
  48 #include <babeltrace/graph/notification-iterator-internal.h>
  49 #include <babeltrace/graph/notification-internal.h>
  50 #include <babeltrace/graph/notification-event.h>
  51 #include <babeltrace/graph/notification-event-internal.h>
  52 #include <babeltrace/graph/notification-packet.h>
  53 #include <babeltrace/graph/notification-packet-internal.h>
  54 #include <babeltrace/graph/notification-stream.h>
  55 #include <babeltrace/graph/notification-stream-internal.h>
  56 #include <babeltrace/graph/notification-discarded-elements-internal.h>
  57 #include <babeltrace/graph/port.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 discarded_elements_state {
  73         struct bt_clock_value *cur_begin;
  74         uint64_t cur_count;
  75 };
  76
  77 struct stream_state {
  78         struct bt_stream *stream; /* owned by this */
  79         struct bt_packet *cur_packet; /* owned by this */
  80         struct discarded_elements_state discarded_packets_state;
  81         struct discarded_elements_state discarded_events_state;
  82         uint64_t expected_notif_seq_num;
  83         bt_bool is_ended;
  84 };
  85
  86 BT_ASSERT_PRE_FUNC
  87 static
  88 void destroy_stream_state(struct stream_state *stream_state)
  89 {
  90         if (!stream_state) {
  91                 return;
  92         }
  93
  94         BT_LOGV("Destroying stream state: stream-state-addr=%p", stream_state);
  95         BT_LOGV_STR("Putting stream state's current packet.");
  96         bt_put(stream_state->cur_packet);
  97         BT_LOGV_STR("Putting stream state's stream.");
  98         bt_put(stream_state->stream);
  99         bt_put(stream_state->discarded_packets_state.cur_begin);
 100         bt_put(stream_state->discarded_events_state.cur_begin);
 101         g_free(stream_state);
 102 }
 103
 104 BT_ASSERT_PRE_FUNC
 105 static
 106 struct stream_state *create_stream_state(struct bt_stream *stream)
 107 {
 108         struct stream_state *stream_state = g_new0(struct stream_state, 1);
 109
 110         if (!stream_state) {
 111                 BT_LOGE_STR("Failed to allocate one stream state.");
 112                 goto end;
 113         }
 114
 115         /*
 116          * The packet index is a monotonic counter which may not start
 117          * at 0 at the beginning of the stream. We therefore need to
 118          * have an internal object initial state of -1ULL to distinguish
 119          * between initial state and having seen a packet with
 120          * the sequence number 0.
 121          */
 122         stream_state->discarded_packets_state.cur_count = -1ULL;
 123
 124         /*
 125          * We keep a reference to the stream until we know it's ended.
 126          */
 127         stream_state->stream = bt_get(stream);
 128         BT_LOGV("Created stream state: stream-addr=%p, stream-name=\"%s\", "
 129                 "stream-state-addr=%p",
 130                 stream, bt_stream_get_name(stream), stream_state);
 131
 132 end:
 133         return stream_state;
 134 }
 135
 136 static
 137 void destroy_base_notification_iterator(struct bt_object *obj)
 138 {
 139         struct bt_notification_iterator *iterator = (void *) obj;
 140
 141         BT_ASSERT(iterator);
 142
 143         if (iterator->notifs) {
 144                 g_ptr_array_free(iterator->notifs, TRUE);
 145         }
 146
 147         g_free(iterator);
 148 }
 149
 150 static
 151 void bt_private_connection_notification_iterator_destroy(struct bt_object *obj)
 152 {
 153         struct bt_notification_iterator_private_connection *iterator;
 154
 155         BT_ASSERT(obj);
 156
 157         /*
 158          * The notification iterator's reference count is 0 if we're
 159          * here. Increment it to avoid a double-destroy (possibly
 160          * infinitely recursive). This could happen for example if the
 161          * notification iterator's finalization function does bt_get()
 162          * (or anything that causes bt_get() to be called) on itself
 163          * (ref. count goes from 0 to 1), and then bt_put(): the
 164          * reference count would go from 1 to 0 again and this function
 165          * would be called again.
 166          */
 167         obj->ref_count++;
 168         iterator = (void *) obj;
 169         BT_LOGD("Destroying private connection notification iterator object: addr=%p",
 170                 iterator);
 171         bt_private_connection_notification_iterator_finalize(iterator);
 172
 173         if (iterator->stream_states) {
 174                 /*
 175                  * Remove our destroy listener from each stream which
 176                  * has a state in this iterator. Otherwise the destroy
 177                  * listener would be called with an invalid/other
 178                  * notification iterator object.
 179                  */
 180                 g_hash_table_destroy(iterator->stream_states);
 181         }
 182
 183         if (iterator->connection) {
 184                 /*
 185                  * Remove ourself from the originating connection so
 186                  * that it does not try to finalize a dangling pointer
 187                  * later.
 188                  */
 189                 bt_connection_remove_iterator(iterator->connection, iterator);
 190         }
 191
 192         destroy_base_notification_iterator(obj);
 193 }
 194
 195 BT_HIDDEN
 196 void bt_private_connection_notification_iterator_finalize(
 197                 struct bt_notification_iterator_private_connection *iterator)
 198 {
 199         struct bt_component_class *comp_class = NULL;
 200         bt_component_class_notification_iterator_finalize_method
 201                 finalize_method = NULL;
 202
 203         BT_ASSERT(iterator);
 204
 205         switch (iterator->state) {
 206         case BT_PRIVATE_CONNECTION_NOTIFICATION_ITERATOR_STATE_NON_INITIALIZED:
 207                 /* Skip user finalization if user initialization failed */
 208                 BT_LOGD("Not finalizing non-initialized notification iterator: "
 209                         "addr=%p", iterator);
 210                 return;
 211         case BT_PRIVATE_CONNECTION_NOTIFICATION_ITERATOR_STATE_FINALIZED:
 212         case BT_PRIVATE_CONNECTION_NOTIFICATION_ITERATOR_STATE_FINALIZED_AND_ENDED:
 213                 /* Already finalized */
 214                 BT_LOGD("Not finalizing notification iterator: already finalized: "
 215                         "addr=%p", iterator);
 216                 return;
 217         default:
 218                 break;
 219         }
 220
 221         BT_LOGD("Finalizing notification iterator: addr=%p", iterator);
 222
 223         if (iterator->state == BT_PRIVATE_CONNECTION_NOTIFICATION_ITERATOR_STATE_ENDED) {
 224                 BT_LOGD("Updating notification iterator's state: "
 225                         "new-state=BT_PRIVATE_CONNECTION_NOTIFICATION_ITERATOR_STATE_FINALIZED_AND_ENDED");
 226                 iterator->state = BT_PRIVATE_CONNECTION_NOTIFICATION_ITERATOR_STATE_FINALIZED_AND_ENDED;
 227         } else {
 228                 BT_LOGD("Updating notification iterator's state: "
 229                         "new-state=BT_PRIVATE_CONNECTION_NOTIFICATION_ITERATOR_STATE_FINALIZED");
 230                 iterator->state = BT_PRIVATE_CONNECTION_NOTIFICATION_ITERATOR_STATE_FINALIZED;
 231         }
 232
 233         BT_ASSERT(iterator->upstream_component);
 234         comp_class = iterator->upstream_component->class;
 235
 236         /* Call user-defined destroy method */
 237         switch (comp_class->type) {
 238         case BT_COMPONENT_CLASS_TYPE_SOURCE:
 239         {
 240                 struct bt_component_class_source *source_class;
 241
 242                 source_class = container_of(comp_class, struct bt_component_class_source, parent);
 243                 finalize_method = source_class->methods.iterator.finalize;
 244                 break;
 245         }
 246         case BT_COMPONENT_CLASS_TYPE_FILTER:
 247         {
 248                 struct bt_component_class_filter *filter_class;
 249
 250                 filter_class = container_of(comp_class, struct bt_component_class_filter, parent);
 251                 finalize_method = filter_class->methods.iterator.finalize;
 252                 break;
 253         }
 254         default:
 255                 /* Unreachable */
 256                 abort();
 257         }
 258
 259         if (finalize_method) {
 260                 BT_LOGD("Calling user's finalization method: addr=%p",
 261                         iterator);
 262                 finalize_method(
 263                         bt_private_connection_private_notification_iterator_from_notification_iterator(iterator));
 264         }
 265
 266         iterator->upstream_component = NULL;
 267         iterator->upstream_port = NULL;
 268         BT_LOGD("Finalized notification iterator: addr=%p", iterator);
 269 }
 270
 271 BT_HIDDEN
 272 void bt_private_connection_notification_iterator_set_connection(
 273                 struct bt_notification_iterator_private_connection *iterator,
 274                 struct bt_connection *connection)
 275 {
 276         BT_ASSERT(iterator);
 277         iterator->connection = connection;
 278         BT_LOGV("Set notification iterator's connection: "
 279                 "iter-addr=%p, conn-addr=%p", iterator, connection);
 280 }
 281
 282 static
 283 int init_notification_iterator(struct bt_notification_iterator *iterator,
 284                 enum bt_notification_iterator_type type,
 285                 bt_object_release_func destroy)
 286 {
 287         int ret = 0;
 288
 289         bt_object_init_shared(&iterator->base, destroy);
 290         iterator->type = type;
 291         iterator->notifs = g_ptr_array_new();
 292         if (!iterator->notifs) {
 293                 BT_LOGE_STR("Failed to allocate a GPtrArray.");
 294                 ret = -1;
 295                 goto end;
 296         }
 297
 298         g_ptr_array_set_size(iterator->notifs, NOTIF_BATCH_SIZE);
 299
 300 end:
 301         return ret;
 302 }
 303
 304 BT_HIDDEN
 305 enum bt_connection_status bt_private_connection_notification_iterator_create(
 306                 struct bt_component *upstream_comp,
 307                 struct bt_port *upstream_port,
 308                 struct bt_connection *connection,
 309                 struct bt_notification_iterator_private_connection **user_iterator)
 310 {
 311         enum bt_connection_status status = BT_CONNECTION_STATUS_OK;
 312         enum bt_component_class_type type;
 313         struct bt_notification_iterator_private_connection *iterator = NULL;
 314         int ret;
 315
 316         BT_ASSERT(upstream_comp);
 317         BT_ASSERT(upstream_port);
 318         BT_ASSERT(bt_port_is_connected(upstream_port));
 319         BT_ASSERT(user_iterator);
 320         BT_LOGD("Creating notification iterator on private connection: "
 321                 "upstream-comp-addr=%p, upstream-comp-name=\"%s\", "
 322                 "upstream-port-addr=%p, upstream-port-name=\"%s\", "
 323                 "conn-addr=%p",
 324                 upstream_comp, bt_component_get_name(upstream_comp),
 325                 upstream_port, bt_port_get_name(upstream_port),
 326                 connection);
 327         type = bt_component_get_class_type(upstream_comp);
 328         BT_ASSERT(type == BT_COMPONENT_CLASS_TYPE_SOURCE ||
 329                 type == BT_COMPONENT_CLASS_TYPE_FILTER);
 330         iterator = g_new0(struct bt_notification_iterator_private_connection, 1);
 331         if (!iterator) {
 332                 BT_LOGE_STR("Failed to allocate one private connection notification iterator.");
 333                 status = BT_CONNECTION_STATUS_NOMEM;
 334                 goto end;
 335         }
 336
 337         ret = init_notification_iterator((void *) iterator,
 338                 BT_NOTIFICATION_ITERATOR_TYPE_PRIVATE_CONNECTION,
 339                 bt_private_connection_notification_iterator_destroy);
 340         if (ret) {
 341                 /* init_notification_iterator() logs errors */
 342                 status = BT_CONNECTION_STATUS_NOMEM;
 343                 goto end;
 344         }
 345
 346         iterator->stream_states = g_hash_table_new_full(g_direct_hash,
 347                 g_direct_equal, NULL, (GDestroyNotify) destroy_stream_state);
 348         if (!iterator->stream_states) {
 349                 BT_LOGE_STR("Failed to allocate a GHashTable.");
 350                 status = BT_CONNECTION_STATUS_NOMEM;
 351                 goto end;
 352         }
 353
 354         iterator->upstream_component = upstream_comp;
 355         iterator->upstream_port = upstream_port;
 356         iterator->connection = connection;
 357         iterator->graph = bt_component_borrow_graph(upstream_comp);
 358         iterator->state = BT_PRIVATE_CONNECTION_NOTIFICATION_ITERATOR_STATE_NON_INITIALIZED;
 359         BT_LOGD("Created notification iterator: "
 360                 "upstream-comp-addr=%p, upstream-comp-name=\"%s\", "
 361                 "upstream-port-addr=%p, upstream-port-name=\"%s\", "
 362                 "conn-addr=%p, iter-addr=%p",
 363                 upstream_comp, bt_component_get_name(upstream_comp),
 364                 upstream_port, bt_port_get_name(upstream_port),
 365                 connection, iterator);
 366
 367         /* Move reference to user */
 368         *user_iterator = iterator;
 369         iterator = NULL;
 370
 371 end:
 372         bt_put(iterator);
 373         return status;
 374 }
 375
 376 void *bt_private_connection_private_notification_iterator_get_user_data(
 377                 struct bt_private_connection_private_notification_iterator *private_iterator)
 378 {
 379         struct bt_notification_iterator_private_connection *iterator = (void *)
 380                 bt_private_connection_notification_iterator_borrow_from_private(private_iterator);
 381
 382         BT_ASSERT_PRE_NON_NULL(private_iterator, "Notification iterator");
 383         return iterator->user_data;
 384 }
 385
 386 enum bt_notification_iterator_status
 387 bt_private_connection_private_notification_iterator_set_user_data(
 388                 struct bt_private_connection_private_notification_iterator *private_iterator,
 389                 void *data)
 390 {
 391         struct bt_notification_iterator_private_connection *iterator = (void *)
 392                 bt_private_connection_notification_iterator_borrow_from_private(private_iterator);
 393
 394         BT_ASSERT_PRE_NON_NULL(iterator, "Notification iterator");
 395         iterator->user_data = data;
 396         BT_LOGV("Set notification iterator's user data: "
 397                 "iter-addr=%p, user-data-addr=%p", iterator, data);
 398         return BT_NOTIFICATION_ITERATOR_STATUS_OK;
 399 }
 400
 401 struct bt_graph *bt_private_connection_private_notification_iterator_borrow_graph(
 402                 struct bt_private_connection_private_notification_iterator *private_iterator)
 403 {
 404         struct bt_notification_iterator_private_connection *iterator = (void *)
 405                 bt_private_connection_notification_iterator_borrow_from_private(
 406                         private_iterator);
 407
 408         BT_ASSERT_PRE_NON_NULL(iterator, "Notification iterator");
 409         return iterator->graph;
 410 }
 411
 412 BT_ASSERT_PRE_FUNC
 413 static inline
 414 void bt_notification_borrow_packet_stream(struct bt_notification *notif,
 415                 struct bt_stream **stream, struct bt_packet **packet)
 416 {
 417         BT_ASSERT(notif);
 418
 419         switch (notif->type) {
 420         case BT_NOTIFICATION_TYPE_EVENT:
 421                 *packet = bt_event_borrow_packet(
 422                         bt_notification_event_borrow_event(notif));
 423                 *stream = bt_packet_borrow_stream(*packet);
 424                 break;
 425         case BT_NOTIFICATION_TYPE_STREAM_BEGIN:
 426                 *stream = bt_notification_stream_begin_borrow_stream(notif);
 427                 break;
 428         case BT_NOTIFICATION_TYPE_STREAM_END:
 429                 *stream = bt_notification_stream_end_borrow_stream(notif);
 430                 break;
 431         case BT_NOTIFICATION_TYPE_PACKET_BEGIN:
 432                 *packet = bt_notification_packet_begin_borrow_packet(notif);
 433                 *stream = bt_packet_borrow_stream(*packet);
 434                 break;
 435         case BT_NOTIFICATION_TYPE_PACKET_END:
 436                 *packet = bt_notification_packet_end_borrow_packet(notif);
 437                 *stream = bt_packet_borrow_stream(*packet);
 438                 break;
 439         default:
 440                 break;
 441         }
 442 }
 443
 444 BT_ASSERT_PRE_FUNC
 445 static inline
 446 bool validate_notification(
 447                 struct bt_notification_iterator_private_connection *iterator,
 448                 struct bt_notification *notif)
 449 {
 450         bool is_valid = true;
 451         struct stream_state *stream_state;
 452         struct bt_stream *stream = NULL;
 453         struct bt_packet *packet = NULL;
 454
 455         BT_ASSERT(notif);
 456         bt_notification_borrow_packet_stream(notif, &stream, &packet);
 457
 458         if (!stream) {
 459                 /* we don't care about notifications not attached to streams */
 460                 goto end;
 461         }
 462
 463         stream_state = g_hash_table_lookup(iterator->stream_states, stream);
 464         if (!stream_state) {
 465                 /*
 466                  * No stream state for this stream: this notification
 467                  * MUST be a BT_NOTIFICATION_TYPE_STREAM_BEGIN notification
 468                  * and its sequence number must be 0.
 469                  */
 470                 if (notif->type != BT_NOTIFICATION_TYPE_STREAM_BEGIN) {
 471                         BT_ASSERT_PRE_MSG("Unexpected notification: missing a "
 472                                 "BT_NOTIFICATION_TYPE_STREAM_BEGIN "
 473                                 "notification prior to this notification: "
 474                                 "%![stream-]+s", stream);
 475                         is_valid = false;
 476                         goto end;
 477                 }
 478
 479                 if (notif->seq_num == -1ULL) {
 480                         notif->seq_num = 0;
 481                 }
 482
 483                 if (notif->seq_num != 0) {
 484                         BT_ASSERT_PRE_MSG("Unexpected notification sequence "
 485                                 "number for this notification iterator: "
 486                                 "this is the first notification for this "
 487                                 "stream, expecting sequence number 0: "
 488                                 "seq-num=%" PRIu64 ", %![stream-]+s",
 489                                 notif->seq_num, stream);
 490                         is_valid = false;
 491                         goto end;
 492                 }
 493
 494                 stream_state = create_stream_state(stream);
 495                 if (!stream_state) {
 496                         abort();
 497                 }
 498
 499                 g_hash_table_insert(iterator->stream_states, stream,
 500                         stream_state);
 501                 stream_state->expected_notif_seq_num++;
 502                 goto end;
 503         }
 504
 505         if (stream_state->is_ended) {
 506                 /*
 507                  * There's a new notification which has a reference to a
 508                  * stream which, from this iterator's point of view, is
 509                  * ended ("end of stream" notification was returned).
 510                  * This is bad: the API guarantees that it can never
 511                  * happen.
 512                  */
 513                 BT_ASSERT_PRE_MSG("Stream is already ended: %![stream-]+s",
 514                         stream);
 515                 is_valid = false;
 516                 goto end;
 517         }
 518
 519         if (notif->seq_num == -1ULL) {
 520                 notif->seq_num = stream_state->expected_notif_seq_num;
 521         }
 522
 523         if (notif->seq_num != -1ULL &&
 524                         notif->seq_num != stream_state->expected_notif_seq_num) {
 525                 BT_ASSERT_PRE_MSG("Unexpected notification sequence number: "
 526                         "seq-num=%" PRIu64 ", "
 527                         "expected-seq-num=%" PRIu64 ", %![stream-]+s",
 528                         notif->seq_num, stream_state->expected_notif_seq_num,
 529                         stream);
 530                 is_valid = false;
 531                 goto end;
 532         }
 533
 534         switch (notif->type) {
 535         case BT_NOTIFICATION_TYPE_STREAM_BEGIN:
 536                 BT_ASSERT_PRE_MSG("Unexpected BT_NOTIFICATION_TYPE_STREAM_BEGIN "
 537                         "notification at this point: notif-seq-num=%" PRIu64 ", "
 538                         "%![stream-]+s", notif->seq_num, stream);
 539                 is_valid = false;
 540                 goto end;
 541         case BT_NOTIFICATION_TYPE_STREAM_END:
 542                 if (stream_state->cur_packet) {
 543                         BT_ASSERT_PRE_MSG("Unexpected BT_NOTIFICATION_TYPE_STREAM_END "
 544                                 "notification: missing a "
 545                                 "BT_NOTIFICATION_TYPE_PACKET_END notification "
 546                                 "prior to this notification: "
 547                                 "notif-seq-num=%" PRIu64 ", "
 548                                 "%![stream-]+s", notif->seq_num, stream);
 549                         is_valid = false;
 550                         goto end;
 551                 }
 552                 stream_state->expected_notif_seq_num++;
 553                 stream_state->is_ended = true;
 554                 goto end;
 555         case BT_NOTIFICATION_TYPE_PACKET_BEGIN:
 556                 if (stream_state->cur_packet) {
 557                         BT_ASSERT_PRE_MSG("Unexpected BT_NOTIFICATION_TYPE_PACKET_BEGIN "
 558                                 "notification at this point: missing a "
 559                                 "BT_NOTIFICATION_TYPE_PACKET_END notification "
 560                                 "prior to this notification: "
 561                                 "notif-seq-num=%" PRIu64 ", %![stream-]+s, "
 562                                 "%![packet-]+a", notif->seq_num, stream,
 563                                 packet);
 564                         is_valid = false;
 565                         goto end;
 566                 }
 567                 stream_state->expected_notif_seq_num++;
 568                 stream_state->cur_packet = bt_get(packet);
 569                 goto end;
 570         case BT_NOTIFICATION_TYPE_PACKET_END:
 571                 if (!stream_state->cur_packet) {
 572                         BT_ASSERT_PRE_MSG("Unexpected BT_NOTIFICATION_TYPE_PACKET_END "
 573                                 "notification at this point: missing a "
 574                                 "BT_NOTIFICATION_TYPE_PACKET_BEGIN notification "
 575                                 "prior to this notification: "
 576                                 "notif-seq-num=%" PRIu64 ", %![stream-]+s, "
 577                                 "%![packet-]+a", notif->seq_num, stream,
 578                                 packet);
 579                         is_valid = false;
 580                         goto end;
 581                 }
 582                 stream_state->expected_notif_seq_num++;
 583                 BT_PUT(stream_state->cur_packet);
 584                 goto end;
 585         case BT_NOTIFICATION_TYPE_EVENT:
 586                 if (packet != stream_state->cur_packet) {
 587                         BT_ASSERT_PRE_MSG("Unexpected packet for "
 588                                 "BT_NOTIFICATION_TYPE_EVENT notification: "
 589                                 "notif-seq-num=%" PRIu64 ", %![stream-]+s, "
 590                                 "%![notif-packet-]+a, %![expected-packet-]+a",
 591                                 notif->seq_num, stream,
 592                                 stream_state->cur_packet, packet);
 593                         is_valid = false;
 594                         goto end;
 595                 }
 596                 stream_state->expected_notif_seq_num++;
 597                 goto end;
 598         default:
 599                 break;
 600         }
 601
 602 end:
 603         return is_valid;
 604 }
 605
 606 BT_ASSERT_PRE_FUNC
 607 static inline
 608 bool validate_notifications(
 609                 struct bt_notification_iterator_private_connection *iterator,
 610                 uint64_t count)
 611 {
 612         bool ret = true;
 613         bt_notification_array notifs = (void *) iterator->base.notifs->pdata;
 614         uint64_t i;
 615
 616         for (i = 0; i < count; i++) {
 617                 ret = validate_notification(iterator, notifs[i]);
 618                 if (!ret) {
 619                         break;
 620                 }
 621         }
 622
 623         return ret;
 624 }
 625
 626 BT_ASSERT_PRE_FUNC
 627 static inline bool priv_conn_notif_iter_can_end(
 628                 struct bt_notification_iterator_private_connection *iterator)
 629 {
 630         GHashTableIter iter;
 631         gpointer stream_key, state_value;
 632         bool ret = true;
 633
 634         /*
 635          * Verify that this iterator received a
 636          * BT_NOTIFICATION_TYPE_STREAM_END notification for each stream
 637          * which has a state.
 638          */
 639
 640         g_hash_table_iter_init(&iter, iterator->stream_states);
 641
 642         while (g_hash_table_iter_next(&iter, &stream_key, &state_value)) {
 643                 struct stream_state *stream_state = (void *) state_value;
 644
 645                 BT_ASSERT(stream_state);
 646                 BT_ASSERT(stream_key);
 647
 648                 if (!stream_state->is_ended) {
 649                         BT_ASSERT_PRE_MSG("Ending notification iterator, "
 650                                 "but stream is not ended: "
 651                                 "%![stream-]s", stream_key);
 652                         ret = false;
 653                         goto end;
 654                 }
 655         }
 656
 657 end:
 658         return ret;
 659 }
 660
 661 enum bt_notification_iterator_status
 662 bt_private_connection_notification_iterator_next(
 663                 struct bt_notification_iterator *user_iterator,
 664                 struct bt_notification ***user_notifs, uint64_t *user_count)
 665 {
 666         struct bt_notification_iterator_private_connection *iterator =
 667                 (void *) user_iterator;
 668         struct bt_private_connection_private_notification_iterator *priv_iterator =
 669                 bt_private_connection_private_notification_iterator_from_notification_iterator(iterator);
 670         bt_component_class_notification_iterator_next_method next_method = NULL;
 671         enum bt_notification_iterator_status status =
 672                 BT_NOTIFICATION_ITERATOR_STATUS_OK;
 673
 674         BT_ASSERT_PRE_NON_NULL(user_iterator, "Notification iterator");
 675         BT_ASSERT_PRE_NON_NULL(user_notifs, "Notification array");
 676         BT_ASSERT_PRE_NON_NULL(user_count, "Notification count");
 677         BT_ASSERT_PRE(user_iterator->type ==
 678                 BT_NOTIFICATION_ITERATOR_TYPE_PRIVATE_CONNECTION,
 679                 "Notification iterator was not created from a private connection: "
 680                 "%!+i", iterator);
 681         BT_LIB_LOGD("Getting next private connection notification iterator's notification: %!+i",
 682                 iterator);
 683         BT_ASSERT_PRE(iterator->state ==
 684                 BT_PRIVATE_CONNECTION_NOTIFICATION_ITERATOR_STATE_ACTIVE,
 685                 "Notification iterator's \"next\" called, but "
 686                 "iterator is in the wrong state: %!+i", iterator);
 687         BT_ASSERT(iterator->upstream_component);
 688         BT_ASSERT(iterator->upstream_component->class);
 689
 690         /* Pick the appropriate "next" method */
 691         switch (iterator->upstream_component->class->type) {
 692         case BT_COMPONENT_CLASS_TYPE_SOURCE:
 693         {
 694                 struct bt_component_class_source *source_class =
 695                         container_of(iterator->upstream_component->class,
 696                                 struct bt_component_class_source, parent);
 697
 698                 BT_ASSERT(source_class->methods.iterator.next);
 699                 next_method = source_class->methods.iterator.next;
 700                 break;
 701         }
 702         case BT_COMPONENT_CLASS_TYPE_FILTER:
 703         {
 704                 struct bt_component_class_filter *filter_class =
 705                         container_of(iterator->upstream_component->class,
 706                                 struct bt_component_class_filter, parent);
 707
 708                 BT_ASSERT(filter_class->methods.iterator.next);
 709                 next_method = filter_class->methods.iterator.next;
 710                 break;
 711         }
 712         default:
 713                 abort();
 714         }
 715
 716         /*
 717          * Call the user's "next" method to get the next notification
 718          * and status.
 719          */
 720         BT_ASSERT(next_method);
 721         BT_LOGD_STR("Calling user's \"next\" method.");
 722         status = next_method(priv_iterator,
 723                 (void *) user_iterator->notifs->pdata,
 724                 NOTIF_BATCH_SIZE, user_count);
 725         BT_LOGD("User method returned: status=%s",
 726                 bt_notification_iterator_status_string(status));
 727         if (status < 0) {
 728                 BT_LOGW_STR("User method failed.");
 729                 goto end;
 730         }
 731
 732         if (iterator->state == BT_PRIVATE_CONNECTION_NOTIFICATION_ITERATOR_STATE_FINALIZED ||
 733                         iterator->state == BT_PRIVATE_CONNECTION_NOTIFICATION_ITERATOR_STATE_FINALIZED_AND_ENDED) {
 734                 /*
 735                  * The user's "next" method, somehow, cancelled its own
 736                  * notification iterator. This can happen, for example,
 737                  * when the user's method removes the port on which
 738                  * there's the connection from which the iterator was
 739                  * created. In this case, said connection is ended, and
 740                  * all its notification iterators are finalized.
 741                  *
 742                  * Only bt_put() the returned notification if
 743                  * the status is
 744                  * BT_NOTIFICATION_ITERATOR_STATUS_OK because
 745                  * otherwise this field could be garbage.
 746                  */
 747                 if (status == BT_NOTIFICATION_ITERATOR_STATUS_OK) {
 748                         uint64_t i;
 749                         bt_notification_array notifs =
 750                                 (void *) user_iterator->notifs->pdata;
 751
 752                         for (i = 0; i < *user_count; i++) {
 753                                 bt_put(notifs[i]);
 754                         }
 755                 }
 756
 757                 status = BT_NOTIFICATION_ITERATOR_STATUS_CANCELED;
 758                 goto end;
 759         }
 760
 761         switch (status) {
 762         case BT_NOTIFICATION_ITERATOR_STATUS_OK:
 763                 BT_ASSERT_PRE(validate_notifications(iterator, *user_count),
 764                         "Notifications are invalid at this point: "
 765                         "%![notif-iter-]+i, count=%" PRIu64,
 766                         iterator, *user_count);
 767                 *user_notifs = (void *) user_iterator->notifs->pdata;
 768                 break;
 769         case BT_NOTIFICATION_ITERATOR_STATUS_AGAIN:
 770                 status = BT_NOTIFICATION_ITERATOR_STATUS_AGAIN;
 771                 goto end;
 772         case BT_NOTIFICATION_ITERATOR_STATUS_END:
 773                 BT_ASSERT_PRE(priv_conn_notif_iter_can_end(iterator),
 774                         "Notification iterator cannot end at this point: "
 775                         "%!+i", iterator);
 776                 BT_ASSERT(iterator->state ==
 777                         BT_PRIVATE_CONNECTION_NOTIFICATION_ITERATOR_STATE_ACTIVE);
 778                 iterator->state = BT_PRIVATE_CONNECTION_NOTIFICATION_ITERATOR_STATE_ENDED;
 779                 status = BT_NOTIFICATION_ITERATOR_STATUS_END;
 780                 BT_LOGD("Set new status: status=%s",
 781                         bt_notification_iterator_status_string(status));
 782                 goto end;
 783         default:
 784                 /* Unknown non-error status */
 785                 abort();
 786         }
 787
 788 end:
 789         return status;
 790 }
 791
 792 enum bt_notification_iterator_status
 793 bt_output_port_notification_iterator_next(
 794                 struct bt_notification_iterator *iterator,
 795                 bt_notification_array *notifs_to_user,
 796                 uint64_t *count_to_user)
 797 {
 798         enum bt_notification_iterator_status status;
 799         struct bt_notification_iterator_output_port *out_port_iter =
 800                 (void *) iterator;
 801         enum bt_graph_status graph_status;
 802
 803         BT_ASSERT_PRE_NON_NULL(iterator, "Notification iterator");
 804         BT_ASSERT_PRE_NON_NULL(notifs_to_user, "Notification array");
 805         BT_ASSERT_PRE_NON_NULL(count_to_user, "Notification count");
 806         BT_ASSERT_PRE(iterator->type ==
 807                 BT_NOTIFICATION_ITERATOR_TYPE_OUTPUT_PORT,
 808                 "Notification iterator was not created from an output port: "
 809                 "%!+i", iterator);
 810         BT_LIB_LOGD("Getting next output port notification iterator's notification: %!+i",
 811                 iterator);
 812
 813         graph_status = bt_graph_consume_sink_no_check(
 814                 out_port_iter->graph, out_port_iter->colander);
 815         switch (graph_status) {
 816         case BT_GRAPH_STATUS_CANCELED:
 817                 status = BT_NOTIFICATION_ITERATOR_STATUS_CANCELED;
 818                 break;
 819         case BT_GRAPH_STATUS_AGAIN:
 820                 status = BT_NOTIFICATION_ITERATOR_STATUS_AGAIN;
 821                 break;
 822         case BT_GRAPH_STATUS_END:
 823                 status = BT_NOTIFICATION_ITERATOR_STATUS_END;
 824                 break;
 825         case BT_GRAPH_STATUS_NOMEM:
 826                 status = BT_NOTIFICATION_ITERATOR_STATUS_NOMEM;
 827                 break;
 828         case BT_GRAPH_STATUS_OK:
 829                 status = BT_NOTIFICATION_ITERATOR_STATUS_OK;
 830
 831                 /*
 832                  * On success, the colander sink moves the notifications
 833                  * to this iterator's array and sets this iterator's
 834                  * notification count: move them to the user.
 835                  */
 836                 *notifs_to_user = (void *) iterator->notifs->pdata;
 837                 *count_to_user = out_port_iter->count;
 838                 break;
 839         default:
 840                 /* Other errors */
 841                 status = BT_NOTIFICATION_ITERATOR_STATUS_ERROR;
 842         }
 843
 844         return status;
 845 }
 846
 847 struct bt_component *bt_private_connection_notification_iterator_get_component(
 848                 struct bt_notification_iterator *iterator)
 849 {
 850         struct bt_notification_iterator_private_connection *iter_priv_conn;
 851
 852         BT_ASSERT_PRE_NON_NULL(iterator, "Notification iterator");
 853         BT_ASSERT_PRE(iterator->type ==
 854                 BT_NOTIFICATION_ITERATOR_TYPE_PRIVATE_CONNECTION,
 855                 "Notification iterator was not created from a private connection: "
 856                 "%!+i", iterator);
 857         iter_priv_conn = (void *) iterator;
 858         return bt_get(iter_priv_conn->upstream_component);
 859 }
 860
 861 struct bt_private_component *
 862 bt_private_connection_private_notification_iterator_get_private_component(
 863                 struct bt_private_connection_private_notification_iterator *private_iterator)
 864 {
 865         return bt_private_component_from_component(
 866                 bt_private_connection_notification_iterator_get_component(
 867                         (void *) bt_private_connection_notification_iterator_borrow_from_private(private_iterator)));
 868 }
 869
 870 static
 871 void bt_output_port_notification_iterator_destroy(struct bt_object *obj)
 872 {
 873         struct bt_notification_iterator_output_port *iterator =
 874                 (void *) container_of(obj, struct bt_notification_iterator, base);
 875
 876         BT_LOGD("Destroying output port notification iterator object: addr=%p",
 877                 iterator);
 878         BT_LOGD_STR("Putting graph.");
 879         bt_put(iterator->graph);
 880         BT_LOGD_STR("Putting colander sink component.");
 881         bt_put(iterator->colander);
 882         destroy_base_notification_iterator(obj);
 883 }
 884
 885 struct bt_notification_iterator *bt_output_port_notification_iterator_create(
 886                 struct bt_port *output_port,
 887                 const char *colander_component_name)
 888 {
 889         struct bt_notification_iterator_output_port *iterator = NULL;
 890         struct bt_component_class *colander_comp_cls = NULL;
 891         struct bt_component *output_port_comp = NULL;
 892         struct bt_component *colander_comp;
 893         struct bt_graph *graph = NULL;
 894         enum bt_graph_status graph_status;
 895         const char *colander_comp_name;
 896         struct bt_port *colander_in_port = NULL;
 897         struct bt_component_class_sink_colander_data colander_data;
 898         int ret;
 899
 900         BT_ASSERT_PRE_NON_NULL(output_port, "Output port");
 901         BT_ASSERT_PRE(bt_port_get_type(output_port) == BT_PORT_TYPE_OUTPUT,
 902                 "Port is not an output port: %!+p", output_port);
 903         output_port_comp = bt_port_get_component(output_port);
 904         BT_ASSERT_PRE(output_port_comp,
 905                 "Output port has no component: %!+p", output_port);
 906         graph = bt_component_get_graph(output_port_comp);
 907         BT_ASSERT(graph);
 908
 909         /* Create notification iterator */
 910         BT_LOGD("Creating notification iterator on output port: "
 911                 "comp-addr=%p, comp-name\"%s\", port-addr=%p, port-name=\"%s\"",
 912                 output_port_comp, bt_component_get_name(output_port_comp),
 913                 output_port, bt_port_get_name(output_port));
 914         iterator = g_new0(struct bt_notification_iterator_output_port, 1);
 915         if (!iterator) {
 916                 BT_LOGE_STR("Failed to allocate one output port notification iterator.");
 917                 goto error;
 918         }
 919
 920         ret = init_notification_iterator((void *) iterator,
 921                 BT_NOTIFICATION_ITERATOR_TYPE_OUTPUT_PORT,
 922                 bt_output_port_notification_iterator_destroy);
 923         if (ret) {
 924                 /* init_notification_iterator() logs errors */
 925                 BT_PUT(iterator);
 926                 goto end;
 927         }
 928
 929         /* Create colander component */
 930         colander_comp_cls = bt_component_class_sink_colander_get();
 931         if (!colander_comp_cls) {
 932                 BT_LOGW("Cannot get colander sink component class.");
 933                 goto error;
 934         }
 935
 936         BT_MOVE(iterator->graph, graph);
 937         colander_comp_name =
 938                 colander_component_name ? colander_component_name : "colander";
 939         colander_data.notifs = (void *) iterator->base.notifs->pdata;
 940         colander_data.count_addr = &iterator->count;
 941
 942         graph_status = bt_graph_add_component_with_init_method_data(
 943                 iterator->graph, colander_comp_cls, colander_comp_name,
 944                 NULL, &colander_data, &iterator->colander);
 945         if (graph_status != BT_GRAPH_STATUS_OK) {
 946                 BT_LOGW("Cannot add colander sink component to graph: "
 947                         "graph-addr=%p, name=\"%s\", graph-status=%s",
 948                         iterator->graph, colander_comp_name,
 949                         bt_graph_status_string(graph_status));
 950                 goto error;
 951         }
 952
 953         /*
 954          * Connect provided output port to the colander component's
 955          * input port.
 956          */
 957         colander_in_port = bt_component_sink_get_input_port_by_index(
 958                 iterator->colander, 0);
 959         BT_ASSERT(colander_in_port);
 960         graph_status = bt_graph_connect_ports(iterator->graph,
 961                 output_port, colander_in_port, NULL);
 962         if (graph_status != BT_GRAPH_STATUS_OK) {
 963                 BT_LOGW("Cannot add colander sink component to graph: "
 964                         "graph-addr=%p, name=\"%s\", graph-status=%s",
 965                         iterator->graph, colander_comp_name,
 966                         bt_graph_status_string(graph_status));
 967                 goto error;
 968         }
 969
 970         /*
 971          * At this point everything went fine. Make the graph
 972          * nonconsumable forever so that only this notification iterator
 973          * can consume (thanks to bt_graph_consume_sink_no_check()).
 974          * This avoids leaking the notification created by the colander
 975          * sink and moved to the notification iterator's notification
 976          * member.
 977          */
 978         bt_graph_set_can_consume(iterator->graph, BT_FALSE);
 979         goto end;
 980
 981 error:
 982         if (iterator && iterator->graph && iterator->colander) {
 983                 int ret;
 984
 985                 /* Remove created colander component from graph if any */
 986                 colander_comp = iterator->colander;
 987                 BT_PUT(iterator->colander);
 988
 989                 /*
 990                  * At this point the colander component's reference
 991                  * count is 0 because iterator->colander was the only
 992                  * owner. We also know that it is not connected because
 993                  * this is the last operation before this function
 994                  * succeeds.
 995                  *
 996                  * Since we honor the preconditions here,
 997                  * bt_graph_remove_unconnected_component() always
 998                  * succeeds.
 999                  */
1000                 ret = bt_graph_remove_unconnected_component(iterator->graph,
1001                         colander_comp);
1002                 BT_ASSERT(ret == 0);
1003         }
1004
1005         BT_PUT(iterator);
1006
1007 end:
1008         bt_put(colander_in_port);
1009         bt_put(colander_comp_cls);
1010         bt_put(output_port_comp);
1011         bt_put(graph);
1012         return (void *) iterator;
1013 }
1014
1015 struct bt_notification_iterator *
1016 bt_private_connection_notification_iterator_borrow_from_private(
1017                 struct bt_private_connection_private_notification_iterator *private_notification_iterator)
1018 {
1019         return (void *) private_notification_iterator;
1020 }