+
+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)->is_configured,
+ "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)->is_configured,
+ "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);
+}
+
+#ifdef BT_DEV_MODE
+# define set_iterator_state_after_seeking _set_iterator_state_after_seeking
+#else
+# define set_iterator_state_after_seeking(_iter, _status)
+#endif
+
+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)->is_configured,
+ "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_self_message_iterator_status_string(status));
+ set_iterator_state_after_seeking(iterator, status);
+ return status;
+}
+
+static inline
+int get_message_ns_from_origin(const struct bt_message *msg,
+ int64_t *ns_from_origin, bool *ignore)
+{
+ const struct bt_clock_snapshot *clk_snapshot = NULL;
+ int ret = 0;
+
+ switch (msg->type) {
+ case BT_MESSAGE_TYPE_EVENT:
+ {
+ const struct bt_message_event *event_msg =
+ (const void *) msg;
+
+ clk_snapshot = event_msg->event->default_cs;
+ BT_ASSERT_PRE(clk_snapshot,
+ "Event has no default clock snapshot: %!+e",
+ event_msg->event);
+ break;
+ }
+ case BT_MESSAGE_TYPE_INACTIVITY:
+ {
+ const struct bt_message_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:
+ /* Ignore */
+ goto end;
+ case BT_MESSAGE_TYPE_DISCARDED_EVENTS:
+ case BT_MESSAGE_TYPE_DISCARDED_PACKETS:
+ {
+ const struct bt_message_discarded_items *disc_items_msg =
+ (const void *) msg;
+
+ clk_snapshot = disc_items_msg->default_begin_cs;
+ BT_ASSERT_PRE(clk_snapshot,
+ "Discarded events/packets message has no default clock snapshot: %!+n",
+ msg);
+ break;
+ }
+ 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 set_ignore;
+ 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 set_ignore;
+ case BT_MESSAGE_STREAM_ACTIVITY_CLOCK_SNAPSHOT_STATE_INFINITE:
+ /*
+ * +inf is always greater than any requested
+ * time.
+ */
+ *ns_from_origin = INT64_MAX;
+ goto end;
+ 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 */
+ break;
+ default:
+ abort();
+ }
+
+set_ignore:
+ if (!clk_snapshot) {
+ *ignore = true;
+ goto end;
+ }
+
+ ret = bt_clock_snapshot_get_ns_from_origin(clk_snapshot,
+ ns_from_origin);
+
+end:
+ return ret;
+}
+
+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;
+
+ 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 (true) {
+ /*
+ * 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));
+
+#ifdef BT_DEV_MODE
+ /*
+ * 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);
+#endif
+
+ 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();
+ }
+
+ /*
+ * Find first message which has a default clock snapshot
+ * that is greater than or equal to the requested value.
+ *
+ * For event and inactivity messages, compare with the
+ * default clock snapshot.
+ *
+ * For packet beginning messages, compare with the
+ * default beginning clock snapshot, if any.
+ *
+ * For packet end messages, compare with the default end
+ * clock snapshot, if any.
+ *
+ * For stream beginning, stream end, ignore.
+ */
+ for (i = 0; i < user_count; i++) {
+ const struct bt_message *msg = messages[i];
+ int64_t msg_ns_from_origin;
+ bool ignore = false;
+ int ret;
+
+ BT_ASSERT(msg);
+ ret = get_message_ns_from_origin(msg, &msg_ns_from_origin,
+ &ignore);
+ if (ret) {
+ status = BT_MESSAGE_ITERATOR_STATUS_ERROR;
+ goto end;
+ }
+
+ if (ignore) {
+ /* Skip message without a clock snapshot */
+ continue;
+ }
+
+ if (msg_ns_from_origin >= ns_from_origin) {
+ /*
+ * We found it: move this message and
+ * the following ones to the iterator's
+ * auto-seek message array.
+ */
+ uint64_t j;
+
+ for (j = i; j < user_count; j++) {
+ iterator->auto_seek_msgs->pdata[j - i] =
+ (void *) messages[j];
+ messages[j] = NULL;
+ }
+
+ iterator->auto_seek_msg_count = user_count - i;
+ goto end;
+ }
+
+ bt_object_put_no_null_check(msg);
+ messages[i] = NULL;
+ }
+ }
+
+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(iterator->auto_seek_msg_count <= capacity);
+ BT_ASSERT(iterator->auto_seek_msg_count > 0);
+
+ /*
+ * Move auto-seek messages to the output array (which is this
+ * iterator's base message array.
+ */
+ memcpy(&msgs[0], &iterator->auto_seek_msgs->pdata[0],
+ sizeof(msgs[0]) * iterator->auto_seek_msg_count);
+ memset(&iterator->auto_seek_msgs->pdata[0], 0,
+ sizeof(iterator->auto_seek_msgs->pdata[0]) *
+ iterator->auto_seek_msg_count);
+ *count = iterator->auto_seek_msg_count;
+
+ /* Restore real user's "next" method */
+ 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)->is_configured,
+ "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_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_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
+ * nanoseconds from origin, and move the received
+ * messages from this point in the batch to this
+ * iterator's auto-seek message array.
+ */
+ status = find_message_ge_ns_from_origin(iterator,
+ ns_from_origin);
+ switch (status) {
+ case BT_MESSAGE_ITERATOR_STATUS_OK:
+ /*
+ * Replace the user's "next" method with a
+ * custom, temporary "next" method which returns
+ * the messages in the iterator's message array.
+ */
+ iterator->methods.next =
+ (bt_self_component_port_input_message_iterator_next_method)
+ post_auto_seek_next;
+ break;
+ case BT_MESSAGE_ITERATOR_STATUS_ERROR:
+ case BT_MESSAGE_ITERATOR_STATUS_NOMEM:
+ case BT_MESSAGE_ITERATOR_STATUS_AGAIN:
+ goto end;
+ case BT_MESSAGE_ITERATOR_STATUS_END:
+ /*
+ * The iterator reached the end: just return
+ * `BT_MESSAGE_ITERATOR_STATUS_OK` here, as if
+ * the seeking operation occured: the next
+ * "next" method will return
+ * `BT_MESSAGE_ITERATOR_STATUS_END` itself.
+ */
+ break;
+ default:
+ abort();
+ }
+ }
+
+end:
+ set_iterator_state_after_seeking(iterator, status);
+
+ if (status == BT_MESSAGE_ITERATOR_STATUS_END) {
+ status = BT_MESSAGE_ITERATOR_STATUS_OK;
+ }
+
+ 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)
+{
+ bt_object_get_ref(iterator);
+}
+
+void bt_port_output_message_iterator_put_ref(
+ const struct bt_port_output_message_iterator *iterator)
+{
+ bt_object_put_ref(iterator);
+}
+
+void bt_self_component_port_input_message_iterator_get_ref(
+ const struct bt_self_component_port_input_message_iterator *iterator)
+{
+ bt_object_get_ref(iterator);
+}
+
+void bt_self_component_port_input_message_iterator_put_ref(
+ const struct bt_self_component_port_input_message_iterator *iterator)
+{
+ bt_object_put_ref(iterator);
+}