X-Git-Url: https://git.lttng.org/?p=lttng-tools.git;a=blobdiff_plain;f=src%2Fbin%2Flttng-relayd%2Fviewer-stream.c;h=8e3a1949288dfddf250b0f5cd6edfb3690318218;hp=333246afb3f4d312874cca9cfc144e6cc0fdf932;hb=4c2717fcf7b66d23540fd0b52d3efa885680619e;hpb=a44ca2ca85e4b64729f7b88b1919fd6737dfff8a diff --git a/src/bin/lttng-relayd/viewer-stream.c b/src/bin/lttng-relayd/viewer-stream.c index 333246afb..8e3a19492 100644 --- a/src/bin/lttng-relayd/viewer-stream.c +++ b/src/bin/lttng-relayd/viewer-stream.c @@ -17,10 +17,10 @@ * Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA. */ -#define _GNU_SOURCE #define _LGPL_SOURCE #include #include +#include #include "lttng-relayd.h" #include "viewer-stream.h" @@ -41,9 +41,16 @@ static void viewer_stream_destroy_rcu(struct rcu_head *head) } struct relay_viewer_stream *viewer_stream_create(struct relay_stream *stream, + struct lttng_trace_chunk *viewer_trace_chunk, enum lttng_viewer_seek seek_t) { - struct relay_viewer_stream *vstream; + struct relay_viewer_stream *vstream = NULL; + const bool acquired_reference = lttng_trace_chunk_get( + viewer_trace_chunk); + + if (!acquired_reference) { + goto error; + } vstream = zmalloc(sizeof(*vstream)); if (!vstream) { @@ -51,12 +58,14 @@ struct relay_viewer_stream *viewer_stream_create(struct relay_stream *stream, goto error; } - vstream->path_name = strndup(stream->path_name, LTTNG_VIEWER_PATH_MAX); + vstream->stream_file.trace_chunk = viewer_trace_chunk; + viewer_trace_chunk = NULL; + vstream->path_name = lttng_strndup(stream->path_name, LTTNG_VIEWER_PATH_MAX); if (vstream->path_name == NULL) { PERROR("relay viewer path_name alloc"); goto error; } - vstream->channel_name = strndup(stream->channel_name, + vstream->channel_name = lttng_strndup(stream->channel_name, LTTNG_VIEWER_NAME_MAX); if (vstream->channel_name == NULL) { PERROR("relay viewer channel_name alloc"); @@ -96,7 +105,7 @@ struct relay_viewer_stream *viewer_stream_create(struct relay_stream *stream, } case LTTNG_VIEWER_SEEK_LAST: vstream->current_tracefile_id = - tracefile_array_get_file_index_head(stream->tfa); + tracefile_array_get_read_file_index_head(stream->tfa); /* * We seek at the very end of each stream, awaiting for * a future packet to eventually come in. @@ -116,29 +125,29 @@ struct relay_viewer_stream *viewer_stream_create(struct relay_stream *stream, * the opening of the index, otherwise open it right now. */ if (stream->index_received_seqcount == 0) { - vstream->index_fd = NULL; + vstream->index_file = NULL; } else { - int read_fd; - - read_fd = index_open(vstream->path_name, vstream->channel_name, - stream->tracefile_count, - vstream->current_tracefile_id); - if (read_fd < 0) { - goto error_unlock; - } - vstream->index_fd = stream_fd_create(read_fd); - if (!vstream->index_fd) { - if (close(read_fd)) { - PERROR("close"); - } + const uint32_t connection_major = stream->trace->session->major; + const uint32_t connection_minor = stream->trace->session->minor; + + vstream->index_file = lttng_index_file_create_from_trace_chunk_read_only( + vstream->stream_file.trace_chunk, + stream->path_name, + stream->channel_name, stream->tracefile_size, + vstream->current_tracefile_id, + lttng_to_index_major(connection_major, + connection_minor), + lttng_to_index_minor(connection_major, + connection_minor)); + if (!vstream->index_file) { goto error_unlock; } } - if (seek_t == LTTNG_VIEWER_SEEK_LAST && vstream->index_fd) { + if (seek_t == LTTNG_VIEWER_SEEK_LAST && vstream->index_file) { off_t lseek_ret; - lseek_ret = lseek(vstream->index_fd->fd, 0, SEEK_END); + lseek_ret = lseek(vstream->index_file->fd, 0, SEEK_END); if (lseek_ret < 0) { goto error_unlock; } @@ -153,7 +162,6 @@ struct relay_viewer_stream *viewer_stream_create(struct relay_stream *stream, lttng_ht_node_init_u64(&vstream->stream_n, stream->stream_handle); lttng_ht_add_unique_u64(viewer_streams_ht, &vstream->stream_n); - pthread_mutex_init(&vstream->reflock, NULL); urcu_ref_init(&vstream->ref); return vstream; @@ -164,6 +172,9 @@ error: if (vstream) { viewer_stream_destroy(vstream); } + if (viewer_trace_chunk && acquired_reference) { + lttng_trace_chunk_put(viewer_trace_chunk); + } return NULL; } @@ -188,34 +199,27 @@ static void viewer_stream_release(struct urcu_ref *ref) viewer_stream_unpublish(vstream); - if (vstream->stream_fd) { - stream_fd_put(vstream->stream_fd); - vstream->stream_fd = NULL; + if (vstream->stream_file.fd) { + stream_fd_put(vstream->stream_file.fd); + vstream->stream_file.fd = NULL; } - if (vstream->index_fd) { - stream_fd_put(vstream->index_fd); - vstream->index_fd = NULL; + if (vstream->index_file) { + lttng_index_file_put(vstream->index_file); + vstream->index_file = NULL; } if (vstream->stream) { stream_put(vstream->stream); vstream->stream = NULL; } + lttng_trace_chunk_put(vstream->stream_file.trace_chunk); + vstream->stream_file.trace_chunk = NULL; call_rcu(&vstream->rcu_node, viewer_stream_destroy_rcu); } /* Must be called with RCU read-side lock held. */ bool viewer_stream_get(struct relay_viewer_stream *vstream) { - bool has_ref = false; - - pthread_mutex_lock(&vstream->reflock); - if (vstream->ref.refcount != 0) { - has_ref = true; - urcu_ref_get(&vstream->ref); - } - pthread_mutex_unlock(&vstream->reflock); - - return has_ref; + return urcu_ref_get_unless_zero(&vstream->ref); } /* @@ -248,9 +252,7 @@ end: void viewer_stream_put(struct relay_viewer_stream *vstream) { rcu_read_lock(); - pthread_mutex_lock(&vstream->reflock); urcu_ref_put(&vstream->ref, viewer_stream_release); - pthread_mutex_unlock(&vstream->reflock); rcu_read_unlock(); } @@ -263,8 +265,10 @@ void viewer_stream_put(struct relay_viewer_stream *vstream) int viewer_stream_rotate(struct relay_viewer_stream *vstream) { int ret; - struct relay_stream *stream = vstream->stream; uint64_t new_id; + const struct relay_stream *stream = vstream->stream; + const uint32_t connection_major = stream->trace->session->major; + const uint32_t connection_minor = stream->trace->session->minor; /* Detect the last tracefile to open. */ if (stream->index_received_seqcount @@ -305,29 +309,30 @@ int viewer_stream_rotate(struct relay_viewer_stream *vstream) vstream->index_sent_seqcount = seq_tail; } - if (vstream->index_fd) { - stream_fd_put(vstream->index_fd); - vstream->index_fd = NULL; + if (vstream->index_file) { + lttng_index_file_put(vstream->index_file); + vstream->index_file = NULL; } - if (vstream->stream_fd) { - stream_fd_put(vstream->stream_fd); - vstream->stream_fd = NULL; + if (vstream->stream_file.fd) { + stream_fd_put(vstream->stream_file.fd); + vstream->stream_file.fd = NULL; } - - ret = index_open(vstream->path_name, vstream->channel_name, - stream->tracefile_count, - vstream->current_tracefile_id); - if (ret < 0) { + vstream->index_file = + lttng_index_file_create_from_trace_chunk_read_only( + vstream->stream_file.trace_chunk, + stream->path_name, + stream->channel_name, + stream->tracefile_size, + vstream->current_tracefile_id, + lttng_to_index_major(connection_major, + connection_minor), + lttng_to_index_minor(connection_major, + connection_minor)); + if (!vstream->index_file) { + ret = -1; goto end; - } - vstream->index_fd = stream_fd_create(ret); - if (vstream->index_fd) { - ret = 0; } else { - if (close(ret)) { - PERROR("close"); - } - ret = -1; + ret = 0; } end: return ret; @@ -338,6 +343,10 @@ void print_viewer_streams(void) struct lttng_ht_iter iter; struct relay_viewer_stream *vstream; + if (!viewer_streams_ht) { + return; + } + rcu_read_lock(); cds_lfht_for_each_entry(viewer_streams_ht->ht, &iter.iter, vstream, stream_n.node) {