Fix: relayd: failure to read index entry or stream packet after clear
[lttng-tools.git] / src / bin / lttng-relayd / viewer-stream.c
CommitLineData
2f8f53af 1/*
ab5be9fa
MJ
2 * Copyright (C) 2013 Julien Desfossez <jdesfossez@efficios.com>
3 * Copyright (C) 2013 David Goulet <dgoulet@efficios.com>
4 * Copyright (C) 2015 Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
2f8f53af 5 *
ab5be9fa 6 * SPDX-License-Identifier: GPL-2.0-only
2f8f53af 7 *
2f8f53af
DG
8 */
9
6c1c0768 10#define _LGPL_SOURCE
2f8f53af
DG
11#include <common/common.h>
12#include <common/index/index.h>
f5436bfc 13#include <common/compat/string.h>
b0d240a2
MD
14#include <common/utils.h>
15#include <sys/types.h>
16#include <sys/stat.h>
17#include <fcntl.h>
2f8f53af
DG
18
19#include "lttng-relayd.h"
20#include "viewer-stream.h"
21
7591bab1 22static void viewer_stream_destroy(struct relay_viewer_stream *vstream)
2f8f53af 23{
80516611 24 lttng_trace_chunk_put(vstream->stream_file.trace_chunk);
7591bab1
MD
25 free(vstream->path_name);
26 free(vstream->channel_name);
27 free(vstream);
2f8f53af
DG
28}
29
7591bab1 30static void viewer_stream_destroy_rcu(struct rcu_head *head)
2f8f53af 31{
7591bab1 32 struct relay_viewer_stream *vstream =
2f8f53af
DG
33 caa_container_of(head, struct relay_viewer_stream, rcu_node);
34
7591bab1 35 viewer_stream_destroy(vstream);
2f8f53af
DG
36}
37
9edaf114 38/* Relay stream's lock must be held by the caller. */
2f8f53af 39struct relay_viewer_stream *viewer_stream_create(struct relay_stream *stream,
9edaf114 40 struct lttng_trace_chunk *trace_chunk,
7591bab1 41 enum lttng_viewer_seek seek_t)
2f8f53af 42{
ebb29c10 43 struct relay_viewer_stream *vstream = NULL;
ebb29c10 44
9edaf114 45 ASSERT_LOCKED(stream->lock);
2f8f53af 46
2f8f53af
DG
47 vstream = zmalloc(sizeof(*vstream));
48 if (!vstream) {
49 PERROR("relay viewer stream zmalloc");
50 goto error;
51 }
52
80516611
JG
53 if (trace_chunk) {
54 const bool acquired_reference = lttng_trace_chunk_get(
55 trace_chunk);
56
57 assert(acquired_reference);
58 }
59
9edaf114 60 vstream->stream_file.trace_chunk = trace_chunk;
f5436bfc 61 vstream->path_name = lttng_strndup(stream->path_name, LTTNG_VIEWER_PATH_MAX);
b272577e
MD
62 if (vstream->path_name == NULL) {
63 PERROR("relay viewer path_name alloc");
64 goto error;
65 }
f5436bfc 66 vstream->channel_name = lttng_strndup(stream->channel_name,
2f8f53af 67 LTTNG_VIEWER_NAME_MAX);
b272577e
MD
68 if (vstream->channel_name == NULL) {
69 PERROR("relay viewer channel_name alloc");
70 goto error;
71 }
2f8f53af 72
a44ca2ca
MD
73 if (!stream_get(stream)) {
74 ERR("Cannot get stream");
75 goto error;
76 }
77 vstream->stream = stream;
78
a44ca2ca
MD
79 if (stream->is_metadata && stream->trace->viewer_metadata_stream) {
80 ERR("Cannot attach viewer metadata stream to trace (busy).");
71381736 81 goto error;
a44ca2ca
MD
82 }
83
2f8f53af 84 switch (seek_t) {
c4e361a4 85 case LTTNG_VIEWER_SEEK_BEGINNING:
a44ca2ca
MD
86 {
87 uint64_t seq_tail = tracefile_array_get_seq_tail(stream->tfa);
88
89 if (seq_tail == -1ULL) {
90 /*
91 * Tail may not be initialized yet. Nonetheless, we know
92 * we want to send the first index once it becomes
93 * available.
94 */
95 seq_tail = 0;
96 }
97 vstream->current_tracefile_id =
98 tracefile_array_get_file_index_tail(stream->tfa);
99 vstream->index_sent_seqcount = seq_tail;
2f8f53af 100 break;
a44ca2ca 101 }
c4e361a4 102 case LTTNG_VIEWER_SEEK_LAST:
a44ca2ca 103 vstream->current_tracefile_id =
78118e3b 104 tracefile_array_get_read_file_index_head(stream->tfa);
a44ca2ca
MD
105 /*
106 * We seek at the very end of each stream, awaiting for
107 * a future packet to eventually come in.
108 *
109 * We don't need to check the head position for -1ULL since the
110 * increment will set it to 0.
111 */
112 vstream->index_sent_seqcount =
113 tracefile_array_get_seq_head(stream->tfa) + 1;
2f8f53af
DG
114 break;
115 default:
71381736 116 goto error;
2f8f53af 117 }
2f8f53af 118
2f8f53af 119 /*
a44ca2ca
MD
120 * If we never received an index for the current stream, delay
121 * the opening of the index, otherwise open it right now.
2f8f53af 122 */
b0d240a2 123 if (stream->index_file == NULL) {
f8f3885c 124 vstream->index_file = NULL;
2f8f53af 125 } else {
ebb29c10
JG
126 const uint32_t connection_major = stream->trace->session->major;
127 const uint32_t connection_minor = stream->trace->session->minor;
3ff5c5db 128 enum lttng_trace_chunk_status chunk_status;
ebb29c10 129
3ff5c5db 130 chunk_status = lttng_index_file_create_from_trace_chunk_read_only(
b66a15d1
JG
131 vstream->stream_file.trace_chunk,
132 stream->path_name,
ebb29c10
JG
133 stream->channel_name, stream->tracefile_size,
134 vstream->current_tracefile_id,
135 lttng_to_index_major(connection_major,
136 connection_minor),
137 lttng_to_index_minor(connection_major,
3ff5c5db
MD
138 connection_minor),
139 true, &vstream->index_file);
140 if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
141 if (chunk_status == LTTNG_TRACE_CHUNK_STATUS_NO_FILE) {
142 vstream->index_file = NULL;
143 } else {
71381736 144 goto error;
3ff5c5db 145 }
2f8f53af 146 }
2f8f53af
DG
147 }
148
b0d240a2
MD
149 /*
150 * If we never received a data file for the current stream, delay the
151 * opening, otherwise open it right now.
152 */
8bb66c3c
JG
153 if (stream->file) {
154 int ret;
b0d240a2
MD
155 char file_path[LTTNG_PATH_MAX];
156 enum lttng_trace_chunk_status status;
157
158 ret = utils_stream_file_path(stream->path_name,
159 stream->channel_name, stream->tracefile_size,
160 vstream->current_tracefile_id, NULL, file_path,
161 sizeof(file_path));
162 if (ret < 0) {
71381736 163 goto error;
b0d240a2
MD
164 }
165
8bb66c3c
JG
166 status = lttng_trace_chunk_open_fs_handle(
167 vstream->stream_file.trace_chunk, file_path,
168 O_RDONLY, 0, &vstream->stream_file.handle,
169 true);
b0d240a2 170 if (status != LTTNG_TRACE_CHUNK_STATUS_OK) {
71381736 171 goto error;
b0d240a2 172 }
b0d240a2
MD
173 }
174
f8f3885c 175 if (seek_t == LTTNG_VIEWER_SEEK_LAST && vstream->index_file) {
2f8f53af
DG
176 off_t lseek_ret;
177
8bb66c3c
JG
178 lseek_ret = fs_handle_seek(
179 vstream->index_file->file, 0, SEEK_END);
2f8f53af 180 if (lseek_ret < 0) {
71381736 181 goto error;
2f8f53af 182 }
7591bab1 183 }
7591bab1
MD
184 if (stream->is_metadata) {
185 rcu_assign_pointer(stream->trace->viewer_metadata_stream,
186 vstream);
2f8f53af
DG
187 }
188
80516611
JG
189 vstream->last_seen_rotation_count = stream->completed_rotation_count;
190
7591bab1
MD
191 /* Globally visible after the add unique. */
192 lttng_ht_node_init_u64(&vstream->stream_n, stream->stream_handle);
7591bab1 193 urcu_ref_init(&vstream->ref);
c51b2177 194 lttng_ht_add_unique_u64(viewer_streams_ht, &vstream->stream_n);
7591bab1 195
2f8f53af
DG
196 return vstream;
197
198error:
199 if (vstream) {
7591bab1 200 viewer_stream_destroy(vstream);
2f8f53af
DG
201 }
202 return NULL;
203}
204
7591bab1 205static void viewer_stream_unpublish(struct relay_viewer_stream *vstream)
2f8f53af
DG
206{
207 int ret;
208 struct lttng_ht_iter iter;
209
7591bab1 210 iter.iter.node = &vstream->stream_n.node;
2f8f53af
DG
211 ret = lttng_ht_del(viewer_streams_ht, &iter);
212 assert(!ret);
213}
214
7591bab1 215static void viewer_stream_release(struct urcu_ref *ref)
2f8f53af 216{
7591bab1
MD
217 struct relay_viewer_stream *vstream = caa_container_of(ref,
218 struct relay_viewer_stream, ref);
2a174661 219
7591bab1
MD
220 if (vstream->stream->is_metadata) {
221 rcu_assign_pointer(vstream->stream->trace->viewer_metadata_stream, NULL);
2a174661 222 }
2f8f53af 223
7591bab1
MD
224 viewer_stream_unpublish(vstream);
225
8bb66c3c
JG
226 if (vstream->stream_file.handle) {
227 fs_handle_close(vstream->stream_file.handle);
228 vstream->stream_file.handle = NULL;
2f8f53af 229 }
f8f3885c
MD
230 if (vstream->index_file) {
231 lttng_index_file_put(vstream->index_file);
232 vstream->index_file = NULL;
7591bab1
MD
233 }
234 if (vstream->stream) {
235 stream_put(vstream->stream);
236 vstream->stream = NULL;
237 }
4c2717fc
JG
238 lttng_trace_chunk_put(vstream->stream_file.trace_chunk);
239 vstream->stream_file.trace_chunk = NULL;
7591bab1
MD
240 call_rcu(&vstream->rcu_node, viewer_stream_destroy_rcu);
241}
242
243/* Must be called with RCU read-side lock held. */
244bool viewer_stream_get(struct relay_viewer_stream *vstream)
245{
ce4d4083 246 return urcu_ref_get_unless_zero(&vstream->ref);
2f8f53af
DG
247}
248
249/*
7591bab1 250 * Get viewer stream by id.
2f8f53af 251 *
7591bab1 252 * Return viewer stream if found else NULL.
2f8f53af 253 */
7591bab1 254struct relay_viewer_stream *viewer_stream_get_by_id(uint64_t id)
2f8f53af
DG
255{
256 struct lttng_ht_node_u64 *node;
257 struct lttng_ht_iter iter;
7591bab1 258 struct relay_viewer_stream *vstream = NULL;
2f8f53af 259
7591bab1 260 rcu_read_lock();
2f8f53af
DG
261 lttng_ht_lookup(viewer_streams_ht, &id, &iter);
262 node = lttng_ht_iter_get_node_u64(&iter);
263 if (!node) {
264 DBG("Relay viewer stream %" PRIu64 " not found", id);
265 goto end;
266 }
7591bab1
MD
267 vstream = caa_container_of(node, struct relay_viewer_stream, stream_n);
268 if (!viewer_stream_get(vstream)) {
269 vstream = NULL;
270 }
2f8f53af 271end:
7591bab1
MD
272 rcu_read_unlock();
273 return vstream;
274}
275
276void viewer_stream_put(struct relay_viewer_stream *vstream)
277{
278 rcu_read_lock();
7591bab1 279 urcu_ref_put(&vstream->ref, viewer_stream_release);
7591bab1
MD
280 rcu_read_unlock();
281}
282
3087b021
MD
283void viewer_stream_close_files(struct relay_viewer_stream *vstream)
284{
285 if (vstream->index_file) {
286 lttng_index_file_put(vstream->index_file);
287 vstream->index_file = NULL;
288 }
8bb66c3c
JG
289 if (vstream->stream_file.handle) {
290 fs_handle_close(vstream->stream_file.handle);
291 vstream->stream_file.handle = NULL;
3087b021
MD
292 }
293}
294
295void viewer_stream_sync_tracefile_array_tail(struct relay_viewer_stream *vstream)
296{
297 const struct relay_stream *stream = vstream->stream;
298 uint64_t seq_tail;
299
300 vstream->current_tracefile_id = tracefile_array_get_file_index_tail(stream->tfa);
301 seq_tail = tracefile_array_get_seq_tail(stream->tfa);
302 if (seq_tail == -1ULL) {
303 seq_tail = 0;
304 }
305 vstream->index_sent_seqcount = seq_tail;
306}
307
2f8f53af
DG
308/*
309 * Rotate a stream to the next tracefile.
310 *
7591bab1 311 * Must be called with the rstream lock held.
b0d240a2 312 * Returns 0 on success, 1 on EOF.
2f8f53af 313 */
7591bab1 314int viewer_stream_rotate(struct relay_viewer_stream *vstream)
2f8f53af
DG
315{
316 int ret;
a44ca2ca 317 uint64_t new_id;
ebb29c10 318 const struct relay_stream *stream = vstream->stream;
2f8f53af 319
7591bab1 320 /* Detect the last tracefile to open. */
a44ca2ca
MD
321 if (stream->index_received_seqcount
322 == vstream->index_sent_seqcount
7591bab1
MD
323 && stream->trace->session->connection_closed) {
324 ret = 1;
2a174661
DG
325 goto end;
326 }
2f8f53af 327
7591bab1
MD
328 if (stream->tracefile_count == 0) {
329 /* Ignore rotation, there is none to do. */
330 ret = 0;
2f8f53af
DG
331 goto end;
332 }
333
a44ca2ca
MD
334 /*
335 * Try to move to the next file.
336 */
337 new_id = (vstream->current_tracefile_id + 1)
338 % stream->tracefile_count;
339 if (tracefile_array_seq_in_file(stream->tfa, new_id,
340 vstream->index_sent_seqcount)) {
341 vstream->current_tracefile_id = new_id;
2f8f53af 342 } else {
a44ca2ca
MD
343 uint64_t seq_tail = tracefile_array_get_seq_tail(stream->tfa);
344
345 /*
346 * This can only be reached on overwrite, which implies there
347 * has been data written at some point, which will have set the
348 * tail.
349 */
350 assert(seq_tail != -1ULL);
351 /*
352 * We need to resync because we lag behind tail.
353 */
7591bab1 354 vstream->current_tracefile_id =
a44ca2ca
MD
355 tracefile_array_get_file_index_tail(stream->tfa);
356 vstream->index_sent_seqcount = seq_tail;
2f8f53af 357 }
b0d240a2
MD
358 viewer_stream_close_files(vstream);
359 ret = 0;
2f8f53af 360end:
2f8f53af
DG
361 return ret;
362}
7591bab1
MD
363
364void print_viewer_streams(void)
365{
366 struct lttng_ht_iter iter;
367 struct relay_viewer_stream *vstream;
368
ce3f3ba3
JG
369 if (!viewer_streams_ht) {
370 return;
371 }
372
7591bab1
MD
373 rcu_read_lock();
374 cds_lfht_for_each_entry(viewer_streams_ht->ht, &iter.iter, vstream,
375 stream_n.node) {
376 if (!viewer_stream_get(vstream)) {
377 continue;
378 }
379 DBG("vstream %p refcount %ld stream %" PRIu64 " trace %" PRIu64
380 " session %" PRIu64,
381 vstream,
382 vstream->ref.refcount,
383 vstream->stream->stream_handle,
384 vstream->stream->trace->id,
385 vstream->stream->trace->session->id);
386 viewer_stream_put(vstream);
387 }
388 rcu_read_unlock();
389}
This page took 0.099242 seconds and 4 git commands to generate.