#define print_dbg(fmt, ...) BT_COMP_LOGD(fmt, ## __VA_ARGS__)
static
-const char *print_live_iterator_status(enum lttng_live_iterator_status status)
+const char *lttng_live_iterator_status_string(
+ enum lttng_live_iterator_status status)
{
switch (status) {
case LTTNG_LIVE_ITERATOR_STATUS_CONTINUE:
}
static
-const char *print_state(struct lttng_live_stream_iterator *s)
+const char *lttng_live_stream_state_string(enum lttng_live_stream_state state)
{
- switch (s->state) {
+ switch (state) {
case LTTNG_LIVE_STREAM_ACTIVE_NO_DATA:
return "ACTIVE_NO_DATA";
case LTTNG_LIVE_STREAM_QUIESCENT_NO_DATA:
}
}
-#define print_stream_state(live_stream_iter) \
+#define LTTNG_LIVE_LOGD_STREAM_ITER(live_stream_iter) \
do { \
- BT_COMP_LOGD("stream state %s last_inact_ts %" PRId64 \
- ", curr_inact_ts %" PRId64, \
- print_state(live_stream_iter), \
+ BT_COMP_LOGD("Live stream iterator state=%s, last-inact-ts=%" PRId64 \
+ ", curr-inact-ts %" PRId64, \
+ lttng_live_stream_state_string(live_stream_iter->state), \
live_stream_iter->last_inactivity_ts, \
live_stream_iter->current_inactivity_ts); \
} while (0);
}
static
-struct lttng_live_trace *lttng_live_find_trace(struct lttng_live_session *session,
+struct lttng_live_trace *lttng_live_session_borrow_trace_by_id(struct lttng_live_session *session,
uint64_t trace_id)
{
uint64_t trace_idx;
bt_logging_level log_level = trace->log_level;
bt_self_component *self_comp = trace->self_comp;
- BT_COMP_LOGD("Destroy lttng_live_trace");
+ BT_COMP_LOGD("Destroying live trace: trace-id=%"PRIu64, trace->id);
BT_ASSERT(trace->stream_iterators);
g_ptr_array_free(trace->stream_iterators, TRUE);
bt_logging_level log_level = session->log_level;
bt_self_component *self_comp = session->self_comp;
+ BT_COMP_LOGD("Creating live trace: "
+ "session-id=%"PRIu64", trace-id=%"PRIu64,
+ session->id, trace_id);
trace = g_new0(struct lttng_live_trace, 1);
if (!trace) {
BT_COMP_LOGE_APPEND_CAUSE(self_comp,
trace->stream_iterators = g_ptr_array_new_with_free_func(
(GDestroyNotify) lttng_live_stream_iterator_destroy);
BT_ASSERT(trace->stream_iterators);
- trace->new_metadata_needed = true;
+ trace->metadata_stream_state = LTTNG_LIVE_METADATA_STREAM_STATE_NEEDED;
g_ptr_array_add(session->traces, trace);
- BT_COMP_LOGI("Create trace");
goto end;
error:
g_free(trace);
}
BT_HIDDEN
-struct lttng_live_trace *lttng_live_borrow_trace(
+struct lttng_live_trace *lttng_live_session_borrow_or_create_trace_by_id(
struct lttng_live_session *session, uint64_t trace_id)
{
struct lttng_live_trace *trace;
- trace = lttng_live_find_trace(session, trace_id);
+ trace = lttng_live_session_borrow_trace_by_id(session, trace_id);
if (trace) {
goto end;
}
bt_logging_level log_level = lttng_live_msg_iter->log_level;
bt_self_component *self_comp = lttng_live_msg_iter->self_comp;
+ BT_COMP_LOGD("Adding live session: "
+ "session-id=%" PRIu64 ", hostname=\"%s\" session-name=\"%s\"",
+ session_id, hostname, session_name);
+
session = g_new0(struct lttng_live_session, 1);
if (!session) {
BT_COMP_LOGE_APPEND_CAUSE(self_comp,
session->session_name = g_string_new(session_name);
BT_ASSERT(session->session_name);
- 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:
log_level = session->log_level;
self_comp = session->self_comp;
- BT_COMP_LOGD("Destroy lttng live session");
+ BT_COMP_LOGD("Destroying live session: "
+ "session-id=%"PRIu64", session-name=\"%s\"",
+ session->id, session->session_name->str);
if (session->id != -1ULL) {
- if (lttng_live_detach_session(session)) {
+ if (lttng_live_session_detach(session)) {
if (!lttng_live_graph_is_canceled(
session->lttng_live_msg_iter)) {
/* Old relayd cannot detach sessions. */
enum lttng_live_stream_state orig_state = lttng_live_stream->state;
struct packet_index index;
- if (lttng_live_stream->trace->new_metadata_needed) {
+ if (lttng_live_stream->trace->metadata_stream_state ==
+ LTTNG_LIVE_METADATA_STREAM_STATE_NEEDED) {
ret = LTTNG_LIVE_ITERATOR_STATUS_CONTINUE;
goto end;
}
if (orig_state == LTTNG_LIVE_STREAM_QUIESCENT_NO_DATA &&
last_inact_ts == curr_inact_ts) {
ret = LTTNG_LIVE_ITERATOR_STATUS_AGAIN;
- print_stream_state(lttng_live_stream);
+ LTTNG_LIVE_LOGD_STREAM_ITER(lttng_live_stream);
} else {
ret = LTTNG_LIVE_ITERATOR_STATUS_CONTINUE;
}
enum lttng_live_iterator_status status;
uint64_t trace_idx;
+ BT_COMP_LOGD("Updating all streams for session: "
+ "session-id=%"PRIu64", session-name=\"%s\"",
+ session->id, session->session_name->str);
+
if (!session->attached) {
enum lttng_live_viewer_status attach_status =
- lttng_live_attach_session(session,
+ lttng_live_session_attach(session,
lttng_live_msg_iter->self_msg_iter);
if (attach_status != LTTNG_LIVE_VIEWER_STATUS_OK) {
if (lttng_live_graph_is_canceled(lttng_live_msg_iter)) {
}
}
- status = lttng_live_get_new_streams(session,
+ status = lttng_live_session_get_new_streams(session,
lttng_live_msg_iter->self_msg_iter);
if (status != LTTNG_LIVE_ITERATOR_STATUS_OK &&
status != LTTNG_LIVE_ITERATOR_STATUS_END) {
status = lttng_live_metadata_update(trace);
switch (status) {
+ case LTTNG_LIVE_ITERATOR_STATUS_END:
case LTTNG_LIVE_ITERATOR_STATUS_OK:
trace_idx++;
break;
- case LTTNG_LIVE_ITERATOR_STATUS_END:
- /*
- * The trace has ended. Remove it of the array an
- * continue the iteration.
- * We can remove the trace safely when using the
- * g_ptr_array_remove_index_fast because it replaces
- * the element at trace_idx with the array's last
- * element. trace_idx is not incremented because of
- * that.
- */
- (void) g_ptr_array_remove_index_fast(session->traces,
- trace_idx);
- break;
+ case LTTNG_LIVE_ITERATOR_STATUS_CONTINUE:
+ case LTTNG_LIVE_ITERATOR_STATUS_AGAIN:
+ goto end;
default:
+ BT_COMP_LOGE_APPEND_CAUSE(self_comp,
+ "Error updating trace metadata: "
+ "stream-iter-status=%s, trace-id=%"PRIu64,
+ lttng_live_iterator_status_string(status),
+ trace->id);
goto end;
}
}
return status;
}
-BT_HIDDEN
-void lttng_live_need_new_streams(struct lttng_live_msg_iter *lttng_live_msg_iter)
-{
- uint64_t session_idx;
-
- for (session_idx = 0; session_idx < lttng_live_msg_iter->sessions->len;
- session_idx++) {
- struct lttng_live_session *session =
- g_ptr_array_index(lttng_live_msg_iter->sessions, session_idx);
- session->new_streams_needed = true;
- }
-}
-
static
void lttng_live_force_new_streams_and_metadata(struct lttng_live_msg_iter *lttng_live_msg_iter)
{
trace_idx++) {
struct lttng_live_trace *trace =
g_ptr_array_index(session->traces, trace_idx);
- trace->new_metadata_needed = true;
+
+ BT_ASSERT(trace->metadata_stream_state !=
+ LTTNG_LIVE_METADATA_STREAM_STATE_CLOSED);
+
+ trace->metadata_stream_state = LTTNG_LIVE_METADATA_STREAM_STATE_NEEDED;
}
}
}
enum session_not_found_action sess_not_found_act =
lttng_live_msg_iter->lttng_live_comp->params.sess_not_found_act;
+ BT_COMP_LOGD("Update data and metadata of all sessions"
+ "live-msg-iter-addr=%p", lttng_live_msg_iter);
/*
* In a remotely distant future, we could add a "new
* session" flag to the protocol, which would tell us that we
*/
if (lttng_live_msg_iter->sessions->len == 0) {
if (sess_not_found_act != SESSION_NOT_FOUND_ACTION_CONTINUE) {
+ BT_COMP_LOGD("No session found. Exiting in accordance with the `session-not-found-action` parameter");
status = LTTNG_LIVE_ITERATOR_STATUS_END;
goto end;
} else {
+ BT_COMP_LOGD("No session found. Try creating a new one in accordance with the `session-not-found-action` parameter");
/*
* Retry to create a viewer session for the requested
* session name.
trace_idx++) {
struct lttng_live_trace *trace =
g_ptr_array_index(session->traces, trace_idx);
- if (trace->new_metadata_needed) {
+ if (trace->metadata_stream_state == LTTNG_LIVE_METADATA_STREAM_STATE_NEEDED) {
ret = LTTNG_LIVE_ITERATOR_STATUS_CONTINUE;
goto end;
}
if (lttng_live_stream->state != LTTNG_LIVE_STREAM_ACTIVE_DATA) {
ret = LTTNG_LIVE_ITERATOR_STATUS_ERROR;
+ BT_COMP_LOGE_APPEND_CAUSE(self_comp,
+ "Invalid state of live stream iterator"
+ "stream-iter-status=%s",
+ lttng_live_stream_state_string(lttng_live_stream->state));
goto end;
}
default:
ret = LTTNG_LIVE_ITERATOR_STATUS_ERROR;
BT_COMP_LOGE_APPEND_CAUSE(self_comp,
- "CTF message iterator return an error or failed: "
- "msg_iter=%p", lttng_live_stream->msg_iter);
+ "CTF message iterator failed to get next message: "
+ "msg-iter=%p, msg-iter-status=%s",
+ lttng_live_stream->msg_iter,
+ ctf_msg_iter_status_string(status));
break;
}
bt_self_component *self_comp = lttng_live_msg_iter->self_comp;
enum lttng_live_iterator_status live_status;
+ BT_COMP_LOGD("Finding the next message for stream iterator: "
+ "stream-name=\"%s\"", stream_iter->name->str);
+
if (stream_iter->has_stream_hung_up) {
/*
* The stream has hung up and the stream was properly closed
}
retry:
- print_stream_state(stream_iter);
+ LTTNG_LIVE_LOGD_STREAM_ITER(stream_iter);
live_status = lttng_live_iterator_handle_new_streams_and_metadata(
lttng_live_msg_iter);
if (live_status != LTTNG_LIVE_ITERATOR_STATUS_OK) {
BT_ASSERT_DBG(live_trace);
BT_ASSERT_DBG(live_trace->stream_iterators);
+
+ BT_COMP_LOGD("Finding the next stream iterator for trace: "
+ "trace-id=%"PRIu64, live_trace->id);
/*
* Update the current message of every stream iterators of this trace.
* The current msg of every stream must have a timestamp equal or
lttng_live_msg_iter, stream_iter, &msg);
BT_COMP_LOGD("live stream iterator returned status :%s",
- print_live_iterator_status(stream_iter_status));
+ lttng_live_iterator_status_string(stream_iter_status));
if (stream_iter_status == LTTNG_LIVE_ITERATOR_STATUS_END) {
stream_iter_is_ended = true;
break;
int64_t youngest_candidate_msg_ts = INT64_MAX;
struct lttng_live_stream_iterator *youngest_candidate_stream_iter = NULL;
+ BT_COMP_LOGD("Finding the next stream iterator for session: "
+ "session-id=%"PRIu64, session->id);
/*
* Make sure we are attached to the session and look for new streams
* and metadata.
status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_END;
break;
case LTTNG_LIVE_ITERATOR_STATUS_NOMEM:
+ BT_COMP_LOGE_APPEND_CAUSE(self_comp,
+ "Memory error preparing the next batch of messages: "
+ "live-iter-status=%s",
+ lttng_live_iterator_status_string(stream_iter_status));
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:
+ BT_COMP_LOGE_APPEND_CAUSE(self_comp,
+ "Error preparing the next batch of messages: "
+ "live-iter-status=%s",
+ lttng_live_iterator_status_string(stream_iter_status));
+
status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_ERROR;
/* Put all existing messages on error. */
put_messages(msgs, *count);
return status;
}
+static
+struct lttng_live_msg_iter *lttng_live_msg_iter_create(
+ struct lttng_live_component *lttng_live_comp,
+ bt_self_message_iterator *self_msg_it)
+{
+ bt_self_component *self_comp = lttng_live_comp->self_comp;
+ bt_logging_level log_level = lttng_live_comp->log_level;
+
+ struct lttng_live_msg_iter *lttng_live_msg_iter =
+ g_new0(struct lttng_live_msg_iter, 1);
+ if (!lttng_live_msg_iter) {
+ BT_COMP_LOGE_APPEND_CAUSE(self_comp,
+ "Failed to allocate lttng_live_msg_iter");
+ goto end;
+ }
+
+ lttng_live_msg_iter->log_level = lttng_live_comp->log_level;
+ lttng_live_msg_iter->self_comp = lttng_live_comp->self_comp;
+ lttng_live_msg_iter->lttng_live_comp = lttng_live_comp;
+ lttng_live_msg_iter->self_msg_iter = self_msg_it;
+
+ lttng_live_msg_iter->active_stream_iter = 0;
+ lttng_live_msg_iter->last_msg_ts_ns = INT64_MIN;
+ lttng_live_msg_iter->was_interrupted = false;
+
+ lttng_live_msg_iter->sessions = g_ptr_array_new_with_free_func(
+ (GDestroyNotify) lttng_live_destroy_session);
+ BT_ASSERT(lttng_live_msg_iter->sessions);
+
+end:
+ return lttng_live_msg_iter;
+
+}
+
BT_HIDDEN
bt_component_class_message_iterator_initialize_method_status lttng_live_msg_iter_init(
bt_self_message_iterator *self_msg_it,
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->has_msg_iter = true;
- lttng_live_msg_iter = g_new0(struct lttng_live_msg_iter, 1);
+ lttng_live_msg_iter = lttng_live_msg_iter_create(lttng_live,
+ self_msg_it);
if (!lttng_live_msg_iter) {
status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_INITIALIZE_METHOD_STATUS_MEMORY_ERROR;
BT_COMP_LOGE_APPEND_CAUSE(self_comp,
- "Failed to allocate lttng_live_msg_iter");
- goto end;
+ "Failed to create lttng_live_msg_iter");
+ goto error;
}
- 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->active_stream_iter = 0;
- lttng_live_msg_iter->last_msg_ts_ns = INT64_MIN;
- lttng_live_msg_iter->was_interrupted = false;
-
- lttng_live_msg_iter->sessions = g_ptr_array_new_with_free_func(
- (GDestroyNotify) lttng_live_destroy_session);
- BT_ASSERT(lttng_live_msg_iter->sessions);
-
viewer_status = live_viewer_connection_create(self_comp, NULL,
log_level, lttng_live->params.url->str, false,
lttng_live_msg_iter, <tng_live_msg_iter->viewer_connection);