Fix: dereferencing null index pointer
[lttng-tools.git] / src / bin / lttng-relayd / index.c
CommitLineData
1c20f0e2
JD
1/*
2 * Copyright (C) 2013 - Julien Desfossez <jdesfossez@efficios.com>
3 * David Goulet <dgoulet@efficios.com>
7591bab1 4 * 2015 - Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
1c20f0e2
JD
5 *
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.
9 *
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
13 * more details.
14 *
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.
18 */
19
20#define _GNU_SOURCE
6c1c0768 21#define _LGPL_SOURCE
1c20f0e2
JD
22#include <assert.h>
23
24#include <common/common.h>
25#include <common/utils.h>
26
0a6518b0 27#include "lttng-relayd.h"
7591bab1 28#include "stream.h"
1c20f0e2
JD
29#include "index.h"
30
31/*
7591bab1
MD
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
34 * table key.
35 *
36 * Called with stream mutex held.
37 * Return allocated object or else NULL on error.
1c20f0e2 38 */
7591bab1
MD
39static struct relay_index *relay_index_create(struct relay_stream *stream,
40 uint64_t net_seq_num)
1c20f0e2 41{
7591bab1 42 struct relay_index *index;
1c20f0e2 43
7591bab1
MD
44 DBG2("Creating relay index for stream id %" PRIu64 " and seqnum %" PRIu64,
45 stream->stream_handle, net_seq_num);
1c20f0e2 46
7591bab1
MD
47 index = zmalloc(sizeof(*index));
48 if (!index) {
49 PERROR("Relay index zmalloc");
50 goto end;
51 }
52 if (!stream_get(stream)) {
53 ERR("Cannot get stream");
54 free(index);
55 index = NULL;
56 goto end;
1c20f0e2 57 }
7591bab1 58 index->stream = stream;
1c20f0e2 59
7591bab1
MD
60 lttng_ht_node_init_u64(&index->index_n, net_seq_num);
61 pthread_mutex_init(&index->lock, NULL);
62 pthread_mutex_init(&index->reflock, NULL);
63 urcu_ref_init(&index->ref);
64
65end:
66 return index;
1c20f0e2
JD
67}
68
69/*
7591bab1
MD
70 * Add unique relay index to the given hash table. In case of a collision, the
71 * already existing object is put in the given _index variable.
1c20f0e2 72 *
7591bab1 73 * RCU read side lock MUST be acquired.
1c20f0e2 74 */
7591bab1
MD
75static struct relay_index *relay_index_add_unique(struct relay_stream *stream,
76 struct relay_index *index)
1c20f0e2 77{
7591bab1
MD
78 struct cds_lfht_node *node_ptr;
79 struct relay_index *_index;
1c20f0e2 80
7591bab1
MD
81 DBG2("Adding relay index with stream id %" PRIu64 " and seqnum %" PRIu64,
82 stream->stream_handle, index->index_n.key);
1c20f0e2 83
7591bab1
MD
84 node_ptr = cds_lfht_add_unique(stream->indexes_ht->ht,
85 stream->indexes_ht->hash_fct(&index->index_n, lttng_ht_seed),
86 stream->indexes_ht->match_fct, &index->index_n,
87 &index->index_n.node);
88 if (node_ptr != &index->index_n.node) {
89 _index = caa_container_of(node_ptr, struct relay_index,
90 index_n.node);
91 } else {
92 _index = NULL;
1c20f0e2 93 }
7591bab1
MD
94 return _index;
95}
96
97/*
98 * Should be called with RCU read-side lock held.
99 */
100static bool relay_index_get(struct relay_index *index)
101{
102 bool has_ref = false;
1c20f0e2 103
7591bab1
MD
104 DBG2("index get for stream id %" PRIu64 " and seqnum %" PRIu64 " refcount %d",
105 index->stream->stream_handle, index->index_n.key,
106 (int) index->ref.refcount);
1c20f0e2 107
7591bab1
MD
108 /* Confirm that the index refcount has not reached 0. */
109 pthread_mutex_lock(&index->reflock);
110 if (index->ref.refcount != 0) {
111 has_ref = true;
112 urcu_ref_get(&index->ref);
113 }
114 pthread_mutex_unlock(&index->reflock);
115
116 return has_ref;
1c20f0e2
JD
117}
118
119/*
7591bab1
MD
120 * Get a relayd index in within the given stream, or create it if not
121 * present.
1c20f0e2 122 *
7591bab1 123 * Called with stream mutex held.
1c20f0e2
JD
124 * Return index object or else NULL on error.
125 */
7591bab1
MD
126struct relay_index *relay_index_get_by_id_or_create(struct relay_stream *stream,
127 uint64_t net_seq_num)
1c20f0e2 128{
7591bab1 129 struct lttng_ht_node_u64 *node;
1c20f0e2 130 struct lttng_ht_iter iter;
1c20f0e2
JD
131 struct relay_index *index = NULL;
132
1c20f0e2 133 DBG3("Finding index for stream id %" PRIu64 " and seq_num %" PRIu64,
7591bab1 134 stream->stream_handle, net_seq_num);
1c20f0e2 135
7591bab1
MD
136 rcu_read_lock();
137 lttng_ht_lookup(stream->indexes_ht, &net_seq_num, &iter);
138 node = lttng_ht_iter_get_node_u64(&iter);
139 if (node) {
140 index = caa_container_of(node, struct relay_index, index_n);
141 } else {
142 struct relay_index *oldindex;
143
144 index = relay_index_create(stream, net_seq_num);
145 if (!index) {
146 ERR("Cannot create index for stream id %" PRIu64 " and seq_num %" PRIu64,
811a4037 147 stream->stream_handle, net_seq_num);
7591bab1
MD
148 goto end;
149 }
150 oldindex = relay_index_add_unique(stream, index);
151 if (oldindex) {
152 /* Added concurrently, keep old. */
153 relay_index_put(index);
154 index = oldindex;
155 if (!relay_index_get(index)) {
156 index = NULL;
157 }
158 } else {
159 stream->indexes_in_flight++;
160 index->in_hash_table = true;
161 }
1c20f0e2 162 }
1c20f0e2 163end:
7591bab1
MD
164 rcu_read_unlock();
165 DBG2("Index %sfound or created in HT for stream ID %" PRIu64 " and seqnum %" PRIu64,
166 (index == NULL) ? "NOT " : "", index->stream->stream_handle, net_seq_num);
1c20f0e2
JD
167 return index;
168}
169
7591bab1
MD
170int relay_index_set_fd(struct relay_index *index, struct stream_fd *index_fd,
171 uint64_t data_offset)
1c20f0e2 172{
7591bab1 173 int ret = 0;
1c20f0e2 174
7591bab1
MD
175 pthread_mutex_lock(&index->lock);
176 if (index->index_fd) {
177 ret = -1;
178 goto end;
179 }
180 stream_fd_get(index_fd);
181 index->index_fd = index_fd;
182 index->index_data.offset = data_offset;
183end:
184 pthread_mutex_unlock(&index->lock);
185 return ret;
186}
1c20f0e2 187
7591bab1
MD
188int relay_index_set_data(struct relay_index *index,
189 const struct ctf_packet_index *data)
190{
191 int ret = 0;
1c20f0e2 192
7591bab1
MD
193 pthread_mutex_lock(&index->lock);
194 if (index->has_index_data) {
195 ret = -1;
196 goto end;
1c20f0e2 197 }
7591bab1
MD
198 /* Set everything except data_offset. */
199 index->index_data.packet_size = data->packet_size;
200 index->index_data.content_size = data->content_size;
201 index->index_data.timestamp_begin = data->timestamp_begin;
202 index->index_data.timestamp_end = data->timestamp_end;
203 index->index_data.events_discarded = data->events_discarded;
204 index->index_data.stream_id = data->stream_id;
205 index->has_index_data = true;
206end:
207 pthread_mutex_unlock(&index->lock);
208 return ret;
1c20f0e2
JD
209}
210
7591bab1 211static void index_destroy(struct relay_index *index)
1c20f0e2 212{
7591bab1
MD
213 free(index);
214}
1c20f0e2 215
7591bab1
MD
216static void index_destroy_rcu(struct rcu_head *rcu_head)
217{
218 struct relay_index *index =
219 caa_container_of(rcu_head, struct relay_index, rcu_node);
1c20f0e2 220
7591bab1 221 index_destroy(index);
1c20f0e2
JD
222}
223
7591bab1
MD
224/* Stream lock must be held by the caller. */
225static void index_release(struct urcu_ref *ref)
1c20f0e2 226{
7591bab1
MD
227 struct relay_index *index = caa_container_of(ref, struct relay_index, ref);
228 struct relay_stream *stream = index->stream;
229 int ret;
230 struct lttng_ht_iter iter;
231
232 if (index->index_fd) {
233 stream_fd_put(index->index_fd);
234 index->index_fd = NULL;
235 }
236 if (index->in_hash_table) {
237 /* Delete index from hash table. */
238 iter.iter.node = &index->index_n.node;
239 ret = lttng_ht_del(stream->indexes_ht, &iter);
240 assert(!ret);
241 stream->indexes_in_flight--;
242 }
243
244 stream_put(index->stream);
245 index->stream = NULL;
246
247 call_rcu(&index->rcu_node, index_destroy_rcu);
1c20f0e2
JD
248}
249
250/*
7591bab1
MD
251 * Called with stream mutex held.
252 *
253 * Stream lock must be held by the caller.
1c20f0e2 254 */
7591bab1 255void relay_index_put(struct relay_index *index)
1c20f0e2 256{
7591bab1
MD
257 DBG2("index put for stream id %" PRIu64 " and seqnum %" PRIu64 " refcount %d",
258 index->stream->stream_handle, index->index_n.key,
259 (int) index->ref.refcount);
260 /*
261 * Ensure existance of index->lock for index unlock.
262 */
263 rcu_read_lock();
264 /*
265 * Index lock ensures that concurrent test and update of stream
266 * ref is atomic.
267 */
268 pthread_mutex_lock(&index->reflock);
269 assert(index->ref.refcount != 0);
270 urcu_ref_put(&index->ref, index_release);
271 pthread_mutex_unlock(&index->reflock);
272 rcu_read_unlock();
1c20f0e2
JD
273}
274
275/*
7591bab1
MD
276 * Try to flush index to disk. Releases self-reference to index once
277 * flush succeeds.
1c20f0e2 278 *
7591bab1
MD
279 * Stream lock must be held by the caller.
280 * Return 0 on successful flush, a negative value on error, or positive
281 * value if no flush was performed.
1c20f0e2 282 */
7591bab1 283int relay_index_try_flush(struct relay_index *index)
1c20f0e2 284{
7591bab1
MD
285 int ret = 1;
286 bool flushed = false;
287 int fd;
1c20f0e2 288
7591bab1
MD
289 pthread_mutex_lock(&index->lock);
290 if (index->flushed) {
291 goto skip;
292 }
293 /* Check if we are ready to flush. */
294 if (!index->has_index_data || !index->index_fd) {
295 goto skip;
296 }
297 fd = index->index_fd->fd;
298 DBG2("Writing index for stream ID %" PRIu64 " and seq num %" PRIu64
299 " on fd %d", index->stream->stream_handle,
300 index->index_n.key, fd);
301 flushed = true;
302 index->flushed = true;
303 ret = index_write(fd, &index->index_data, sizeof(index->index_data));
304 if (ret == sizeof(index->index_data)) {
305 ret = 0;
306 } else {
307 ret = -1;
308 }
309skip:
310 pthread_mutex_unlock(&index->lock);
1c20f0e2 311
7591bab1
MD
312 if (flushed) {
313 /* Put self-ref from index now that it has been flushed. */
314 relay_index_put(index);
315 }
316 return ret;
1c20f0e2
JD
317}
318
319/*
7591bab1
MD
320 * Close every relay index within a given stream, without flushing
321 * them.
1c20f0e2 322 */
7591bab1 323void relay_index_close_all(struct relay_stream *stream)
1c20f0e2
JD
324{
325 struct lttng_ht_iter iter;
326 struct relay_index *index;
327
1c20f0e2 328 rcu_read_lock();
7591bab1
MD
329 cds_lfht_for_each_entry(stream->indexes_ht->ht, &iter.iter,
330 index, index_n.node) {
331 /* Put self-ref from index. */
332 relay_index_put(index);
1c20f0e2
JD
333 }
334 rcu_read_unlock();
335}
3d07a857
MD
336
337void relay_index_close_partial_fd(struct relay_stream *stream)
338{
339 struct lttng_ht_iter iter;
340 struct relay_index *index;
341
342 rcu_read_lock();
343 cds_lfht_for_each_entry(stream->indexes_ht->ht, &iter.iter,
344 index, index_n.node) {
345 if (!index->index_fd) {
346 continue;
347 }
348 /*
349 * Partial index has its index_fd: we have only
350 * received its info from the data socket.
351 * Put self-ref from index.
352 */
353 relay_index_put(index);
354 }
355 rcu_read_unlock();
356}
357
358uint64_t relay_index_find_last(struct relay_stream *stream)
359{
360 struct lttng_ht_iter iter;
361 struct relay_index *index;
362 uint64_t net_seq_num = -1ULL;
363
364 rcu_read_lock();
365 cds_lfht_for_each_entry(stream->indexes_ht->ht, &iter.iter,
366 index, index_n.node) {
367 if (net_seq_num == -1ULL ||
368 index->index_n.key > net_seq_num) {
369 net_seq_num = index->index_n.key;
370 }
371 }
372 rcu_read_unlock();
373 return net_seq_num;
374}
This page took 0.071973 seconds and 4 git commands to generate.