X-Git-Url: http://git.efficios.com/?a=blobdiff_plain;f=lib%2Fgraph%2Fiterator.c;h=a2111746793fa71a8def54e06bbc022a37cc66d7;hb=56e18c4ce186892c36d7f2cb5078087425e60134;hp=2428c72b68d4296433b6c201e416b0449d450619;hpb=5af447e5102d6226ee198e4db52b51e12075f1bb;p=babeltrace.git diff --git a/lib/graph/iterator.c b/lib/graph/iterator.c index 2428c72b..a2111746 100644 --- a/lib/graph/iterator.c +++ b/lib/graph/iterator.c @@ -30,13 +30,19 @@ #include #include -#include -#include -#include +#include +#include +#include +#include +#include +#include +#include #include #include #include #include +#include +#include #include #include #include @@ -48,66 +54,28 @@ #include #include #include +#include #include +#include +#include #include +#include +#include + +/* + * TODO: Use graph's state (number of active iterators, etc.) and + * possibly system specifications to make a better guess than this. + */ +#define NOTIF_BATCH_SIZE 15 struct stream_state { - struct bt_ctf_stream *stream; /* owned by this */ - struct bt_ctf_packet *cur_packet; /* owned by this */ + struct bt_stream *stream; /* owned by this */ + struct bt_packet *cur_packet; /* owned by this */ + uint64_t expected_notif_seq_num; bt_bool is_ended; }; -enum action_type { - ACTION_TYPE_PUSH_NOTIF, - ACTION_TYPE_MAP_PORT_TO_COMP_IN_STREAM, - ACTION_TYPE_ADD_STREAM_STATE, - ACTION_TYPE_SET_STREAM_STATE_IS_ENDED, - ACTION_TYPE_SET_STREAM_STATE_CUR_PACKET, -}; - -struct action { - enum action_type type; - union { - /* ACTION_TYPE_PUSH_NOTIF */ - struct { - struct bt_notification *notif; /* owned by this */ - } push_notif; - - /* ACTION_TYPE_MAP_PORT_TO_COMP_IN_STREAM */ - struct { - struct bt_ctf_stream *stream; /* owned by this */ - struct bt_component *component; /* owned by this */ - struct bt_port *port; /* owned by this */ - } map_port_to_comp_in_stream; - - /* ACTION_TYPE_ADD_STREAM_STATE */ - struct { - struct bt_ctf_stream *stream; /* owned by this */ - struct stream_state *stream_state; /* owned by this */ - } add_stream_state; - - /* ACTION_TYPE_SET_STREAM_STATE_IS_ENDED */ - struct { - struct stream_state *stream_state; /* weak */ - } set_stream_state_is_ended; - - /* ACTION_TYPE_SET_STREAM_STATE_CUR_PACKET */ - struct { - struct stream_state *stream_state; /* weak */ - struct bt_ctf_packet *packet; /* owned by this */ - } set_stream_state_cur_packet; - } payload; -}; - -static -void stream_destroy_listener(struct bt_ctf_stream *stream, void *data) -{ - struct bt_notification_iterator *iterator = data; - - /* Remove associated stream state */ - g_hash_table_remove(iterator->stream_states, stream); -} - +BT_ASSERT_PRE_FUNC static void destroy_stream_state(struct stream_state *stream_state) { @@ -123,153 +91,9 @@ void destroy_stream_state(struct stream_state *stream_state) g_free(stream_state); } +BT_ASSERT_PRE_FUNC static -void destroy_action(struct action *action) -{ - assert(action); - - switch (action->type) { - case ACTION_TYPE_PUSH_NOTIF: - BT_PUT(action->payload.push_notif.notif); - break; - case ACTION_TYPE_MAP_PORT_TO_COMP_IN_STREAM: - BT_PUT(action->payload.map_port_to_comp_in_stream.stream); - BT_PUT(action->payload.map_port_to_comp_in_stream.component); - BT_PUT(action->payload.map_port_to_comp_in_stream.port); - break; - case ACTION_TYPE_ADD_STREAM_STATE: - BT_PUT(action->payload.add_stream_state.stream); - destroy_stream_state( - action->payload.add_stream_state.stream_state); - action->payload.add_stream_state.stream_state = NULL; - break; - case ACTION_TYPE_SET_STREAM_STATE_CUR_PACKET: - BT_PUT(action->payload.set_stream_state_cur_packet.packet); - break; - case ACTION_TYPE_SET_STREAM_STATE_IS_ENDED: - break; - default: - assert(BT_FALSE); - } -} - -static -void add_action(struct bt_notification_iterator *iterator, - struct action *action) -{ - g_array_append_val(iterator->actions, *action); -} - -static -void clear_actions(struct bt_notification_iterator *iterator) -{ - size_t i; - - for (i = 0; i < iterator->actions->len; i++) { - struct action *action = &g_array_index(iterator->actions, - struct action, i); - - destroy_action(action); - } - - g_array_set_size(iterator->actions, 0); -} - -static inline -const char *action_type_string(enum action_type type) -{ - switch (type) { - case ACTION_TYPE_PUSH_NOTIF: - return "ACTION_TYPE_PUSH_NOTIF"; - case ACTION_TYPE_MAP_PORT_TO_COMP_IN_STREAM: - return "ACTION_TYPE_MAP_PORT_TO_COMP_IN_STREAM"; - case ACTION_TYPE_ADD_STREAM_STATE: - return "ACTION_TYPE_ADD_STREAM_STATE"; - case ACTION_TYPE_SET_STREAM_STATE_IS_ENDED: - return "ACTION_TYPE_SET_STREAM_STATE_IS_ENDED"; - case ACTION_TYPE_SET_STREAM_STATE_CUR_PACKET: - return "ACTION_TYPE_SET_STREAM_STATE_CUR_PACKET"; - default: - return "(unknown)"; - } -} - -static -void apply_actions(struct bt_notification_iterator *iterator) -{ - size_t i; - - BT_LOGV("Applying notification's iterator current actions: " - "count=%u", iterator->actions->len); - - for (i = 0; i < iterator->actions->len; i++) { - struct action *action = &g_array_index(iterator->actions, - struct action, i); - - BT_LOGV("Applying action: index=%zu, type=%s", - i, action_type_string(action->type)); - - switch (action->type) { - case ACTION_TYPE_PUSH_NOTIF: - /* Move notification to queue */ - g_queue_push_head(iterator->queue, - action->payload.push_notif.notif); - bt_notification_freeze( - action->payload.push_notif.notif); - action->payload.push_notif.notif = NULL; - break; - case ACTION_TYPE_MAP_PORT_TO_COMP_IN_STREAM: - bt_ctf_stream_map_component_to_port( - action->payload.map_port_to_comp_in_stream.stream, - action->payload.map_port_to_comp_in_stream.component, - action->payload.map_port_to_comp_in_stream.port); - break; - case ACTION_TYPE_ADD_STREAM_STATE: - /* Move stream state to hash table */ - g_hash_table_insert(iterator->stream_states, - action->payload.add_stream_state.stream, - action->payload.add_stream_state.stream_state); - - action->payload.add_stream_state.stream_state = NULL; - break; - case ACTION_TYPE_SET_STREAM_STATE_IS_ENDED: - /* - * We know that this stream is ended. We need to - * remember this as long as the stream exists to - * enforce that the same stream does not end - * twice. - * - * Here we add a destroy listener to the stream - * which we put after (becomes weak as the hash - * table key). If we were the last object to own - * this stream, the destroy listener is called - * when we call bt_put() which removes this - * stream state completely. This is important - * because the memory used by this stream object - * could be reused for another stream, and they - * must have different states. - */ - bt_ctf_stream_add_destroy_listener( - action->payload.set_stream_state_is_ended.stream_state->stream, - stream_destroy_listener, iterator); - action->payload.set_stream_state_is_ended.stream_state->is_ended = BT_TRUE; - BT_PUT(action->payload.set_stream_state_is_ended.stream_state->stream); - break; - case ACTION_TYPE_SET_STREAM_STATE_CUR_PACKET: - /* Move packet to stream state's current packet */ - BT_MOVE(action->payload.set_stream_state_cur_packet.stream_state->cur_packet, - action->payload.set_stream_state_cur_packet.packet); - break; - default: - assert(BT_FALSE); - } - } - - clear_actions(iterator); -} - -static -struct stream_state *create_stream_state(struct bt_ctf_stream *stream) +struct stream_state *create_stream_state(struct bt_stream *stream) { struct stream_state *stream_state = g_new0(struct stream_state, 1); @@ -279,28 +103,37 @@ struct stream_state *create_stream_state(struct bt_ctf_stream *stream) } /* - * We keep a reference to the stream until we know it's ended - * because we need to be able to create an automatic "stream - * end" notification when the user's "next" method returns - * BT_NOTIFICATION_ITERATOR_STATUS_END. - * - * We put this reference when the stream is marked as ended. + * We keep a reference to the stream until we know it's ended. */ stream_state->stream = bt_get(stream); BT_LOGV("Created stream state: stream-addr=%p, stream-name=\"%s\", " "stream-state-addr=%p", - stream, bt_ctf_stream_get_name(stream), stream_state); + stream, bt_stream_get_name(stream), stream_state); end: return stream_state; } static -void bt_notification_iterator_destroy(struct bt_object *obj) +void destroy_base_notification_iterator(struct bt_object *obj) +{ + struct bt_notification_iterator *iterator = (void *) obj; + + BT_ASSERT(iterator); + + if (iterator->notifs) { + g_ptr_array_free(iterator->notifs, TRUE); + } + + g_free(iterator); +} + +static +void bt_private_connection_notification_iterator_destroy(struct bt_object *obj) { - struct bt_notification_iterator *iterator; + struct bt_notification_iterator_private_connection *iterator; - assert(obj); + BT_ASSERT(obj); /* * The notification iterator's reference count is 0 if we're @@ -312,23 +145,11 @@ void bt_notification_iterator_destroy(struct bt_object *obj) * reference count would go from 1 to 0 again and this function * would be called again. */ - obj->ref_count.count++; - iterator = container_of(obj, struct bt_notification_iterator, base); - BT_LOGD("Destroying notification iterator object: addr=%p", + obj->ref_count++; + iterator = (void *) obj; + BT_LOGD("Destroying private connection notification iterator object: addr=%p", iterator); - bt_notification_iterator_finalize(iterator); - - if (iterator->queue) { - struct bt_notification *notif; - - BT_LOGD("Putting notifications in queue."); - - while ((notif = g_queue_pop_tail(iterator->queue))) { - bt_put(notif); - } - - g_queue_free(iterator->queue); - } + bt_private_connection_notification_iterator_finalize(iterator); if (iterator->stream_states) { /* @@ -337,27 +158,9 @@ void bt_notification_iterator_destroy(struct bt_object *obj) * listener would be called with an invalid/other * notification iterator object. */ - GHashTableIter ht_iter; - gpointer stream_gptr, stream_state_gptr; - - g_hash_table_iter_init(&ht_iter, iterator->stream_states); - - while (g_hash_table_iter_next(&ht_iter, &stream_gptr, &stream_state_gptr)) { - assert(stream_gptr); - - BT_LOGD_STR("Removing stream's destroy listener for notification iterator."); - bt_ctf_stream_remove_destroy_listener( - (void *) stream_gptr, stream_destroy_listener, - iterator); - } - g_hash_table_destroy(iterator->stream_states); } - if (iterator->actions) { - g_array_free(iterator->actions, TRUE); - } - if (iterator->connection) { /* * Remove ourself from the originating connection so @@ -367,24 +170,27 @@ void bt_notification_iterator_destroy(struct bt_object *obj) bt_connection_remove_iterator(iterator->connection, iterator); } - BT_LOGD_STR("Putting current notification."); - bt_put(iterator->current_notification); - g_free(iterator); + destroy_base_notification_iterator(obj); } BT_HIDDEN -void bt_notification_iterator_finalize( - struct bt_notification_iterator *iterator) +void bt_private_connection_notification_iterator_finalize( + struct bt_notification_iterator_private_connection *iterator) { struct bt_component_class *comp_class = NULL; bt_component_class_notification_iterator_finalize_method finalize_method = NULL; - assert(iterator); + BT_ASSERT(iterator); switch (iterator->state) { - case BT_NOTIFICATION_ITERATOR_STATE_FINALIZED: - case BT_NOTIFICATION_ITERATOR_STATE_FINALIZED_AND_ENDED: + case BT_PRIVATE_CONNECTION_NOTIFICATION_ITERATOR_STATE_NON_INITIALIZED: + /* Skip user finalization if user initialization failed */ + BT_LOGD("Not finalizing non-initialized notification iterator: " + "addr=%p", iterator); + return; + case BT_PRIVATE_CONNECTION_NOTIFICATION_ITERATOR_STATE_FINALIZED: + case BT_PRIVATE_CONNECTION_NOTIFICATION_ITERATOR_STATE_FINALIZED_AND_ENDED: /* Already finalized */ BT_LOGD("Not finalizing notification iterator: already finalized: " "addr=%p", iterator); @@ -395,17 +201,17 @@ void bt_notification_iterator_finalize( BT_LOGD("Finalizing notification iterator: addr=%p", iterator); - if (iterator->state == BT_NOTIFICATION_ITERATOR_STATE_ENDED) { + if (iterator->state == BT_PRIVATE_CONNECTION_NOTIFICATION_ITERATOR_STATE_ENDED) { BT_LOGD("Updating notification iterator's state: " - "new-state=BT_NOTIFICATION_ITERATOR_STATE_FINALIZED_AND_ENDED"); - iterator->state = BT_NOTIFICATION_ITERATOR_STATE_FINALIZED_AND_ENDED; + "new-state=BT_PRIVATE_CONNECTION_NOTIFICATION_ITERATOR_STATE_FINALIZED_AND_ENDED"); + iterator->state = BT_PRIVATE_CONNECTION_NOTIFICATION_ITERATOR_STATE_FINALIZED_AND_ENDED; } else { BT_LOGD("Updating notification iterator's state: " - "new-state=BT_NOTIFICATION_ITERATOR_STATE_FINALIZED"); - iterator->state = BT_NOTIFICATION_ITERATOR_STATE_FINALIZED; + "new-state=BT_PRIVATE_CONNECTION_NOTIFICATION_ITERATOR_STATE_FINALIZED"); + iterator->state = BT_PRIVATE_CONNECTION_NOTIFICATION_ITERATOR_STATE_FINALIZED; } - assert(iterator->upstream_component); + BT_ASSERT(iterator->upstream_component); comp_class = iterator->upstream_component->class; /* Call user-defined destroy method */ @@ -428,14 +234,14 @@ void bt_notification_iterator_finalize( } default: /* Unreachable */ - assert(0); + abort(); } if (finalize_method) { BT_LOGD("Calling user's finalization method: addr=%p", iterator); finalize_method( - bt_private_notification_iterator_from_notification_iterator(iterator)); + bt_private_connection_private_notification_iterator_from_notification_iterator(iterator)); } iterator->upstream_component = NULL; @@ -444,88 +250,55 @@ void bt_notification_iterator_finalize( } BT_HIDDEN -void bt_notification_iterator_set_connection( - struct bt_notification_iterator *iterator, +void bt_private_connection_notification_iterator_set_connection( + struct bt_notification_iterator_private_connection *iterator, struct bt_connection *connection) { - assert(iterator); + BT_ASSERT(iterator); iterator->connection = connection; BT_LOGV("Set notification iterator's connection: " "iter-addr=%p, conn-addr=%p", iterator, connection); } static -int create_subscription_mask_from_notification_types( - struct bt_notification_iterator *iterator, - const enum bt_notification_type *notif_types) +int init_notification_iterator(struct bt_notification_iterator *iterator, + enum bt_notification_iterator_type type, + bt_object_release_func destroy) { - const enum bt_notification_type *notif_type; int ret = 0; - assert(notif_types); - iterator->subscription_mask = 0; - - for (notif_type = notif_types; - *notif_type != BT_NOTIFICATION_TYPE_SENTINEL; - notif_type++) { - switch (*notif_type) { - case BT_NOTIFICATION_TYPE_ALL: - iterator->subscription_mask |= - BT_NOTIFICATION_ITERATOR_NOTIF_TYPE_EVENT | - BT_NOTIFICATION_ITERATOR_NOTIF_TYPE_INACTIVITY | - BT_NOTIFICATION_ITERATOR_NOTIF_TYPE_STREAM_BEGIN | - BT_NOTIFICATION_ITERATOR_NOTIF_TYPE_STREAM_END | - BT_NOTIFICATION_ITERATOR_NOTIF_TYPE_PACKET_BEGIN | - BT_NOTIFICATION_ITERATOR_NOTIF_TYPE_PACKET_END; - break; - case BT_NOTIFICATION_TYPE_EVENT: - iterator->subscription_mask |= BT_NOTIFICATION_ITERATOR_NOTIF_TYPE_EVENT; - break; - case BT_NOTIFICATION_TYPE_INACTIVITY: - iterator->subscription_mask |= BT_NOTIFICATION_ITERATOR_NOTIF_TYPE_INACTIVITY; - break; - case BT_NOTIFICATION_TYPE_STREAM_BEGIN: - iterator->subscription_mask |= BT_NOTIFICATION_ITERATOR_NOTIF_TYPE_STREAM_BEGIN; - break; - case BT_NOTIFICATION_TYPE_STREAM_END: - iterator->subscription_mask |= BT_NOTIFICATION_ITERATOR_NOTIF_TYPE_STREAM_END; - break; - case BT_NOTIFICATION_TYPE_PACKET_BEGIN: - iterator->subscription_mask |= BT_NOTIFICATION_ITERATOR_NOTIF_TYPE_PACKET_BEGIN; - break; - case BT_NOTIFICATION_TYPE_PACKET_END: - iterator->subscription_mask |= BT_NOTIFICATION_ITERATOR_NOTIF_TYPE_PACKET_END; - break; - default: - ret = -1; - goto end; - } - - BT_LOGV("Added notification type to subscription mask: " - "type=%s, mask=%x", - bt_notification_type_string(*notif_type), - iterator->subscription_mask); + bt_object_init_shared(&iterator->base, destroy); + iterator->type = type; + iterator->notifs = g_ptr_array_new(); + if (!iterator->notifs) { + BT_LOGE_STR("Failed to allocate a GPtrArray."); + ret = -1; + goto end; } + g_ptr_array_set_size(iterator->notifs, NOTIF_BATCH_SIZE); + end: return ret; } BT_HIDDEN -struct bt_notification_iterator *bt_notification_iterator_create( +enum bt_connection_status bt_private_connection_notification_iterator_create( struct bt_component *upstream_comp, struct bt_port *upstream_port, - const enum bt_notification_type *notification_types, - struct bt_connection *connection) + struct bt_connection *connection, + struct bt_notification_iterator_private_connection **user_iterator) { + enum bt_connection_status status = BT_CONNECTION_STATUS_OK; enum bt_component_class_type type; - struct bt_notification_iterator *iterator = NULL; + struct bt_notification_iterator_private_connection *iterator = NULL; + int ret; - assert(upstream_comp); - assert(upstream_port); - assert(notification_types); - assert(bt_port_is_connected(upstream_port)); - BT_LOGD("Creating notification iterator: " + BT_ASSERT(upstream_comp); + BT_ASSERT(upstream_port); + BT_ASSERT(bt_port_is_connected(upstream_port)); + BT_ASSERT(user_iterator); + BT_LOGD("Creating notification iterator on private connection: " "upstream-comp-addr=%p, upstream-comp-name=\"%s\", " "upstream-port-addr=%p, upstream-port-name=\"%s\", " "conn-addr=%p", @@ -533,45 +306,37 @@ struct bt_notification_iterator *bt_notification_iterator_create( upstream_port, bt_port_get_name(upstream_port), connection); type = bt_component_get_class_type(upstream_comp); - assert(type == BT_COMPONENT_CLASS_TYPE_SOURCE || + BT_ASSERT(type == BT_COMPONENT_CLASS_TYPE_SOURCE || type == BT_COMPONENT_CLASS_TYPE_FILTER); - iterator = g_new0(struct bt_notification_iterator, 1); + iterator = g_new0(struct bt_notification_iterator_private_connection, 1); if (!iterator) { - BT_LOGE_STR("Failed to allocate one notification iterator."); - goto error; + BT_LOGE_STR("Failed to allocate one private connection notification iterator."); + status = BT_CONNECTION_STATUS_NOMEM; + goto end; } - bt_object_init(iterator, bt_notification_iterator_destroy); - - if (create_subscription_mask_from_notification_types(iterator, - notification_types)) { - BT_LOGW_STR("Cannot create subscription mask from notification types."); - goto error; + ret = init_notification_iterator((void *) iterator, + BT_NOTIFICATION_ITERATOR_TYPE_PRIVATE_CONNECTION, + bt_private_connection_notification_iterator_destroy); + if (ret) { + /* init_notification_iterator() logs errors */ + status = BT_CONNECTION_STATUS_NOMEM; + goto end; } iterator->stream_states = g_hash_table_new_full(g_direct_hash, g_direct_equal, NULL, (GDestroyNotify) destroy_stream_state); if (!iterator->stream_states) { BT_LOGE_STR("Failed to allocate a GHashTable."); - goto error; - } - - iterator->queue = g_queue_new(); - if (!iterator->queue) { - BT_LOGE_STR("Failed to allocate a GQueue."); - goto error; - } - - iterator->actions = g_array_new(FALSE, FALSE, sizeof(struct action)); - if (!iterator->actions) { - BT_LOGE_STR("Failed to allocate a GArray."); - goto error; + status = BT_CONNECTION_STATUS_NOMEM; + goto end; } iterator->upstream_component = upstream_comp; iterator->upstream_port = upstream_port; iterator->connection = connection; - iterator->state = BT_NOTIFICATION_ITERATOR_STATE_ACTIVE; + iterator->graph = bt_component_borrow_graph(upstream_comp); + iterator->state = BT_PRIVATE_CONNECTION_NOTIFICATION_ITERATOR_STATE_NON_INITIALIZED; BT_LOGD("Created notification iterator: " "upstream-comp-addr=%p, upstream-comp-name=\"%s\", " "upstream-port-addr=%p, upstream-port-name=\"%s\", " @@ -579,1088 +344,658 @@ struct bt_notification_iterator *bt_notification_iterator_create( upstream_comp, bt_component_get_name(upstream_comp), upstream_port, bt_port_get_name(upstream_port), connection, iterator); - goto end; -error: - BT_PUT(iterator); + /* Move reference to user */ + *user_iterator = iterator; + iterator = NULL; end: - return iterator; + bt_put(iterator); + return status; } -void *bt_private_notification_iterator_get_user_data( - struct bt_private_notification_iterator *private_iterator) +void *bt_private_connection_private_notification_iterator_get_user_data( + struct bt_private_connection_private_notification_iterator *private_iterator) { - struct bt_notification_iterator *iterator = - bt_notification_iterator_from_private(private_iterator); + struct bt_notification_iterator_private_connection *iterator = (void *) + bt_private_connection_notification_iterator_borrow_from_private(private_iterator); - return iterator ? iterator->user_data : NULL; + BT_ASSERT_PRE_NON_NULL(private_iterator, "Notification iterator"); + return iterator->user_data; } enum bt_notification_iterator_status -bt_private_notification_iterator_set_user_data( - struct bt_private_notification_iterator *private_iterator, +bt_private_connection_private_notification_iterator_set_user_data( + struct bt_private_connection_private_notification_iterator *private_iterator, void *data) { - enum bt_notification_iterator_status ret = - BT_NOTIFICATION_ITERATOR_STATUS_OK; - struct bt_notification_iterator *iterator = - bt_notification_iterator_from_private(private_iterator); - - if (!iterator) { - BT_LOGW_STR("Invalid parameter: notification iterator is NULL."); - ret = BT_NOTIFICATION_ITERATOR_STATUS_INVALID; - goto end; - } + struct bt_notification_iterator_private_connection *iterator = (void *) + bt_private_connection_notification_iterator_borrow_from_private(private_iterator); + BT_ASSERT_PRE_NON_NULL(iterator, "Notification iterator"); iterator->user_data = data; BT_LOGV("Set notification iterator's user data: " "iter-addr=%p, user-data-addr=%p", iterator, data); - -end: - return ret; + return BT_NOTIFICATION_ITERATOR_STATUS_OK; } -struct bt_notification *bt_notification_iterator_get_notification( - struct bt_notification_iterator *iterator) +struct bt_graph *bt_private_connection_private_notification_iterator_borrow_graph( + struct bt_private_connection_private_notification_iterator *private_iterator) { - struct bt_notification *notification = NULL; - - if (!iterator) { - BT_LOGW_STR("Invalid parameter: notification iterator is NULL."); - goto end; - } - - notification = bt_get(iterator->current_notification); + struct bt_notification_iterator_private_connection *iterator = (void *) + bt_private_connection_notification_iterator_borrow_from_private( + private_iterator); -end: - return notification; + BT_ASSERT_PRE_NON_NULL(iterator, "Notification iterator"); + return iterator->graph; } -static -enum bt_notification_iterator_notif_type -bt_notification_iterator_notif_type_from_notif_type( - enum bt_notification_type notif_type) +BT_ASSERT_PRE_FUNC +static inline +void bt_notification_borrow_packet_stream(struct bt_notification *notif, + struct bt_stream **stream, struct bt_packet **packet) { - enum bt_notification_iterator_notif_type iter_notif_type; + BT_ASSERT(notif); - switch (notif_type) { + switch (notif->type) { case BT_NOTIFICATION_TYPE_EVENT: - iter_notif_type = BT_NOTIFICATION_ITERATOR_NOTIF_TYPE_EVENT; - break; - case BT_NOTIFICATION_TYPE_INACTIVITY: - iter_notif_type = BT_NOTIFICATION_ITERATOR_NOTIF_TYPE_INACTIVITY; + *packet = bt_event_borrow_packet( + bt_notification_event_borrow_event(notif)); + *stream = bt_packet_borrow_stream(*packet); break; case BT_NOTIFICATION_TYPE_STREAM_BEGIN: - iter_notif_type = BT_NOTIFICATION_ITERATOR_NOTIF_TYPE_STREAM_BEGIN; + *stream = bt_notification_stream_begin_borrow_stream(notif); break; case BT_NOTIFICATION_TYPE_STREAM_END: - iter_notif_type = BT_NOTIFICATION_ITERATOR_NOTIF_TYPE_STREAM_END; + *stream = bt_notification_stream_end_borrow_stream(notif); break; case BT_NOTIFICATION_TYPE_PACKET_BEGIN: - iter_notif_type = BT_NOTIFICATION_ITERATOR_NOTIF_TYPE_PACKET_BEGIN; + *packet = bt_notification_packet_begin_borrow_packet(notif); + *stream = bt_packet_borrow_stream(*packet); break; case BT_NOTIFICATION_TYPE_PACKET_END: - iter_notif_type = BT_NOTIFICATION_ITERATOR_NOTIF_TYPE_PACKET_END; + *packet = bt_notification_packet_end_borrow_packet(notif); + *stream = bt_packet_borrow_stream(*packet); break; default: - assert(BT_FALSE); + break; } - - return iter_notif_type; } -static -bt_bool validate_notification(struct bt_notification_iterator *iterator, - struct bt_notification *notif, - struct bt_ctf_stream *notif_stream, - struct bt_ctf_packet *notif_packet) +BT_ASSERT_PRE_FUNC +static inline +bool validate_notification( + struct bt_notification_iterator_private_connection *iterator, + struct bt_notification *notif) { - bt_bool is_valid = BT_TRUE; + bool is_valid = true; struct stream_state *stream_state; - struct bt_port *stream_comp_cur_port; + struct bt_stream *stream = NULL; + struct bt_packet *packet = NULL; + + BT_ASSERT(notif); + bt_notification_borrow_packet_stream(notif, &stream, &packet); + + if (!stream) { + /* we don't care about notifications not attached to streams */ + goto end; + } - assert(notif_stream); - stream_comp_cur_port = - bt_ctf_stream_port_for_component(notif_stream, - iterator->upstream_component); - if (!stream_comp_cur_port) { + stream_state = g_hash_table_lookup(iterator->stream_states, stream); + if (!stream_state) { /* - * This is the first time this notification iterator - * bumps into this stream. Add an action to map the - * iterator's upstream component to the iterator's - * upstream port in this stream. + * No stream state for this stream: this notification + * MUST be a BT_NOTIFICATION_TYPE_STREAM_BEGIN notification + * and its sequence number must be 0. */ - struct action action = { - .type = ACTION_TYPE_MAP_PORT_TO_COMP_IN_STREAM, - .payload.map_port_to_comp_in_stream = { - .stream = bt_get(notif_stream), - .component = bt_get(iterator->upstream_component), - .port = bt_get(iterator->upstream_port), - }, - }; - - add_action(iterator, &action); - } else { - if (stream_comp_cur_port != iterator->upstream_port) { - /* - * It looks like two different ports of the same - * component are emitting notifications which - * have references to the same stream. This is - * bad: the API guarantees that it can never - * happen. - */ - BT_LOGW("Two different ports of the same component are emitting notifications which refer to the same stream: " - "stream-addr=%p, stream-name=\"%s\", " - "stream-comp-cur-port-addr=%p, " - "stream-comp-cur-port-name=%p, " - "iter-upstream-port-addr=%p, " - "iter-upstream-port-name=%s", - notif_stream, - bt_ctf_stream_get_name(notif_stream), - stream_comp_cur_port, - bt_port_get_name(stream_comp_cur_port), - iterator->upstream_port, - bt_port_get_name(iterator->upstream_port)); - is_valid = BT_FALSE; + if (notif->type != BT_NOTIFICATION_TYPE_STREAM_BEGIN) { + BT_ASSERT_PRE_MSG("Unexpected notification: missing a " + "BT_NOTIFICATION_TYPE_STREAM_BEGIN " + "notification prior to this notification: " + "%![stream-]+s", stream); + is_valid = false; goto end; } - } - - stream_state = g_hash_table_lookup(iterator->stream_states, - notif_stream); - if (stream_state) { - BT_LOGV("Stream state already exists: " - "stream-addr=%p, stream-name=\"%s\", " - "stream-state-addr=%p", - notif_stream, - bt_ctf_stream_get_name(notif_stream), stream_state); - - if (stream_state->is_ended) { - /* - * There's a new notification which has a - * reference to a stream which, from this - * iterator's point of view, is ended ("end of - * stream" notification was returned). This is - * bad: the API guarantees that it can never - * happen. - */ - BT_LOGW("Stream is already ended: " - "stream-addr=%p, stream-name=\"%s\"", - notif_stream, - bt_ctf_stream_get_name(notif_stream)); - is_valid = BT_FALSE; - goto end; + if (notif->seq_num == -1ULL) { + notif->seq_num = 0; } - switch (notif->type) { - case BT_NOTIFICATION_TYPE_STREAM_BEGIN: - /* - * We already have a stream state, which means - * we already returned a "stream begin" - * notification: this is an invalid duplicate. - */ - BT_LOGW("Duplicate stream beginning notification: " - "stream-addr=%p, stream-name=\"%s\"", - notif_stream, - bt_ctf_stream_get_name(notif_stream)); - is_valid = BT_FALSE; + if (notif->seq_num != 0) { + BT_ASSERT_PRE_MSG("Unexpected notification sequence " + "number for this notification iterator: " + "this is the first notification for this " + "stream, expecting sequence number 0: " + "seq-num=%" PRIu64 ", %![stream-]+s", + notif->seq_num, stream); + is_valid = false; goto end; - case BT_NOTIFICATION_TYPE_PACKET_BEGIN: - if (notif_packet == stream_state->cur_packet) { - /* Duplicate "packet begin" notification */ - BT_LOGW("Duplicate stream beginning notification: " - "stream-addr=%p, stream-name=\"%s\", " - "packet-addr=%p", - notif_stream, - bt_ctf_stream_get_name(notif_stream), - notif_packet); - is_valid = BT_FALSE; - goto end; - } - break; - default: - break; } - } - -end: - return is_valid; -} - -static -bt_bool is_subscribed_to_notification_type(struct bt_notification_iterator *iterator, - enum bt_notification_type notif_type) -{ - uint32_t iter_notif_type = - (uint32_t) bt_notification_iterator_notif_type_from_notif_type( - notif_type); - - return (iter_notif_type & iterator->subscription_mask) ? BT_TRUE : BT_FALSE; -} -static -void add_action_push_notif(struct bt_notification_iterator *iterator, - struct bt_notification *notif) -{ - struct action action = { - .type = ACTION_TYPE_PUSH_NOTIF, - }; - - assert(notif); + stream_state = create_stream_state(stream); + if (!stream_state) { + abort(); + } - if (!is_subscribed_to_notification_type(iterator, notif->type)) { - return; + g_hash_table_insert(iterator->stream_states, stream, + stream_state); + stream_state->expected_notif_seq_num++; + goto end; } - action.payload.push_notif.notif = bt_get(notif); - add_action(iterator, &action); - BT_LOGV("Added \"push notification\" action: notif-addr=%p", notif); -} - -static -int add_action_push_notif_stream_begin( - struct bt_notification_iterator *iterator, - struct bt_ctf_stream *stream) -{ - int ret = 0; - struct bt_notification *stream_begin_notif = NULL; - - if (!is_subscribed_to_notification_type(iterator, - BT_NOTIFICATION_TYPE_STREAM_BEGIN)) { - BT_LOGV("Not adding \"push stream beginning notification\" action: " - "notification iterator is not subscribed: addr=%p", - iterator); + if (stream_state->is_ended) { + /* + * There's a new notification which has a reference to a + * stream which, from this iterator's point of view, is + * ended ("end of stream" notification was returned). + * This is bad: the API guarantees that it can never + * happen. + */ + BT_ASSERT_PRE_MSG("Stream is already ended: %![stream-]+s", + stream); + is_valid = false; goto end; } - assert(stream); - stream_begin_notif = bt_notification_stream_begin_create(stream); - if (!stream_begin_notif) { - BT_LOGE_STR("Cannot create stream beginning notification."); - goto error; + if (notif->seq_num == -1ULL) { + notif->seq_num = stream_state->expected_notif_seq_num; } - add_action_push_notif(iterator, stream_begin_notif); - BT_LOGV("Added \"push stream beginning notification\" action: " - "stream-addr=%p, stream-name=\"%s\"", - stream, bt_ctf_stream_get_name(stream)); - goto end; - -error: - ret = -1; - -end: - bt_put(stream_begin_notif); - return ret; -} - -static -int add_action_push_notif_stream_end( - struct bt_notification_iterator *iterator, - struct bt_ctf_stream *stream) -{ - int ret = 0; - struct bt_notification *stream_end_notif = NULL; - - if (!is_subscribed_to_notification_type(iterator, - BT_NOTIFICATION_TYPE_STREAM_END)) { - BT_LOGV("Not adding \"push stream end notification\" action: " - "notification iterator is not subscribed: addr=%p", - iterator); + if (notif->seq_num != -1ULL && + notif->seq_num != stream_state->expected_notif_seq_num) { + BT_ASSERT_PRE_MSG("Unexpected notification sequence number: " + "seq-num=%" PRIu64 ", " + "expected-seq-num=%" PRIu64 ", %![stream-]+s", + notif->seq_num, stream_state->expected_notif_seq_num, + stream); + is_valid = false; goto end; } - assert(stream); - stream_end_notif = bt_notification_stream_end_create(stream); - if (!stream_end_notif) { - BT_LOGE_STR("Cannot create stream end notification."); - goto error; + switch (notif->type) { + case BT_NOTIFICATION_TYPE_STREAM_BEGIN: + BT_ASSERT_PRE_MSG("Unexpected BT_NOTIFICATION_TYPE_STREAM_BEGIN " + "notification at this point: notif-seq-num=%" PRIu64 ", " + "%![stream-]+s", notif->seq_num, stream); + is_valid = false; + goto end; + case BT_NOTIFICATION_TYPE_STREAM_END: + if (stream_state->cur_packet) { + BT_ASSERT_PRE_MSG("Unexpected BT_NOTIFICATION_TYPE_STREAM_END " + "notification: missing a " + "BT_NOTIFICATION_TYPE_PACKET_END notification " + "prior to this notification: " + "notif-seq-num=%" PRIu64 ", " + "%![stream-]+s", notif->seq_num, stream); + is_valid = false; + goto end; + } + stream_state->expected_notif_seq_num++; + stream_state->is_ended = true; + goto end; + case BT_NOTIFICATION_TYPE_PACKET_BEGIN: + if (stream_state->cur_packet) { + BT_ASSERT_PRE_MSG("Unexpected BT_NOTIFICATION_TYPE_PACKET_BEGIN " + "notification at this point: missing a " + "BT_NOTIFICATION_TYPE_PACKET_END notification " + "prior to this notification: " + "notif-seq-num=%" PRIu64 ", %![stream-]+s, " + "%![packet-]+a", notif->seq_num, stream, + packet); + is_valid = false; + goto end; + } + stream_state->expected_notif_seq_num++; + stream_state->cur_packet = bt_get(packet); + goto end; + case BT_NOTIFICATION_TYPE_PACKET_END: + if (!stream_state->cur_packet) { + BT_ASSERT_PRE_MSG("Unexpected BT_NOTIFICATION_TYPE_PACKET_END " + "notification at this point: missing a " + "BT_NOTIFICATION_TYPE_PACKET_BEGIN notification " + "prior to this notification: " + "notif-seq-num=%" PRIu64 ", %![stream-]+s, " + "%![packet-]+a", notif->seq_num, stream, + packet); + is_valid = false; + goto end; + } + stream_state->expected_notif_seq_num++; + BT_PUT(stream_state->cur_packet); + goto end; + case BT_NOTIFICATION_TYPE_EVENT: + if (packet != stream_state->cur_packet) { + BT_ASSERT_PRE_MSG("Unexpected packet for " + "BT_NOTIFICATION_TYPE_EVENT notification: " + "notif-seq-num=%" PRIu64 ", %![stream-]+s, " + "%![notif-packet-]+a, %![expected-packet-]+a", + notif->seq_num, stream, + stream_state->cur_packet, packet); + is_valid = false; + goto end; + } + stream_state->expected_notif_seq_num++; + goto end; + default: + break; } - add_action_push_notif(iterator, stream_end_notif); - BT_LOGV("Added \"push stream end notification\" action: " - "stream-addr=%p, stream-name=\"%s\"", - stream, bt_ctf_stream_get_name(stream)); - goto end; - -error: - ret = -1; - end: - bt_put(stream_end_notif); - return ret; + return is_valid; } -static -int add_action_push_notif_packet_begin( - struct bt_notification_iterator *iterator, - struct bt_ctf_packet *packet) +BT_ASSERT_PRE_FUNC +static inline +bool validate_notifications( + struct bt_notification_iterator_private_connection *iterator, + uint64_t count) { - int ret = 0; - struct bt_notification *packet_begin_notif = NULL; - - if (!is_subscribed_to_notification_type(iterator, - BT_NOTIFICATION_TYPE_PACKET_BEGIN)) { - BT_LOGV("Not adding \"push packet beginning notification\" action: " - "notification iterator is not subscribed: addr=%p", - iterator); - goto end; - } + bool ret = true; + bt_notification_array notifs = (void *) iterator->base.notifs->pdata; + uint64_t i; - assert(packet); - packet_begin_notif = bt_notification_packet_begin_create(packet); - if (!packet_begin_notif) { - BT_LOGE_STR("Cannot create packet beginning notification."); - goto error; + for (i = 0; i < count; i++) { + ret = validate_notification(iterator, notifs[i]); + if (!ret) { + break; + } } - add_action_push_notif(iterator, packet_begin_notif); - BT_LOGV("Added \"push packet beginning notification\" action: " - "packet-addr=%p", packet); - goto end; - -error: - ret = -1; - -end: - bt_put(packet_begin_notif); return ret; } -static -int add_action_push_notif_packet_end( - struct bt_notification_iterator *iterator, - struct bt_ctf_packet *packet) +BT_ASSERT_PRE_FUNC +static inline bool priv_conn_notif_iter_can_end( + struct bt_notification_iterator_private_connection *iterator) { - int ret = 0; - struct bt_notification *packet_end_notif = NULL; + GHashTableIter iter; + gpointer stream_key, state_value; + bool ret = true; - if (!is_subscribed_to_notification_type(iterator, - BT_NOTIFICATION_TYPE_PACKET_END)) { - BT_LOGV("Not adding \"push packet end notification\" action: " - "notification iterator is not subscribed: addr=%p", - iterator); - goto end; - } + /* + * Verify that this iterator received a + * BT_NOTIFICATION_TYPE_STREAM_END notification for each stream + * which has a state. + */ - assert(packet); - packet_end_notif = bt_notification_packet_end_create(packet); - if (!packet_end_notif) { - BT_LOGE_STR("Cannot create packet end notification."); - goto error; - } + g_hash_table_iter_init(&iter, iterator->stream_states); - add_action_push_notif(iterator, packet_end_notif); - BT_LOGV("Added \"push packet end notification\" action: " - "packet-addr=%p", packet); - goto end; + while (g_hash_table_iter_next(&iter, &stream_key, &state_value)) { + struct stream_state *stream_state = (void *) state_value; -error: - ret = -1; + BT_ASSERT(stream_state); + BT_ASSERT(stream_key); + + if (!stream_state->is_ended) { + BT_ASSERT_PRE_MSG("Ending notification iterator, " + "but stream is not ended: " + "%![stream-]s", stream_key); + ret = false; + goto end; + } + } end: - bt_put(packet_end_notif); return ret; } -static -void add_action_set_stream_state_is_ended( - struct bt_notification_iterator *iterator, - struct stream_state *stream_state) -{ - struct action action = { - .type = ACTION_TYPE_SET_STREAM_STATE_IS_ENDED, - .payload.set_stream_state_is_ended = { - .stream_state = stream_state, - }, - }; - - assert(stream_state); - add_action(iterator, &action); - BT_LOGV("Added \"set stream state's ended\" action: " - "stream-state-addr=%p", stream_state); -} +enum bt_notification_iterator_status +bt_private_connection_notification_iterator_next( + struct bt_notification_iterator *user_iterator, + struct bt_notification ***user_notifs, uint64_t *user_count) +{ + struct bt_notification_iterator_private_connection *iterator = + (void *) user_iterator; + struct bt_private_connection_private_notification_iterator *priv_iterator = + bt_private_connection_private_notification_iterator_from_notification_iterator(iterator); + bt_component_class_notification_iterator_next_method next_method = NULL; + enum bt_notification_iterator_status status = + BT_NOTIFICATION_ITERATOR_STATUS_OK; -static -void add_action_set_stream_state_cur_packet( - struct bt_notification_iterator *iterator, - struct stream_state *stream_state, - struct bt_ctf_packet *packet) -{ - struct action action = { - .type = ACTION_TYPE_SET_STREAM_STATE_CUR_PACKET, - .payload.set_stream_state_cur_packet = { - .stream_state = stream_state, - .packet = bt_get(packet), - }, - }; - - assert(stream_state); - add_action(iterator, &action); - BT_LOGV("Added \"set stream state's current packet\" action: " - "stream-state-addr=%p, packet-addr=%p", - stream_state, packet); -} + BT_ASSERT_PRE_NON_NULL(user_iterator, "Notification iterator"); + BT_ASSERT_PRE_NON_NULL(user_notifs, "Notification array"); + BT_ASSERT_PRE_NON_NULL(user_count, "Notification count"); + BT_ASSERT_PRE(user_iterator->type == + BT_NOTIFICATION_ITERATOR_TYPE_PRIVATE_CONNECTION, + "Notification iterator was not created from a private connection: " + "%!+i", iterator); + BT_LIB_LOGD("Getting next private connection notification iterator's notification: %!+i", + iterator); + BT_ASSERT_PRE(iterator->state == + BT_PRIVATE_CONNECTION_NOTIFICATION_ITERATOR_STATE_ACTIVE, + "Notification iterator's \"next\" called, but " + "iterator is in the wrong state: %!+i", iterator); + BT_ASSERT(iterator->upstream_component); + BT_ASSERT(iterator->upstream_component->class); -static -int ensure_stream_state_exists(struct bt_notification_iterator *iterator, - struct bt_notification *stream_begin_notif, - struct bt_ctf_stream *notif_stream, - struct stream_state **stream_state) -{ - int ret = 0; + /* Pick the appropriate "next" method */ + switch (iterator->upstream_component->class->type) { + case BT_COMPONENT_CLASS_TYPE_SOURCE: + { + struct bt_component_class_source *source_class = + container_of(iterator->upstream_component->class, + struct bt_component_class_source, parent); - if (!notif_stream) { - /* - * The notification does not reference any stream: no - * need to get or create a stream state. - */ - goto end; + BT_ASSERT(source_class->methods.iterator.next); + next_method = source_class->methods.iterator.next; + break; } + case BT_COMPONENT_CLASS_TYPE_FILTER: + { + struct bt_component_class_filter *filter_class = + container_of(iterator->upstream_component->class, + struct bt_component_class_filter, parent); - *stream_state = g_hash_table_lookup(iterator->stream_states, - notif_stream); - if (!*stream_state) { - /* - * This iterator did not bump into this stream yet: - * create a stream state and a "stream begin" - * notification. - */ - struct action action = { - .type = ACTION_TYPE_ADD_STREAM_STATE, - .payload.add_stream_state = { - .stream = bt_get(notif_stream), - .stream_state = NULL, - }, - }; - - *stream_state = create_stream_state(notif_stream); - if (!stream_state) { - BT_LOGE_STR("Cannot create stream state."); - goto error; - } - - action.payload.add_stream_state.stream_state = - *stream_state; - add_action(iterator, &action); - - if (stream_begin_notif) { - add_action_push_notif(iterator, stream_begin_notif); - } else { - ret = add_action_push_notif_stream_begin(iterator, - notif_stream); - if (ret) { - BT_LOGE_STR("Cannot add \"push stream beginning notification\" action."); - goto error; - } - } + BT_ASSERT(filter_class->methods.iterator.next); + next_method = filter_class->methods.iterator.next; + break; + } + default: + abort(); } - goto end; - -error: - destroy_stream_state(*stream_state); - ret = -1; - -end: - return ret; -} - -static -int handle_packet_switch(struct bt_notification_iterator *iterator, - struct bt_notification *packet_begin_notif, - struct bt_ctf_packet *new_packet, - struct stream_state *stream_state) -{ - int ret = 0; - - if (stream_state->cur_packet == new_packet) { + /* + * Call the user's "next" method to get the next notification + * and status. + */ + BT_ASSERT(next_method); + BT_LOGD_STR("Calling user's \"next\" method."); + status = next_method(priv_iterator, + (void *) user_iterator->notifs->pdata, + NOTIF_BATCH_SIZE, user_count); + BT_LOGD("User method returned: status=%s", + bt_notification_iterator_status_string(status)); + if (status < 0) { + BT_LOGW_STR("User method failed."); goto end; } - BT_LOGV("Handling packet switch: " - "cur-packet-addr=%p, new-packet-addr=%p", - stream_state->cur_packet, new_packet); - - if (stream_state->cur_packet) { - /* End of the current packet */ - ret = add_action_push_notif_packet_end(iterator, - stream_state->cur_packet); - if (ret) { - BT_LOGE_STR("Cannot add \"push packet end notification\" action."); - goto error; - } - } + if (iterator->state == BT_PRIVATE_CONNECTION_NOTIFICATION_ITERATOR_STATE_FINALIZED || + iterator->state == BT_PRIVATE_CONNECTION_NOTIFICATION_ITERATOR_STATE_FINALIZED_AND_ENDED) { + /* + * The user's "next" method, somehow, cancelled its own + * notification iterator. This can happen, for example, + * when the user's method removes the port on which + * there's the connection from which the iterator was + * created. In this case, said connection is ended, and + * all its notification iterators are finalized. + * + * Only bt_put() the returned notification if + * the status is + * BT_NOTIFICATION_ITERATOR_STATUS_OK because + * otherwise this field could be garbage. + */ + if (status == BT_NOTIFICATION_ITERATOR_STATUS_OK) { + uint64_t i; + bt_notification_array notifs = + (void *) user_iterator->notifs->pdata; - /* Beginning of the new packet */ - if (packet_begin_notif) { - add_action_push_notif(iterator, packet_begin_notif); - } else if (new_packet) { - ret = add_action_push_notif_packet_begin(iterator, - new_packet); - if (ret) { - BT_LOGE_STR("Cannot add \"push packet beginning notification\" action."); - goto error; + for (i = 0; i < *user_count; i++) { + bt_put(notifs[i]); + } } - } - - add_action_set_stream_state_cur_packet(iterator, stream_state, - new_packet); - goto end; - -error: - ret = -1; - -end: - return ret; -} - -static -int handle_notif_stream_begin( - struct bt_notification_iterator *iterator, - struct bt_notification *notif, - struct bt_ctf_stream *notif_stream) -{ - int ret = 0; - struct stream_state *stream_state; - - assert(notif->type == BT_NOTIFICATION_TYPE_STREAM_BEGIN); - assert(notif_stream); - ret = ensure_stream_state_exists(iterator, notif, notif_stream, - &stream_state); - if (ret) { - BT_LOGE_STR("Cannot ensure that stream state exists."); - goto error; - } - - goto end; - -error: - ret = -1; - -end: - return ret; -} - -static -int handle_notif_stream_end( - struct bt_notification_iterator *iterator, - struct bt_notification *notif, - struct bt_ctf_stream *notif_stream) -{ - int ret = 0; - struct stream_state *stream_state; - - assert(notif->type == BT_NOTIFICATION_TYPE_STREAM_END); - assert(notif_stream); - ret = ensure_stream_state_exists(iterator, NULL, notif_stream, - &stream_state); - if (ret) { - BT_LOGE_STR("Cannot ensure that stream state exists."); - goto error; - } - - ret = handle_packet_switch(iterator, NULL, NULL, stream_state); - if (ret) { - BT_LOGE_STR("Cannot handle packet switch."); - goto error; - } - - add_action_push_notif(iterator, notif); - add_action_set_stream_state_is_ended(iterator, stream_state); - goto end; - -error: - ret = -1; - -end: - return ret; -} - -static -int handle_notif_packet_begin( - struct bt_notification_iterator *iterator, - struct bt_notification *notif, - struct bt_ctf_stream *notif_stream, - struct bt_ctf_packet *notif_packet) -{ - int ret = 0; - struct stream_state *stream_state; - - assert(notif->type == BT_NOTIFICATION_TYPE_PACKET_BEGIN); - assert(notif_packet); - ret = ensure_stream_state_exists(iterator, NULL, notif_stream, - &stream_state); - if (ret) { - BT_LOGE_STR("Cannot ensure that stream state exists."); - goto error; - } - - ret = handle_packet_switch(iterator, notif, notif_packet, stream_state); - if (ret) { - BT_LOGE_STR("Cannot handle packet switch."); - goto error; - } - - goto end; - -error: - ret = -1; - -end: - return ret; -} - -static -int handle_notif_packet_end( - struct bt_notification_iterator *iterator, - struct bt_notification *notif, - struct bt_ctf_stream *notif_stream, - struct bt_ctf_packet *notif_packet) -{ - int ret = 0; - struct stream_state *stream_state; - - assert(notif->type == BT_NOTIFICATION_TYPE_PACKET_END); - assert(notif_packet); - ret = ensure_stream_state_exists(iterator, NULL, notif_stream, - &stream_state); - if (ret) { - BT_LOGE_STR("Cannot ensure that stream state exists."); - goto error; - } - - ret = handle_packet_switch(iterator, NULL, notif_packet, stream_state); - if (ret) { - BT_LOGE_STR("Cannot handle packet switch."); - goto error; - } - - /* End of the current packet */ - add_action_push_notif(iterator, notif); - add_action_set_stream_state_cur_packet(iterator, stream_state, NULL); - goto end; - -error: - ret = -1; - -end: - return ret; -} -static -int handle_notif_event( - struct bt_notification_iterator *iterator, - struct bt_notification *notif, - struct bt_ctf_stream *notif_stream, - struct bt_ctf_packet *notif_packet) -{ - int ret = 0; - struct stream_state *stream_state; - - assert(notif->type == BT_NOTIFICATION_TYPE_EVENT); - assert(notif_packet); - ret = ensure_stream_state_exists(iterator, NULL, notif_stream, - &stream_state); - if (ret) { - BT_LOGE_STR("Cannot ensure that stream state exists."); - goto error; + status = BT_NOTIFICATION_ITERATOR_STATUS_CANCELED; + goto end; } - ret = handle_packet_switch(iterator, NULL, notif_packet, stream_state); - if (ret) { - BT_LOGE_STR("Cannot handle packet switch."); - goto error; + switch (status) { + case BT_NOTIFICATION_ITERATOR_STATUS_OK: + BT_ASSERT_PRE(validate_notifications(iterator, *user_count), + "Notifications are invalid at this point: " + "%![notif-iter-]+i, count=%" PRIu64, + iterator, *user_count); + *user_notifs = (void *) user_iterator->notifs->pdata; + break; + case BT_NOTIFICATION_ITERATOR_STATUS_AGAIN: + status = BT_NOTIFICATION_ITERATOR_STATUS_AGAIN; + goto end; + case BT_NOTIFICATION_ITERATOR_STATUS_END: + BT_ASSERT_PRE(priv_conn_notif_iter_can_end(iterator), + "Notification iterator cannot end at this point: " + "%!+i", iterator); + BT_ASSERT(iterator->state == + BT_PRIVATE_CONNECTION_NOTIFICATION_ITERATOR_STATE_ACTIVE); + iterator->state = BT_PRIVATE_CONNECTION_NOTIFICATION_ITERATOR_STATE_ENDED; + status = BT_NOTIFICATION_ITERATOR_STATUS_END; + BT_LOGD("Set new status: status=%s", + bt_notification_iterator_status_string(status)); + goto end; + default: + /* Unknown non-error status */ + abort(); } - add_action_push_notif(iterator, notif); - goto end; - -error: - ret = -1; - end: - return ret; + return status; } -static -int enqueue_notification_and_automatic( +enum bt_notification_iterator_status +bt_output_port_notification_iterator_next( struct bt_notification_iterator *iterator, - struct bt_notification *notif) + bt_notification_array *notifs_to_user, + uint64_t *count_to_user) { - int ret = 0; - struct bt_ctf_event *notif_event = NULL; - struct bt_ctf_stream *notif_stream = NULL; - struct bt_ctf_packet *notif_packet = NULL; - - assert(notif); - - BT_LOGV("Enqueuing user notification and automatic notifications: " - "iter-addr=%p, notif-addr=%p", iterator, notif); - - // TODO: Skip most of this if the iterator is only subscribed - // to event/inactivity notifications. + enum bt_notification_iterator_status status; + struct bt_notification_iterator_output_port *out_port_iter = + (void *) iterator; + enum bt_graph_status graph_status; + + BT_ASSERT_PRE_NON_NULL(iterator, "Notification iterator"); + BT_ASSERT_PRE_NON_NULL(notifs_to_user, "Notification array"); + BT_ASSERT_PRE_NON_NULL(count_to_user, "Notification count"); + BT_ASSERT_PRE(iterator->type == + BT_NOTIFICATION_ITERATOR_TYPE_OUTPUT_PORT, + "Notification iterator was not created from an output port: " + "%!+i", iterator); + BT_LIB_LOGD("Getting next output port notification iterator's notification: %!+i", + iterator); - /* Get the stream and packet referred by the notification */ - switch (notif->type) { - case BT_NOTIFICATION_TYPE_EVENT: - notif_event = bt_notification_event_borrow_event(notif); - assert(notif_event); - notif_packet = bt_ctf_event_borrow_packet(notif_event); - assert(notif_packet); - break; - case BT_NOTIFICATION_TYPE_STREAM_BEGIN: - notif_stream = - bt_notification_stream_begin_borrow_stream(notif); - assert(notif_stream); + graph_status = bt_graph_consume_sink_no_check( + out_port_iter->graph, out_port_iter->colander); + switch (graph_status) { + case BT_GRAPH_STATUS_CANCELED: + status = BT_NOTIFICATION_ITERATOR_STATUS_CANCELED; break; - case BT_NOTIFICATION_TYPE_STREAM_END: - notif_stream = bt_notification_stream_end_borrow_stream(notif); - assert(notif_stream); + case BT_GRAPH_STATUS_AGAIN: + status = BT_NOTIFICATION_ITERATOR_STATUS_AGAIN; break; - case BT_NOTIFICATION_TYPE_PACKET_BEGIN: - notif_packet = - bt_notification_packet_begin_borrow_packet(notif); - assert(notif_packet); + case BT_GRAPH_STATUS_END: + status = BT_NOTIFICATION_ITERATOR_STATUS_END; break; - case BT_NOTIFICATION_TYPE_PACKET_END: - notif_packet = bt_notification_packet_end_borrow_packet(notif); - assert(notif_packet); + case BT_GRAPH_STATUS_NOMEM: + status = BT_NOTIFICATION_ITERATOR_STATUS_NOMEM; break; - case BT_NOTIFICATION_TYPE_INACTIVITY: - /* Always valid */ - goto handle_notif; - default: - /* - * Invalid type of notification. Only the notification - * types above are allowed to be returned by a user - * component. - */ - goto error; - } + case BT_GRAPH_STATUS_OK: + status = BT_NOTIFICATION_ITERATOR_STATUS_OK; - if (notif_packet) { - notif_stream = bt_ctf_packet_borrow_stream(notif_packet); - assert(notif_stream); - } - - if (!notif_stream) { /* - * The notification has no reference to a stream: it - * cannot cause the creation of automatic notifications. + * On success, the colander sink moves the notifications + * to this iterator's array and sets this iterator's + * notification count: move them to the user. */ - BT_LOGV_STR("Notification has no reference to any stream: skipping automatic notification generation."); - goto end; - } - - if (!validate_notification(iterator, notif, notif_stream, - notif_packet)) { - BT_LOGW_STR("Invalid notification."); - goto error; - } - -handle_notif: - switch (notif->type) { - case BT_NOTIFICATION_TYPE_EVENT: - ret = handle_notif_event(iterator, notif, notif_stream, - notif_packet); - break; - case BT_NOTIFICATION_TYPE_STREAM_BEGIN: - ret = handle_notif_stream_begin(iterator, notif, notif_stream); - break; - case BT_NOTIFICATION_TYPE_STREAM_END: - ret = handle_notif_stream_end(iterator, notif, notif_stream); - break; - case BT_NOTIFICATION_TYPE_PACKET_BEGIN: - ret = handle_notif_packet_begin(iterator, notif, notif_stream, - notif_packet); - break; - case BT_NOTIFICATION_TYPE_PACKET_END: - ret = handle_notif_packet_end(iterator, notif, notif_stream, - notif_packet); - break; - case BT_NOTIFICATION_TYPE_INACTIVITY: - add_action_push_notif(iterator, notif); + *notifs_to_user = (void *) iterator->notifs->pdata; + *count_to_user = out_port_iter->count; break; default: - break; - } - - if (ret) { - BT_LOGW_STR("Failed to handle notification for automatic notification generation."); - goto error; + /* Other errors */ + status = BT_NOTIFICATION_ITERATOR_STATUS_ERROR; } - apply_actions(iterator); - BT_LOGV("Enqueued user notification and automatic notifications: " - "iter-addr=%p, notif-addr=%p", iterator, notif); - goto end; - -error: - ret = -1; - -end: - return ret; + return status; } -static -int handle_end(struct bt_notification_iterator *iterator) +struct bt_component *bt_private_connection_notification_iterator_get_component( + struct bt_notification_iterator *iterator) { - GHashTableIter stream_state_iter; - gpointer stream_gptr, stream_state_gptr; - int ret = 0; - - BT_LOGV("Handling end of iteration: addr=%p", iterator); - - /* - * Emit a "stream end" notification for each non-ended stream - * known by this iterator and mark them as ended. - */ - g_hash_table_iter_init(&stream_state_iter, iterator->stream_states); - - while (g_hash_table_iter_next(&stream_state_iter, &stream_gptr, - &stream_state_gptr)) { - struct stream_state *stream_state = stream_state_gptr; - - assert(stream_state_gptr); - - if (stream_state->is_ended) { - continue; - } - - ret = handle_packet_switch(iterator, NULL, NULL, stream_state); - if (ret) { - BT_LOGE_STR("Cannot handle packet switch."); - goto error; - } + struct bt_notification_iterator_private_connection *iter_priv_conn; - ret = add_action_push_notif_stream_end(iterator, stream_gptr); - if (ret) { - BT_LOGE_STR("Cannot add \"push stream end notification\" action."); - goto error; - } - - add_action_set_stream_state_is_ended(iterator, stream_state); - } - - apply_actions(iterator); - BT_LOGV("Handled end of iteration: addr=%p", iterator); - goto end; - -error: - ret = -1; + BT_ASSERT_PRE_NON_NULL(iterator, "Notification iterator"); + BT_ASSERT_PRE(iterator->type == + BT_NOTIFICATION_ITERATOR_TYPE_PRIVATE_CONNECTION, + "Notification iterator was not created from a private connection: " + "%!+i", iterator); + iter_priv_conn = (void *) iterator; + return bt_get(iter_priv_conn->upstream_component); +} -end: - return ret; +struct bt_private_component * +bt_private_connection_private_notification_iterator_get_private_component( + struct bt_private_connection_private_notification_iterator *private_iterator) +{ + return bt_private_component_from_component( + bt_private_connection_notification_iterator_get_component( + (void *) bt_private_connection_notification_iterator_borrow_from_private(private_iterator))); } static -enum bt_notification_iterator_status ensure_queue_has_notifications( - struct bt_notification_iterator *iterator) +void bt_output_port_notification_iterator_destroy(struct bt_object *obj) { - struct bt_private_notification_iterator *priv_iterator = - bt_private_notification_iterator_from_notification_iterator(iterator); - bt_component_class_notification_iterator_next_method next_method = NULL; - struct bt_notification_iterator_next_return next_return = { - .status = BT_NOTIFICATION_ITERATOR_STATUS_OK, - .notification = NULL, - }; - enum bt_notification_iterator_status status = - BT_NOTIFICATION_ITERATOR_STATUS_OK; - int ret; + struct bt_notification_iterator_output_port *iterator = + (void *) container_of(obj, struct bt_notification_iterator, base); - assert(iterator); - BT_LOGD("Ensuring that notification iterator's queue has at least one notification: " - "iter-addr=%p, queue-size=%u", - iterator, iterator->queue->length); + BT_LOGD("Destroying output port notification iterator object: addr=%p", + iterator); + BT_LOGD_STR("Putting graph."); + bt_put(iterator->graph); + BT_LOGD_STR("Putting colander sink component."); + bt_put(iterator->colander); + destroy_base_notification_iterator(obj); +} + +struct bt_notification_iterator *bt_output_port_notification_iterator_create( + struct bt_port *output_port, + const char *colander_component_name) +{ + struct bt_notification_iterator_output_port *iterator = NULL; + struct bt_component_class *colander_comp_cls = NULL; + struct bt_component *output_port_comp = NULL; + struct bt_component *colander_comp; + struct bt_graph *graph = NULL; + enum bt_graph_status graph_status; + const char *colander_comp_name; + struct bt_port *colander_in_port = NULL; + struct bt_component_class_sink_colander_data colander_data; + int ret; - if (iterator->queue->length > 0) { - /* We already have enough */ - BT_LOGD_STR("Queue already has at least one notification."); - goto end; + BT_ASSERT_PRE_NON_NULL(output_port, "Output port"); + BT_ASSERT_PRE(bt_port_get_type(output_port) == BT_PORT_TYPE_OUTPUT, + "Port is not an output port: %!+p", output_port); + output_port_comp = bt_port_get_component(output_port); + BT_ASSERT_PRE(output_port_comp, + "Output port has no component: %!+p", output_port); + graph = bt_component_get_graph(output_port_comp); + BT_ASSERT(graph); + + /* Create notification iterator */ + BT_LOGD("Creating notification iterator on output port: " + "comp-addr=%p, comp-name\"%s\", port-addr=%p, port-name=\"%s\"", + output_port_comp, bt_component_get_name(output_port_comp), + output_port, bt_port_get_name(output_port)); + iterator = g_new0(struct bt_notification_iterator_output_port, 1); + if (!iterator) { + BT_LOGE_STR("Failed to allocate one output port notification iterator."); + goto error; } - switch (iterator->state) { - case BT_NOTIFICATION_ITERATOR_STATE_FINALIZED_AND_ENDED: - BT_LOGD_STR("Notification iterator's \"next\" called, but it is finalized."); - status = BT_NOTIFICATION_ITERATOR_STATUS_CANCELED; - goto end; - case BT_NOTIFICATION_ITERATOR_STATE_ENDED: - BT_LOGD_STR("Notification iterator is ended."); - status = BT_NOTIFICATION_ITERATOR_STATUS_END; + ret = init_notification_iterator((void *) iterator, + BT_NOTIFICATION_ITERATOR_TYPE_OUTPUT_PORT, + bt_output_port_notification_iterator_destroy); + if (ret) { + /* init_notification_iterator() logs errors */ + BT_PUT(iterator); goto end; - default: - break; } - assert(iterator->upstream_component); - assert(iterator->upstream_component->class); - - /* Pick the appropriate "next" method */ - switch (iterator->upstream_component->class->type) { - case BT_COMPONENT_CLASS_TYPE_SOURCE: - { - struct bt_component_class_source *source_class = - container_of(iterator->upstream_component->class, - struct bt_component_class_source, parent); - - assert(source_class->methods.iterator.next); - next_method = source_class->methods.iterator.next; - break; + /* Create colander component */ + colander_comp_cls = bt_component_class_sink_colander_get(); + if (!colander_comp_cls) { + BT_LOGW("Cannot get colander sink component class."); + goto error; } - case BT_COMPONENT_CLASS_TYPE_FILTER: - { - struct bt_component_class_filter *filter_class = - container_of(iterator->upstream_component->class, - struct bt_component_class_filter, parent); - assert(filter_class->methods.iterator.next); - next_method = filter_class->methods.iterator.next; - break; - } - default: - assert(BT_FALSE); - break; + BT_MOVE(iterator->graph, graph); + colander_comp_name = + colander_component_name ? colander_component_name : "colander"; + colander_data.notifs = (void *) iterator->base.notifs->pdata; + colander_data.count_addr = &iterator->count; + + graph_status = bt_graph_add_component_with_init_method_data( + iterator->graph, colander_comp_cls, colander_comp_name, + NULL, &colander_data, &iterator->colander); + if (graph_status != BT_GRAPH_STATUS_OK) { + BT_LOGW("Cannot add colander sink component to graph: " + "graph-addr=%p, name=\"%s\", graph-status=%s", + iterator->graph, colander_comp_name, + bt_graph_status_string(graph_status)); + goto error; } /* - * Call the user's "next" method to get the next notification - * and status. + * Connect provided output port to the colander component's + * input port. */ - assert(next_method); - - while (iterator->queue->length == 0) { - BT_LOGD_STR("Calling user's \"next\" method."); - next_return = next_method(priv_iterator); - BT_LOGD("User method returned: status=%s", - bt_notification_iterator_status_string(next_return.status)); - if (next_return.status < 0) { - BT_LOGW_STR("User method failed."); - status = next_return.status; - goto end; - } - - switch (next_return.status) { - case BT_NOTIFICATION_ITERATOR_STATUS_END: - ret = handle_end(iterator); - if (ret) { - BT_LOGW_STR("Cannot handle end of iteration."); - status = BT_NOTIFICATION_ITERATOR_STATUS_ERROR; - goto end; - } - - if (iterator->state == BT_NOTIFICATION_ITERATOR_STATE_FINALIZED) { - iterator->state = - BT_NOTIFICATION_ITERATOR_STATE_FINALIZED_AND_ENDED; - - if (iterator->queue->length == 0) { - status = BT_NOTIFICATION_ITERATOR_STATUS_CANCELED; - } - } else { - iterator->state = - BT_NOTIFICATION_ITERATOR_STATE_ENDED; - - if (iterator->queue->length == 0) { - status = BT_NOTIFICATION_ITERATOR_STATUS_END; - } - } - - BT_LOGD("Set new status: status=%s", - bt_notification_iterator_status_string(status)); - goto end; - case BT_NOTIFICATION_ITERATOR_STATUS_AGAIN: - status = BT_NOTIFICATION_ITERATOR_STATUS_AGAIN; - goto end; - case BT_NOTIFICATION_ITERATOR_STATUS_OK: - if (!next_return.notification) { - BT_LOGW_STR("User method returned BT_NOTIFICATION_ITERATOR_STATUS_OK, but notification is NULL."); - status = BT_NOTIFICATION_ITERATOR_STATUS_ERROR; - goto end; - } - - /* - * We know the notification is valid. Before we - * push it to the head of the queue, push the - * appropriate automatic notifications if any. - */ - ret = enqueue_notification_and_automatic(iterator, - next_return.notification); - BT_PUT(next_return.notification); - if (ret) { - BT_LOGW("Cannot enqueue notification and automatic notifications."); - status = BT_NOTIFICATION_ITERATOR_STATUS_ERROR; - goto end; - } - break; - default: - /* Unknown non-error status */ - assert(BT_FALSE); - } + colander_in_port = bt_component_sink_get_input_port_by_index( + iterator->colander, 0); + BT_ASSERT(colander_in_port); + graph_status = bt_graph_connect_ports(iterator->graph, + output_port, colander_in_port, NULL); + if (graph_status != BT_GRAPH_STATUS_OK) { + BT_LOGW("Cannot add colander sink component to graph: " + "graph-addr=%p, name=\"%s\", graph-status=%s", + iterator->graph, colander_comp_name, + bt_graph_status_string(graph_status)); + goto error; } -end: - return status; -} - -enum bt_notification_iterator_status -bt_notification_iterator_next(struct bt_notification_iterator *iterator) -{ - enum bt_notification_iterator_status status; + /* + * At this point everything went fine. Make the graph + * nonconsumable forever so that only this notification iterator + * can consume (thanks to bt_graph_consume_sink_no_check()). + * This avoids leaking the notification created by the colander + * sink and moved to the notification iterator's notification + * member. + */ + bt_graph_set_can_consume(iterator->graph, BT_FALSE); + goto end; - if (!iterator) { - BT_LOGW_STR("Invalid parameter: notification iterator is NULL."); - status = BT_NOTIFICATION_ITERATOR_STATUS_INVALID; - goto end; - } +error: + if (iterator && iterator->graph && iterator->colander) { + int ret; - BT_LOGD("Notification iterator's \"next\": iter-addr=%p", iterator); + /* Remove created colander component from graph if any */ + colander_comp = iterator->colander; + BT_PUT(iterator->colander); - /* - * Make sure that the iterator's queue contains at least one - * notification. - */ - status = ensure_queue_has_notifications(iterator); - if (status != BT_NOTIFICATION_ITERATOR_STATUS_OK) { - /* Not an error */ - goto end; + /* + * At this point the colander component's reference + * count is 0 because iterator->colander was the only + * owner. We also know that it is not connected because + * this is the last operation before this function + * succeeds. + * + * Since we honor the preconditions here, + * bt_graph_remove_unconnected_component() always + * succeeds. + */ + ret = bt_graph_remove_unconnected_component(iterator->graph, + colander_comp); + BT_ASSERT(ret == 0); } - /* - * Move the notification at the tail of the queue to the - * iterator's current notification. - */ - assert(iterator->queue->length > 0); - bt_put(iterator->current_notification); - iterator->current_notification = g_queue_pop_tail(iterator->queue); - assert(iterator->current_notification); + BT_PUT(iterator); end: - return status; + bt_put(colander_in_port); + bt_put(colander_comp_cls); + bt_put(output_port_comp); + bt_put(graph); + return (void *) iterator; } -struct bt_component *bt_notification_iterator_get_component( - struct bt_notification_iterator *iterator) +struct bt_notification_iterator * +bt_private_connection_notification_iterator_borrow_from_private( + struct bt_private_connection_private_notification_iterator *private_notification_iterator) { - return bt_get(iterator->upstream_component); -} - -struct bt_private_component * -bt_private_notification_iterator_get_private_component( - struct bt_private_notification_iterator *private_iterator) -{ - return bt_private_component_from_component( - bt_notification_iterator_get_component( - bt_notification_iterator_from_private(private_iterator))); -} - -enum bt_notification_iterator_status bt_notification_iterator_seek_time( - struct bt_notification_iterator *iterator, - enum bt_notification_iterator_seek_origin seek_origin, - int64_t time) -{ - enum bt_notification_iterator_status ret = - BT_NOTIFICATION_ITERATOR_STATUS_UNSUPPORTED; - return ret; + return (void *) private_notification_iterator; }