Fix: relayd: live connection fails to open file during 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{
7591bab1
MD
24 free(vstream->path_name);
25 free(vstream->channel_name);
26 free(vstream);
2f8f53af
DG
27}
28
7591bab1 29static void viewer_stream_destroy_rcu(struct rcu_head *head)
2f8f53af 30{
7591bab1 31 struct relay_viewer_stream *vstream =
2f8f53af
DG
32 caa_container_of(head, struct relay_viewer_stream, rcu_node);
33
7591bab1 34 viewer_stream_destroy(vstream);
2f8f53af
DG
35}
36
9edaf114 37/* Relay stream's lock must be held by the caller. */
2f8f53af 38struct relay_viewer_stream *viewer_stream_create(struct relay_stream *stream,
9edaf114 39 struct lttng_trace_chunk *trace_chunk,
7591bab1 40 enum lttng_viewer_seek seek_t)
2f8f53af 41{
ebb29c10 42 struct relay_viewer_stream *vstream = NULL;
9edaf114 43 const bool acquired_reference = lttng_trace_chunk_get(trace_chunk);
ebb29c10 44
9edaf114 45 ASSERT_LOCKED(stream->lock);
ebb29c10
JG
46 if (!acquired_reference) {
47 goto error;
48 }
2f8f53af 49
2f8f53af
DG
50 vstream = zmalloc(sizeof(*vstream));
51 if (!vstream) {
52 PERROR("relay viewer stream zmalloc");
53 goto error;
54 }
55
9edaf114
JG
56 vstream->stream_file.trace_chunk = trace_chunk;
57 trace_chunk = NULL;
f5436bfc 58 vstream->path_name = lttng_strndup(stream->path_name, LTTNG_VIEWER_PATH_MAX);
b272577e
MD
59 if (vstream->path_name == NULL) {
60 PERROR("relay viewer path_name alloc");
61 goto error;
62 }
f5436bfc 63 vstream->channel_name = lttng_strndup(stream->channel_name,
2f8f53af 64 LTTNG_VIEWER_NAME_MAX);
b272577e
MD
65 if (vstream->channel_name == NULL) {
66 PERROR("relay viewer channel_name alloc");
67 goto error;
68 }
2f8f53af 69
a44ca2ca
MD
70 if (!stream_get(stream)) {
71 ERR("Cannot get stream");
72 goto error;
73 }
74 vstream->stream = stream;
75
a44ca2ca
MD
76 if (stream->is_metadata && stream->trace->viewer_metadata_stream) {
77 ERR("Cannot attach viewer metadata stream to trace (busy).");
78 goto error_unlock;
79 }
80
2f8f53af 81 switch (seek_t) {
c4e361a4 82 case LTTNG_VIEWER_SEEK_BEGINNING:
a44ca2ca
MD
83 {
84 uint64_t seq_tail = tracefile_array_get_seq_tail(stream->tfa);
85
86 if (seq_tail == -1ULL) {
87 /*
88 * Tail may not be initialized yet. Nonetheless, we know
89 * we want to send the first index once it becomes
90 * available.
91 */
92 seq_tail = 0;
93 }
94 vstream->current_tracefile_id =
95 tracefile_array_get_file_index_tail(stream->tfa);
96 vstream->index_sent_seqcount = seq_tail;
2f8f53af 97 break;
a44ca2ca 98 }
c4e361a4 99 case LTTNG_VIEWER_SEEK_LAST:
a44ca2ca 100 vstream->current_tracefile_id =
78118e3b 101 tracefile_array_get_read_file_index_head(stream->tfa);
a44ca2ca
MD
102 /*
103 * We seek at the very end of each stream, awaiting for
104 * a future packet to eventually come in.
105 *
106 * We don't need to check the head position for -1ULL since the
107 * increment will set it to 0.
108 */
109 vstream->index_sent_seqcount =
110 tracefile_array_get_seq_head(stream->tfa) + 1;
2f8f53af
DG
111 break;
112 default:
a44ca2ca 113 goto error_unlock;
2f8f53af 114 }
2f8f53af 115
2f8f53af 116 /*
a44ca2ca
MD
117 * If we never received an index for the current stream, delay
118 * the opening of the index, otherwise open it right now.
2f8f53af 119 */
b0d240a2 120 if (stream->index_file == NULL) {
f8f3885c 121 vstream->index_file = NULL;
2f8f53af 122 } else {
ebb29c10
JG
123 const uint32_t connection_major = stream->trace->session->major;
124 const uint32_t connection_minor = stream->trace->session->minor;
3ff5c5db 125 enum lttng_trace_chunk_status chunk_status;
ebb29c10 126
3ff5c5db 127 chunk_status = lttng_index_file_create_from_trace_chunk_read_only(
b66a15d1
JG
128 vstream->stream_file.trace_chunk,
129 stream->path_name,
ebb29c10
JG
130 stream->channel_name, stream->tracefile_size,
131 vstream->current_tracefile_id,
132 lttng_to_index_major(connection_major,
133 connection_minor),
134 lttng_to_index_minor(connection_major,
3ff5c5db
MD
135 connection_minor),
136 true, &vstream->index_file);
137 if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
138 if (chunk_status == LTTNG_TRACE_CHUNK_STATUS_NO_FILE) {
139 vstream->index_file = NULL;
140 } else {
141 goto error_unlock;
142 }
2f8f53af 143 }
2f8f53af
DG
144 }
145
b0d240a2
MD
146 /*
147 * If we never received a data file for the current stream, delay the
148 * opening, otherwise open it right now.
149 */
8bb66c3c
JG
150 if (stream->file) {
151 int ret;
b0d240a2
MD
152 char file_path[LTTNG_PATH_MAX];
153 enum lttng_trace_chunk_status status;
154
155 ret = utils_stream_file_path(stream->path_name,
156 stream->channel_name, stream->tracefile_size,
157 vstream->current_tracefile_id, NULL, file_path,
158 sizeof(file_path));
159 if (ret < 0) {
160 goto error_unlock;
161 }
162
8bb66c3c
JG
163 status = lttng_trace_chunk_open_fs_handle(
164 vstream->stream_file.trace_chunk, file_path,
165 O_RDONLY, 0, &vstream->stream_file.handle,
166 true);
b0d240a2
MD
167 if (status != LTTNG_TRACE_CHUNK_STATUS_OK) {
168 goto error_unlock;
169 }
b0d240a2
MD
170 }
171
f8f3885c 172 if (seek_t == LTTNG_VIEWER_SEEK_LAST && vstream->index_file) {
2f8f53af
DG
173 off_t lseek_ret;
174
8bb66c3c
JG
175 lseek_ret = fs_handle_seek(
176 vstream->index_file->file, 0, SEEK_END);
2f8f53af 177 if (lseek_ret < 0) {
7591bab1 178 goto error_unlock;
2f8f53af 179 }
7591bab1 180 }
7591bab1
MD
181 if (stream->is_metadata) {
182 rcu_assign_pointer(stream->trace->viewer_metadata_stream,
183 vstream);
2f8f53af
DG
184 }
185
7591bab1
MD
186 /* Globally visible after the add unique. */
187 lttng_ht_node_init_u64(&vstream->stream_n, stream->stream_handle);
7591bab1 188 urcu_ref_init(&vstream->ref);
c51b2177 189 lttng_ht_add_unique_u64(viewer_streams_ht, &vstream->stream_n);
7591bab1 190
2f8f53af
DG
191 return vstream;
192
7591bab1
MD
193error_unlock:
194 pthread_mutex_unlock(&stream->lock);
2f8f53af
DG
195error:
196 if (vstream) {
7591bab1 197 viewer_stream_destroy(vstream);
2f8f53af 198 }
9edaf114
JG
199 if (trace_chunk && acquired_reference) {
200 lttng_trace_chunk_put(trace_chunk);
b66a15d1 201 }
2f8f53af
DG
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.098757 seconds and 4 git commands to generate.