X-Git-Url: http://git.efficios.com/?a=blobdiff_plain;f=src%2Fbin%2Flttng-relayd%2Findex.c;h=b15bbcd7702954e675a98c06e02ab31058ff7499;hb=d30b2041e3a5db2d1671219ea845057cb3dd3082;hp=7182e36cc8e461daf666d6dbc32079dd11b7e5f3;hpb=7591bab11eceedc6a0d1e02fd6f85592267a63b5;p=lttng-tools.git diff --git a/src/bin/lttng-relayd/index.c b/src/bin/lttng-relayd/index.c index 7182e36cc..b15bbcd77 100644 --- a/src/bin/lttng-relayd/index.c +++ b/src/bin/lttng-relayd/index.c @@ -17,7 +17,6 @@ * Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA. */ -#define _GNU_SOURCE #define _LGPL_SOURCE #include @@ -144,7 +143,7 @@ struct relay_index *relay_index_get_by_id_or_create(struct relay_stream *stream, index = relay_index_create(stream, net_seq_num); if (!index) { ERR("Cannot create index for stream id %" PRIu64 " and seq_num %" PRIu64, - index->stream->stream_handle, net_seq_num); + stream->stream_handle, net_seq_num); goto end; } oldindex = relay_index_add_unique(stream, index); @@ -163,22 +162,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); @@ -229,9 +229,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. */ @@ -291,21 +291,16 @@ int relay_index_try_flush(struct relay_index *index) 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; + fd = index->index_file->fd; DBG2("Writing index for stream ID %" PRIu64 " and seq num %" PRIu64 " on fd %d", index->stream->stream_handle, index->index_n.key, fd); 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); @@ -333,3 +328,42 @@ void relay_index_close_all(struct relay_stream *stream) } rcu_read_unlock(); } + +void relay_index_close_partial_fd(struct relay_stream *stream) +{ + struct lttng_ht_iter iter; + struct relay_index *index; + + rcu_read_lock(); + cds_lfht_for_each_entry(stream->indexes_ht->ht, &iter.iter, + index, index_n.node) { + if (!index->index_file) { + continue; + } + /* + * Partial index has its index_file: we have only + * received its info from the data socket. + * Put self-ref from index. + */ + relay_index_put(index); + } + rcu_read_unlock(); +} + +uint64_t relay_index_find_last(struct relay_stream *stream) +{ + struct lttng_ht_iter iter; + struct relay_index *index; + uint64_t net_seq_num = -1ULL; + + rcu_read_lock(); + cds_lfht_for_each_entry(stream->indexes_ht->ht, &iter.iter, + index, index_n.node) { + if (net_seq_num == -1ULL || + index->index_n.key > net_seq_num) { + net_seq_num = index->index_n.key; + } + } + rcu_read_unlock(); + return net_seq_num; +}