2 * Copyright (C) 2013 - Julien Desfossez <jdesfossez@efficios.com>
3 * David Goulet <dgoulet@efficios.com>
4 * 2015 - Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
6 * This program is free software; you can redistribute it and/or modify it
7 * under the terms of the GNU General Public License, version 2 only, as
8 * published by the Free Software Foundation.
10 * This program is distributed in the hope that it will be useful, but WITHOUT
11 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
12 * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
15 * You should have received a copy of the GNU General Public License along with
16 * this program; if not, write to the Free Software Foundation, Inc., 51
17 * Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
23 #include <common/common.h>
24 #include <common/utils.h>
25 #include <common/compat/endian.h>
27 #include "lttng-relayd.h"
32 * Allocate a new relay index object. Pass the stream in which it is
33 * contained as parameter. The sequence number will be used as the hash
36 * Called with stream mutex held.
37 * Return allocated object or else NULL on error.
39 static struct relay_index
*relay_index_create(struct relay_stream
*stream
,
42 struct relay_index
*index
;
44 DBG2("Creating relay index for stream id %" PRIu64
" and seqnum %" PRIu64
,
45 stream
->stream_handle
, net_seq_num
);
47 index
= zmalloc(sizeof(*index
));
49 PERROR("Relay index zmalloc");
52 if (!stream_get(stream
)) {
53 ERR("Cannot get stream");
58 index
->stream
= stream
;
60 lttng_ht_node_init_u64(&index
->index_n
, net_seq_num
);
61 pthread_mutex_init(&index
->lock
, NULL
);
62 urcu_ref_init(&index
->ref
);
69 * Add unique relay index to the given hash table. In case of a collision, the
70 * already existing object is put in the given _index variable.
72 * RCU read side lock MUST be acquired.
74 static struct relay_index
*relay_index_add_unique(struct relay_stream
*stream
,
75 struct relay_index
*index
)
77 struct cds_lfht_node
*node_ptr
;
78 struct relay_index
*_index
;
80 DBG2("Adding relay index with stream id %" PRIu64
" and seqnum %" PRIu64
,
81 stream
->stream_handle
, index
->index_n
.key
);
83 node_ptr
= cds_lfht_add_unique(stream
->indexes_ht
->ht
,
84 stream
->indexes_ht
->hash_fct(&index
->index_n
, lttng_ht_seed
),
85 stream
->indexes_ht
->match_fct
, &index
->index_n
,
86 &index
->index_n
.node
);
87 if (node_ptr
!= &index
->index_n
.node
) {
88 _index
= caa_container_of(node_ptr
, struct relay_index
,
97 * Should be called with RCU read-side lock held.
99 static bool relay_index_get(struct relay_index
*index
)
101 DBG2("index get for stream id %" PRIu64
" and seqnum %" PRIu64
" refcount %d",
102 index
->stream
->stream_handle
, index
->index_n
.key
,
103 (int) index
->ref
.refcount
);
105 return urcu_ref_get_unless_zero(&index
->ref
);
109 * Get a relayd index in within the given stream, or create it if not
112 * Called with stream mutex held.
113 * Return index object or else NULL on error.
115 struct relay_index
*relay_index_get_by_id_or_create(struct relay_stream
*stream
,
116 uint64_t net_seq_num
)
118 struct lttng_ht_node_u64
*node
;
119 struct lttng_ht_iter iter
;
120 struct relay_index
*index
= NULL
;
122 DBG3("Finding index for stream id %" PRIu64
" and seq_num %" PRIu64
,
123 stream
->stream_handle
, net_seq_num
);
126 lttng_ht_lookup(stream
->indexes_ht
, &net_seq_num
, &iter
);
127 node
= lttng_ht_iter_get_node_u64(&iter
);
129 index
= caa_container_of(node
, struct relay_index
, index_n
);
131 struct relay_index
*oldindex
;
133 index
= relay_index_create(stream
, net_seq_num
);
135 ERR("Cannot create index for stream id %" PRIu64
" and seq_num %" PRIu64
,
136 stream
->stream_handle
, net_seq_num
);
139 oldindex
= relay_index_add_unique(stream
, index
);
141 /* Added concurrently, keep old. */
142 relay_index_put(index
);
144 if (!relay_index_get(index
)) {
148 stream
->indexes_in_flight
++;
149 index
->in_hash_table
= true;
154 DBG2("Index %sfound or created in HT for stream ID %" PRIu64
" and seqnum %" PRIu64
,
155 (index
== NULL
) ? "NOT " : "", stream
->stream_handle
, net_seq_num
);
159 int relay_index_set_file(struct relay_index
*index
,
160 struct lttng_index_file
*index_file
,
161 uint64_t data_offset
)
165 pthread_mutex_lock(&index
->lock
);
166 if (index
->index_file
) {
170 lttng_index_file_get(index_file
);
171 index
->index_file
= index_file
;
172 index
->index_data
.offset
= data_offset
;
174 pthread_mutex_unlock(&index
->lock
);
178 int relay_index_set_data(struct relay_index
*index
,
179 const struct ctf_packet_index
*data
)
183 pthread_mutex_lock(&index
->lock
);
184 if (index
->has_index_data
) {
188 /* Set everything except data_offset. */
189 index
->index_data
.packet_size
= data
->packet_size
;
190 index
->index_data
.content_size
= data
->content_size
;
191 index
->index_data
.timestamp_begin
= data
->timestamp_begin
;
192 index
->index_data
.timestamp_end
= data
->timestamp_end
;
193 index
->index_data
.events_discarded
= data
->events_discarded
;
194 index
->index_data
.stream_id
= data
->stream_id
;
195 index
->has_index_data
= true;
197 pthread_mutex_unlock(&index
->lock
);
201 static void index_destroy(struct relay_index
*index
)
206 static void index_destroy_rcu(struct rcu_head
*rcu_head
)
208 struct relay_index
*index
=
209 caa_container_of(rcu_head
, struct relay_index
, rcu_node
);
211 index_destroy(index
);
214 /* Stream lock must be held by the caller. */
215 static void index_release(struct urcu_ref
*ref
)
217 struct relay_index
*index
= caa_container_of(ref
, struct relay_index
, ref
);
218 struct relay_stream
*stream
= index
->stream
;
220 struct lttng_ht_iter iter
;
222 if (index
->index_file
) {
223 lttng_index_file_put(index
->index_file
);
224 index
->index_file
= NULL
;
226 if (index
->in_hash_table
) {
227 /* Delete index from hash table. */
228 iter
.iter
.node
= &index
->index_n
.node
;
229 ret
= lttng_ht_del(stream
->indexes_ht
, &iter
);
231 stream
->indexes_in_flight
--;
234 stream_put(index
->stream
);
235 index
->stream
= NULL
;
237 call_rcu(&index
->rcu_node
, index_destroy_rcu
);
241 * Called with stream mutex held.
243 * Stream lock must be held by the caller.
245 void relay_index_put(struct relay_index
*index
)
247 DBG2("index put for stream id %" PRIu64
" and seqnum %" PRIu64
" refcount %d",
248 index
->stream
->stream_handle
, index
->index_n
.key
,
249 (int) index
->ref
.refcount
);
251 * Ensure existance of index->lock for index unlock.
255 * Index lock ensures that concurrent test and update of stream
258 assert(index
->ref
.refcount
!= 0);
259 urcu_ref_put(&index
->ref
, index_release
);
264 * Try to flush index to disk. Releases self-reference to index once
267 * Stream lock must be held by the caller.
268 * Return 0 on successful flush, a negative value on error, or positive
269 * value if no flush was performed.
271 int relay_index_try_flush(struct relay_index
*index
)
274 bool flushed
= false;
277 pthread_mutex_lock(&index
->lock
);
278 if (index
->flushed
) {
281 /* Check if we are ready to flush. */
282 if (!index
->has_index_data
|| !index
->index_file
) {
285 fd
= index
->index_file
->fd
;
286 DBG2("Writing index for stream ID %" PRIu64
" and seq num %" PRIu64
287 " on fd %d", index
->stream
->stream_handle
,
288 index
->index_n
.key
, fd
);
290 index
->flushed
= true;
291 ret
= lttng_index_file_write(index
->index_file
, &index
->index_data
);
293 pthread_mutex_unlock(&index
->lock
);
296 /* Put self-ref from index now that it has been flushed. */
297 relay_index_put(index
);
303 * Close every relay index within a given stream, without flushing
306 void relay_index_close_all(struct relay_stream
*stream
)
308 struct lttng_ht_iter iter
;
309 struct relay_index
*index
;
312 cds_lfht_for_each_entry(stream
->indexes_ht
->ht
, &iter
.iter
,
313 index
, index_n
.node
) {
314 /* Put self-ref from index. */
315 relay_index_put(index
);
320 void relay_index_close_partial_fd(struct relay_stream
*stream
)
322 struct lttng_ht_iter iter
;
323 struct relay_index
*index
;
326 cds_lfht_for_each_entry(stream
->indexes_ht
->ht
, &iter
.iter
,
327 index
, index_n
.node
) {
328 if (!index
->index_file
) {
332 * Partial index has its index_file: we have only
333 * received its info from the data socket.
334 * Put self-ref from index.
336 relay_index_put(index
);
341 uint64_t relay_index_find_last(struct relay_stream
*stream
)
343 struct lttng_ht_iter iter
;
344 struct relay_index
*index
;
345 uint64_t net_seq_num
= -1ULL;
348 cds_lfht_for_each_entry(stream
->indexes_ht
->ht
, &iter
.iter
,
349 index
, index_n
.node
) {
350 if (net_seq_num
== -1ULL ||
351 index
->index_n
.key
> net_seq_num
) {
352 net_seq_num
= index
->index_n
.key
;
360 * Update the index file of an already existing relay_index.
361 * Offsets by 'removed_data_count' the offset field of an index.
364 int relay_index_switch_file(struct relay_index
*index
,
365 struct lttng_index_file
*new_index_file
,
366 uint64_t removed_data_count
)
371 pthread_mutex_lock(&index
->lock
);
372 if (!index
->index_file
) {
373 ERR("No index_file");
378 lttng_index_file_put(index
->index_file
);
379 lttng_index_file_get(new_index_file
);
380 index
->index_file
= new_index_file
;
381 offset
= be64toh(index
->index_data
.offset
);
382 index
->index_data
.offset
= htobe64(offset
- removed_data_count
);
385 pthread_mutex_unlock(&index
->lock
);
390 * Switch the index file of all pending indexes for a stream and update the
391 * data offset by substracting the last safe position.
392 * Stream lock must be held.
394 int relay_index_switch_all_files(struct relay_stream
*stream
)
396 struct lttng_ht_iter iter
;
397 struct relay_index
*index
;
401 cds_lfht_for_each_entry(stream
->indexes_ht
->ht
, &iter
.iter
,
402 index
, index_n
.node
) {
403 DBG("Update index to fd %d", stream
->index_file
->fd
);
404 ret
= relay_index_switch_file(index
, stream
->index_file
,
405 stream
->pos_after_last_complete_data_index
);