X-Git-Url: http://git.efficios.com/?a=blobdiff_plain;f=plugins%2Futils%2Fmuxer%2Fmuxer.c;h=e0553423a927ead2e2d0cf6a13bd06e24e349b82;hb=beed0223d4bed1b12c89e5bb0491dfbe2eaf0859;hp=75b195971ac77049b0e0d5ca9d1f320d39dd3cbd;hpb=544d0515ffa2d011247e4f1d7cad5770b8ec7033;p=babeltrace.git diff --git a/plugins/utils/muxer/muxer.c b/plugins/utils/muxer/muxer.c index 75b19597..e0553423 100644 --- a/plugins/utils/muxer/muxer.c +++ b/plugins/utils/muxer/muxer.c @@ -39,8 +39,11 @@ #include #include #include +#include #include +#define IGNORE_ABSOLUTE_PARAM_NAME "ignore-absolute" + struct muxer_comp { /* Array of struct bt_private_notification_iterator * (weak refs) */ GPtrArray *muxer_notif_iters; @@ -51,14 +54,28 @@ struct muxer_comp { size_t available_input_ports; bool error; bool initializing_muxer_notif_iter; + bool ignore_absolute; }; struct muxer_upstream_notif_iter { /* Owned by this, NULL if ended */ struct bt_notification_iterator *notif_iter; - /* Owned by this*/ + /* Weak */ struct bt_private_port *priv_port; + + /* + * This flag is true if the upstream notification iterator's + * current notification must be considered for the multiplexing + * operations. If the upstream iterator returns + * BT_NOTIFICATION_ITERATOR_STATUS_AGAIN, then this object + * is considered invalid, because its current notification is + * still the previous one, but we already took it into account. + * + * The value of this flag is not important if notif_iter above + * is NULL (which means the upstream iterator is finished). + */ + bool is_valid; }; struct muxer_notif_iter { @@ -72,11 +89,8 @@ struct muxer_notif_iter { */ GPtrArray *muxer_upstream_notif_iters; - /* Array of struct muxer_upstream_notif_iter * (weak refs) */ - GList *muxer_upstream_notif_iters_to_retry; - /* - * List of "recently" connected input ports (owned by this) to + * List of "recently" connected input ports (weak) to * handle by this muxer notification iterator. * muxer_port_connected() adds entries to this list, and the * entries are removed when a notification iterator is created @@ -88,7 +102,6 @@ struct muxer_notif_iter { /* Next thing to return by the "next" method */ struct bt_notification_iterator_next_return next_next_return; - int64_t next_next_return_ts_ns; /* Last time returned in a notification */ int64_t last_returned_ts_ns; @@ -103,7 +116,6 @@ void destroy_muxer_upstream_notif_iter( } bt_put(muxer_upstream_notif_iter->notif_iter); - bt_put(muxer_upstream_notif_iter->priv_port); g_free(muxer_upstream_notif_iter); } @@ -121,7 +133,8 @@ struct muxer_upstream_notif_iter *muxer_notif_iter_add_upstream_notif_iter( } muxer_upstream_notif_iter->notif_iter = bt_get(notif_iter); - muxer_upstream_notif_iter->priv_port = bt_get(priv_port); + muxer_upstream_notif_iter->priv_port = priv_port; + muxer_upstream_notif_iter->is_valid = false; g_ptr_array_add(muxer_notif_iter->muxer_upstream_notif_iters, muxer_upstream_notif_iter); @@ -129,27 +142,6 @@ end: return muxer_upstream_notif_iter; } -static inline -bool muxer_notif_iter_has_upstream_notif_iter_to_retry( - struct muxer_notif_iter *muxer_notif_iter) -{ - assert(muxer_notif_iter); - return muxer_notif_iter->muxer_upstream_notif_iters_to_retry != NULL; -} - -static -void muxer_notif_iter_add_upstream_notif_iter_to_retry( - struct muxer_notif_iter *muxer_notif_iter, - struct muxer_upstream_notif_iter *muxer_upstream_notif_iter) -{ - assert(muxer_notif_iter); - assert(muxer_upstream_notif_iter); - muxer_notif_iter->muxer_upstream_notif_iters_to_retry = - g_list_append( - muxer_notif_iter->muxer_upstream_notif_iters_to_retry, - muxer_upstream_notif_iter); -} - static int ensure_available_input_port(struct bt_private_component *priv_comp) { @@ -173,7 +165,7 @@ int ensure_available_input_port(struct bt_private_component *priv_comp) g_string_append_printf(port_name, "%u", muxer_comp->next_port_num); priv_port = bt_private_component_filter_add_input_private_port( - priv_comp, port_name->str); + priv_comp, port_name->str, NULL); if (!priv_port) { ret = -1; goto end; @@ -187,36 +179,6 @@ end: g_string_free(port_name, TRUE); } - BT_PUT(priv_port); - return ret; -} - -static -int remove_default_ports(struct bt_private_component *priv_comp) -{ - struct bt_private_port *priv_port; - int ret = 0; - - priv_port = bt_private_component_filter_get_default_input_private_port( - priv_comp); - if (priv_port) { - ret = bt_private_port_remove_from_component(priv_port); - if (ret) { - goto end; - } - } - - bt_put(priv_port); - priv_port = bt_private_component_filter_get_default_output_private_port( - priv_comp); - if (priv_port) { - ret = bt_private_port_remove_from_component(priv_port); - if (ret) { - goto end; - } - } - -end: bt_put(priv_port); return ret; } @@ -228,7 +190,7 @@ int create_output_port(struct bt_private_component *priv_comp) int ret = 0; priv_port = bt_private_component_filter_add_output_private_port( - priv_comp, "out"); + priv_comp, "out", NULL); if (!priv_port) { ret = -1; } @@ -251,6 +213,75 @@ void destroy_muxer_comp(struct muxer_comp *muxer_comp) g_free(muxer_comp); } +static +struct bt_value *get_default_params(void) +{ + struct bt_value *params; + int ret; + + params = bt_value_map_create(); + if (!params) { + goto error; + } + + ret = bt_value_map_insert_bool(params, IGNORE_ABSOLUTE_PARAM_NAME, + false); + if (ret) { + goto error; + } + + goto end; + +error: + BT_PUT(params); + +end: + return params; +} + +static +int configure_muxer_comp(struct muxer_comp *muxer_comp, struct bt_value *params) +{ + struct bt_value *default_params = NULL; + struct bt_value *real_params = NULL; + struct bt_value *ignore_absolute = NULL; + int ret = 0; + bt_bool bool_val; + + default_params = get_default_params(); + if (!default_params) { + goto error; + } + + real_params = bt_value_map_extend(default_params, params); + if (!real_params) { + goto error; + } + + ignore_absolute = bt_value_map_get(real_params, + IGNORE_ABSOLUTE_PARAM_NAME); + if (!bt_value_is_bool(ignore_absolute)) { + goto error; + } + + if (bt_value_bool_get(ignore_absolute, &bool_val)) { + goto error; + } + + muxer_comp->ignore_absolute = (bool) bool_val; + + goto end; + +error: + ret = -1; + +end: + bt_put(default_params); + bt_put(real_params); + bt_put(ignore_absolute); + return ret; +} + BT_HIDDEN enum bt_component_status muxer_init( struct bt_private_component *priv_comp, @@ -264,6 +295,11 @@ enum bt_component_status muxer_init( goto error; } + ret = configure_muxer_comp(muxer_comp, params); + if (ret) { + goto error; + } + muxer_comp->muxer_notif_iters = g_ptr_array_new(); if (!muxer_comp->muxer_notif_iters) { goto error; @@ -272,11 +308,6 @@ enum bt_component_status muxer_init( muxer_comp->priv_comp = priv_comp; ret = bt_private_component_set_user_data(priv_comp, muxer_comp); assert(ret == 0); - ret = remove_default_ports(priv_comp); - if (ret) { - goto error; - } - ret = ensure_available_input_port(priv_comp); if (ret) { goto error; @@ -331,7 +362,7 @@ struct bt_notification_iterator *create_notif_iter_on_input_port( // returned notification by the muxer notification // iterator which creates it. notif_iter = bt_private_connection_create_notification_iterator( - priv_conn); + priv_conn, NULL); if (!notif_iter) { *ret = -1; goto end; @@ -344,81 +375,64 @@ end: } static -int muxer_upstream_notif_iter_next(struct muxer_notif_iter *muxer_notif_iter, +enum bt_notification_iterator_status muxer_upstream_notif_iter_next( struct muxer_upstream_notif_iter *muxer_upstream_notif_iter) { - int ret = 0; - enum bt_notification_iterator_status next_status; + enum bt_notification_iterator_status status; - next_status = bt_notification_iterator_next( + status = bt_notification_iterator_next( muxer_upstream_notif_iter->notif_iter); - switch (next_status) { + switch (status) { case BT_NOTIFICATION_ITERATOR_STATUS_OK: - /* Everything okay */ + /* + * Notification iterator's current notification is valid: + * it must be considered for muxing operations. + */ + muxer_upstream_notif_iter->is_valid = true; break; case BT_NOTIFICATION_ITERATOR_STATUS_AGAIN: - muxer_notif_iter_add_upstream_notif_iter_to_retry( - muxer_notif_iter, muxer_upstream_notif_iter); + /* + * Notification iterator's current notification is not + * valid anymore. Return + * BT_NOTIFICATION_ITERATOR_STATUS_AGAIN + * immediately. + */ + muxer_upstream_notif_iter->is_valid = false; break; - case BT_NOTIFICATION_ITERATOR_STATUS_END: + case BT_NOTIFICATION_ITERATOR_STATUS_END: /* Fall-through. */ + case BT_NOTIFICATION_ITERATOR_STATUS_CANCELED: /* * Notification iterator reached the end: release it. It * won't be considered again to find the youngest * notification. */ BT_PUT(muxer_upstream_notif_iter->notif_iter); - goto end; + muxer_upstream_notif_iter->is_valid = false; + status = BT_NOTIFICATION_ITERATOR_STATUS_OK; + break; default: /* Error or unsupported status code */ - ret = next_status; + status = BT_NOTIFICATION_ITERATOR_STATUS_ERROR; + break; } -end: - return ret; + return status; } static -int muxer_notif_iter_handle_newly_connected_ports(struct muxer_comp *muxer_comp, +int muxer_notif_iter_handle_newly_connected_ports( struct muxer_notif_iter *muxer_notif_iter) { - struct bt_component *comp = NULL; int ret = 0; - comp = bt_component_from_private_component(muxer_comp->priv_comp); - assert(comp); - /* * Here we create one upstream notification iterator for each - * newly connected port. The list of newly connected ports to - * handle here is updated by muxer_port_connected(). - * - * An initial "next" operation is performed on each new upstream - * notification iterator. The possible return values of this - * initial "next" operation are: - * - * * BT_NOTIFICATION_ITERATOR_STATUS_OK: Perfect, we have a - * current notification. - * - * * BT_NOTIFICATION_ITERATOR_STATUS_AGAIN: No notification so - * far, but the muxer upstream notification iterator is added - * to the list of upstream notification iterators to retry - * before finding the next youngest notification. - * - * * BT_NOTIFICATION_ITERATOR_STATUS_END: No notification, and - * we immediately release the upstream notification iterator - * because it's useless. - * - * A possible side effect of this initial "next" operation, on - * each notification iterator, is the connection of a new port. - * In this case the list of newly connected ports is updated and - * this loop continues. - * - * Once this loop finishes successfully, the set of upstream - * notification iterators is considered _stable_, that is, it is - * safe, if no notification iterators must be retried, to select - * the youngest notification amongst them to be returned by the - * next "next" method call. + * newly connected port. We do not perform an initial "next" on + * those new upstream notification iterators: they are + * invalidated, to be validated later. The list of newly + * connected ports to handle here is updated by + * muxer_port_connected(). */ while (true) { GList *node = muxer_notif_iter->newly_connected_priv_ports; @@ -439,8 +453,8 @@ int muxer_notif_iter_handle_newly_connected_ports(struct muxer_comp *muxer_comp, /* * Looks like this port is not connected * anymore: we can't create an upstream - * notification iterator on its connection in - * this case. + * notification iterator on its (non-existing) + * connection in this case. */ goto remove_node; } @@ -450,7 +464,6 @@ int muxer_notif_iter_handle_newly_connected_ports(struct muxer_comp *muxer_comp, &ret); if (ret) { assert(!upstream_notif_iter); - bt_put(priv_port); goto error; } @@ -458,22 +471,14 @@ int muxer_notif_iter_handle_newly_connected_ports(struct muxer_comp *muxer_comp, muxer_notif_iter_add_upstream_notif_iter( muxer_notif_iter, upstream_notif_iter, priv_port); - BT_PUT(priv_port); BT_PUT(upstream_notif_iter); if (!muxer_upstream_notif_iter) { goto error; } - ret = muxer_upstream_notif_iter_next(muxer_notif_iter, - muxer_upstream_notif_iter); - if (ret) { - goto error; - } - remove_node: bt_put(upstream_notif_iter); bt_put(port); - bt_put(priv_port); muxer_notif_iter->newly_connected_priv_ports = g_list_delete_link( muxer_notif_iter->newly_connected_priv_ports, @@ -483,12 +488,11 @@ remove_node: goto end; error: - if (ret == 0) { + if (ret >= 0) { ret = -1; } end: - bt_put(comp); return ret; } @@ -519,10 +523,7 @@ int get_notif_ts_ns(struct muxer_comp *muxer_comp, notif); break; default: - /* - * All the other notifications have a higher - * priority. - */ + /* All the other notifications have a higher priority */ *ts_ns = last_returned_ts_ns; goto end; } @@ -548,8 +549,8 @@ int get_notif_ts_ns(struct muxer_comp *muxer_comp, goto error; } - if (!bt_ctf_clock_class_is_absolute(clock_class)) { - // TODO: Allow this with an explicit parameter + if (!muxer_comp->ignore_absolute && + !bt_ctf_clock_class_is_absolute(clock_class)) { goto error; } @@ -637,6 +638,7 @@ muxer_notif_iter_youngest_upstream_notif_iter( continue; } + assert(cur_muxer_upstream_notif_iter->is_valid); notif = bt_notification_iterator_get_notification( cur_muxer_upstream_notif_iter->notif_iter); assert(notif); @@ -667,114 +669,133 @@ end: } static -int muxer_notif_iter_set_next_next_return(struct muxer_comp *muxer_comp, - struct muxer_notif_iter *muxer_notif_iter) +enum bt_notification_iterator_status validate_muxer_upstream_notif_iter( + struct muxer_upstream_notif_iter *muxer_upstream_notif_iter) { - struct muxer_upstream_notif_iter *muxer_upstream_notif_iter; - enum bt_notification_iterator_status notif_iter_status; - int ret = 0; + enum bt_notification_iterator_status status = + BT_NOTIFICATION_ITERATOR_STATUS_OK; - /* - * Previous operations might have connected ports. They must be - * considered when finding the youngest notification because - * their upstream notification iterator does not exist yet. - */ - ret = muxer_notif_iter_handle_newly_connected_ports(muxer_comp, - muxer_notif_iter); - if (ret) { - muxer_notif_iter->next_next_return.status = - BT_NOTIFICATION_ITERATOR_STATUS_ERROR; - BT_PUT(muxer_notif_iter->next_next_return.notification); + if (muxer_upstream_notif_iter->is_valid || + !muxer_upstream_notif_iter->notif_iter) { goto end; } - assert(!muxer_notif_iter->newly_connected_priv_ports); + status = muxer_upstream_notif_iter_next(muxer_upstream_notif_iter); + +end: + return status; +} + +static +enum bt_notification_iterator_status validate_muxer_upstream_notif_iters( + struct muxer_notif_iter *muxer_notif_iter) +{ + enum bt_notification_iterator_status status = + BT_NOTIFICATION_ITERATOR_STATUS_OK; + size_t i; + + for (i = 0; i < muxer_notif_iter->muxer_upstream_notif_iters->len; i++) { + struct muxer_upstream_notif_iter *muxer_upstream_notif_iter = + g_ptr_array_index( + muxer_notif_iter->muxer_upstream_notif_iters, + i); + + status = validate_muxer_upstream_notif_iter( + muxer_upstream_notif_iter); + if (status != BT_NOTIFICATION_ITERATOR_STATUS_OK) { + goto end; + } + } + +end: + return status; +} + +static +struct bt_notification_iterator_next_return muxer_notif_iter_do_next( + struct muxer_comp *muxer_comp, + struct muxer_notif_iter *muxer_notif_iter) +{ + struct muxer_upstream_notif_iter *muxer_upstream_notif_iter = NULL; + struct bt_notification_iterator_next_return next_return = { + .notification = NULL, + .status = BT_NOTIFICATION_ITERATOR_STATUS_OK, + }; + int64_t next_return_ts; + + while (true) { + int ret = muxer_notif_iter_handle_newly_connected_ports( + muxer_notif_iter); + + if (ret) { + next_return.status = + BT_NOTIFICATION_ITERATOR_STATUS_ERROR; + goto end; + } + + next_return.status = + validate_muxer_upstream_notif_iters(muxer_notif_iter); + if (next_return.status != BT_NOTIFICATION_ITERATOR_STATUS_OK) { + goto end; + } - if (muxer_notif_iter_has_upstream_notif_iter_to_retry( - muxer_notif_iter)) { /* - * At least one upstream notification iterator to retry: - * try again later, because we cannot find the youngest - * notification if we don't have the current - * notification of each upstream notification iterator. + * At this point, we know that all the existing upstream + * notification iterators are valid. However the + * operations to validate them (during + * validate_muxer_upstream_notif_iters()) may have + * connected new ports. If no ports were connected + * during this operation, exit the loop. */ - muxer_notif_iter->next_next_return.status = - BT_NOTIFICATION_ITERATOR_STATUS_AGAIN; - BT_PUT(muxer_notif_iter->next_next_return.notification); - goto end; + if (!muxer_notif_iter->newly_connected_priv_ports) { + break; + } } + assert(!muxer_notif_iter->newly_connected_priv_ports); + /* - * At this point we know that all our connected ports have an - * upstream notification iterator, and that all those iterators - * have a current notification (stable state). It is safe to - * find the youngest notification. It is possible that calling - * "next" on its iterator will connect new ports. This will be - * handled by the next call to - * muxer_notif_iter_set_next_next_return(). + * At this point we know that all the existing upstream + * notification iterators are valid. We can find the one, + * amongst those, of which the current notification is the + * youngest. */ - notif_iter_status = + next_return.status = muxer_notif_iter_youngest_upstream_notif_iter(muxer_comp, muxer_notif_iter, &muxer_upstream_notif_iter, - &muxer_notif_iter->next_next_return_ts_ns); - if (notif_iter_status == BT_NOTIFICATION_ITERATOR_STATUS_END) { - /* No more active upstream notification iterator */ - muxer_notif_iter->next_next_return.status = - BT_NOTIFICATION_ITERATOR_STATUS_END; - BT_PUT(muxer_notif_iter->next_next_return.notification); + &next_return_ts); + if (next_return.status < 0 || + next_return.status == BT_NOTIFICATION_ITERATOR_STATUS_END || + next_return.status == BT_NOTIFICATION_ITERATOR_STATUS_CANCELED) { goto end; } - if (notif_iter_status < 0) { - ret = -1; + if (next_return_ts < muxer_notif_iter->last_returned_ts_ns) { + next_return.status = BT_NOTIFICATION_ITERATOR_STATUS_ERROR; goto end; } - assert(notif_iter_status == BT_NOTIFICATION_ITERATOR_STATUS_OK); - BT_PUT(muxer_notif_iter->next_next_return.notification); - muxer_notif_iter->next_next_return.notification = - bt_notification_iterator_get_notification( - muxer_upstream_notif_iter->notif_iter); - assert(muxer_notif_iter->next_next_return.notification); - muxer_notif_iter->next_next_return.status = - BT_NOTIFICATION_ITERATOR_STATUS_OK; - ret = muxer_upstream_notif_iter_next(muxer_notif_iter, - muxer_upstream_notif_iter); - if (ret) { - goto end; - } - - /* - * Here we have the next "next" return value. It won't change - * until it is returned by the next call to our "next" method. - * If its time is less than the time of the last notification - * that our "next" method returned, then fail because the - * muxer's output wouldn't be monotonic. - */ - if (muxer_notif_iter->next_next_return_ts_ns < - muxer_notif_iter->last_returned_ts_ns) { - ret = -1; - goto end; - } + assert(next_return.status == BT_NOTIFICATION_ITERATOR_STATUS_OK); + assert(muxer_upstream_notif_iter); + next_return.notification = bt_notification_iterator_get_notification( + muxer_upstream_notif_iter->notif_iter); + assert(next_return.notification); /* - * We are now sure that the next "next" return value will not - * change until it is returned by this muxer notification - * iterator (unless there's a fatal error). It is now safe to - * set the last returned time to this one. + * We invalidate the upstream notification iterator so that, the + * next time this function is called, + * validate_muxer_upstream_notif_iters() will make it valid. */ - muxer_notif_iter->last_returned_ts_ns = - muxer_notif_iter->next_next_return_ts_ns; + muxer_upstream_notif_iter->is_valid = false; + muxer_notif_iter->last_returned_ts_ns = next_return_ts; end: - return ret; + return next_return; } static void destroy_muxer_notif_iter(struct muxer_notif_iter *muxer_notif_iter) { - GList *node; - if (!muxer_notif_iter) { return; } @@ -784,15 +805,6 @@ void destroy_muxer_notif_iter(struct muxer_notif_iter *muxer_notif_iter) muxer_notif_iter->muxer_upstream_notif_iters, TRUE); } - if (muxer_notif_iter->muxer_upstream_notif_iters_to_retry) { - g_list_free(muxer_notif_iter->muxer_upstream_notif_iters_to_retry); - } - - for (node = muxer_notif_iter->newly_connected_priv_ports; - node; node = g_list_next(node)) { - bt_put(node->data); - } - g_list_free(muxer_notif_iter->newly_connected_priv_ports); g_free(muxer_notif_iter); } @@ -820,7 +832,7 @@ int muxer_notif_iter_init_newly_connected_ports(struct muxer_comp *muxer_comp, for (i = 0; i < count; i++) { struct bt_private_port *priv_port = - bt_private_component_filter_get_input_private_port_at_index( + bt_private_component_filter_get_input_private_port_by_index( muxer_comp->priv_comp, i); struct bt_port *port; @@ -835,12 +847,12 @@ int muxer_notif_iter_init_newly_connected_ports(struct muxer_comp *muxer_comp, } bt_put(port); + bt_put(priv_port); muxer_notif_iter->newly_connected_priv_ports = g_list_append( muxer_notif_iter->newly_connected_priv_ports, priv_port); if (!muxer_notif_iter->newly_connected_priv_ports) { - bt_put(priv_port); ret = -1; goto end; } @@ -871,9 +883,9 @@ enum bt_notification_iterator_status muxer_notif_iter_init( if (muxer_comp->initializing_muxer_notif_iter) { /* - * Weird, unhandled situation: downstream creates a - * muxer notification iterator while creating another - * muxer notification iterator (same component). + * Weird, unhandled situation detected: downstream + * creates a muxer notification iterator while creating + * another muxer notification iterator (same component). */ goto error; } @@ -907,13 +919,6 @@ enum bt_notification_iterator_status muxer_notif_iter_init( goto error; } - /* Set the initial "next" return value */ - ret = muxer_notif_iter_set_next_next_return(muxer_comp, - muxer_notif_iter); - if (ret) { - goto error; - } - ret = bt_private_notification_iterator_set_user_data(priv_notif_iter, muxer_notif_iter); assert(ret == 0); @@ -965,15 +970,11 @@ BT_HIDDEN struct bt_notification_iterator_next_return muxer_notif_iter_next( struct bt_private_notification_iterator *priv_notif_iter) { - struct bt_notification_iterator_next_return next_ret = { - .notification = NULL, - }; + struct bt_notification_iterator_next_return next_ret; struct muxer_notif_iter *muxer_notif_iter = bt_private_notification_iterator_get_user_data(priv_notif_iter); struct bt_private_component *priv_comp = NULL; struct muxer_comp *muxer_comp = NULL; - GList *retry_node; - int ret; assert(muxer_notif_iter); priv_comp = bt_private_notification_iterator_get_private_component( @@ -984,83 +985,12 @@ struct bt_notification_iterator_next_return muxer_notif_iter_next( /* Are we in an error state set elsewhere? */ if (unlikely(muxer_comp->error)) { - goto error; - } - - /* - * If we have upstream notification iterators to retry, retry - * them now. Each one we find which now has a notification or - * is in "end" state, we set it to NULL in this array. Then - * we remove all the NULL values from this array. - */ - retry_node = muxer_notif_iter->muxer_upstream_notif_iters_to_retry; - while (retry_node) { - struct muxer_upstream_notif_iter *muxer_upstream_notif_iter = - retry_node->data; - enum bt_notification_iterator_status status; - GList *next_retry_node = g_list_next(retry_node); - - assert(muxer_upstream_notif_iter->notif_iter); - status = bt_notification_iterator_next( - muxer_upstream_notif_iter->notif_iter); - if (status < 0) { - goto error; - } - - if (status == BT_NOTIFICATION_ITERATOR_STATUS_END) { - /* - * This upstream notification iterator is done. - * Put the iterator and remove node from list. - */ - BT_PUT(muxer_upstream_notif_iter->notif_iter); - muxer_notif_iter->muxer_upstream_notif_iters_to_retry = - g_list_delete_link( - muxer_notif_iter->muxer_upstream_notif_iters_to_retry, - retry_node); - retry_node = next_retry_node; - continue; - } - - assert(status == BT_NOTIFICATION_ITERATOR_STATUS_OK || - status == BT_NOTIFICATION_ITERATOR_STATUS_AGAIN); - - if (status == BT_NOTIFICATION_ITERATOR_STATUS_OK) { - /* - * This upstream notification iterator now has. - * a notification. Remove it from this list. - */ - muxer_notif_iter->muxer_upstream_notif_iters_to_retry = - g_list_delete_link( - muxer_notif_iter->muxer_upstream_notif_iters_to_retry, - retry_node); - } - - retry_node = next_retry_node; - } - - /* Take our next "next" next return value */ - next_ret = muxer_notif_iter->next_next_return; - muxer_notif_iter->next_next_return.status = - BT_NOTIFICATION_ITERATOR_STATUS_ERROR; - muxer_notif_iter->next_next_return.notification = NULL; - - /* Set the next "next" return value */ - ret = muxer_notif_iter_set_next_next_return(muxer_comp, - muxer_notif_iter); - if (ret) { - goto error; + next_ret.notification = NULL; + next_ret.status = BT_NOTIFICATION_ITERATOR_STATUS_ERROR; + goto end; } - goto end; - -error: - /* - * Technically we already have a next "next" return value which - * is ready for this call, but we're failing within this call, - * so discard this buffer and return the error ASAP. - */ - BT_PUT(next_ret.notification); - next_ret.status = BT_NOTIFICATION_ITERATOR_STATUS_ERROR; + next_ret = muxer_notif_iter_do_next(muxer_comp, muxer_notif_iter); end: bt_put(priv_comp); @@ -1078,20 +1008,13 @@ void muxer_port_connected( struct muxer_comp *muxer_comp = bt_private_component_get_user_data(priv_comp); size_t i; + int ret; assert(self_port); assert(muxer_comp); - if (bt_port_get_type(self_port) == BT_PORT_TYPE_INPUT) { - int ret; - - /* One less available input port */ - muxer_comp->available_input_ports--; - ret = ensure_available_input_port(priv_comp); - if (ret) { - muxer_comp->error = true; - goto end; - } + if (bt_port_get_type(self_port) == BT_PORT_TYPE_OUTPUT) { + goto end; } for (i = 0; i < muxer_comp->muxer_notif_iters->len; i++) { @@ -1110,14 +1033,26 @@ void muxer_port_connected( muxer_notif_iter->newly_connected_priv_ports = g_list_append( muxer_notif_iter->newly_connected_priv_ports, - bt_get(self_private_port)); + self_private_port); if (!muxer_notif_iter->newly_connected_priv_ports) { - bt_put(self_private_port); + /* Put reference taken by bt_get() above */ muxer_comp->error = true; goto end; } } + /* One less available input port */ + muxer_comp->available_input_ports--; + ret = ensure_available_input_port(priv_comp); + if (ret) { + /* + * Only way to report an error later since this + * method does not return anything. + */ + muxer_comp->error = true; + goto end; + } + end: bt_put(self_port); } @@ -1139,9 +1074,10 @@ void muxer_port_disconnected(struct bt_private_component *priv_comp, * iterators which were already created thanks to connected * ports. The fact that the port is disconnected does not cancel * the upstream notification iterators created using its - * connection: they still exist. The only way to remove an - * upstream notification iterator is for its "next" operation to - * return BT_NOTIFICATION_ITERATOR_STATUS_END. + * connection: they still exist, even if the connection is dead. + * The only way to remove an upstream notification iterator is + * for its "next" operation to return + * BT_NOTIFICATION_ITERATOR_STATUS_END. */ if (bt_port_get_type(port) == BT_PORT_TYPE_INPUT) { /* One more available input port */