#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 "plugins/comp-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"
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()
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(
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;
goto end;
}
- if (candidate_stream_iter->current_msg_ts_ns <= next_msg_ts_ns) {
+ if (G_UNLIKELY(next_stream_iter == NULL) ||
+ 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;
+ } else if (candidate_stream_iter->current_msg_ts_ns == next_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
+ * determinitic way.
+ */
+ int ret = common_muxing_compare_messages(candidate_stream_iter->current_msg,
+ next_stream_iter->current_msg);
+ if (ret < 0) {
+ /*
+ * The `candidate_stream_iter->current_msg`
+ * should go first. Update the next
+ * iterator and the current timestamp.
+ */
+ next_msg_ts_ns = candidate_stream_iter->current_msg_ts_ns;
+ next_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",
+ next_stream_iter, candidate_stream_iter);
+ }
}
session_idx++;
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;
}
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;
}
}
url_value = bt_value_map_borrow_entry_value_const(params, URL_PARAM);
if (!url_value) {
- BT_COMP_LOGW("Mandatory `%s` parameter missing", URL_PARAM);
- status = BT_COMPONENT_CLASS_QUERY_METHOD_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_COMP_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_COMPONENT_CLASS_QUERY_METHOD_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;
}
BT_HIDDEN
bt_component_class_query_method_status lttng_live_query(
bt_self_component_class_source *comp_class,
- const bt_query_executor *query_exec,
+ bt_private_query_executor *priv_query_exec,
const char *object, const bt_value *params,
- bt_logging_level log_level, const bt_value **result)
+ __attribute__((unused)) void *method_data,
+ const bt_value **result)
{
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,
log_level);
} else {
BT_COMP_LOGI("Unknown query object `%s`", object);
- status = BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_INVALID_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;
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) {
+ 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));
projects / babeltrace.git / blobdiff