X-Git-Url: http://git.efficios.com/?a=blobdiff_plain;f=src%2Fplugins%2Fctf%2Flttng-live%2Flttng-live.c;h=bd2023d23f6c5816e6fc0863afa39969ce8419e7;hb=b607a4f79b18d92c2af08eb0df80b602fc7653e4;hp=af7237bdc2ed721c5d7d3f10e34678384ac735d8;hpb=0757916e5e17e126cd57be0f9d7810de37cc01c8;p=babeltrace.git diff --git a/src/plugins/ctf/lttng-live/lttng-live.c b/src/plugins/ctf/lttng-live/lttng-live.c index af7237bd..bd2023d2 100644 --- a/src/plugins/ctf/lttng-live/lttng-live.c +++ b/src/plugins/ctf/lttng-live/lttng-live.c @@ -33,10 +33,12 @@ #define BT_LOG_TAG "PLUGIN/SRC.CTF.LTTNG-LIVE" #include "logging/comp-logging.h" -#include #include +#include #include +#include + #include "common/assert.h" #include #include "compat/compiler.h" @@ -60,7 +62,8 @@ #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: @@ -80,7 +83,7 @@ const char *print_live_iterator_status(enum lttng_live_iterator_status status) case LTTNG_LIVE_ITERATOR_STATUS_UNSUPPORTED: return "LTTNG_LIVE_ITERATOR_STATUS_UNSUPPORTED"; default: - abort(); + bt_common_abort(); } } @@ -177,6 +180,8 @@ struct lttng_live_trace *lttng_live_create_trace(struct lttng_live_session *sess trace = g_new0(struct lttng_live_trace, 1); if (!trace) { + BT_COMP_LOGE_APPEND_CAUSE(self_comp, + "Failed to allocate live trace"); goto error; } trace->log_level = session->log_level; @@ -230,6 +235,8 @@ int lttng_live_add_session(struct lttng_live_msg_iter *lttng_live_msg_iter, session = g_new0(struct lttng_live_session, 1); if (!session) { + BT_COMP_LOGE_APPEND_CAUSE(self_comp, + "Failed to allocate live session"); goto error; } @@ -252,7 +259,6 @@ int lttng_live_add_session(struct lttng_live_msg_iter *lttng_live_msg_iter, g_ptr_array_add(lttng_live_msg_iter->sessions, session); goto end; error: - BT_COMP_LOGE("Error adding session"); g_free(session); ret = -1; end: @@ -311,7 +317,9 @@ void lttng_live_msg_iter_destroy(struct lttng_live_msg_iter *lttng_live_msg_iter g_ptr_array_free(lttng_live_msg_iter->sessions, TRUE); } - BT_OBJECT_PUT_REF_AND_RESET(lttng_live_msg_iter->viewer_connection); + if (lttng_live_msg_iter->viewer_connection) { + live_viewer_connection_destroy(lttng_live_msg_iter->viewer_connection); + } BT_ASSERT(lttng_live_msg_iter->lttng_live_comp); BT_ASSERT(lttng_live_msg_iter->lttng_live_comp->has_msg_iter); @@ -351,11 +359,11 @@ enum lttng_live_iterator_status lttng_live_iterator_next_check_stream_state( case LTTNG_LIVE_STREAM_ACTIVE_NO_DATA: /* Invalid state. */ BT_COMP_LOGF("Unexpected stream state \"ACTIVE_NO_DATA\""); - abort(); + bt_common_abort(); case LTTNG_LIVE_STREAM_QUIESCENT_NO_DATA: /* Invalid state. */ BT_COMP_LOGF("Unexpected stream state \"QUIESCENT_NO_DATA\""); - abort(); + bt_common_abort(); case LTTNG_LIVE_STREAM_EOF: break; } @@ -434,38 +442,69 @@ enum lttng_live_iterator_status lttng_live_get_session( struct lttng_live_msg_iter *lttng_live_msg_iter, 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; enum lttng_live_iterator_status status; uint64_t trace_idx; - int ret = 0; if (!session->attached) { - ret = lttng_live_attach_session(session); - if (ret) { + enum lttng_live_viewer_status attach_status = + lttng_live_attach_session(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)) { + /* + * Clear any causes appended in + * `lttng_live_attach_session()` as we want to + * return gracefully since the graph was + * cancelled. + */ + bt_current_thread_clear_error(); status = LTTNG_LIVE_ITERATOR_STATUS_AGAIN; } else { status = LTTNG_LIVE_ITERATOR_STATUS_ERROR; + BT_COMP_LOGE_APPEND_CAUSE(self_comp, + "Error attaching to LTTng live session"); } goto end; } } - status = lttng_live_get_new_streams(session); + status = lttng_live_get_new_streams(session, + lttng_live_msg_iter->self_msg_iter); if (status != LTTNG_LIVE_ITERATOR_STATUS_OK && status != LTTNG_LIVE_ITERATOR_STATUS_END) { goto end; } - for (trace_idx = 0; trace_idx < session->traces->len; trace_idx++) { + trace_idx = 0; + while (trace_idx < session->traces->len) { struct lttng_live_trace *trace = g_ptr_array_index(session->traces, trace_idx); status = lttng_live_metadata_update(trace); - if (status != LTTNG_LIVE_ITERATOR_STATUS_OK && - status != LTTNG_LIVE_ITERATOR_STATUS_END) { + switch (status) { + 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; + default: goto end; } } - status = lttng_live_lazy_msg_init(session); + status = lttng_live_lazy_msg_init(session, + lttng_live_msg_iter->self_msg_iter); end: return status; @@ -508,7 +547,10 @@ enum lttng_live_iterator_status lttng_live_iterator_handle_new_streams_and_metadata( struct lttng_live_msg_iter *lttng_live_msg_iter) { - enum lttng_live_iterator_status ret = LTTNG_LIVE_ITERATOR_STATUS_OK; + enum lttng_live_iterator_status status; + enum lttng_live_viewer_status viewer_status; + bt_logging_level log_level = lttng_live_msg_iter->log_level; + bt_self_component *self_comp = lttng_live_msg_iter->self_comp; uint64_t session_idx = 0, nr_sessions_opened = 0; struct lttng_live_session *session; enum session_not_found_action sess_not_found_act = @@ -522,15 +564,24 @@ lttng_live_iterator_handle_new_streams_and_metadata( */ if (lttng_live_msg_iter->sessions->len == 0) { if (sess_not_found_act != SESSION_NOT_FOUND_ACTION_CONTINUE) { - ret = LTTNG_LIVE_ITERATOR_STATUS_END; + status = LTTNG_LIVE_ITERATOR_STATUS_END; goto end; } else { /* * Retry to create a viewer session for the requested * session name. */ - if (lttng_live_create_viewer_session(lttng_live_msg_iter)) { - ret = LTTNG_LIVE_ITERATOR_STATUS_ERROR; + viewer_status = lttng_live_create_viewer_session(lttng_live_msg_iter); + if (viewer_status != LTTNG_LIVE_VIEWER_STATUS_OK) { + if (viewer_status == LTTNG_LIVE_VIEWER_STATUS_ERROR) { + status = LTTNG_LIVE_ITERATOR_STATUS_ERROR; + BT_COMP_LOGE_APPEND_CAUSE(self_comp, + "Error creating LTTng live viewer session"); + } else if (viewer_status == LTTNG_LIVE_VIEWER_STATUS_INTERRUPTED) { + status = LTTNG_LIVE_ITERATOR_STATUS_AGAIN; + } else { + bt_common_abort(); + } goto end; } } @@ -540,12 +591,12 @@ lttng_live_iterator_handle_new_streams_and_metadata( session_idx++) { session = g_ptr_array_index(lttng_live_msg_iter->sessions, session_idx); - ret = lttng_live_get_session(lttng_live_msg_iter, session); - switch (ret) { + status = lttng_live_get_session(lttng_live_msg_iter, session); + switch (status) { case LTTNG_LIVE_ITERATOR_STATUS_OK: break; case LTTNG_LIVE_ITERATOR_STATUS_END: - ret = LTTNG_LIVE_ITERATOR_STATUS_OK; + status = LTTNG_LIVE_ITERATOR_STATUS_OK; break; default: goto end; @@ -554,13 +605,16 @@ lttng_live_iterator_handle_new_streams_and_metadata( nr_sessions_opened++; } } -end: - if (ret == LTTNG_LIVE_ITERATOR_STATUS_OK && - sess_not_found_act != SESSION_NOT_FOUND_ACTION_CONTINUE && + + if (sess_not_found_act != SESSION_NOT_FOUND_ACTION_CONTINUE && nr_sessions_opened == 0) { - ret = LTTNG_LIVE_ITERATOR_STATUS_END; + status = LTTNG_LIVE_ITERATOR_STATUS_END; + } else { + status = LTTNG_LIVE_ITERATOR_STATUS_OK; } - return ret; + +end: + return status; } static @@ -570,6 +624,8 @@ enum lttng_live_iterator_status emit_inactivity_message( bt_message **message, uint64_t timestamp) { enum lttng_live_iterator_status ret = LTTNG_LIVE_ITERATOR_STATUS_OK; + bt_logging_level log_level = lttng_live_msg_iter->log_level; + bt_self_component *self_comp = lttng_live_msg_iter->self_comp; bt_message *msg = NULL; BT_ASSERT(stream_iter->trace->clock_class); @@ -578,6 +634,8 @@ enum lttng_live_iterator_status emit_inactivity_message( lttng_live_msg_iter->self_msg_iter, stream_iter->trace->clock_class, timestamp); if (!msg) { + BT_COMP_LOGE_APPEND_CAUSE(self_comp, + "Error emitting message iterator inactivity message"); goto error; } @@ -695,7 +753,8 @@ int live_get_msg_ts_ns(struct lttng_live_stream_iterator *stream_iter, ret = bt_clock_snapshot_get_ns_from_origin(clock_snapshot, ts_ns); if (ret) { - BT_COMP_LOGE("Cannot get nanoseconds from Epoch of clock snapshot: " + BT_COMP_LOGE_APPEND_CAUSE(self_comp, + "Cannot get nanoseconds from Epoch of clock snapshot: " "clock-snapshot-addr=%p", clock_snapshot); goto error; } @@ -725,7 +784,7 @@ enum lttng_live_iterator_status lttng_live_iterator_next_handle_one_active_data_ enum lttng_live_iterator_status ret = LTTNG_LIVE_ITERATOR_STATUS_OK; bt_logging_level log_level = lttng_live_msg_iter->log_level; bt_self_component *self_comp = lttng_live_msg_iter->self_comp; - enum bt_msg_iter_status status; + enum ctf_msg_iter_status status; uint64_t session_idx, trace_idx; for (session_idx = 0; session_idx < lttng_live_msg_iter->sessions->len; @@ -753,16 +812,16 @@ enum lttng_live_iterator_status lttng_live_iterator_next_handle_one_active_data_ goto end; } - status = bt_msg_iter_get_next_message(lttng_live_stream->msg_iter, - lttng_live_msg_iter->self_msg_iter, message); + status = ctf_msg_iter_get_next_message( + lttng_live_stream->msg_iter, message); switch (status) { - case BT_MSG_ITER_STATUS_EOF: + case CTF_MSG_ITER_STATUS_EOF: ret = LTTNG_LIVE_ITERATOR_STATUS_END; break; - case BT_MSG_ITER_STATUS_OK: + case CTF_MSG_ITER_STATUS_OK: ret = LTTNG_LIVE_ITERATOR_STATUS_OK; break; - case BT_MSG_ITER_STATUS_AGAIN: + case CTF_MSG_ITER_STATUS_AGAIN: /* * Continue immediately (end of packet). The next * get_index may return AGAIN to delay the following @@ -770,13 +829,14 @@ enum lttng_live_iterator_status lttng_live_iterator_next_handle_one_active_data_ */ ret = LTTNG_LIVE_ITERATOR_STATUS_CONTINUE; break; - case BT_MSG_ITER_STATUS_INVAL: + case CTF_MSG_ITER_STATUS_INVAL: /* No argument provided by the user, so don't return INVAL. */ - case BT_MSG_ITER_STATUS_ERROR: + case CTF_MSG_ITER_STATUS_ERROR: default: ret = LTTNG_LIVE_ITERATOR_STATUS_ERROR; - BT_COMP_LOGW("CTF msg iterator return an error or failed msg_iter=%p", - lttng_live_stream->msg_iter); + BT_COMP_LOGE_APPEND_CAUSE(self_comp, + "CTF message iterator return an error or failed: " + "msg_iter=%p", lttng_live_stream->msg_iter); break; } @@ -792,21 +852,24 @@ enum lttng_live_iterator_status lttng_live_iterator_close_stream( { enum lttng_live_iterator_status live_status = LTTNG_LIVE_ITERATOR_STATUS_OK; + bt_logging_level log_level = lttng_live_msg_iter->log_level; + bt_self_component *self_comp = lttng_live_msg_iter->self_comp; /* * 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 + * `ctf_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); + enum ctf_msg_iter_status status = ctf_msg_iter_get_next_message( + stream_iter->msg_iter, curr_msg); - if (status == BT_MSG_ITER_STATUS_ERROR) { + if (status == CTF_MSG_ITER_STATUS_ERROR) { + BT_COMP_LOGE_APPEND_CAUSE(self_comp, + "Error getting the next message from CTF message iterator"); live_status = LTTNG_LIVE_ITERATOR_STATUS_ERROR; goto end; } - BT_ASSERT(status == BT_MSG_ITER_STATUS_OK); + BT_ASSERT(status == CTF_MSG_ITER_STATUS_OK); end: return live_status; @@ -964,7 +1027,7 @@ enum lttng_live_iterator_status next_stream_iterator_for_trace( 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; @@ -998,7 +1061,8 @@ enum lttng_live_iterator_status next_stream_iterator_for_trace( * We received a message in the past. To ensure * monotonicity, we can't send it forward. */ - BT_COMP_LOGE("Message's timestamp is less than " + BT_COMP_LOGE_APPEND_CAUSE(self_comp, + "Message's timestamp is less than " "lttng-live's message iterator's last " "returned timestamp: " "lttng-live-msg-iter-addr=%p, ts=%" PRId64 ", " @@ -1207,6 +1271,7 @@ bt_component_class_message_iterator_next_method_status lttng_live_msg_iter_next( uint64_t *count) { bt_component_class_message_iterator_next_method_status status; + enum lttng_live_viewer_status viewer_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 = @@ -1220,6 +1285,21 @@ bt_component_class_message_iterator_next_method_status lttng_live_msg_iter_next( BT_ASSERT_DBG(lttng_live_msg_iter); + if (G_UNLIKELY(lttng_live_msg_iter->was_interrupted)) { + /* + * The iterator was interrupted in a previous call to the + * `_next()` method. We currently do not support generating + * messages after such event. The babeltrace2 CLI should never + * be running the graph after being interrupted. So this check + * is to prevent other graph users from using this live + * iterator in an messed up internal state. + */ + status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_ERROR; + BT_COMP_LOGE_APPEND_CAUSE(self_comp, + "Message iterator was interrupted during a previous call to the `next()` and currently does not support continuing after such event."); + goto end; + } + /* * Clear all the invalid message reference that might be left over in * the output array. @@ -1235,16 +1315,25 @@ bt_component_class_message_iterator_next_method_status lttng_live_msg_iter_next( if (lttng_live->params.sess_not_found_act != SESSION_NOT_FOUND_ACTION_CONTINUE) { status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_END; - goto no_session; + goto end; } else { /* * The are no more active session for this session * name. Retry to create a viewer session for the * requested session name. */ - if (lttng_live_create_viewer_session(lttng_live_msg_iter)) { - status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_ERROR; - goto no_session; + viewer_status = lttng_live_create_viewer_session(lttng_live_msg_iter); + if (viewer_status != LTTNG_LIVE_VIEWER_STATUS_OK) { + if (viewer_status == LTTNG_LIVE_VIEWER_STATUS_ERROR) { + status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_ERROR; + BT_COMP_LOGE_APPEND_CAUSE(self_comp, + "Error creating LTTng live viewer session"); + } else if (viewer_status == LTTNG_LIVE_VIEWER_STATUS_INTERRUPTED) { + status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_AGAIN; + } else { + bt_common_abort(); + } + goto end; } } } @@ -1312,7 +1401,7 @@ bt_component_class_message_iterator_next_method_status lttng_live_msg_iter_next( } if (stream_iter_status != LTTNG_LIVE_ITERATOR_STATUS_OK) { - goto end; + goto return_status; } if (G_UNLIKELY(youngest_stream_iter == NULL) || @@ -1355,7 +1444,7 @@ bt_component_class_message_iterator_next_method_status lttng_live_msg_iter_next( if (!youngest_stream_iter) { stream_iter_status = LTTNG_LIVE_ITERATOR_STATUS_AGAIN; - goto end; + goto return_status; } BT_ASSERT_DBG(youngest_stream_iter->current_msg); @@ -1377,17 +1466,20 @@ bt_component_class_message_iterator_next_method_status lttng_live_msg_iter_next( stream_iter_status = LTTNG_LIVE_ITERATOR_STATUS_OK; } -end: + +return_status: switch (stream_iter_status) { case LTTNG_LIVE_ITERATOR_STATUS_OK: case LTTNG_LIVE_ITERATOR_STATUS_AGAIN: + /* + * If we gathered messages, return _OK even if the graph was + * interrupted. This allows for the components downstream to at + * least get the thoses messages. If the graph was indeed + * interrupted there should not be another _next() call as the + * application will tear down the graph. This component class + * doesn't support restarting after an interruption. + */ if (*count > 0) { - /* - * We received a again status but we have some messages - * to send downstream. We send them and return OK for - * now. On the next call we return again if there are - * still no new message to send. - */ status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK; } else { status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_AGAIN; @@ -1397,23 +1489,66 @@ end: 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); break; default: - abort(); + bt_common_abort(); } -no_session: +end: 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, @@ -1421,12 +1556,12 @@ bt_component_class_message_iterator_initialize_method_status lttng_live_msg_iter bt_self_component_source *self_comp_src, bt_self_component_port_output *self_port) { - bt_component_class_message_iterator_initialize_method_status ret = - BT_COMPONENT_CLASS_MESSAGE_ITERATOR_INITIALIZE_METHOD_STATUS_OK; + bt_component_class_message_iterator_initialize_method_status status; bt_self_component *self_comp = 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; + enum lttng_live_viewer_status viewer_status; bt_logging_level log_level; BT_ASSERT(self_msg_it); @@ -1435,77 +1570,92 @@ bt_component_class_message_iterator_initialize_method_status lttng_live_msg_iter 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) { - ret = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_INITIALIZE_METHOD_STATUS_ERROR; - goto end; + status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_INITIALIZE_METHOD_STATUS_MEMORY_ERROR; + BT_COMP_LOGE_APPEND_CAUSE(self_comp, + "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->sessions = g_ptr_array_new_with_free_func( - (GDestroyNotify) lttng_live_destroy_session); - BT_ASSERT(lttng_live_msg_iter->sessions); - - lttng_live_msg_iter->viewer_connection = - live_viewer_connection_create(lttng_live->params.url->str, false, - lttng_live_msg_iter, log_level); - if (!lttng_live_msg_iter->viewer_connection) { + 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); + if (viewer_status != LTTNG_LIVE_VIEWER_STATUS_OK) { + if (viewer_status == LTTNG_LIVE_VIEWER_STATUS_ERROR) { + BT_COMP_LOGE_APPEND_CAUSE(self_comp, + "Failed to create viewer connection"); + } else if (viewer_status == LTTNG_LIVE_VIEWER_STATUS_INTERRUPTED) { + /* + * Interruption in the _iter_init() method is not + * supported. Return an error. + */ + BT_COMP_LOGE_APPEND_CAUSE(self_comp, + "Interrupted while creating viewer connection"); + } goto error; } - if (lttng_live_create_viewer_session(lttng_live_msg_iter)) { + viewer_status = lttng_live_create_viewer_session(lttng_live_msg_iter); + if (viewer_status != LTTNG_LIVE_VIEWER_STATUS_OK) { + if (viewer_status == LTTNG_LIVE_VIEWER_STATUS_ERROR) { + BT_COMP_LOGE_APPEND_CAUSE(self_comp, + "Failed to create viewer session"); + } else if (viewer_status == LTTNG_LIVE_VIEWER_STATUS_INTERRUPTED) { + /* + * Interruption in the _iter_init() method is not + * supported. Return an error. + */ + BT_COMP_LOGE_APPEND_CAUSE(self_comp, + "Interrupted when creating viewer session"); + } 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_COMP_LOGI("Unable to connect to the requested live viewer " - "session. Keep trying to connect because of " + 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, SESS_NOT_FOUND_ACTION_CONTINUE_STR, lttng_live->params.url->str); break; case SESSION_NOT_FOUND_ACTION_FAIL: - BT_COMP_LOGE("Unable to connect to the requested live viewer " - "session. Fail the message iterator" - "initialization because of %s=\"%s\" " + BT_COMP_LOGE_APPEND_CAUSE(self_comp, + "Unable to connect to the requested live viewer session. 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); goto error; case SESSION_NOT_FOUND_ACTION_END: - BT_COMP_LOGI("Unable to connect to the requested live viewer " - "session. End gracefully at the first _next() " + 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; default: - abort(); + bt_common_abort(); } } bt_self_message_iterator_set_data(self_msg_it, lttng_live_msg_iter); - + status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_INITIALIZE_METHOD_STATUS_OK; goto end; + error: - ret = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_INITIALIZE_METHOD_STATUS_ERROR; + status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_INITIALIZE_METHOD_STATUS_ERROR; lttng_live_msg_iter_destroy(lttng_live_msg_iter); end: - return ret; + return status; } static struct bt_param_validation_map_value_entry_descr list_sessions_params[] = { @@ -1523,6 +1673,7 @@ bt_component_class_query_method_status lttng_live_query_list_sessions( const bt_value *url_value = NULL; const char *url; struct live_viewer_connection *viewer_connection = NULL; + enum lttng_live_viewer_status viewer_status; enum bt_param_validation_status validation_status; gchar *validate_error = NULL; @@ -1533,23 +1684,34 @@ bt_component_class_query_method_status lttng_live_query_list_sessions( goto error; } else if (validation_status == BT_PARAM_VALIDATION_STATUS_VALIDATION_ERROR) { status = BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_ERROR; - BT_COMP_CLASS_LOGE_APPEND_CAUSE(self_comp_class, "%s", validate_error); + BT_COMP_CLASS_LOGE_APPEND_CAUSE(self_comp_class, "%s", + validate_error); goto error; } url_value = bt_value_map_borrow_entry_value_const(params, URL_PARAM); url = bt_value_string_get(url_value); - viewer_connection = live_viewer_connection_create(url, true, NULL, - log_level); - if (!viewer_connection) { - status = BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_ERROR; + viewer_status = live_viewer_connection_create(NULL, self_comp_class, + log_level, url, true, NULL, &viewer_connection); + if (viewer_status != LTTNG_LIVE_VIEWER_STATUS_OK) { + if (viewer_status == LTTNG_LIVE_VIEWER_STATUS_ERROR) { + BT_COMP_CLASS_LOGE_APPEND_CAUSE(self_comp_class, + "Failed to create viewer connection"); + status = BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_ERROR; + } else if (viewer_status == LTTNG_LIVE_VIEWER_STATUS_INTERRUPTED) { + status = BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_AGAIN; + } else { + bt_common_abort(); + } goto error; } status = live_viewer_connection_list_sessions(viewer_connection, result); if (status != BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_OK) { + BT_COMP_CLASS_LOGE_APPEND_CAUSE(self_comp_class, + "Failed to list viewer sessions"); goto error; } @@ -1575,6 +1737,7 @@ end: static bt_component_class_query_method_status lttng_live_query_support_info( const bt_value *params, const bt_value **result, + bt_self_component_class *self_comp_class, bt_logging_level log_level) { bt_component_class_query_method_status status = @@ -1591,12 +1754,14 @@ bt_component_class_query_method_status lttng_live_query_support_info( input_type_value = bt_value_map_borrow_entry_value_const(params, "type"); if (!input_type_value) { - BT_COMP_LOGE("Missing expected `type` parameter."); + BT_COMP_CLASS_LOGE_APPEND_CAUSE(self_comp_class, + "Missing expected `type` parameter."); goto error; } if (!bt_value_is_string(input_type_value)) { - BT_COMP_LOGE("`type` parameter is not a string value."); + BT_COMP_CLASS_LOGE_APPEND_CAUSE(self_comp_class, + "`type` parameter is not a string value."); goto error; } @@ -1607,12 +1772,14 @@ bt_component_class_query_method_status lttng_live_query_support_info( input_value = bt_value_map_borrow_entry_value_const(params, "input"); if (!input_value) { - BT_COMP_LOGE("Missing expected `input` parameter."); + BT_COMP_CLASS_LOGE_APPEND_CAUSE(self_comp_class, + "Missing expected `input` parameter."); goto error; } if (!bt_value_is_string(input_value)) { - BT_COMP_LOGE("`input` parameter is not a string value."); + BT_COMP_CLASS_LOGE_APPEND_CAUSE(self_comp_class, + "`input` parameter is not a string value."); goto error; } @@ -1669,7 +1836,7 @@ bt_component_class_query_method_status lttng_live_query( self_comp_class, log_level); } else if (strcmp(object, "babeltrace.support-info") == 0) { status = lttng_live_query_support_info(params, result, - log_level); + self_comp_class, log_level); } else { BT_COMP_LOGI("Unknown query object `%s`", object); status = BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_UNKNOWN_OBJECT;