X-Git-Url: http://git.efficios.com/?a=blobdiff_plain;ds=sidebyside;f=src%2Fplugins%2Fctf%2Flttng-live%2Flttng-live.c;h=fa493cbb3853fc7960764ecaf34ee00f1b33a9ee;hb=477ede8e64839e70251d7cee2ba43e72ec730723;hp=4586d9ccc6732a6d22a4f78a4bc807d100e69d36;hpb=0b6986b4940800145c85d72024b8ca10e6560c62;p=babeltrace.git diff --git a/src/plugins/ctf/lttng-live/lttng-live.c b/src/plugins/ctf/lttng-live/lttng-live.c index 4586d9cc..fa493cbb 100644 --- a/src/plugins/ctf/lttng-live/lttng-live.c +++ b/src/plugins/ctf/lttng-live/lttng-live.c @@ -35,6 +35,7 @@ #include #include +#include #include #include "common/assert.h" @@ -274,9 +275,8 @@ void lttng_live_destroy_session(struct lttng_live_session *session) BT_COMP_LOGD("Destroy lttng live session"); if (session->id != -1ULL) { if (lttng_live_detach_session(session)) { - if (session->lttng_live_msg_iter && - !lttng_live_graph_is_canceled( - session->lttng_live_msg_iter)) { + if (!lttng_live_graph_is_canceled( + session->lttng_live_msg_iter)) { /* Old relayd cannot detach sessions. */ BT_COMP_LOGD("Unable to detach lttng live session %" PRIu64, session->id); @@ -399,7 +399,7 @@ enum lttng_live_iterator_status lttng_live_iterator_next_handle_one_no_data_stre if (ret != LTTNG_LIVE_ITERATOR_STATUS_OK) { goto end; } - BT_ASSERT(lttng_live_stream->state != LTTNG_LIVE_STREAM_EOF); + BT_ASSERT_DBG(lttng_live_stream->state != LTTNG_LIVE_STREAM_EOF); if (lttng_live_stream->state == LTTNG_LIVE_STREAM_QUIESCENT) { uint64_t last_inact_ts = lttng_live_stream->last_inactivity_ts, curr_inact_ts = lttng_live_stream->current_inactivity_ts; @@ -437,13 +437,12 @@ enum lttng_live_iterator_status lttng_live_get_session( { enum lttng_live_iterator_status status; uint64_t trace_idx; - int ret = 0; if (!session->attached) { - ret = lttng_live_attach_session(session); - if (ret) { - if (lttng_live_msg_iter && lttng_live_graph_is_canceled( - lttng_live_msg_iter)) { + enum lttng_live_attach_session_status attach_status = + lttng_live_attach_session(session); + if (attach_status != LTTNG_LIVE_ATTACH_SESSION_STATUS_OK) { + if (lttng_live_graph_is_canceled(lttng_live_msg_iter)) { status = LTTNG_LIVE_ITERATOR_STATUS_AGAIN; } else { status = LTTNG_LIVE_ITERATOR_STATUS_ERROR; @@ -457,13 +456,30 @@ enum lttng_live_iterator_status lttng_live_get_session( 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; } } @@ -633,8 +649,8 @@ int live_get_msg_ts_ns(struct lttng_live_stream_iterator *stream_iter, 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_ASSERT_DBG(msg); + BT_ASSERT_DBG(ts_ns); BT_COMP_LOGD("Getting message's timestamp: iter-data-addr=%p, msg-addr=%p, " "last-msg-ts=%" PRId64, lttng_live_msg_iter, msg, @@ -644,7 +660,7 @@ int live_get_msg_ts_ns(struct lttng_live_stream_iterator *stream_iter, case BT_MESSAGE_TYPE_EVENT: clock_class = bt_message_event_borrow_stream_class_default_clock_class_const( msg); - BT_ASSERT(clock_class); + BT_ASSERT_DBG(clock_class); clock_snapshot = bt_message_event_borrow_default_clock_snapshot_const( msg); @@ -693,7 +709,7 @@ int live_get_msg_ts_ns(struct lttng_live_stream_iterator *stream_iter, } clock_class = bt_clock_snapshot_borrow_clock_class_const(clock_snapshot); - BT_ASSERT(clock_class); + BT_ASSERT_DBG(clock_class); ret = bt_clock_snapshot_get_ns_from_origin(clock_snapshot, ts_ns); if (ret) { @@ -940,8 +956,8 @@ enum lttng_live_iterator_status next_stream_iterator_for_trace( int64_t youngest_candidate_msg_ts = INT64_MAX; uint64_t stream_iter_idx; - BT_ASSERT(live_trace); - BT_ASSERT(live_trace->stream_iterators); + BT_ASSERT_DBG(live_trace); + BT_ASSERT_DBG(live_trace->stream_iterators); /* * Update the current message of every stream iterators of this trace. * The current msg of every stream must have a timestamp equal or @@ -955,16 +971,6 @@ enum lttng_live_iterator_status next_stream_iterator_for_trace( 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. @@ -986,7 +992,7 @@ enum lttng_live_iterator_status next_stream_iterator_for_trace( goto end; } - BT_ASSERT(msg); + BT_ASSERT_DBG(msg); /* * Get the timestamp in nanoseconds from origin of this @@ -1022,7 +1028,7 @@ enum lttng_live_iterator_status next_stream_iterator_for_trace( } } - BT_ASSERT(stream_iter != youngest_candidate_stream_iter); + BT_ASSERT_DBG(stream_iter != youngest_candidate_stream_iter); if (!stream_iter_is_ended) { if (G_UNLIKELY(youngest_candidate_stream_iter == NULL) || @@ -1039,7 +1045,7 @@ enum lttng_live_iterator_status next_stream_iterator_for_trace( * Order the messages in an arbitrary but * deterministic way. */ - BT_ASSERT(stream_iter != youngest_candidate_stream_iter); + BT_ASSERT_DBG(stream_iter != youngest_candidate_stream_iter); int ret = common_muxing_compare_messages( stream_iter->current_msg, youngest_candidate_stream_iter->current_msg); @@ -1067,14 +1073,17 @@ enum lttng_live_iterator_status next_stream_iterator_for_trace( stream_iter_idx++; } else { /* - * The live stream iterator is ENDed. We remove that - * iterator from the list and we restart the iteration - * at the beginning of the live stream iterator array - * to because the removal will shuffle the array. + * The live stream iterator has ended. That + * iterator is removed from the array, but + * there is no need to increment + * stream_iter_idx as + * g_ptr_array_remove_index_fast replaces the + * removed element with the array's last + * element. */ - g_ptr_array_remove_index_fast(live_trace->stream_iterators, + g_ptr_array_remove_index_fast( + live_trace->stream_iterators, stream_iter_idx); - stream_iter_idx = 0; } } @@ -1118,13 +1127,8 @@ enum lttng_live_iterator_status next_stream_iterator_for_session( goto end; } - BT_ASSERT(session->traces); + BT_ASSERT_DBG(session->traces); - /* - * Use while loops here rather then for loops so we can restart the - * iteration if an element is removed from the array during the - * looping. - */ while (trace_idx < session->traces->len) { bool trace_is_ended = false; struct lttng_live_stream_iterator *stream_iter; @@ -1144,7 +1148,7 @@ enum lttng_live_iterator_status next_stream_iterator_for_session( } if (!trace_is_ended) { - BT_ASSERT(stream_iter); + BT_ASSERT_DBG(stream_iter); if (G_UNLIKELY(youngest_candidate_stream_iter == NULL) || stream_iter->current_msg_ts_ns < youngest_candidate_msg_ts) { @@ -1175,8 +1179,13 @@ enum lttng_live_iterator_status next_stream_iterator_for_session( } trace_idx++; } else { - g_ptr_array_remove_index_fast(session->traces, trace_idx); - trace_idx = 0; + /* + * trace_idx is not incremented since + * g_ptr_array_remove_index_fast replaces the + * element at trace_idx with the array's last element. + */ + g_ptr_array_remove_index_fast(session->traces, + trace_idx); } } if (youngest_candidate_stream_iter) { @@ -1227,7 +1236,7 @@ bt_component_class_message_iterator_next_method_status lttng_live_msg_iter_next( *count = 0; - BT_ASSERT(lttng_live_msg_iter); + BT_ASSERT_DBG(lttng_live_msg_iter); /* * Clear all the invalid message reference that might be left over in @@ -1286,18 +1295,8 @@ bt_component_class_message_iterator_next_method_status lttng_live_msg_iter_next( *candidate_stream_iter = NULL; int64_t youngest_msg_ts_ns = INT64_MAX; - BT_ASSERT(lttng_live_msg_iter->sessions); + BT_ASSERT_DBG(lttng_live_msg_iter->sessions); session_idx = 0; - /* - * Use a while loop instead of a for loop so we can restart the - * iteration if we remove an element. We can safely call - * next_stream_iterator_for_session() multiple times on the - * same session as we only fetch a new message if there is no - * current next message for each live stream iterator. - * If all live stream iterator of that session already have a - * current next message, the function will simply exit return - * the same candidate live stream iterator every time. - */ while (session_idx < lttng_live_msg_iter->sessions->len) { struct lttng_live_session *session = g_ptr_array_index(lttng_live_msg_iter->sessions, @@ -1315,14 +1314,15 @@ bt_component_class_message_iterator_next_method_status lttng_live_msg_iter_next( if (stream_iter_status == LTTNG_LIVE_ITERATOR_STATUS_END) { if (session->closed && session->traces->len == 0) { /* - * Remove the session from the list and restart the - * iteration at the beginning of the array since the - * removal shuffle the elements of the array. + * Remove the session from the list. + * session_idx is not modified since + * g_ptr_array_remove_index_fast + * replaces the the removed element with + * the array's last element. */ g_ptr_array_remove_index_fast( lttng_live_msg_iter->sessions, session_idx); - session_idx = 0; } else { session_idx++; } @@ -1376,9 +1376,9 @@ bt_component_class_message_iterator_next_method_status lttng_live_msg_iter_next( goto end; } - BT_ASSERT(youngest_stream_iter->current_msg); + BT_ASSERT_DBG(youngest_stream_iter->current_msg); /* Ensure monotonicity. */ - BT_ASSERT(lttng_live_msg_iter->last_msg_ts_ns <= + BT_ASSERT_DBG(lttng_live_msg_iter->last_msg_ts_ns <= youngest_stream_iter->current_msg_ts_ns); /*