* SOFTWARE.
*/
+#define BT_COMP_LOG_SELF_COMP self_comp
+#define BT_LOG_OUTPUT_LEVEL log_level
#define BT_LOG_TAG "PLUGIN/SRC.CTF.LTTNG-LIVE"
-#include "logging.h"
+#include "logging/comp-logging.h"
#include <glib.h>
#include <inttypes.h>
#include "compat/compiler.h"
#include <babeltrace2/types.h>
+#include "plugins/common/muxing/muxing.h"
+
#include "data-stream.h"
#include "metadata.h"
#include "lttng-live.h"
#define MAX_QUERY_SIZE (256*1024)
#define URL_PARAM "url"
+#define INPUTS_PARAM "inputs"
#define SESS_NOT_FOUND_ACTION_PARAM "session-not-found-action"
#define SESS_NOT_FOUND_ACTION_CONTINUE_STR "continue"
#define SESS_NOT_FOUND_ACTION_FAIL_STR "fail"
#define SESS_NOT_FOUND_ACTION_END_STR "end"
-#define print_dbg(fmt, ...) BT_LOGD(fmt, ## __VA_ARGS__)
+#define print_dbg(fmt, ...) BT_COMP_LOGD(fmt, ## __VA_ARGS__)
static
const char *print_live_iterator_status(enum lttng_live_iterator_status status)
#define print_stream_state(live_stream_iter) \
do { \
- BT_LOGD("stream state %s last_inact_ts %" PRId64 \
+ BT_COMP_LOGD("stream state %s last_inact_ts %" PRId64 \
", curr_inact_ts %" PRId64, \
print_state(live_stream_iter), \
live_stream_iter->last_inactivity_ts, \
} while (0);
BT_HIDDEN
-bool lttng_live_graph_is_canceled(struct lttng_live_component *lttng_live)
+bool lttng_live_graph_is_canceled(struct lttng_live_msg_iter *msg_iter)
{
- const bt_component *component;
bool ret;
- if (!lttng_live) {
+ if (!msg_iter) {
ret = false;
goto end;
}
- component = bt_component_source_as_component_const(
- bt_self_component_source_as_component_source(
- lttng_live->self_comp));
-
- ret = bt_component_graph_is_canceled(component);
+ ret = bt_self_message_iterator_is_interrupted(
+ msg_iter->self_msg_iter);
end:
return ret;
static
void lttng_live_destroy_trace(struct lttng_live_trace *trace)
{
- BT_LOGD("Destroy lttng_live_trace");
+ bt_logging_level log_level = trace->log_level;
+ bt_self_component *self_comp = trace->self_comp;
+
+ BT_COMP_LOGD("Destroy lttng_live_trace");
BT_ASSERT(trace->stream_iterators);
g_ptr_array_free(trace->stream_iterators, TRUE);
uint64_t trace_id)
{
struct lttng_live_trace *trace = NULL;
+ bt_logging_level log_level = session->log_level;
+ bt_self_component *self_comp = session->self_comp;
trace = g_new0(struct lttng_live_trace, 1);
if (!trace) {
goto error;
}
+ trace->log_level = session->log_level;
+ trace->self_comp = session->self_comp;
trace->session = session;
trace->id = trace_id;
trace->trace_class = NULL;
trace->new_metadata_needed = true;
g_ptr_array_add(session->traces, trace);
- BT_LOGI("Create trace");
+ BT_COMP_LOGI("Create trace");
goto end;
error:
g_free(trace);
{
int ret = 0;
struct lttng_live_session *session;
+ bt_logging_level log_level = lttng_live_msg_iter->log_level;
+ bt_self_component *self_comp = lttng_live_msg_iter->self_comp;
session = g_new0(struct lttng_live_session, 1);
if (!session) {
goto error;
}
+ session->log_level = lttng_live_msg_iter->log_level;
+ session->self_comp = lttng_live_msg_iter->self_comp;
session->id = session_id;
session->traces = g_ptr_array_new_with_free_func(
(GDestroyNotify) lttng_live_destroy_trace);
session->session_name = g_string_new(session_name);
BT_ASSERT(session->session_name);
- BT_LOGI("Reading from session: %" PRIu64 " hostname: %s session_name: %s",
+ BT_COMP_LOGI("Reading from session: %" PRIu64 " hostname: %s session_name: %s",
session->id, hostname, session_name);
g_ptr_array_add(lttng_live_msg_iter->sessions, session);
goto end;
error:
- BT_LOGE("Error adding session");
+ BT_COMP_LOGE("Error adding session");
g_free(session);
ret = -1;
end:
static
void lttng_live_destroy_session(struct lttng_live_session *session)
{
- struct lttng_live_component *live_comp;
+ bt_logging_level log_level;
+ bt_self_component *self_comp;
if (!session) {
goto end;
}
- BT_LOGD("Destroy lttng live session");
+ log_level = session->log_level;
+ self_comp = session->self_comp;
+ BT_COMP_LOGD("Destroy lttng live session");
if (session->id != -1ULL) {
if (lttng_live_detach_session(session)) {
- live_comp = session->lttng_live_msg_iter->lttng_live_comp;
if (session->lttng_live_msg_iter &&
- !lttng_live_graph_is_canceled(live_comp)) {
+ !lttng_live_graph_is_canceled(
+ session->lttng_live_msg_iter)) {
/* Old relayd cannot detach sessions. */
- BT_LOGD("Unable to detach lttng live session %" PRIu64,
+ BT_COMP_LOGD("Unable to detach lttng live session %" PRIu64,
session->id);
}
}
enum lttng_live_iterator_status lttng_live_iterator_next_check_stream_state(
struct lttng_live_stream_iterator *lttng_live_stream)
{
+ bt_logging_level log_level = lttng_live_stream->log_level;
+ bt_self_component *self_comp = lttng_live_stream->self_comp;
+
switch (lttng_live_stream->state) {
case LTTNG_LIVE_STREAM_QUIESCENT:
case LTTNG_LIVE_STREAM_ACTIVE_DATA:
break;
case LTTNG_LIVE_STREAM_ACTIVE_NO_DATA:
/* Invalid state. */
- BT_LOGF("Unexpected stream state \"ACTIVE_NO_DATA\"");
+ BT_COMP_LOGF("Unexpected stream state \"ACTIVE_NO_DATA\"");
abort();
case LTTNG_LIVE_STREAM_QUIESCENT_NO_DATA:
/* Invalid state. */
- BT_LOGF("Unexpected stream state \"QUIESCENT_NO_DATA\"");
+ BT_COMP_LOGF("Unexpected stream state \"QUIESCENT_NO_DATA\"");
abort();
case LTTNG_LIVE_STREAM_EOF:
break;
struct lttng_live_msg_iter *lttng_live_msg_iter,
struct lttng_live_stream_iterator *lttng_live_stream)
{
+ bt_logging_level log_level = lttng_live_msg_iter->log_level;
+ bt_self_component *self_comp = lttng_live_msg_iter->self_comp;
enum lttng_live_iterator_status ret =
LTTNG_LIVE_ITERATOR_STATUS_OK;
struct packet_index index;
ret = lttng_live_attach_session(session);
if (ret) {
if (lttng_live_msg_iter && lttng_live_graph_is_canceled(
- lttng_live_msg_iter->lttng_live_comp)) {
+ lttng_live_msg_iter)) {
status = LTTNG_LIVE_ITERATOR_STATUS_AGAIN;
} else {
status = LTTNG_LIVE_ITERATOR_STATUS_ERROR;
const bt_clock_class *clock_class = NULL;
const bt_clock_snapshot *clock_snapshot = NULL;
int ret = 0;
- bt_message_stream_activity_clock_snapshot_state sa_cs_state;
+ bt_logging_level log_level = lttng_live_msg_iter->log_level;
+ bt_self_component *self_comp = lttng_live_msg_iter->self_comp;
BT_ASSERT(msg);
BT_ASSERT(ts_ns);
- BT_LOGD("Getting message's timestamp: iter-data-addr=%p, msg-addr=%p, "
+ BT_COMP_LOGD("Getting message's timestamp: iter-data-addr=%p, msg-addr=%p, "
"last-msg-ts=%" PRId64, lttng_live_msg_iter, msg,
last_msg_ts_ns);
clock_snapshot = bt_message_discarded_packets_borrow_beginning_default_clock_snapshot_const(
msg);
break;
- case BT_MESSAGE_TYPE_STREAM_ACTIVITY_BEGINNING:
- clock_class =
- bt_message_stream_activity_beginning_borrow_stream_class_default_clock_class_const(
- msg);
- BT_ASSERT(clock_class);
-
- 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) {
- goto no_clock_snapshot;
- }
-
- break;
- case BT_MESSAGE_TYPE_STREAM_ACTIVITY_END:
- clock_class =
- bt_message_stream_activity_end_borrow_stream_class_default_clock_class_const(
- msg);
- BT_ASSERT(clock_class);
-
- 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) {
- goto no_clock_snapshot;
- }
-
- break;
case BT_MESSAGE_TYPE_MESSAGE_ITERATOR_INACTIVITY:
clock_snapshot =
bt_message_message_iterator_inactivity_borrow_default_clock_snapshot_const(
break;
default:
/* All the other messages have a higher priority */
- BT_LOGD_STR("Message has no timestamp: using the last message timestamp.");
+ BT_COMP_LOGD_STR("Message has no timestamp: using the last message timestamp.");
*ts_ns = last_msg_ts_ns;
goto end;
}
ret = bt_clock_snapshot_get_ns_from_origin(clock_snapshot, ts_ns);
if (ret) {
- BT_LOGE("Cannot get nanoseconds from Epoch of clock snapshot: "
+ BT_COMP_LOGE("Cannot get nanoseconds from Epoch of clock snapshot: "
"clock-snapshot-addr=%p", clock_snapshot);
goto error;
}
goto end;
-no_clock_snapshot:
- BT_LOGD_STR("Message's default clock snapshot is missing: "
- "using the last message timestamp.");
- *ts_ns = last_msg_ts_ns;
- goto end;
-
error:
ret = -1;
end:
if (ret == 0) {
- BT_LOGD("Found message's timestamp: "
+ BT_COMP_LOGD("Found message's timestamp: "
"iter-data-addr=%p, msg-addr=%p, "
"last-msg-ts=%" PRId64 ", ts=%" PRId64,
lttng_live_msg_iter, msg, last_msg_ts_ns, *ts_ns);
enum lttng_live_iterator_status ret = LTTNG_LIVE_ITERATOR_STATUS_OK;
enum bt_msg_iter_status status;
uint64_t session_idx, trace_idx;
+ bt_logging_level log_level = lttng_live_msg_iter->log_level;
+ bt_self_component *self_comp = lttng_live_msg_iter->self_comp;
for (session_idx = 0; session_idx < lttng_live_msg_iter->sessions->len;
session_idx++) {
case BT_MSG_ITER_STATUS_ERROR:
default:
ret = LTTNG_LIVE_ITERATOR_STATUS_ERROR;
- BT_LOGW("CTF msg iterator return an error or failed msg_iter=%p",
+ BT_COMP_LOGW("CTF msg iterator return an error or failed msg_iter=%p",
lttng_live_stream->msg_iter);
break;
}
return ret;
}
+static
+enum lttng_live_iterator_status lttng_live_iterator_close_stream(
+ struct lttng_live_msg_iter *lttng_live_msg_iter,
+ struct lttng_live_stream_iterator *stream_iter,
+ bt_message **curr_msg)
+{
+ enum lttng_live_iterator_status live_status =
+ LTTNG_LIVE_ITERATOR_STATUS_OK;
+ /*
+ * The viewer has hung up on us so we are closing the stream. The
+ * `bt_msg_iter` should simply realize that it needs to close the
+ * stream properly by emitting the necessary stream end message.
+ */
+ enum bt_msg_iter_status status =
+ bt_msg_iter_get_next_message(stream_iter->msg_iter,
+ lttng_live_msg_iter->self_msg_iter, curr_msg);
+
+ if (status == BT_MSG_ITER_STATUS_ERROR) {
+ live_status = LTTNG_LIVE_ITERATOR_STATUS_ERROR;
+ goto end;
+ }
+
+ BT_ASSERT(status == BT_MSG_ITER_STATUS_OK);
+
+end:
+ return live_status;
+}
+
/*
* helper function:
* handle_no_data_streams()
* When disconnected from relayd: try to re-connect endlessly.
*/
static
-enum lttng_live_iterator_status lttng_live_iterator_next_on_stream(
+enum lttng_live_iterator_status lttng_live_iterator_next_msg_on_stream(
struct lttng_live_msg_iter *lttng_live_msg_iter,
struct lttng_live_stream_iterator *stream_iter,
bt_message **curr_msg)
{
+ bt_logging_level log_level = lttng_live_msg_iter->log_level;
+ bt_self_component *self_comp = lttng_live_msg_iter->self_comp;
enum lttng_live_iterator_status live_status;
+ if (stream_iter->has_stream_hung_up) {
+ /*
+ * The stream has hung up and the stream was properly closed
+ * during the last call to the current function. Return _END
+ * status now so that this stream iterator is removed for the
+ * stream iterator list.
+ */
+ live_status = LTTNG_LIVE_ITERATOR_STATUS_END;
+ goto end;
+ }
+
retry:
print_stream_state(stream_iter);
live_status = lttng_live_iterator_handle_new_streams_and_metadata(
}
live_status = lttng_live_iterator_next_handle_one_no_data_stream(
lttng_live_msg_iter, stream_iter);
+
if (live_status != LTTNG_LIVE_ITERATOR_STATUS_OK) {
+ if (live_status == LTTNG_LIVE_ITERATOR_STATUS_END) {
+ /*
+ * We overwrite `live_status` since `curr_msg` is
+ * likely set to a valid message in this function.
+ */
+ live_status = lttng_live_iterator_close_stream(
+ lttng_live_msg_iter, stream_iter, curr_msg);
+ }
goto end;
}
live_status = lttng_live_iterator_next_handle_one_quiescent_stream(
lttng_live_msg_iter, stream_iter, curr_msg);
if (live_status != LTTNG_LIVE_ITERATOR_STATUS_OK) {
- BT_ASSERT(*curr_msg == NULL);
+ BT_ASSERT(!*curr_msg);
goto end;
}
if (*curr_msg) {
live_status = lttng_live_iterator_next_handle_one_active_data_stream(
lttng_live_msg_iter, stream_iter, curr_msg);
if (live_status != LTTNG_LIVE_ITERATOR_STATUS_OK) {
- BT_ASSERT(*curr_msg == NULL);
+ BT_ASSERT(!*curr_msg);
}
end:
enum lttng_live_iterator_status next_stream_iterator_for_trace(
struct lttng_live_msg_iter *lttng_live_msg_iter,
struct lttng_live_trace *live_trace,
- struct lttng_live_stream_iterator **candidate_stream_iter)
+ struct lttng_live_stream_iterator **youngest_trace_stream_iter)
{
- struct lttng_live_stream_iterator *curr_candidate_stream_iter = NULL;
+ struct lttng_live_stream_iterator *youngest_candidate_stream_iter = NULL;
enum lttng_live_iterator_status stream_iter_status;;
- int64_t curr_candidate_msg_ts = INT64_MAX;
+ int64_t youngest_candidate_msg_ts = INT64_MAX;
uint64_t stream_iter_idx;
+ bt_logging_level log_level = lttng_live_msg_iter->log_level;
+ bt_self_component *self_comp = lttng_live_msg_iter->self_comp;
BT_ASSERT(live_trace);
BT_ASSERT(live_trace->stream_iterators);
g_ptr_array_index(live_trace->stream_iterators,
stream_iter_idx);
+ /*
+ * Since we may remove elements from the GPtrArray as we
+ * iterate over it, it's possible to see the same element more
+ * than once.
+ */
+ if (stream_iter == youngest_candidate_stream_iter) {
+ stream_iter_idx++;
+ continue;
+ }
+
/*
* Find if there is are now current message for this stream
* iterator get it.
while (!stream_iter->current_msg) {
bt_message *msg = NULL;
int64_t curr_msg_ts_ns = INT64_MAX;
- stream_iter_status = lttng_live_iterator_next_on_stream(
+ stream_iter_status = lttng_live_iterator_next_msg_on_stream(
lttng_live_msg_iter, stream_iter, &msg);
- BT_LOGD("live stream iterator returned status :%s",
+ BT_COMP_LOGD("live stream iterator returned status :%s",
print_live_iterator_status(stream_iter_status));
if (stream_iter_status == LTTNG_LIVE_ITERATOR_STATUS_END) {
stream_iter_is_ended = true;
* We received a message in the past. To ensure
* monotonicity, we can't send it forward.
*/
- BT_LOGE("Message's timestamp is less than "
+ BT_COMP_LOGE("Message's timestamp is less than "
"lttng-live's message iterator's last "
"returned timestamp: "
"lttng-live-msg-iter-addr=%p, ts=%" PRId64 ", "
}
}
- if (!stream_iter_is_ended &&
- stream_iter->current_msg_ts_ns <= curr_candidate_msg_ts) {
- /*
- * Update the current best candidate message for the
- * stream iterator of thise live trace to be forwarded
- * downstream.
- */
- curr_candidate_msg_ts = stream_iter->current_msg_ts_ns;
- curr_candidate_stream_iter = stream_iter;
- }
+ BT_ASSERT(stream_iter != youngest_candidate_stream_iter);
- if (stream_iter_is_ended) {
+ if (!stream_iter_is_ended) {
+ if (G_UNLIKELY(youngest_candidate_stream_iter == NULL) ||
+ stream_iter->current_msg_ts_ns < youngest_candidate_msg_ts) {
+ /*
+ * Update the current best candidate message
+ * for the stream iterator of this live trace
+ * to be forwarded downstream.
+ */
+ youngest_candidate_msg_ts = stream_iter->current_msg_ts_ns;
+ youngest_candidate_stream_iter = stream_iter;
+ } else if (stream_iter->current_msg_ts_ns == youngest_candidate_msg_ts) {
+ /*
+ * Order the messages in an arbitrary but
+ * deterministic way.
+ */
+ BT_ASSERT(stream_iter != youngest_candidate_stream_iter);
+ int ret = common_muxing_compare_messages(
+ stream_iter->current_msg,
+ youngest_candidate_stream_iter->current_msg);
+ if (ret < 0) {
+ /*
+ * The `youngest_candidate_stream_iter->current_msg`
+ * should go first. Update the next
+ * iterator and the current timestamp.
+ */
+ youngest_candidate_msg_ts = stream_iter->current_msg_ts_ns;
+ youngest_candidate_stream_iter = stream_iter;
+ } else if (ret == 0) {
+ /*
+ * Unable to pick which one should go
+ * first.
+ */
+ BT_COMP_LOGW("Cannot deterministically pick next live stream message iterator because they have identical next messages: "
+ "stream-iter-addr=%p"
+ "stream-iter-addr=%p",
+ stream_iter,
+ youngest_candidate_stream_iter);
+ }
+ }
+
+ stream_iter_idx++;
+ } else {
/*
* The live stream iterator is ENDed. We remove that
* iterator from the list and we restart the iteration
g_ptr_array_remove_index_fast(live_trace->stream_iterators,
stream_iter_idx);
stream_iter_idx = 0;
- } else {
- stream_iter_idx++;
}
}
- if (curr_candidate_stream_iter) {
- *candidate_stream_iter = curr_candidate_stream_iter;
+ if (youngest_candidate_stream_iter) {
+ *youngest_trace_stream_iter = youngest_candidate_stream_iter;
stream_iter_status = LTTNG_LIVE_ITERATOR_STATUS_OK;
} else {
/*
enum lttng_live_iterator_status next_stream_iterator_for_session(
struct lttng_live_msg_iter *lttng_live_msg_iter,
struct lttng_live_session *session,
- struct lttng_live_stream_iterator **candidate_session_stream_iter)
+ struct lttng_live_stream_iterator **youngest_session_stream_iter)
{
+ bt_self_component *self_comp = lttng_live_msg_iter->self_comp;
+ bt_logging_level log_level = lttng_live_msg_iter->log_level;
enum lttng_live_iterator_status stream_iter_status;
uint64_t trace_idx = 0;
- int64_t curr_candidate_msg_ts = INT64_MAX;
- struct lttng_live_stream_iterator *curr_candidate_stream_iter = NULL;
+ int64_t youngest_candidate_msg_ts = INT64_MAX;
+ struct lttng_live_stream_iterator *youngest_candidate_stream_iter = NULL;
/*
* Make sure we are attached to the session and look for new streams
if (!trace_is_ended) {
BT_ASSERT(stream_iter);
- if (stream_iter->current_msg_ts_ns <= curr_candidate_msg_ts) {
- curr_candidate_msg_ts = stream_iter->current_msg_ts_ns;
- curr_candidate_stream_iter = stream_iter;
+ if (G_UNLIKELY(youngest_candidate_stream_iter == NULL) ||
+ stream_iter->current_msg_ts_ns < youngest_candidate_msg_ts) {
+ youngest_candidate_msg_ts = stream_iter->current_msg_ts_ns;
+ youngest_candidate_stream_iter = stream_iter;
+ } else if (stream_iter->current_msg_ts_ns == youngest_candidate_msg_ts) {
+ /*
+ * Order the messages in an arbitrary but
+ * deterministic way.
+ */
+ int ret = common_muxing_compare_messages(
+ stream_iter->current_msg,
+ youngest_candidate_stream_iter->current_msg);
+ if (ret < 0) {
+ /*
+ * The `youngest_candidate_stream_iter->current_msg`
+ * should go first. Update the next iterator
+ * and the current timestamp.
+ */
+ youngest_candidate_msg_ts = stream_iter->current_msg_ts_ns;
+ youngest_candidate_stream_iter = stream_iter;
+ } else if (ret == 0) {
+ /* Unable to pick which one should go first. */
+ BT_COMP_LOGW("Cannot deterministically pick next live stream message iterator because they have identical next messages: "
+ "stream-iter-addr=%p" "stream-iter-addr=%p",
+ stream_iter, youngest_candidate_stream_iter);
+ }
}
trace_idx++;
} else {
trace_idx = 0;
}
}
- if (curr_candidate_stream_iter) {
- *candidate_session_stream_iter = curr_candidate_stream_iter;
+ if (youngest_candidate_stream_iter) {
+ *youngest_session_stream_iter = youngest_candidate_stream_iter;
stream_iter_status = LTTNG_LIVE_ITERATOR_STATUS_OK;
} else {
/*
}
BT_HIDDEN
-bt_self_message_iterator_status lttng_live_msg_iter_next(
+bt_component_class_message_iterator_next_method_status lttng_live_msg_iter_next(
bt_self_message_iterator *self_msg_it,
bt_message_array_const msgs, uint64_t capacity,
uint64_t *count)
{
- bt_self_message_iterator_status status;
+ bt_component_class_message_iterator_next_method_status status;
struct lttng_live_msg_iter *lttng_live_msg_iter =
bt_self_message_iterator_get_data(self_msg_it);
struct lttng_live_component *lttng_live =
lttng_live_msg_iter->lttng_live_comp;
+ bt_self_component *self_comp = lttng_live_msg_iter->self_comp;
+ bt_logging_level log_level = lttng_live_msg_iter->log_level;
enum lttng_live_iterator_status stream_iter_status;
uint64_t session_idx;
if (lttng_live_msg_iter->sessions->len == 0) {
if (lttng_live->params.sess_not_found_act !=
SESSION_NOT_FOUND_ACTION_CONTINUE) {
- status = BT_SELF_MESSAGE_ITERATOR_STATUS_END;
+ status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_END;
goto no_session;
} else {
/*
* requested session name.
*/
if (lttng_live_create_viewer_session(lttng_live_msg_iter)) {
- status = BT_SELF_MESSAGE_ITERATOR_STATUS_ERROR;
+ status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_ERROR;
goto no_session;
}
}
* return it.
*/
while (*count < capacity) {
- struct lttng_live_stream_iterator *next_stream_iter = NULL,
- *candidate_stream_iter = NULL;
- int64_t next_msg_ts_ns = INT64_MAX;
+ struct lttng_live_stream_iterator *youngest_stream_iter = NULL,
+ *candidate_stream_iter = NULL;
+ int64_t youngest_msg_ts_ns = INT64_MAX;
BT_ASSERT(lttng_live_msg_iter->sessions);
session_idx = 0;
goto end;
}
- if (candidate_stream_iter->current_msg_ts_ns <= next_msg_ts_ns) {
- next_msg_ts_ns = candidate_stream_iter->current_msg_ts_ns;
- next_stream_iter = candidate_stream_iter;
+ if (G_UNLIKELY(youngest_stream_iter == NULL) ||
+ candidate_stream_iter->current_msg_ts_ns < youngest_msg_ts_ns) {
+ youngest_msg_ts_ns = candidate_stream_iter->current_msg_ts_ns;
+ youngest_stream_iter = candidate_stream_iter;
+ } else if (candidate_stream_iter->current_msg_ts_ns == youngest_msg_ts_ns) {
+ /*
+ * The currently selected message to be sent
+ * downstream next has the exact same timestamp
+ * that of the current candidate message. We
+ * must break the tie in a predictable manner.
+ */
+ BT_COMP_LOGD_STR("Two of the next message candidates have the same timestamps, pick one deterministically.");
+ /*
+ * Order the messages in an arbitrary but
+ * deterministic way.
+ */
+ int ret = common_muxing_compare_messages(
+ candidate_stream_iter->current_msg,
+ youngest_stream_iter->current_msg);
+ if (ret < 0) {
+ /*
+ * The `candidate_stream_iter->current_msg`
+ * should go first. Update the next
+ * iterator and the current timestamp.
+ */
+ youngest_msg_ts_ns = candidate_stream_iter->current_msg_ts_ns;
+ youngest_stream_iter = candidate_stream_iter;
+ } else if (ret == 0) {
+ /* Unable to pick which one should go first. */
+ BT_COMP_LOGW("Cannot deterministically pick next live stream message iterator because they have identical next messages: "
+ "next-stream-iter-addr=%p" "candidate-stream-iter-addr=%p",
+ youngest_stream_iter, candidate_stream_iter);
+ }
}
session_idx++;
}
- if (!next_stream_iter) {
+ if (!youngest_stream_iter) {
stream_iter_status = LTTNG_LIVE_ITERATOR_STATUS_AGAIN;
goto end;
}
- BT_ASSERT(next_stream_iter->current_msg);
+ BT_ASSERT(youngest_stream_iter->current_msg);
/* Ensure monotonicity. */
BT_ASSERT(lttng_live_msg_iter->last_msg_ts_ns <=
- next_stream_iter->current_msg_ts_ns);
+ youngest_stream_iter->current_msg_ts_ns);
/*
* Insert the next message to the message batch. This will set
* stream iterator current messsage to NULL so that next time
* we fetch the next message of that stream iterator
*/
- BT_MESSAGE_MOVE_REF(msgs[*count], next_stream_iter->current_msg);
+ BT_MESSAGE_MOVE_REF(msgs[*count], youngest_stream_iter->current_msg);
(*count)++;
/* Update the last timestamp in nanoseconds sent downstream. */
- lttng_live_msg_iter->last_msg_ts_ns = next_msg_ts_ns;
- next_stream_iter->current_msg_ts_ns = INT64_MAX;
+ lttng_live_msg_iter->last_msg_ts_ns = youngest_msg_ts_ns;
+ youngest_stream_iter->current_msg_ts_ns = INT64_MAX;
stream_iter_status = LTTNG_LIVE_ITERATOR_STATUS_OK;
}
* now. On the next call we return again if there are
* still no new message to send.
*/
- status = BT_SELF_MESSAGE_ITERATOR_STATUS_OK;
+ status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK;
} else {
- status = BT_SELF_MESSAGE_ITERATOR_STATUS_AGAIN;
+ status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_AGAIN;
}
break;
case LTTNG_LIVE_ITERATOR_STATUS_END:
- status = BT_SELF_MESSAGE_ITERATOR_STATUS_END;
+ status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_END;
break;
case LTTNG_LIVE_ITERATOR_STATUS_NOMEM:
- status = BT_SELF_MESSAGE_ITERATOR_STATUS_NOMEM;
+ status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_MEMORY_ERROR;
break;
case LTTNG_LIVE_ITERATOR_STATUS_ERROR:
case LTTNG_LIVE_ITERATOR_STATUS_INVAL:
case LTTNG_LIVE_ITERATOR_STATUS_UNSUPPORTED:
- status = BT_SELF_MESSAGE_ITERATOR_STATUS_ERROR;
+ status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_ERROR;
/* Put all existing messages on error. */
put_messages(msgs, *count);
break;
}
BT_HIDDEN
-bt_self_message_iterator_status lttng_live_msg_iter_init(
+bt_component_class_message_iterator_init_method_status lttng_live_msg_iter_init(
bt_self_message_iterator *self_msg_it,
+ bt_self_message_iterator_configuration *config,
bt_self_component_source *self_comp_src,
bt_self_component_port_output *self_port)
{
- bt_self_message_iterator_status ret =
- BT_SELF_MESSAGE_ITERATOR_STATUS_OK;
+ bt_component_class_message_iterator_init_method_status ret =
+ BT_COMPONENT_CLASS_MESSAGE_ITERATOR_INIT_METHOD_STATUS_OK;
bt_self_component *self_comp =
- bt_self_component_source_as_self_component(self_comp_src);
+ bt_self_component_source_as_self_component(self_comp_src);
struct lttng_live_component *lttng_live;
struct lttng_live_msg_iter *lttng_live_msg_iter;
+ bt_logging_level log_level;
BT_ASSERT(self_msg_it);
lttng_live = bt_self_component_get_data(self_comp);
+ log_level = lttng_live->log_level;
+ self_comp = lttng_live->self_comp;
/* There can be only one downstream iterator at the same time. */
BT_ASSERT(!lttng_live->has_msg_iter);
lttng_live_msg_iter = g_new0(struct lttng_live_msg_iter, 1);
if (!lttng_live_msg_iter) {
- ret = BT_SELF_MESSAGE_ITERATOR_STATUS_ERROR;
+ ret = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_INIT_METHOD_STATUS_ERROR;
goto end;
}
+ lttng_live_msg_iter->log_level = lttng_live->log_level;
+ lttng_live_msg_iter->self_comp = lttng_live->self_comp;
lttng_live_msg_iter->lttng_live_comp = lttng_live;
lttng_live_msg_iter->self_msg_iter = self_msg_it;
lttng_live_msg_iter->viewer_connection =
live_viewer_connection_create(lttng_live->params.url->str, false,
- lttng_live_msg_iter);
+ lttng_live_msg_iter, log_level);
if (!lttng_live_msg_iter->viewer_connection) {
goto error;
}
if (lttng_live_msg_iter->sessions->len == 0) {
switch (lttng_live->params.sess_not_found_act) {
case SESSION_NOT_FOUND_ACTION_CONTINUE:
- BT_LOGI("Unable to connect to the requested live viewer "
+ BT_COMP_LOGI("Unable to connect to the requested live viewer "
"session. Keep trying to connect because of "
"%s=\"%s\" component parameter: url=\"%s\"",
SESS_NOT_FOUND_ACTION_PARAM,
lttng_live->params.url->str);
break;
case SESSION_NOT_FOUND_ACTION_FAIL:
- BT_LOGE("Unable to connect to the requested live viewer "
+ BT_COMP_LOGE("Unable to connect to the requested live viewer "
"session. Fail the message iterator"
"initialization because of %s=\"%s\" "
"component parameter: url =\"%s\"",
lttng_live->params.url->str);
goto error;
case SESSION_NOT_FOUND_ACTION_END:
- BT_LOGI("Unable to connect to the requested live viewer "
+ BT_COMP_LOGI("Unable to connect to the requested live viewer "
"session. End gracefully at the first _next() "
"call because of %s=\"%s\" component parameter: "
"url=\"%s\"", SESS_NOT_FOUND_ACTION_PARAM,
SESS_NOT_FOUND_ACTION_END_STR,
lttng_live->params.url->str);
break;
+ case SESSION_NOT_FOUND_ACTION_UNKNOWN:
+ default:
+ /* Fallthrough */
+ BT_COMP_LOGE("Unknown action for session not found"
+ "error. Fail the message iterator"
+ "initialization because of %s=\"%s\" "
+ "component parameter: url =\"%s\"",
+ SESS_NOT_FOUND_ACTION_PARAM,
+ SESS_NOT_FOUND_ACTION_FAIL_STR,
+ lttng_live->params.url->str);
+ break;
}
}
goto end;
error:
- ret = BT_SELF_MESSAGE_ITERATOR_STATUS_ERROR;
+ ret = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_INIT_METHOD_STATUS_ERROR;
lttng_live_msg_iter_destroy(lttng_live_msg_iter);
end:
return ret;
}
static
-bt_query_status lttng_live_query_list_sessions(const bt_value *params,
- const bt_value **result)
+bt_component_class_query_method_status lttng_live_query_list_sessions(
+ const bt_value *params, const bt_value **result,
+ bt_logging_level log_level)
{
- bt_query_status status = BT_QUERY_STATUS_OK;
+ bt_component_class_query_method_status status =
+ BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_OK;
const bt_value *url_value = NULL;
const char *url;
struct live_viewer_connection *viewer_connection = NULL;
+ bt_self_component *self_comp = NULL;
url_value = bt_value_map_borrow_entry_value_const(params, URL_PARAM);
if (!url_value) {
- BT_LOGW("Mandatory `%s` parameter missing", URL_PARAM);
- status = BT_QUERY_STATUS_INVALID_PARAMS;
+ BT_COMP_LOGE("Mandatory `%s` parameter missing", URL_PARAM);
+ status = BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_ERROR;
goto error;
}
if (!bt_value_is_string(url_value)) {
- BT_LOGW("`%s` parameter is required to be a string value",
+ BT_COMP_LOGE("`%s` parameter is required to be a string value",
URL_PARAM);
- status = BT_QUERY_STATUS_INVALID_PARAMS;
+ status = BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_ERROR;
goto error;
}
url = bt_value_string_get(url_value);
- viewer_connection = live_viewer_connection_create(url, true, NULL);
+ viewer_connection = live_viewer_connection_create(url, true, NULL,
+ log_level);
if (!viewer_connection) {
goto error;
}
status = live_viewer_connection_list_sessions(viewer_connection,
result);
- if (status != BT_QUERY_STATUS_OK) {
+ if (status != BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_OK) {
goto error;
}
BT_VALUE_PUT_REF_AND_RESET(*result);
if (status >= 0) {
- status = BT_QUERY_STATUS_ERROR;
+ status = BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_ERROR;
}
end:
return status;
}
+static
+bt_component_class_query_method_status lttng_live_query_support_info(
+ const bt_value *params, const bt_value **result,
+ bt_logging_level log_level)
+{
+ bt_component_class_query_method_status status =
+ BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_OK;
+ const bt_value *input_type_value;
+ const bt_value *input_value;
+ double weight = 0;
+ struct bt_common_lttng_live_url_parts parts = { 0 };
+
+ /* Used by the logging macros */
+ __attribute__((unused)) bt_self_component *self_comp = NULL;
+
+ *result = NULL;
+ input_type_value = bt_value_map_borrow_entry_value_const(params,
+ "type");
+ if (!input_type_value) {
+ BT_COMP_LOGE("Missing expected `type` parameter.");
+ goto error;
+ }
+
+ if (!bt_value_is_string(input_type_value)) {
+ BT_COMP_LOGE("`type` parameter is not a string value.");
+ goto error;
+ }
+
+ if (strcmp(bt_value_string_get(input_type_value), "string") != 0) {
+ /* We don't handle file system paths */
+ goto create_result;
+ }
+
+ input_value = bt_value_map_borrow_entry_value_const(params, "input");
+ if (!input_value) {
+ BT_COMP_LOGE("Missing expected `input` parameter.");
+ goto error;
+ }
+
+ if (!bt_value_is_string(input_value)) {
+ BT_COMP_LOGE("`input` parameter is not a string value.");
+ goto error;
+ }
+
+ parts = bt_common_parse_lttng_live_url(bt_value_string_get(input_value),
+ NULL, 0);
+ if (parts.session_name) {
+ /*
+ * Looks pretty much like an LTTng live URL: we got the
+ * session name part, which forms a complete URL.
+ */
+ weight = .75;
+ }
+
+create_result:
+ *result = bt_value_real_create_init(weight);
+ if (!*result) {
+ status = BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_MEMORY_ERROR;
+ goto error;
+ }
+
+ goto end;
+
+error:
+ if (status >= 0) {
+ status = BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_ERROR;
+ }
+
+ BT_ASSERT(!*result);
+
+end:
+ bt_common_destroy_lttng_live_url_parts(&parts);
+ return status;
+}
+
BT_HIDDEN
-bt_query_status lttng_live_query(bt_self_component_class_source *comp_class,
- const bt_query_executor *query_exec,
+bt_component_class_query_method_status lttng_live_query(
+ bt_self_component_class_source *comp_class,
+ bt_private_query_executor *priv_query_exec,
const char *object, const bt_value *params,
- __attribute__((unused)) bt_logging_level log_level,
+ __attribute__((unused)) void *method_data,
const bt_value **result)
{
- bt_query_status status = BT_QUERY_STATUS_OK;
+ bt_component_class_query_method_status status =
+ BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_OK;
+ bt_self_component *self_comp = NULL;
+ bt_logging_level log_level = bt_query_executor_get_logging_level(
+ bt_private_query_executor_as_query_executor_const(
+ priv_query_exec));
if (strcmp(object, "sessions") == 0) {
- status = lttng_live_query_list_sessions(params, result);
+ status = lttng_live_query_list_sessions(params, result,
+ log_level);
+ } else if (strcmp(object, "babeltrace.support-info") == 0) {
+ status = lttng_live_query_support_info(params, result,
+ log_level);
} else {
- BT_LOGW("Unknown query object `%s`", object);
- status = BT_QUERY_STATUS_INVALID_OBJECT;
+ BT_COMP_LOGI("Unknown query object `%s`", object);
+ status = BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_UNKNOWN_OBJECT;
goto end;
}
} else if (strcmp(no_session_act_str, SESS_NOT_FOUND_ACTION_END_STR) == 0) {
action = SESSION_NOT_FOUND_ACTION_END;
} else {
- action = -1;
+ action = SESSION_NOT_FOUND_ACTION_UNKNOWN;
}
return action;
}
-struct lttng_live_component *lttng_live_component_create(const bt_value *params)
+struct lttng_live_component *lttng_live_component_create(const bt_value *params,
+ bt_logging_level log_level, bt_self_component *self_comp)
{
struct lttng_live_component *lttng_live;
- const bt_value *value = NULL;
+ const bt_value *inputs_value;
+ const bt_value *url_value;
+ const bt_value *value;
const char *url;
lttng_live = g_new0(struct lttng_live_component, 1);
if (!lttng_live) {
goto end;
}
+ lttng_live->log_level = log_level;
+ lttng_live->self_comp = self_comp;
lttng_live->max_query_size = MAX_QUERY_SIZE;
lttng_live->has_msg_iter = false;
- value = bt_value_map_borrow_entry_value_const(params, URL_PARAM);
- if (!value || !bt_value_is_string(value)) {
- BT_LOGW("Mandatory `%s` parameter missing or not a string",
- URL_PARAM);
+ inputs_value = bt_value_map_borrow_entry_value_const(params,
+ INPUTS_PARAM);
+ if (!inputs_value) {
+ BT_COMP_LOGE("Mandatory `%s` parameter missing", INPUTS_PARAM);
goto error;
}
- url = bt_value_string_get(value);
+
+ if (!bt_value_is_array(inputs_value)) {
+ BT_COMP_LOGE("`%s` parameter is required to be an array value",
+ INPUTS_PARAM);
+ goto error;
+ }
+
+ if (bt_value_array_get_length(inputs_value) != 1) {
+ BT_COMP_LOGE("`%s` parameter's length is required to be 1",
+ INPUTS_PARAM);
+ goto error;
+ }
+
+ url_value = bt_value_array_borrow_element_by_index_const(inputs_value,
+ 0);
+ BT_ASSERT(url_value);
+
+ if (!bt_value_is_string(url_value)) {
+ BT_COMP_LOGE("First element of `%s` parameter is required to be a string value (URL)",
+ INPUTS_PARAM);
+ goto error;
+ }
+
+ url = bt_value_string_get(url_value);
lttng_live->params.url = g_string_new(url);
if (!lttng_live->params.url) {
goto error;
value = bt_value_map_borrow_entry_value_const(params,
SESS_NOT_FOUND_ACTION_PARAM);
+ if (value) {
+ if (!bt_value_is_string(value)) {
+ BT_COMP_LOGE("`%s` parameter is required to be a string value",
+ SESS_NOT_FOUND_ACTION_PARAM);
+ goto error;
+ }
- if (value && bt_value_is_string(value)) {
lttng_live->params.sess_not_found_act =
parse_session_not_found_action_param(value);
- if (lttng_live->params.sess_not_found_act == -1) {
- BT_LOGE("Unexpected value for `%s` parameter: "
+ if (lttng_live->params.sess_not_found_act == SESSION_NOT_FOUND_ACTION_UNKNOWN) {
+ BT_COMP_LOGE("Unexpected value for `%s` parameter: "
"value=\"%s\"", SESS_NOT_FOUND_ACTION_PARAM,
bt_value_string_get(value));
goto error;
}
} else {
- BT_LOGW("Optional `%s` parameter is missing or "
- "not a string value. Defaulting to %s=\"%s\".",
- SESS_NOT_FOUND_ACTION_PARAM,
+ BT_COMP_LOGI("Optional `%s` parameter is missing: "
+ "defaulting to `%s`.",
SESS_NOT_FOUND_ACTION_PARAM,
SESS_NOT_FOUND_ACTION_CONTINUE_STR);
lttng_live->params.sess_not_found_act =
}
BT_HIDDEN
-bt_self_component_status lttng_live_component_init(
- bt_self_component_source *self_comp,
- const bt_value *params, __attribute__((unused)) void *init_method_data)
+bt_component_class_init_method_status lttng_live_component_init(
+ bt_self_component_source *self_comp_src,
+ bt_self_component_source_configuration *config,
+ const bt_value *params,
+ __attribute__((unused)) void *init_method_data)
{
struct lttng_live_component *lttng_live;
- bt_self_component_status ret = BT_SELF_COMPONENT_STATUS_OK;
+ bt_component_class_init_method_status ret =
+ BT_COMPONENT_CLASS_INIT_METHOD_STATUS_OK;
+ bt_self_component *self_comp =
+ bt_self_component_source_as_self_component(self_comp_src);
+ bt_logging_level log_level = bt_component_get_logging_level(
+ bt_self_component_as_component(self_comp));
+ bt_self_component_add_port_status add_port_status;
- lttng_live = lttng_live_component_create(params);
+ lttng_live = lttng_live_component_create(params, log_level, self_comp);
if (!lttng_live) {
- ret = BT_SELF_COMPONENT_STATUS_NOMEM;
+ ret = BT_COMPONENT_CLASS_INIT_METHOD_STATUS_MEMORY_ERROR;
goto error;
}
- lttng_live->self_comp = self_comp;
- if (lttng_live_graph_is_canceled(lttng_live)) {
- ret = BT_SELF_COMPONENT_STATUS_END;
+ add_port_status = bt_self_component_source_add_output_port(
+ self_comp_src, "out", NULL, NULL);
+ switch (add_port_status) {
+ case BT_SELF_COMPONENT_ADD_PORT_STATUS_ERROR:
+ ret = BT_COMPONENT_CLASS_INIT_METHOD_STATUS_ERROR;
goto error;
- }
-
- ret = bt_self_component_source_add_output_port(
- lttng_live->self_comp, "out",
- NULL, NULL);
- if (ret != BT_SELF_COMPONENT_STATUS_OK) {
+ case BT_SELF_COMPONENT_ADD_PORT_STATUS_MEMORY_ERROR:
+ ret = BT_COMPONENT_CLASS_INIT_METHOD_STATUS_MEMORY_ERROR;
goto error;
+ default:
+ break;
}
- bt_self_component_set_data(
- bt_self_component_source_as_self_component(self_comp),
- lttng_live);
+ bt_self_component_set_data(self_comp, lttng_live);
goto end;
error:
projects / babeltrace.git / blobdiff