X-Git-Url: http://git.liburcu.org/?a=blobdiff_plain;f=src%2Fbin%2Flttng-relayd%2Flive.cpp;h=699778d0e4ef2c144e4bcaba390387750b2f7606;hb=bb9f214082555e181b5a0dc7a7dfdc03a5b827cd;hp=99037e417c0db58cd76e267383a9f3f7dc1bd068;hpb=0359d56c6cbb7a41c36ba6c387acb9dd89c54628;p=lttng-tools.git diff --git a/src/bin/lttng-relayd/live.cpp b/src/bin/lttng-relayd/live.cpp index 99037e417..699778d0e 100644 --- a/src/bin/lttng-relayd/live.cpp +++ b/src/bin/lttng-relayd/live.cpp @@ -238,7 +238,6 @@ static ssize_t send_response(struct lttcomm_sock *sock, void *buf, size_t size) static int check_new_streams(struct relay_connection *conn) { struct relay_session *session; - unsigned long current_val; int ret = 0; if (!conn->viewer_session) { @@ -253,15 +252,19 @@ static int check_new_streams(struct relay_connection *conn) if (!session_get(session)) { continue; } - current_val = uatomic_cmpxchg(&session->new_streams, 1, 0); - ret = current_val; + + ret = uatomic_read(&session->new_streams); session_put(session); if (ret == 1) { goto end; } } } + end: + DBG("Viewer connection has%s new streams: socket_fd = %d", + ret == 0 ? " no" : "", + conn->sock->fd); return ret; } @@ -1218,6 +1221,8 @@ static int viewer_get_new_streams(struct relay_connection *conn) response.status = htobe32(LTTNG_VIEWER_NEW_STREAMS_ERR); goto send_reply_unlock; } + + uatomic_set(&session->new_streams, 0); send_streams = 1; response.status = htobe32(LTTNG_VIEWER_NEW_STREAMS_OK); @@ -1661,6 +1666,7 @@ static int viewer_get_next_index(struct relay_connection *conn) bool viewer_stream_and_session_in_same_chunk, viewer_stream_one_rotation_behind; uint64_t stream_file_chunk_id = -1ULL, viewer_session_chunk_id = -1ULL; enum lttng_trace_chunk_status status; + bool attached_sessions_have_new_streams = false; LTTNG_ASSERT(conn); @@ -1712,6 +1718,17 @@ static int viewer_get_next_index(struct relay_connection *conn) goto send_reply; } + ret = check_new_streams(conn); + if (ret < 0) { + viewer_index.status = LTTNG_VIEWER_INDEX_ERR; + ERR("Error checking for new streams in the attached sessions, returning status=%s", + lttng_viewer_next_index_return_code_str( + (enum lttng_viewer_next_index_return_code) viewer_index.status)); + goto send_reply; + } else if (ret == 1) { + attached_sessions_have_new_streams = true; + } + if (rstream->ongoing_rotation.is_set) { /* Rotation is ongoing, try again later. */ viewer_index.status = LTTNG_VIEWER_INDEX_RETRY; @@ -1821,6 +1838,7 @@ static int viewer_get_next_index(struct relay_connection *conn) */ goto send_reply; } + /* At this point, ret is 0 thus we will be able to read the index. */ LTTNG_ASSERT(!ret); @@ -1899,19 +1917,6 @@ static int viewer_get_next_index(struct relay_connection *conn) vstream->stream_file.handle = fs_handle; } - ret = check_new_streams(conn); - if (ret < 0) { - viewer_index.status = LTTNG_VIEWER_INDEX_ERR; - ERR("Error checking for new streams before sending new index to stream id %" PRIu64 - ", returning status=%s", - (uint64_t) be64toh(request_index.stream_id), - lttng_viewer_next_index_return_code_str( - (enum lttng_viewer_next_index_return_code) viewer_index.status)); - goto send_reply; - } else if (ret == 1) { - viewer_index.flags |= LTTNG_VIEWER_FLAG_NEW_STREAM; - } - ret = lttng_index_file_read(vstream->index_file, &packet_index); if (ret) { viewer_index.status = LTTNG_VIEWER_INDEX_ERR; @@ -1962,6 +1967,10 @@ send_reply: pthread_mutex_unlock(&metadata_viewer_stream->stream->lock); } + if (attached_sessions_have_new_streams) { + viewer_index.flags |= LTTNG_VIEWER_FLAG_NEW_STREAM; + } + viewer_index.flags = htobe32(viewer_index.flags); viewer_index.status = htobe32(viewer_index.status); health_code_update(); @@ -2128,6 +2137,7 @@ static int viewer_get_metadata(struct relay_connection *conn) struct lttng_viewer_get_metadata request; struct lttng_viewer_metadata_packet reply; struct relay_viewer_stream *vstream = nullptr; + bool dispose_of_stream = false; LTTNG_ASSERT(conn); @@ -2156,6 +2166,9 @@ static int viewer_get_metadata(struct relay_connection *conn) reply.status = htobe32(LTTNG_VIEWER_METADATA_ERR); goto send_reply; } + + pthread_mutex_lock(&vstream->stream->trace->session->lock); + pthread_mutex_lock(&vstream->stream->trace->lock); pthread_mutex_lock(&vstream->stream->lock); if (!vstream->stream->is_metadata) { ERR("Invalid metadata stream"); @@ -2164,10 +2177,6 @@ static int viewer_get_metadata(struct relay_connection *conn) if (vstream->metadata_sent >= vstream->stream->metadata_received) { /* - * The live viewers expect to receive a NO_NEW_METADATA - * status before a stream disappears, otherwise they abort the - * entire live connection when receiving an error status. - * * Clear feature resets the metadata_received to 0 until the * same metadata is received again. */ @@ -2176,20 +2185,7 @@ static int viewer_get_metadata(struct relay_connection *conn) * The live viewer considers a closed 0 byte metadata stream as * an error. */ - if (vstream->metadata_sent > 0) { - if (vstream->stream->closed && vstream->stream->no_new_metadata_notified) { - /* - * Release ownership for the viewer metadata - * stream. Note that this reference is the - * viewer's reference. The vstream still exists - * until the end of the function as - * viewer_stream_get_by_id() took a reference. - */ - viewer_stream_put(vstream); - } - - vstream->stream->no_new_metadata_notified = true; - } + dispose_of_stream = vstream->metadata_sent > 0 && vstream->stream->closed; goto send_reply; } @@ -2227,6 +2223,19 @@ static int viewer_get_metadata(struct relay_connection *conn) len = vstream->stream->metadata_received - vstream->metadata_sent; if (!vstream->stream_file.trace_chunk) { + if (vstream->stream->trace->session->connection_closed) { + /* + * If the connection is closed, there is no way for the metadata stream + * to ever transition back to an active chunk. As such, signal to the viewer + * that there is no new metadata available. + * + * The stream can be disposed-of. On the next execution of this command, + * the relay daemon will reply with an error status since the stream can't + * be found. + */ + dispose_of_stream = true; + } + reply.status = htobe32(LTTNG_VIEWER_NO_NEW_METADATA); len = 0; goto send_reply; @@ -2357,6 +2366,8 @@ send_reply: health_code_update(); if (vstream) { pthread_mutex_unlock(&vstream->stream->lock); + pthread_mutex_unlock(&vstream->stream->trace->lock); + pthread_mutex_unlock(&vstream->stream->trace->session->lock); } ret = send_response(conn->sock, &reply, sizeof(reply)); if (ret < 0) { @@ -2382,7 +2393,22 @@ end_free: end: if (vstream) { viewer_stream_put(vstream); + if (dispose_of_stream) { + /* + * Trigger the destruction of the viewer stream + * by releasing its global reference. + * + * The live viewers expect to receive a NO_NEW_METADATA + * status before a stream disappears, otherwise they abort the + * entire live connection when receiving an error status. + * + * On the next query for this stream, an error will be reported to the + * client. + */ + viewer_stream_put(vstream); + } } + return ret; }