sink.text.pretty: use fast clock class access

[babeltrace.git] / plugins / utils / muxer / muxer.c

diff --git a/plugins/utils/muxer/muxer.c b/plugins/utils/muxer/muxer.c
index 19c6b768f6d7079a21c6c0399e76d9e39787b56f..4b3669726cd7c28f99bcccb24809e95fb27ca738 100644 (file)
--- a/plugins/utils/muxer/muxer.c
+++ b/plugins/utils/muxer/muxer.c
@@ -610,11 +610,11 @@ int get_msg_ts_ns(struct muxer_comp *muxer_comp,
 {
        const bt_clock_class *clock_class = NULL;
        const bt_clock_snapshot *clock_snapshot = NULL;
-       const bt_event *event = NULL;
        int ret = 0;
        const unsigned char *cc_uuid;
        const char *cc_name;
        bt_clock_snapshot_state cs_state = BT_CLOCK_SNAPSHOT_STATE_KNOWN;
+       bt_message_stream_activity_clock_snapshot_state sa_cs_state;
 
        BT_ASSERT(msg);
        BT_ASSERT(ts_ns);
@@ -626,12 +626,45 @@ int get_msg_ts_ns(struct muxer_comp *muxer_comp,
 
        switch (bt_message_get_type(msg)) {
        case BT_MESSAGE_TYPE_EVENT:
-               event = bt_message_event_borrow_event_const(msg);
-               BT_ASSERT(event);
-               cs_state = bt_event_borrow_default_clock_snapshot_const(event,
-                       &clock_snapshot);
+               cs_state = bt_message_event_borrow_default_clock_snapshot_const(
+                       msg, &clock_snapshot);
                break;
+       case BT_MESSAGE_TYPE_PACKET_BEGINNING:
+               cs_state = bt_message_packet_beginning_borrow_default_clock_snapshot_const(
+                       msg, &clock_snapshot);
+               break;
+       case BT_MESSAGE_TYPE_PACKET_END:
+               cs_state = bt_message_packet_end_borrow_default_clock_snapshot_const(
+                       msg, &clock_snapshot);
+               break;
+       case BT_MESSAGE_TYPE_DISCARDED_EVENTS:
+               cs_state = bt_message_discarded_events_borrow_default_beginning_clock_snapshot_const(
+                       msg, &clock_snapshot);
+               break;
+       case BT_MESSAGE_TYPE_DISCARDED_PACKETS:
+               cs_state = bt_message_discarded_packets_borrow_default_beginning_clock_snapshot_const(
+                       msg, &clock_snapshot);
+               break;
+       case BT_MESSAGE_TYPE_STREAM_ACTIVITY_BEGINNING:
+               sa_cs_state = bt_message_stream_activity_beginning_borrow_default_clock_snapshot_const(
+                       msg, &clock_snapshot);
+               if (sa_cs_state != BT_MESSAGE_STREAM_ACTIVITY_CLOCK_SNAPSHOT_STATE_KNOWN) {
+                       /* No timestamp: high priority */
+                       *ts_ns = last_returned_ts_ns;
+                       goto end;
+               }
 
+               break;
+       case BT_MESSAGE_TYPE_STREAM_ACTIVITY_END:
+               sa_cs_state = bt_message_stream_activity_end_borrow_default_clock_snapshot_const(
+                       msg, &clock_snapshot);
+               if (sa_cs_state != BT_MESSAGE_STREAM_ACTIVITY_CLOCK_SNAPSHOT_STATE_KNOWN) {
+                       /* No timestamp: high priority */
+                       *ts_ns = last_returned_ts_ns;
+                       goto end;
+               }
+
+               break;
        case BT_MESSAGE_TYPE_INACTIVITY:
                cs_state =
                        bt_message_inactivity_borrow_default_clock_snapshot_const(
@@ -1117,10 +1150,10 @@ bt_self_message_iterator_status muxer_msg_iter_do_next(
                BT_SELF_MESSAGE_ITERATOR_STATUS_OK;
        uint64_t i = 0;
 
-       while (i < capacity && status == BT_MESSAGE_ITERATOR_STATUS_OK) {
+       while (i < capacity && status == BT_SELF_MESSAGE_ITERATOR_STATUS_OK) {
                status = muxer_msg_iter_do_next_one(muxer_comp,
                        muxer_msg_iter, &msgs[i]);
-               if (status == BT_MESSAGE_ITERATOR_STATUS_OK) {
+               if (status == BT_SELF_MESSAGE_ITERATOR_STATUS_OK) {
                        i++;
                }
        }
@@ -1464,27 +1497,51 @@ end:
 }
 
 BT_HIDDEN
-void muxer_input_port_disconnected(
-               bt_self_component_filter *self_component,
-               bt_self_component_port_input *self_port)
+bt_bool muxer_msg_iter_can_seek_beginning(
+               bt_self_message_iterator *self_msg_iter)
 {
-       struct muxer_comp *muxer_comp =
-               bt_self_component_get_data(
-                       bt_self_component_filter_as_self_component(
-                               self_component));
-       const bt_port *port =
-               bt_self_component_port_as_port(
-                       bt_self_component_port_input_as_self_component_port(
-                               self_port));
+       struct muxer_msg_iter *muxer_msg_iter =
+               bt_self_message_iterator_get_data(self_msg_iter);
+       uint64_t i;
+       bt_bool ret = BT_TRUE;
 
-       BT_ASSERT(port);
-       BT_ASSERT(muxer_comp);
+       for (i = 0; i < muxer_msg_iter->muxer_upstream_msg_iters->len; i++) {
+               struct muxer_upstream_msg_iter *upstream_msg_iter =
+                       muxer_msg_iter->muxer_upstream_msg_iters->pdata[i];
 
-       /* One more available input port */
-       muxer_comp->available_input_ports++;
-       BT_LOGD("Leaving disconnected input port available for future connections: "
-               "comp-addr=%p, muxer-comp-addr=%p, port-addr=%p, "
-               "port-name=\"%s\", avail-input-port-count=%zu",
-               self_component, muxer_comp, port, bt_port_get_name(port),
-               muxer_comp->available_input_ports);
+               if (!bt_self_component_port_input_message_iterator_can_seek_beginning(
+                               upstream_msg_iter->msg_iter)) {
+                       ret = BT_FALSE;
+                       goto end;
+               }
+       }
+
+end:
+       return ret;
+}
+
+BT_HIDDEN
+bt_self_message_iterator_status muxer_msg_iter_seek_beginning(
+               bt_self_message_iterator *self_msg_iter)
+{
+       struct muxer_msg_iter *muxer_msg_iter =
+               bt_self_message_iterator_get_data(self_msg_iter);
+       int status;
+       uint64_t i;
+
+       for (i = 0; i < muxer_msg_iter->muxer_upstream_msg_iters->len; i++) {
+               struct muxer_upstream_msg_iter *upstream_msg_iter =
+                       muxer_msg_iter->muxer_upstream_msg_iters->pdata[i];
+
+               status = bt_self_component_port_input_message_iterator_seek_beginning(
+                       upstream_msg_iter->msg_iter);
+               if (status != BT_MESSAGE_ITERATOR_STATUS_OK) {
+                       goto end;
+               }
+       }
+
+       muxer_msg_iter->last_returned_ts_ns = INT64_MIN;
+
+end:
+       return status;
 }