X-Git-Url: https://git.lttng.org/?p=lttng-tools.git;a=blobdiff_plain;f=src%2Fbin%2Flttng-relayd%2Findex.c;h=ff27d958a0aebd6eb0f7816544eaa1f686ee192d;hp=9c307790ac5f3f332e7695ff9790782b76683400;hb=211d28db4ccebf1a2abcd04eabba31cf46af3060;hpb=890d8fe47755c3bad936389cf48ffa141cff41c9 diff --git a/src/bin/lttng-relayd/index.c b/src/bin/lttng-relayd/index.c index 9c307790a..ff27d958a 100644 --- a/src/bin/lttng-relayd/index.c +++ b/src/bin/lttng-relayd/index.c @@ -1,20 +1,10 @@ /* - * Copyright (C) 2013 - Julien Desfossez - * David Goulet - * 2015 - Mathieu Desnoyers + * Copyright (C) 2013 Julien Desfossez + * Copyright (C) 2013 David Goulet + * Copyright (C) 2015 Mathieu Desnoyers * - * This program is free software; you can redistribute it and/or modify it - * under the terms of the GNU General Public License, version 2 only, as - * published by the Free Software Foundation. + * SPDX-License-Identifier: GPL-2.0-only * - * This program is distributed in the hope that it will be useful, but WITHOUT - * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or - * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for - * more details. - * - * You should have received a copy of the GNU General Public License along with - * this program; if not, write to the Free Software Foundation, Inc., 51 - * Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA. */ #define _LGPL_SOURCE @@ -22,10 +12,12 @@ #include #include +#include #include "lttng-relayd.h" #include "stream.h" #include "index.h" +#include "connection.h" /* * Allocate a new relay index object. Pass the stream in which it is @@ -58,7 +50,6 @@ static struct relay_index *relay_index_create(struct relay_stream *stream, lttng_ht_node_init_u64(&index->index_n, net_seq_num); pthread_mutex_init(&index->lock, NULL); - pthread_mutex_init(&index->reflock, NULL); urcu_ref_init(&index->ref); end: @@ -98,21 +89,11 @@ static struct relay_index *relay_index_add_unique(struct relay_stream *stream, */ static bool relay_index_get(struct relay_index *index) { - bool has_ref = false; - DBG2("index get for stream id %" PRIu64 " and seqnum %" PRIu64 " refcount %d", index->stream->stream_handle, index->index_n.key, (int) index->ref.refcount); - /* Confirm that the index refcount has not reached 0. */ - pthread_mutex_lock(&index->reflock); - if (index->ref.refcount != 0) { - has_ref = true; - urcu_ref_get(&index->ref); - } - pthread_mutex_unlock(&index->reflock); - - return has_ref; + return urcu_ref_get_unless_zero(&index->ref); } /* @@ -162,22 +143,23 @@ struct relay_index *relay_index_get_by_id_or_create(struct relay_stream *stream, end: rcu_read_unlock(); DBG2("Index %sfound or created in HT for stream ID %" PRIu64 " and seqnum %" PRIu64, - (index == NULL) ? "NOT " : "", index->stream->stream_handle, net_seq_num); + (index == NULL) ? "NOT " : "", stream->stream_handle, net_seq_num); return index; } -int relay_index_set_fd(struct relay_index *index, struct stream_fd *index_fd, +int relay_index_set_file(struct relay_index *index, + struct lttng_index_file *index_file, uint64_t data_offset) { int ret = 0; pthread_mutex_lock(&index->lock); - if (index->index_fd) { + if (index->index_file) { ret = -1; goto end; } - stream_fd_get(index_fd); - index->index_fd = index_fd; + lttng_index_file_get(index_file); + index->index_file = index_file; index->index_data.offset = data_offset; end: pthread_mutex_unlock(&index->lock); @@ -185,7 +167,7 @@ end: } int relay_index_set_data(struct relay_index *index, - const struct ctf_packet_index *data) + const struct ctf_packet_index *data) { int ret = 0; @@ -228,9 +210,9 @@ static void index_release(struct urcu_ref *ref) int ret; struct lttng_ht_iter iter; - if (index->index_fd) { - stream_fd_put(index->index_fd); - index->index_fd = NULL; + if (index->index_file) { + lttng_index_file_put(index->index_file); + index->index_file = NULL; } if (index->in_hash_table) { /* Delete index from hash table. */ @@ -264,10 +246,8 @@ void relay_index_put(struct relay_index *index) * Index lock ensures that concurrent test and update of stream * ref is atomic. */ - pthread_mutex_lock(&index->reflock); assert(index->ref.refcount != 0); urcu_ref_put(&index->ref, index_release); - pthread_mutex_unlock(&index->reflock); rcu_read_unlock(); } @@ -283,28 +263,21 @@ int relay_index_try_flush(struct relay_index *index) { int ret = 1; bool flushed = false; - int fd; pthread_mutex_lock(&index->lock); if (index->flushed) { goto skip; } /* Check if we are ready to flush. */ - if (!index->has_index_data || !index->index_fd) { + if (!index->has_index_data || !index->index_file) { goto skip; } - fd = index->index_fd->fd; - DBG2("Writing index for stream ID %" PRIu64 " and seq num %" PRIu64 - " on fd %d", index->stream->stream_handle, - index->index_n.key, fd); + + DBG2("Writing index for stream ID %" PRIu64 " and seq num %" PRIu64, + index->stream->stream_handle, index->index_n.key); flushed = true; index->flushed = true; - ret = index_write(fd, &index->index_data, sizeof(index->index_data)); - if (ret == sizeof(index->index_data)) { - ret = 0; - } else { - ret = -1; - } + ret = lttng_index_file_write(index->index_file, &index->index_data); skip: pthread_mutex_unlock(&index->lock); @@ -341,11 +314,11 @@ void relay_index_close_partial_fd(struct relay_stream *stream) rcu_read_lock(); cds_lfht_for_each_entry(stream->indexes_ht->ht, &iter.iter, index, index_n.node) { - if (!index->index_fd) { + if (!index->index_file) { continue; } /* - * Partial index has its index_fd: we have only + * Partial index has its index_file: we have only * received its info from the data socket. * Put self-ref from index. */ @@ -371,3 +344,88 @@ uint64_t relay_index_find_last(struct relay_stream *stream) rcu_read_unlock(); return net_seq_num; } + +/* + * Update the index file of an already existing relay_index. + * Offsets by 'removed_data_count' the offset field of an index. + */ +static +int relay_index_switch_file(struct relay_index *index, + struct lttng_index_file *new_index_file, + uint64_t removed_data_count) +{ + int ret = 0; + uint64_t offset; + + pthread_mutex_lock(&index->lock); + if (!index->index_file) { + ERR("No index_file"); + ret = 0; + goto end; + } + + lttng_index_file_put(index->index_file); + lttng_index_file_get(new_index_file); + index->index_file = new_index_file; + offset = be64toh(index->index_data.offset); + index->index_data.offset = htobe64(offset - removed_data_count); + +end: + pthread_mutex_unlock(&index->lock); + return ret; +} + +/* + * Switch the index file of all pending indexes for a stream and update the + * data offset by substracting the last safe position. + * Stream lock must be held. + */ +int relay_index_switch_all_files(struct relay_stream *stream) +{ + struct lttng_ht_iter iter; + struct relay_index *index; + int ret = 0; + + rcu_read_lock(); + cds_lfht_for_each_entry(stream->indexes_ht->ht, &iter.iter, + index, index_n.node) { + ret = relay_index_switch_file(index, stream->index_file, + stream->pos_after_last_complete_data_index); + if (ret) { + goto end; + } + } +end: + rcu_read_unlock(); + return ret; +} + +/* + * Set index data from the control port to a given index object. + */ +int relay_index_set_control_data(struct relay_index *index, + const struct lttcomm_relayd_index *data, + unsigned int minor_version) +{ + /* The index on disk is encoded in big endian. */ + const struct ctf_packet_index index_data = { + .packet_size = htobe64(data->packet_size), + .content_size = htobe64(data->content_size), + .timestamp_begin = htobe64(data->timestamp_begin), + .timestamp_end = htobe64(data->timestamp_end), + .events_discarded = htobe64(data->events_discarded), + .stream_id = htobe64(data->stream_id), + }; + + if (minor_version >= 8) { + index->index_data.stream_instance_id = htobe64(data->stream_instance_id); + index->index_data.packet_seq_num = htobe64(data->packet_seq_num); + } else { + uint64_t unset_value = -1ULL; + + index->index_data.stream_instance_id = htobe64(unset_value); + index->index_data.packet_seq_num = htobe64(unset_value); + } + + return relay_index_set_data(index, &index_data); +}