X-Git-Url: http://git.efficios.com/?a=blobdiff_plain;f=lib%2Fgraph%2Fiterator.c;h=aec91f114315dbe9eb83533588e92746988fc202;hb=3fadfbc0c91f82c46bd36e6e0657ea93570c9db1;hp=95429fb509b9cfdfcb6940b345951addf6571f76;hpb=d0fea13089e4ea4825826b1022ff0d8110ef2898;p=babeltrace.git diff --git a/lib/graph/iterator.c b/lib/graph/iterator.c index 95429fb5..aec91f11 100644 --- a/lib/graph/iterator.c +++ b/lib/graph/iterator.c @@ -22,42 +22,50 @@ */ #define BT_LOG_TAG "MSG-ITER" -#include - -#include -#include -#include -#include -#include -#include -#include -#include -#include -#include -#include -#include -#include -#include -#include -#include -#include -#include -#include -#include -#include -#include -#include -#include -#include -#include -#include -#include -#include -#include -#include -#include -#include -#include +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include #include #include #include @@ -68,71 +76,24 @@ */ #define MSG_BATCH_SIZE 15 -struct stream_state { - const struct bt_stream *stream; /* owned by this */ - const struct bt_packet *cur_packet; /* owned by this */ - uint64_t expected_msg_seq_num; - bt_bool is_ended; -}; - -BT_ASSERT_PRE_FUNC -static -void destroy_stream_state(struct stream_state *stream_state) -{ - if (!stream_state) { - return; - } - - BT_LOGV("Destroying stream state: stream-state-addr=%p", stream_state); - BT_LOGV_STR("Putting stream state's current packet."); - BT_OBJECT_PUT_REF_AND_RESET(stream_state->cur_packet); - BT_LOGV_STR("Putting stream state's stream."); - BT_OBJECT_PUT_REF_AND_RESET(stream_state->stream); - g_free(stream_state); -} - -BT_ASSERT_PRE_FUNC -static -struct stream_state *create_stream_state(const struct bt_stream *stream) -{ - struct stream_state *stream_state = g_new0(struct stream_state, 1); - - if (!stream_state) { - BT_LOGE_STR("Failed to allocate one stream state."); - goto end; - } - - /* - * We keep a reference to the stream until we know it's ended. - */ - stream_state->stream = stream; - bt_object_get_no_null_check(stream_state->stream); - BT_LIB_LOGV("Created stream state: %![stream-]+s, " - "stream-state-addr=%p", - stream, stream_state); - -end: - return stream_state; -} +#define BT_ASSERT_PRE_ITER_HAS_STATE_TO_SEEK(_iter) \ + BT_ASSERT_PRE((_iter)->state == BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_ACTIVE || \ + (_iter)->state == BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_ENDED || \ + (_iter)->state == BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_LAST_SEEKING_RETURNED_AGAIN || \ + (_iter)->state == BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_LAST_SEEKING_RETURNED_ERROR, \ + "Message iterator is in the wrong state: %!+i", _iter) static inline -void _set_self_comp_port_input_msg_iterator_state( +void set_self_comp_port_input_msg_iterator_state( struct bt_self_component_port_input_message_iterator *iterator, enum bt_self_component_port_input_message_iterator_state state) { BT_ASSERT(iterator); - BT_LIB_LOGD("Updating message iterator's state: " - "new-state=%s", + BT_LIB_LOGD("Updating message iterator's state: new-state=%s", bt_self_component_port_input_message_iterator_state_string(state)); iterator->state = state; } -#ifdef BT_DEV_MODE -# define set_self_comp_port_input_msg_iterator_state _set_self_comp_port_input_msg_iterator_state -#else -# define set_self_comp_port_input_msg_iterator_state(_a, _b) -#endif - static void destroy_base_message_iterator(struct bt_object *obj) { @@ -172,17 +133,6 @@ void bt_self_component_port_input_message_iterator_destroy(struct bt_object *obj "%!+i", iterator); bt_self_component_port_input_message_iterator_try_finalize(iterator); - if (iterator->stream_states) { - /* - * Remove our destroy listener from each stream which - * has a state in this iterator. Otherwise the destroy - * listener would be called with an invalid/other - * message iterator object. - */ - g_hash_table_destroy(iterator->stream_states); - iterator->stream_states = NULL; - } - if (iterator->connection) { /* * Remove ourself from the originating connection so @@ -193,6 +143,16 @@ void bt_self_component_port_input_message_iterator_destroy(struct bt_object *obj iterator->connection = NULL; } + if (iterator->auto_seek_msgs) { + while (!g_queue_is_empty(iterator->auto_seek_msgs)) { + bt_object_put_no_null_check( + g_queue_pop_tail(iterator->auto_seek_msgs)); + } + + g_queue_free(iterator->auto_seek_msgs); + iterator->auto_seek_msgs = NULL; + } + destroy_base_message_iterator(obj); } @@ -305,6 +265,21 @@ end: return ret; } +static +bt_bool can_seek_ns_from_origin_true( + struct bt_self_component_port_input_message_iterator *iterator, + int64_t ns_from_origin) +{ + return BT_TRUE; +} + +static +bt_bool can_seek_beginning_true( + struct bt_self_component_port_input_message_iterator *iterator) +{ + return BT_TRUE; +} + static struct bt_self_component_port_input_message_iterator * bt_self_component_port_input_message_iterator_create_initial( @@ -340,11 +315,10 @@ bt_self_component_port_input_message_iterator_create_initial( 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."); - BT_OBJECT_PUT_REF_AND_RESET(iterator); + iterator->auto_seek_msgs = g_queue_new(); + if (!iterator->auto_seek_msgs) { + BT_LOGE_STR("Failed to allocate a GQueue."); + ret = -1; goto end; } @@ -354,6 +328,70 @@ bt_self_component_port_input_message_iterator_create_initial( iterator->graph = bt_component_borrow_graph(upstream_comp); set_self_comp_port_input_msg_iterator_state(iterator, BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_NON_INITIALIZED); + + switch (iterator->upstream_component->class->type) { + case BT_COMPONENT_CLASS_TYPE_SOURCE: + { + struct bt_component_class_source *src_comp_cls = + (void *) iterator->upstream_component->class; + + iterator->methods.next = + (bt_self_component_port_input_message_iterator_next_method) + src_comp_cls->methods.msg_iter_next; + iterator->methods.seek_ns_from_origin = + (bt_self_component_port_input_message_iterator_seek_ns_from_origin_method) + src_comp_cls->methods.msg_iter_seek_ns_from_origin; + iterator->methods.seek_beginning = + (bt_self_component_port_input_message_iterator_seek_beginning_method) + src_comp_cls->methods.msg_iter_seek_beginning; + iterator->methods.can_seek_ns_from_origin = + (bt_self_component_port_input_message_iterator_can_seek_ns_from_origin_method) + src_comp_cls->methods.msg_iter_can_seek_ns_from_origin; + iterator->methods.can_seek_beginning = + (bt_self_component_port_input_message_iterator_can_seek_beginning_method) + src_comp_cls->methods.msg_iter_can_seek_beginning; + break; + } + case BT_COMPONENT_CLASS_TYPE_FILTER: + { + struct bt_component_class_filter *flt_comp_cls = + (void *) iterator->upstream_component->class; + + iterator->methods.next = + (bt_self_component_port_input_message_iterator_next_method) + flt_comp_cls->methods.msg_iter_next; + iterator->methods.seek_ns_from_origin = + (bt_self_component_port_input_message_iterator_seek_ns_from_origin_method) + flt_comp_cls->methods.msg_iter_seek_ns_from_origin; + iterator->methods.seek_beginning = + (bt_self_component_port_input_message_iterator_seek_beginning_method) + flt_comp_cls->methods.msg_iter_seek_beginning; + iterator->methods.can_seek_ns_from_origin = + (bt_self_component_port_input_message_iterator_can_seek_ns_from_origin_method) + flt_comp_cls->methods.msg_iter_can_seek_ns_from_origin; + iterator->methods.can_seek_beginning = + (bt_self_component_port_input_message_iterator_can_seek_beginning_method) + flt_comp_cls->methods.msg_iter_can_seek_beginning; + break; + } + default: + abort(); + } + + if (iterator->methods.seek_ns_from_origin && + !iterator->methods.can_seek_ns_from_origin) { + iterator->methods.can_seek_ns_from_origin = + (bt_self_component_port_input_message_iterator_can_seek_ns_from_origin_method) + can_seek_ns_from_origin_true; + } + + if (iterator->methods.seek_beginning && + !iterator->methods.can_seek_beginning) { + iterator->methods.can_seek_beginning = + (bt_self_component_port_input_message_iterator_seek_beginning_method) + can_seek_beginning_true; + } + BT_LIB_LOGD("Created initial message iterator on self component input port: " "%![up-port-]+p, %![up-comp-]+c, %![iter-]+i", upstream_port, upstream_comp, iterator); @@ -392,6 +430,11 @@ bt_self_component_port_input_message_iterator_create( BT_ASSERT(upstream_port); upstream_comp = bt_port_borrow_component_inline(upstream_port); BT_ASSERT(upstream_comp); + BT_ASSERT_PRE( + bt_component_borrow_graph(upstream_comp)->config_state != + BT_GRAPH_CONFIGURATION_STATE_CONFIGURING, + "Graph is not configured: %!+g", + bt_component_borrow_graph(upstream_comp)); upstream_comp_cls = upstream_comp->class; BT_ASSERT(upstream_comp->class->type == BT_COMPONENT_CLASS_TYPE_SOURCE || @@ -477,269 +520,11 @@ void bt_self_message_iterator_set_data( "%!+i, user-data-addr=%p", iterator, data); } -BT_ASSERT_PRE_FUNC -static inline -void bt_message_borrow_packet_stream(const struct bt_message *msg, - const struct bt_stream **stream, - const struct bt_packet **packet) -{ - BT_ASSERT(msg); - - switch (msg->type) { - case BT_MESSAGE_TYPE_EVENT: - *packet = bt_event_borrow_packet_const( - bt_message_event_borrow_event_const(msg)); - *stream = bt_packet_borrow_stream_const(*packet); - break; - case BT_MESSAGE_TYPE_STREAM_BEGINNING: - *stream = bt_message_stream_beginning_borrow_stream_const(msg); - break; - case BT_MESSAGE_TYPE_STREAM_END: - *stream = bt_message_stream_end_borrow_stream_const(msg); - break; - case BT_MESSAGE_TYPE_PACKET_BEGINNING: - *packet = bt_message_packet_beginning_borrow_packet_const(msg); - *stream = bt_packet_borrow_stream_const(*packet); - break; - case BT_MESSAGE_TYPE_PACKET_END: - *packet = bt_message_packet_end_borrow_packet_const(msg); - *stream = bt_packet_borrow_stream_const(*packet); - break; - default: - break; - } -} - -BT_ASSERT_PRE_FUNC -static inline -bool validate_message( - struct bt_self_component_port_input_message_iterator *iterator, - const struct bt_message *c_msg) -{ - bool is_valid = true; - struct stream_state *stream_state; - const struct bt_stream *stream = NULL; - const struct bt_packet *packet = NULL; - struct bt_message *msg = (void *) c_msg; - - BT_ASSERT(msg); - bt_message_borrow_packet_stream(c_msg, &stream, &packet); - - if (!stream) { - /* we don't care about messages not attached to streams */ - goto end; - } - - stream_state = g_hash_table_lookup(iterator->stream_states, stream); - if (!stream_state) { - /* - * No stream state for this stream: this message - * MUST be a BT_MESSAGE_TYPE_STREAM_BEGINNING message - * and its sequence number must be 0. - */ - if (c_msg->type != BT_MESSAGE_TYPE_STREAM_BEGINNING) { - BT_ASSERT_PRE_MSG("Unexpected message: missing a " - "BT_MESSAGE_TYPE_STREAM_BEGINNING " - "message prior to this message: " - "%![stream-]+s", stream); - is_valid = false; - goto end; - } - - if (c_msg->seq_num == -1ULL) { - msg->seq_num = 0; - } - - if (c_msg->seq_num != 0) { - BT_ASSERT_PRE_MSG("Unexpected message sequence " - "number for this message iterator: " - "this is the first message for this " - "stream, expecting sequence number 0: " - "seq-num=%" PRIu64 ", %![stream-]+s", - c_msg->seq_num, stream); - is_valid = false; - goto end; - } - - stream_state = create_stream_state(stream); - if (!stream_state) { - abort(); - } - - g_hash_table_insert(iterator->stream_states, - (void *) stream, stream_state); - stream_state->expected_msg_seq_num++; - goto end; - } - - if (stream_state->is_ended) { - /* - * There's a new message which has a reference to a - * stream which, from this iterator's point of view, is - * ended ("end of stream" message 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; - } - - if (c_msg->seq_num == -1ULL) { - msg->seq_num = stream_state->expected_msg_seq_num; - } - - if (c_msg->seq_num != -1ULL && - c_msg->seq_num != stream_state->expected_msg_seq_num) { - BT_ASSERT_PRE_MSG("Unexpected message sequence number: " - "seq-num=%" PRIu64 ", " - "expected-seq-num=%" PRIu64 ", %![stream-]+s", - c_msg->seq_num, stream_state->expected_msg_seq_num, - stream); - is_valid = false; - goto end; - } - - switch (c_msg->type) { - case BT_MESSAGE_TYPE_STREAM_BEGINNING: - BT_ASSERT_PRE_MSG("Unexpected BT_MESSAGE_TYPE_STREAM_BEGINNING " - "message at this point: msg-seq-num=%" PRIu64 ", " - "%![stream-]+s", c_msg->seq_num, stream); - is_valid = false; - goto end; - case BT_MESSAGE_TYPE_STREAM_END: - if (stream_state->cur_packet) { - BT_ASSERT_PRE_MSG("Unexpected BT_MESSAGE_TYPE_STREAM_END " - "message: missing a " - "BT_MESSAGE_TYPE_PACKET_END message " - "prior to this message: " - "msg-seq-num=%" PRIu64 ", " - "%![stream-]+s", c_msg->seq_num, stream); - is_valid = false; - goto end; - } - stream_state->expected_msg_seq_num++; - stream_state->is_ended = true; - goto end; - case BT_MESSAGE_TYPE_PACKET_BEGINNING: - if (stream_state->cur_packet) { - BT_ASSERT_PRE_MSG("Unexpected BT_MESSAGE_TYPE_PACKET_BEGINNING " - "message at this point: missing a " - "BT_MESSAGE_TYPE_PACKET_END message " - "prior to this message: " - "msg-seq-num=%" PRIu64 ", %![stream-]+s, " - "%![packet-]+a", c_msg->seq_num, stream, - packet); - is_valid = false; - goto end; - } - stream_state->expected_msg_seq_num++; - stream_state->cur_packet = packet; - bt_object_get_no_null_check(stream_state->cur_packet); - goto end; - case BT_MESSAGE_TYPE_PACKET_END: - if (!stream_state->cur_packet) { - BT_ASSERT_PRE_MSG("Unexpected BT_MESSAGE_TYPE_PACKET_END " - "message at this point: missing a " - "BT_MESSAGE_TYPE_PACKET_BEGINNING message " - "prior to this message: " - "msg-seq-num=%" PRIu64 ", %![stream-]+s, " - "%![packet-]+a", c_msg->seq_num, stream, - packet); - is_valid = false; - goto end; - } - stream_state->expected_msg_seq_num++; - BT_OBJECT_PUT_REF_AND_RESET(stream_state->cur_packet); - goto end; - case BT_MESSAGE_TYPE_EVENT: - if (packet != stream_state->cur_packet) { - BT_ASSERT_PRE_MSG("Unexpected packet for " - "BT_MESSAGE_TYPE_EVENT message: " - "msg-seq-num=%" PRIu64 ", %![stream-]+s, " - "%![msg-packet-]+a, %![expected-packet-]+a", - c_msg->seq_num, stream, - stream_state->cur_packet, packet); - is_valid = false; - goto end; - } - stream_state->expected_msg_seq_num++; - goto end; - default: - break; - } - -end: - return is_valid; -} - -BT_ASSERT_PRE_FUNC -static inline -bool validate_messages( - struct bt_self_component_port_input_message_iterator *iterator, - uint64_t count) -{ - bool ret = true; - bt_message_array_const msgs = - (void *) iterator->base.msgs->pdata; - uint64_t i; - - for (i = 0; i < count; i++) { - ret = validate_message(iterator, msgs[i]); - if (!ret) { - break; - } - } - - return ret; -} - -BT_ASSERT_PRE_FUNC -static inline bool self_comp_port_input_msg_iter_can_end( - struct bt_self_component_port_input_message_iterator *iterator) -{ - GHashTableIter iter; - gpointer stream_key, state_value; - bool ret = true; - - /* - * Verify that this iterator received a - * BT_MESSAGE_TYPE_STREAM_END message for each stream - * which has a state. - */ - - g_hash_table_iter_init(&iter, iterator->stream_states); - - while (g_hash_table_iter_next(&iter, &stream_key, &state_value)) { - struct stream_state *stream_state = (void *) state_value; - - BT_ASSERT(stream_state); - BT_ASSERT(stream_key); - - if (!stream_state->is_ended) { - BT_ASSERT_PRE_MSG("Ending message iterator, " - "but stream is not ended: " - "%![stream-]s", stream_key); - ret = false; - goto end; - } - } - -end: - return ret; -} - enum bt_message_iterator_status bt_self_component_port_input_message_iterator_next( struct bt_self_component_port_input_message_iterator *iterator, bt_message_array_const *msgs, uint64_t *user_count) { - typedef enum bt_self_message_iterator_status (*method_t)( - void *, bt_message_array_const, uint64_t, uint64_t *); - - method_t method = NULL; - struct bt_component_class *comp_cls; int status = BT_MESSAGE_ITERATOR_STATUS_OK; BT_ASSERT_PRE_NON_NULL(iterator, "Message iterator"); @@ -748,43 +533,26 @@ bt_self_component_port_input_message_iterator_next( BT_ASSERT_PRE(iterator->state == BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_ACTIVE, "Message iterator's \"next\" called, but " - "iterator is in the wrong state: %!+i", iterator); + "message iterator is in the wrong state: %!+i", iterator); BT_ASSERT(iterator->upstream_component); BT_ASSERT(iterator->upstream_component->class); + BT_ASSERT_PRE( + bt_component_borrow_graph(iterator->upstream_component)->config_state != + BT_GRAPH_CONFIGURATION_STATE_CONFIGURING, + "Graph is not configured: %!+g", + bt_component_borrow_graph(iterator->upstream_component)); BT_LIB_LOGD("Getting next self component input port " "message iterator's messages: %!+i", iterator); - comp_cls = iterator->upstream_component->class; - - /* Pick the appropriate "next" method */ - switch (comp_cls->type) { - case BT_COMPONENT_CLASS_TYPE_SOURCE: - { - struct bt_component_class_source *src_comp_cls = - (void *) comp_cls; - - method = (method_t) src_comp_cls->methods.msg_iter_next; - break; - } - case BT_COMPONENT_CLASS_TYPE_FILTER: - { - struct bt_component_class_filter *flt_comp_cls = - (void *) comp_cls; - - method = (method_t) flt_comp_cls->methods.msg_iter_next; - break; - } - default: - abort(); - } /* * Call the user's "next" method to get the next messages * and status. */ - BT_ASSERT(method); + BT_ASSERT(iterator->methods.next); BT_LOGD_STR("Calling user's \"next\" method."); - status = method(iterator, (void *) iterator->base.msgs->pdata, - MSG_BATCH_SIZE, user_count); + status = iterator->methods.next(iterator, + (void *) iterator->base.msgs->pdata, MSG_BATCH_SIZE, + user_count); BT_LOGD("User method returned: status=%s", bt_message_iterator_status_string(status)); if (status < 0) { @@ -792,31 +560,29 @@ bt_self_component_port_input_message_iterator_next( goto end; } -#ifdef BT_DEV_MODE /* * There is no way that this iterator could have been finalized * during its "next" method, as the only way to do this is to * put the last iterator's reference, and this can only be done * by its downstream owner. + * + * For the same reason, there is no way that this iterator could + * have seeked (cannot seek a self message iterator). */ BT_ASSERT(iterator->state == BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_ACTIVE); -#endif switch (status) { case BT_MESSAGE_ITERATOR_STATUS_OK: - BT_ASSERT_PRE(validate_messages(iterator, *user_count), - "Messages are invalid at this point: " - "%![msg-iter-]+i, count=%" PRIu64, - iterator, *user_count); + BT_ASSERT_PRE(*user_count <= MSG_BATCH_SIZE, + "Invalid returned message count: greater than " + "batch size: count=%" PRIu64 ", batch-size=%u", + *user_count, MSG_BATCH_SIZE); *msgs = (void *) iterator->base.msgs->pdata; break; case BT_MESSAGE_ITERATOR_STATUS_AGAIN: goto end; case BT_MESSAGE_ITERATOR_STATUS_END: - BT_ASSERT_PRE(self_comp_port_input_msg_iter_can_end(iterator), - "Message iterator cannot end at this point: " - "%!+i", iterator); set_self_comp_port_input_msg_iterator_state(iterator, BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_ENDED); goto end; @@ -842,7 +608,6 @@ enum bt_message_iterator_status bt_port_output_message_iterator_next( BT_ASSERT_PRE_NON_NULL(count_to_user, "Message count (output)"); BT_LIB_LOGD("Getting next output port message iterator's messages: " "%!+i", iterator); - graph_status = bt_graph_consume_sink_no_check(iterator->graph, iterator->colander); switch (graph_status) { @@ -871,13 +636,22 @@ enum bt_message_iterator_status bt_port_output_message_iterator_next( return status; } -struct bt_component *bt_self_component_port_input_message_iterator_borrow_component( +struct bt_component * +bt_self_component_port_input_message_iterator_borrow_component( struct bt_self_component_port_input_message_iterator *iterator) { BT_ASSERT_PRE_NON_NULL(iterator, "Message iterator"); return iterator->upstream_component; } +const struct bt_component * +bt_self_component_port_input_message_iterator_borrow_component_const( + const struct bt_self_component_port_input_message_iterator *iterator) +{ + BT_ASSERT_PRE_NON_NULL(iterator, "Message iterator"); + return iterator->upstream_component; +} + struct bt_self_component *bt_self_message_iterator_borrow_component( struct bt_self_message_iterator *self_iterator) { @@ -913,8 +687,7 @@ void bt_port_output_message_iterator_destroy(struct bt_object *obj) } struct bt_port_output_message_iterator * -bt_port_output_message_iterator_create( - struct bt_graph *graph, +bt_port_output_message_iterator_create(struct bt_graph *graph, const struct bt_port_output *output_port) { struct bt_port_output_message_iterator *iterator = NULL; @@ -936,6 +709,8 @@ bt_port_output_message_iterator_create( (void *) graph, "Output port is not part of graph: %![graph-]+g, %![port-]+p", graph, output_port); + BT_ASSERT_PRE(!graph->has_sink, + "Graph already has a sink component: %![graph-]+g"); /* Create message iterator */ BT_LIB_LOGD("Creating message iterator on output port: " @@ -1007,6 +782,15 @@ bt_port_output_message_iterator_create( * member. */ bt_graph_set_can_consume(iterator->graph, false); + + /* Also set the graph as being configured. */ + graph_status = bt_graph_configure(graph); + if (graph_status != BT_GRAPH_STATUS_OK) { + BT_LIB_LOGW("Cannot configure graph after having added colander: " + "%![graph-]+g, status=%s", graph, + bt_graph_status_string(graph_status)); + goto error; + } goto end; error: @@ -1040,6 +824,629 @@ end: return (void *) iterator; } +bt_bool bt_self_component_port_input_message_iterator_can_seek_ns_from_origin( + struct bt_self_component_port_input_message_iterator *iterator, + int64_t ns_from_origin) +{ + bt_bool can = BT_FALSE; + + BT_ASSERT_PRE_NON_NULL(iterator, "Message iterator"); + BT_ASSERT_PRE_ITER_HAS_STATE_TO_SEEK(iterator); + BT_ASSERT_PRE( + bt_component_borrow_graph(iterator->upstream_component)->config_state != + BT_GRAPH_CONFIGURATION_STATE_CONFIGURING, + "Graph is not configured: %!+g", + bt_component_borrow_graph(iterator->upstream_component)); + + if (iterator->methods.can_seek_ns_from_origin) { + can = iterator->methods.can_seek_ns_from_origin(iterator, + ns_from_origin); + goto end; + } + + /* + * Automatic seeking fall back: if we can seek to the beginning, + * then we can automatically seek to any message. + */ + if (iterator->methods.can_seek_beginning) { + can = iterator->methods.can_seek_beginning(iterator); + } + +end: + return can; +} + +bt_bool bt_self_component_port_input_message_iterator_can_seek_beginning( + struct bt_self_component_port_input_message_iterator *iterator) +{ + bt_bool can = BT_FALSE; + + BT_ASSERT_PRE_NON_NULL(iterator, "Message iterator"); + BT_ASSERT_PRE_ITER_HAS_STATE_TO_SEEK(iterator); + BT_ASSERT_PRE( + bt_component_borrow_graph(iterator->upstream_component)->config_state != + BT_GRAPH_CONFIGURATION_STATE_CONFIGURING, + "Graph is not configured: %!+g", + bt_component_borrow_graph(iterator->upstream_component)); + + if (iterator->methods.can_seek_beginning) { + can = iterator->methods.can_seek_beginning(iterator); + } + + return can; +} + +static inline +void set_iterator_state_after_seeking( + struct bt_self_component_port_input_message_iterator *iterator, + enum bt_message_iterator_status status) +{ + enum bt_self_component_port_input_message_iterator_state new_state = 0; + + /* Set iterator's state depending on seeking status */ + switch (status) { + case BT_MESSAGE_ITERATOR_STATUS_OK: + new_state = BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_ACTIVE; + break; + case BT_MESSAGE_ITERATOR_STATUS_AGAIN: + new_state = BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_LAST_SEEKING_RETURNED_AGAIN; + break; + case BT_MESSAGE_ITERATOR_STATUS_ERROR: + case BT_MESSAGE_ITERATOR_STATUS_NOMEM: + new_state = BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_LAST_SEEKING_RETURNED_ERROR; + break; + case BT_MESSAGE_ITERATOR_STATUS_END: + new_state = BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_ENDED; + break; + default: + abort(); + } + + set_self_comp_port_input_msg_iterator_state(iterator, new_state); +} + +enum bt_message_iterator_status +bt_self_component_port_input_message_iterator_seek_beginning( + struct bt_self_component_port_input_message_iterator *iterator) +{ + int status; + + BT_ASSERT_PRE_NON_NULL(iterator, "Message iterator"); + BT_ASSERT_PRE_ITER_HAS_STATE_TO_SEEK(iterator); + BT_ASSERT_PRE( + bt_component_borrow_graph(iterator->upstream_component)->config_state != + BT_GRAPH_CONFIGURATION_STATE_CONFIGURING, + "Graph is not configured: %!+g", + bt_component_borrow_graph(iterator->upstream_component)); + BT_ASSERT_PRE( + bt_self_component_port_input_message_iterator_can_seek_beginning( + iterator), + "Message iterator cannot seek beginning: %!+i", iterator); + BT_LIB_LOGD("Calling user's \"seek beginning\" method: %!+i", iterator); + set_self_comp_port_input_msg_iterator_state(iterator, + BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_SEEKING); + status = iterator->methods.seek_beginning(iterator); + BT_LOGD("User method returned: status=%s", + bt_message_iterator_status_string(status)); + BT_ASSERT_PRE(status == BT_MESSAGE_ITERATOR_STATUS_OK || + status == BT_MESSAGE_ITERATOR_STATUS_ERROR || + status == BT_MESSAGE_ITERATOR_STATUS_NOMEM || + status == BT_MESSAGE_ITERATOR_STATUS_AGAIN, + "Unexpected status: %![iter-]+i, status=%s", + iterator, bt_common_self_message_iterator_status_string(status)); + set_iterator_state_after_seeking(iterator, status); + return status; +} + +static inline +enum bt_message_iterator_status auto_seek_handle_message( + struct bt_self_component_port_input_message_iterator *iterator, + int64_t ns_from_origin, const struct bt_message *msg, + bool *got_first) +{ + enum bt_message_iterator_status status = BT_MESSAGE_ITERATOR_STATUS_OK; + int64_t msg_ns_from_origin; + const struct bt_clock_snapshot *clk_snapshot = NULL; + int ret; + + BT_ASSERT(msg); + BT_ASSERT(got_first); + + switch (msg->type) { + case BT_MESSAGE_TYPE_EVENT: + { + const struct bt_message_event *event_msg = + (const void *) msg; + + clk_snapshot = event_msg->default_cs; + BT_ASSERT_PRE(clk_snapshot, + "Event message has no default clock snapshot: %!+n", + event_msg); + break; + } + case BT_MESSAGE_TYPE_MESSAGE_ITERATOR_INACTIVITY: + { + const struct bt_message_message_iterator_inactivity *inactivity_msg = + (const void *) msg; + + clk_snapshot = inactivity_msg->default_cs; + BT_ASSERT(clk_snapshot); + break; + } + case BT_MESSAGE_TYPE_PACKET_BEGINNING: + case BT_MESSAGE_TYPE_PACKET_END: + { + const struct bt_message_packet *packet_msg = + (const void *) msg; + + clk_snapshot = packet_msg->default_cs; + BT_ASSERT_PRE(clk_snapshot, + "Packet message has no default clock snapshot: %!+n", + packet_msg); + break; + } + case BT_MESSAGE_TYPE_DISCARDED_EVENTS: + case BT_MESSAGE_TYPE_DISCARDED_PACKETS: + { + struct bt_message_discarded_items *msg_disc_items = + (void *) msg; + + BT_ASSERT_PRE(msg_disc_items->default_begin_cs && + msg_disc_items->default_end_cs, + "Discarded events/packets message has no default clock snapshots: %!+n", + msg_disc_items); + ret = bt_clock_snapshot_get_ns_from_origin( + msg_disc_items->default_begin_cs, + &msg_ns_from_origin); + if (ret) { + status = BT_MESSAGE_ITERATOR_STATUS_ERROR; + goto end; + } + + if (msg_ns_from_origin >= ns_from_origin) { + *got_first = true; + goto push_msg; + } + + ret = bt_clock_snapshot_get_ns_from_origin( + msg_disc_items->default_end_cs, + &msg_ns_from_origin); + if (ret) { + status = BT_MESSAGE_ITERATOR_STATUS_ERROR; + goto end; + } + + if (msg_ns_from_origin >= ns_from_origin) { + /* + * The discarded items message's beginning time + * is before the requested seeking time, but its + * end time is after. Modify the message so as + * to set its beginning time to the requested + * seeking time, and make its item count unknown + * as we don't know if items were really + * discarded within the new time range. + */ + uint64_t new_begin_raw_value; + + ret = bt_clock_class_clock_value_from_ns_from_origin( + msg_disc_items->default_end_cs->clock_class, + ns_from_origin, &new_begin_raw_value); + if (ret) { + status = BT_MESSAGE_ITERATOR_STATUS_ERROR; + goto end; + } + + bt_clock_snapshot_set_raw_value( + msg_disc_items->default_begin_cs, + new_begin_raw_value); + msg_disc_items->count.base.avail = + BT_PROPERTY_AVAILABILITY_NOT_AVAILABLE; + + /* + * It is safe to push it because its beginning + * time is exactly the requested seeking time. + */ + goto push_msg; + } else { + goto skip_msg; + } + } + case BT_MESSAGE_TYPE_STREAM_ACTIVITY_BEGINNING: + { + const struct bt_message_stream_activity *stream_act_msg = + (const void *) msg; + + switch (stream_act_msg->default_cs_state) { + case BT_MESSAGE_STREAM_ACTIVITY_CLOCK_SNAPSHOT_STATE_UNKNOWN: + case BT_MESSAGE_STREAM_ACTIVITY_CLOCK_SNAPSHOT_STATE_INFINITE: + /* + * -inf is always less than any requested time, + * and we can't assume any specific time for an + * unknown clock snapshot, so skip this. + */ + goto skip_msg; + case BT_MESSAGE_STREAM_ACTIVITY_CLOCK_SNAPSHOT_STATE_KNOWN: + clk_snapshot = stream_act_msg->default_cs; + BT_ASSERT(clk_snapshot); + break; + default: + abort(); + } + + break; + } + case BT_MESSAGE_TYPE_STREAM_ACTIVITY_END: + { + const struct bt_message_stream_activity *stream_act_msg = + (const void *) msg; + + switch (stream_act_msg->default_cs_state) { + case BT_MESSAGE_STREAM_ACTIVITY_CLOCK_SNAPSHOT_STATE_UNKNOWN: + /* + * We can't assume any specific time for an + * unknown clock snapshot, so skip this. + */ + goto skip_msg; + case BT_MESSAGE_STREAM_ACTIVITY_CLOCK_SNAPSHOT_STATE_INFINITE: + /* + * +inf is always greater than any requested + * time. + */ + *got_first = true; + goto push_msg; + case BT_MESSAGE_STREAM_ACTIVITY_CLOCK_SNAPSHOT_STATE_KNOWN: + clk_snapshot = stream_act_msg->default_cs; + BT_ASSERT(clk_snapshot); + break; + default: + abort(); + } + + break; + } + case BT_MESSAGE_TYPE_STREAM_BEGINNING: + case BT_MESSAGE_TYPE_STREAM_END: + /* Ignore */ + goto skip_msg; + default: + abort(); + } + + BT_ASSERT(clk_snapshot); + ret = bt_clock_snapshot_get_ns_from_origin(clk_snapshot, + &msg_ns_from_origin); + if (ret) { + status = BT_MESSAGE_ITERATOR_STATUS_ERROR; + goto end; + } + + if (msg_ns_from_origin >= ns_from_origin) { + *got_first = true; + goto push_msg; + } + +skip_msg: + bt_object_put_no_null_check(msg); + msg = NULL; + goto end; + +push_msg: + g_queue_push_head(iterator->auto_seek_msgs, (void *) msg); + msg = NULL; + +end: + BT_ASSERT(!msg || status != BT_MESSAGE_ITERATOR_STATUS_OK); + return status; +} + +static +enum bt_message_iterator_status find_message_ge_ns_from_origin( + struct bt_self_component_port_input_message_iterator *iterator, + int64_t ns_from_origin) +{ + int status; + enum bt_self_component_port_input_message_iterator_state init_state = + iterator->state; + const struct bt_message *messages[MSG_BATCH_SIZE]; + uint64_t user_count = 0; + uint64_t i; + bool got_first = false; + + BT_ASSERT(iterator); + memset(&messages[0], 0, sizeof(messages[0]) * MSG_BATCH_SIZE); + + /* + * Make this iterator temporarily active (not seeking) to call + * the "next" method. + */ + set_self_comp_port_input_msg_iterator_state(iterator, + BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_ACTIVE); + + BT_ASSERT(iterator->methods.next); + + while (!got_first) { + /* + * Call the user's "next" method to get the next + * messages and status. + */ + BT_LOGD_STR("Calling user's \"next\" method."); + status = iterator->methods.next(iterator, + &messages[0], MSG_BATCH_SIZE, &user_count); + BT_LOGD("User method returned: status=%s", + bt_message_iterator_status_string(status)); + + /* + * The user's "next" method must not do any action which + * would change the iterator's state. + */ + BT_ASSERT(iterator->state == + BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_ACTIVE); + + switch (status) { + case BT_MESSAGE_ITERATOR_STATUS_OK: + BT_ASSERT_PRE(user_count <= MSG_BATCH_SIZE, + "Invalid returned message count: greater than " + "batch size: count=%" PRIu64 ", batch-size=%u", + user_count, MSG_BATCH_SIZE); + break; + case BT_MESSAGE_ITERATOR_STATUS_AGAIN: + case BT_MESSAGE_ITERATOR_STATUS_ERROR: + case BT_MESSAGE_ITERATOR_STATUS_NOMEM: + case BT_MESSAGE_ITERATOR_STATUS_END: + goto end; + default: + abort(); + } + + for (i = 0; i < user_count; i++) { + if (got_first) { + g_queue_push_head(iterator->auto_seek_msgs, + (void *) messages[i]); + messages[i] = NULL; + continue; + } + + status = auto_seek_handle_message(iterator, + ns_from_origin, messages[i], &got_first); + if (status == BT_MESSAGE_ITERATOR_STATUS_OK) { + /* Message was either pushed or moved */ + messages[i] = NULL; + } else { + goto end; + } + } + } + +end: + for (i = 0; i < user_count; i++) { + if (messages[i]) { + bt_object_put_no_null_check(messages[i]); + } + } + + set_self_comp_port_input_msg_iterator_state(iterator, init_state); + return status; +} + +static +enum bt_self_message_iterator_status post_auto_seek_next( + struct bt_self_component_port_input_message_iterator *iterator, + bt_message_array_const msgs, uint64_t capacity, + uint64_t *count) +{ + BT_ASSERT(!g_queue_is_empty(iterator->auto_seek_msgs)); + *count = 0; + + /* + * Move auto-seek messages to the output array (which is this + * iterator's base message array). + */ + while (capacity > 0 && !g_queue_is_empty(iterator->auto_seek_msgs)) { + msgs[*count] = g_queue_pop_tail(iterator->auto_seek_msgs); + capacity--; + (*count)++; + } + + BT_ASSERT(*count > 0); + + if (g_queue_is_empty(iterator->auto_seek_msgs)) { + /* No more auto-seek messages */ + switch (iterator->upstream_component->class->type) { + case BT_COMPONENT_CLASS_TYPE_SOURCE: + { + struct bt_component_class_source *src_comp_cls = + (void *) iterator->upstream_component->class; + + iterator->methods.next = + (bt_self_component_port_input_message_iterator_next_method) + src_comp_cls->methods.msg_iter_next; + break; + } + case BT_COMPONENT_CLASS_TYPE_FILTER: + { + struct bt_component_class_filter *flt_comp_cls = + (void *) iterator->upstream_component->class; + + iterator->methods.next = + (bt_self_component_port_input_message_iterator_next_method) + flt_comp_cls->methods.msg_iter_next; + break; + } + default: + abort(); + } + } + + return BT_SELF_MESSAGE_ITERATOR_STATUS_OK; +} + +enum bt_message_iterator_status +bt_self_component_port_input_message_iterator_seek_ns_from_origin( + struct bt_self_component_port_input_message_iterator *iterator, + int64_t ns_from_origin) +{ + int status; + + BT_ASSERT_PRE_NON_NULL(iterator, "Message iterator"); + BT_ASSERT_PRE_ITER_HAS_STATE_TO_SEEK(iterator); + BT_ASSERT_PRE( + bt_component_borrow_graph(iterator->upstream_component)->config_state != + BT_GRAPH_CONFIGURATION_STATE_CONFIGURING, + "Graph is not configured: %!+g", + bt_component_borrow_graph(iterator->upstream_component)); + BT_ASSERT_PRE( + bt_self_component_port_input_message_iterator_can_seek_ns_from_origin( + iterator, ns_from_origin), + "Message iterator cannot seek nanoseconds from origin: %!+i, " + "ns-from-origin=%" PRId64, iterator, ns_from_origin); + set_self_comp_port_input_msg_iterator_state(iterator, + BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_SEEKING); + + if (iterator->methods.seek_ns_from_origin) { + BT_LIB_LOGD("Calling user's \"seek nanoseconds from origin\" method: " + "%![iter-]+i, ns=%" PRId64, iterator, ns_from_origin); + status = iterator->methods.seek_ns_from_origin(iterator, + ns_from_origin); + BT_LOGD("User method returned: status=%s", + bt_message_iterator_status_string(status)); + BT_ASSERT_PRE(status == BT_MESSAGE_ITERATOR_STATUS_OK || + status == BT_MESSAGE_ITERATOR_STATUS_ERROR || + status == BT_MESSAGE_ITERATOR_STATUS_NOMEM || + status == BT_MESSAGE_ITERATOR_STATUS_AGAIN, + "Unexpected status: %![iter-]+i, status=%s", + iterator, + bt_common_self_message_iterator_status_string(status)); + } else { + /* Start automatic seeking: seek beginning first */ + BT_ASSERT(iterator->methods.can_seek_beginning(iterator)); + BT_ASSERT(iterator->methods.seek_beginning); + BT_LIB_LOGD("Calling user's \"seek beginning\" method: %!+i", + iterator); + status = iterator->methods.seek_beginning(iterator); + BT_LOGD("User method returned: status=%s", + bt_message_iterator_status_string(status)); + BT_ASSERT_PRE(status == BT_MESSAGE_ITERATOR_STATUS_OK || + status == BT_MESSAGE_ITERATOR_STATUS_ERROR || + status == BT_MESSAGE_ITERATOR_STATUS_NOMEM || + status == BT_MESSAGE_ITERATOR_STATUS_AGAIN, + "Unexpected status: %![iter-]+i, status=%s", + iterator, + bt_common_self_message_iterator_status_string(status)); + switch (status) { + case BT_MESSAGE_ITERATOR_STATUS_OK: + break; + case BT_MESSAGE_ITERATOR_STATUS_ERROR: + case BT_MESSAGE_ITERATOR_STATUS_NOMEM: + case BT_MESSAGE_ITERATOR_STATUS_AGAIN: + goto end; + default: + abort(); + } + + /* + * Find the first message which has a default clock + * snapshot greater than or equal to the requested + * seeking time, and move the received messages from + * this point in the batch to this iterator's auto-seek + * message queue. + */ + while (!g_queue_is_empty(iterator->auto_seek_msgs)) { + bt_object_put_no_null_check( + g_queue_pop_tail(iterator->auto_seek_msgs)); + } + + status = find_message_ge_ns_from_origin(iterator, + ns_from_origin); + switch (status) { + case BT_MESSAGE_ITERATOR_STATUS_OK: + case BT_MESSAGE_ITERATOR_STATUS_END: + /* + * If there are messages in the auto-seek + * message queue, replace the user's "next" + * method with a custom, temporary "next" method + * which returns them. + */ + if (!g_queue_is_empty(iterator->auto_seek_msgs)) { + iterator->methods.next = + (bt_self_component_port_input_message_iterator_next_method) + post_auto_seek_next; + } + + /* + * `BT_MESSAGE_ITERATOR_STATUS_END` becomes + * `BT_MESSAGE_ITERATOR_STATUS_OK`: the next + * time this iterator's "next" method is called, + * it will return + * `BT_MESSAGE_ITERATOR_STATUS_END`. + */ + status = BT_MESSAGE_ITERATOR_STATUS_OK; + break; + case BT_MESSAGE_ITERATOR_STATUS_ERROR: + case BT_MESSAGE_ITERATOR_STATUS_NOMEM: + case BT_MESSAGE_ITERATOR_STATUS_AGAIN: + goto end; + default: + abort(); + } + } + +end: + set_iterator_state_after_seeking(iterator, status); + return status; +} + +static inline +bt_self_component_port_input_message_iterator * +borrow_output_port_message_iterator_upstream_iterator( + struct bt_port_output_message_iterator *iterator) +{ + struct bt_component_class_sink_colander_priv_data *colander_data; + + BT_ASSERT(iterator); + colander_data = (void *) iterator->colander->parent.user_data; + BT_ASSERT(colander_data); + BT_ASSERT(colander_data->msg_iter); + return colander_data->msg_iter; +} + +bt_bool bt_port_output_message_iterator_can_seek_ns_from_origin( + struct bt_port_output_message_iterator *iterator, + int64_t ns_from_origin) +{ + BT_ASSERT_PRE_NON_NULL(iterator, "Message iterator"); + return bt_self_component_port_input_message_iterator_can_seek_ns_from_origin( + borrow_output_port_message_iterator_upstream_iterator( + iterator), ns_from_origin); +} + +bt_bool bt_port_output_message_iterator_can_seek_beginning( + struct bt_port_output_message_iterator *iterator) +{ + BT_ASSERT_PRE_NON_NULL(iterator, "Message iterator"); + return bt_self_component_port_input_message_iterator_can_seek_beginning( + borrow_output_port_message_iterator_upstream_iterator( + iterator)); +} + +enum bt_message_iterator_status bt_port_output_message_iterator_seek_ns_from_origin( + struct bt_port_output_message_iterator *iterator, + int64_t ns_from_origin) +{ + BT_ASSERT_PRE_NON_NULL(iterator, "Message iterator"); + return bt_self_component_port_input_message_iterator_seek_ns_from_origin( + borrow_output_port_message_iterator_upstream_iterator(iterator), + ns_from_origin); +} + +enum bt_message_iterator_status bt_port_output_message_iterator_seek_beginning( + struct bt_port_output_message_iterator *iterator) +{ + BT_ASSERT_PRE_NON_NULL(iterator, "Message iterator"); + return bt_self_component_port_input_message_iterator_seek_beginning( + borrow_output_port_message_iterator_upstream_iterator( + iterator)); +} + void bt_port_output_message_iterator_get_ref( const struct bt_port_output_message_iterator *iterator) {