static
int make_viewer_streams(struct relay_session *session,
enum lttng_viewer_seek seek_t, uint32_t *nb_total, uint32_t *nb_unsent,
- uint32_t *nb_created)
+ uint32_t *nb_created, bool *closed)
{
int ret;
struct lttng_ht_iter iter;
*/
pthread_mutex_lock(&session->lock);
+ if (session->connection_closed) {
+ *closed = true;
+ }
+
/*
* Create viewer streams for relay streams that are ready to be
* used for a the given session id only.
/* Update number of created stream counter. */
(*nb_created)++;
}
+ /*
+ * Ensure a self-reference is preserved even
+ * after we have put our local reference.
+ */
+ viewer_stream_get(vstream);
} else {
if (!vstream->sent_flag && nb_unsent) {
/* Update number of unsent stream counter. */
(*nb_unsent)++;
}
- viewer_stream_put(vstream);
}
/* Update number of total stream counter. */
if (nb_total) {
- (*nb_total)++;
+ if (stream->is_metadata) {
+ if (!stream->closed ||
+ stream->metadata_received > vstream->metadata_sent) {
+ (*nb_total)++;
+ }
+ } else {
+ if (!stream->closed ||
+ !(((int64_t) (stream->prev_seq - stream->last_net_seq_num)) >= 0)) {
+
+ (*nb_total)++;
+ }
+ }
}
+ /* Put local reference. */
+ viewer_stream_put(vstream);
next:
stream_put(stream);
}
struct lttng_viewer_new_streams_response response;
struct relay_session *session;
uint64_t session_id;
+ bool closed = false;
assert(conn);
response.status = htobe32(LTTNG_VIEWER_NEW_STREAMS_OK);
ret = make_viewer_streams(session, LTTNG_VIEWER_SEEK_LAST, &nb_total, &nb_unsent,
- &nb_created);
+ &nb_created, &closed);
if (ret < 0) {
goto end_put_session;
}
response.streams_count = htobe32(nb_streams);
/*
- * If the session is closed and we have no new streams to send,
- * it means that the viewer has already received the whole trace
- * for this session and should now close it.
+ * If the session is closed, HUP when there are no more streams
+ * with data.
*/
- if (nb_total == 0 && session->connection_closed) {
+ if (closed && nb_total == 0) {
send_streams = 0;
+ response.streams_count = 0;
response.status = htobe32(LTTNG_VIEWER_NEW_STREAMS_HUP);
goto send_reply;
}
struct lttng_viewer_attach_session_request request;
struct lttng_viewer_attach_session_response response;
struct relay_session *session = NULL;
+ bool closed = false;
assert(conn);
goto send_reply;
}
- ret = make_viewer_streams(session, seek_type, &nb_streams, NULL, NULL);
+ ret = make_viewer_streams(session, seek_type, &nb_streams, NULL,
+ NULL, &closed);
if (ret < 0) {
goto end_put_session;
}
response.streams_count = htobe32(nb_streams);
+ /*
+ * If the session is closed when the viewer is attaching, it
+ * means some of the streams may have been concurrently removed,
+ * so we don't allow the viewer to attach, even if there are
+ * streams available.
+ */
+ if (closed) {
+ send_streams = 0;
+ response.streams_count = 0;
+ response.status = htobe32(LTTNG_VIEWER_NEW_STREAMS_HUP);
+ goto send_reply;
+ }
+
send_reply:
health_code_update();
ret = send_response(conn->sock, &response, sizeof(response));
vstream = viewer_stream_get_by_id(be64toh(request_index.stream_id));
if (!vstream) {
- ret = -1;
- goto end;
+ viewer_index.status = htobe32(LTTNG_VIEWER_INDEX_ERR);
+ goto send_reply;
}
/* Use back. ref. Protected by refcounts. */
viewer_index.stream_id = packet_index.stream_id;
send_reply:
- pthread_mutex_unlock(&rstream->lock);
+ if (rstream) {
+ pthread_mutex_unlock(&rstream->lock);
+ }
if (metadata_viewer_stream) {
pthread_mutex_lock(&metadata_viewer_stream->stream->lock);
vstream = viewer_stream_get_by_id(be64toh(get_packet_info.stream_id));
if (!vstream) {
- goto error;
+ reply.status = htobe32(LTTNG_VIEWER_GET_PACKET_ERR);
+ goto send_reply_nolock;
}
ctf_trace = vstream->stream->trace;
}
pthread_mutex_lock(&vstream->stream->lock);
- /*
- * The vstream->stream_fd used here has been opened by
- * get_next_index. It is opened there because this is what
- * allows us to grab a reference to the file with stream lock
- * held, thus protecting us against overwrite caused by
- * tracefile rotation. Since tracefile rotation unlinks the old
- * data file, we are ensured that we won't have our data
- * overwritten under us.
- */
- ret = check_new_streams(conn);
- if (ret < 0) {
- goto end_free;
- } else if (ret == 1) {
- reply.status = htobe32(LTTNG_VIEWER_GET_PACKET_ERR);
- reply.flags |= LTTNG_VIEWER_FLAG_NEW_STREAM;
- goto send_reply;
- }
len = be32toh(get_packet_info.len);
data = zmalloc(len);
vstream = viewer_stream_get_by_id(be64toh(request.stream_id));
if (!vstream) {
- reply.status = htobe32(LTTNG_VIEWER_NO_NEW_METADATA);
+ /*
+ * The metadata stream can be closed by a CLOSE command
+ * just before we attach. It can also be closed by
+ * per-pid tracing during tracing. Therefore, it is
+ * possible that we cannot find this viewer stream.
+ * Reply back to the client with an error if we cannot
+ * find it.
+ */
+ reply.status = htobe32(LTTNG_VIEWER_METADATA_ERR);
goto send_reply;
}
pthread_mutex_lock(&vstream->stream->lock);
projects / lttng-tools.git / blobdiff