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