Fix: relayd: live connection fails to open file during clear
[lttng-tools.git] / src / bin / lttng-relayd / viewer-stream.c
1 /*
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>
5 *
6 * SPDX-License-Identifier: GPL-2.0-only
7 *
8 */
9
10 #define _LGPL_SOURCE
11 #include <common/common.h>
12 #include <common/index/index.h>
13 #include <common/compat/string.h>
14 #include <common/utils.h>
15 #include <sys/types.h>
16 #include <sys/stat.h>
17 #include <fcntl.h>
18
19 #include "lttng-relayd.h"
20 #include "viewer-stream.h"
21
22 static void viewer_stream_destroy(struct relay_viewer_stream *vstream)
23 {
24 free(vstream->path_name);
25 free(vstream->channel_name);
26 free(vstream);
27 }
28
29 static void viewer_stream_destroy_rcu(struct rcu_head *head)
30 {
31 struct relay_viewer_stream *vstream =
32 caa_container_of(head, struct relay_viewer_stream, rcu_node);
33
34 viewer_stream_destroy(vstream);
35 }
36
37 /* Relay stream's lock must be held by the caller. */
38 struct relay_viewer_stream *viewer_stream_create(struct relay_stream *stream,
39 struct lttng_trace_chunk *trace_chunk,
40 enum lttng_viewer_seek seek_t)
41 {
42 struct relay_viewer_stream *vstream = NULL;
43 const bool acquired_reference = lttng_trace_chunk_get(trace_chunk);
44
45 ASSERT_LOCKED(stream->lock);
46 if (!acquired_reference) {
47 goto error;
48 }
49
50 vstream = zmalloc(sizeof(*vstream));
51 if (!vstream) {
52 PERROR("relay viewer stream zmalloc");
53 goto error;
54 }
55
56 vstream->stream_file.trace_chunk = trace_chunk;
57 trace_chunk = NULL;
58 vstream->path_name = lttng_strndup(stream->path_name, LTTNG_VIEWER_PATH_MAX);
59 if (vstream->path_name == NULL) {
60 PERROR("relay viewer path_name alloc");
61 goto error;
62 }
63 vstream->channel_name = lttng_strndup(stream->channel_name,
64 LTTNG_VIEWER_NAME_MAX);
65 if (vstream->channel_name == NULL) {
66 PERROR("relay viewer channel_name alloc");
67 goto error;
68 }
69
70 if (!stream_get(stream)) {
71 ERR("Cannot get stream");
72 goto error;
73 }
74 vstream->stream = stream;
75
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
81 switch (seek_t) {
82 case LTTNG_VIEWER_SEEK_BEGINNING:
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;
97 break;
98 }
99 case LTTNG_VIEWER_SEEK_LAST:
100 vstream->current_tracefile_id =
101 tracefile_array_get_read_file_index_head(stream->tfa);
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;
111 break;
112 default:
113 goto error_unlock;
114 }
115
116 /*
117 * If we never received an index for the current stream, delay
118 * the opening of the index, otherwise open it right now.
119 */
120 if (stream->index_file == NULL) {
121 vstream->index_file = NULL;
122 } else {
123 const uint32_t connection_major = stream->trace->session->major;
124 const uint32_t connection_minor = stream->trace->session->minor;
125 enum lttng_trace_chunk_status chunk_status;
126
127 chunk_status = lttng_index_file_create_from_trace_chunk_read_only(
128 vstream->stream_file.trace_chunk,
129 stream->path_name,
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,
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 }
143 }
144 }
145
146 /*
147 * If we never received a data file for the current stream, delay the
148 * opening, otherwise open it right now.
149 */
150 if (stream->file) {
151 int ret;
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
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);
167 if (status != LTTNG_TRACE_CHUNK_STATUS_OK) {
168 goto error_unlock;
169 }
170 }
171
172 if (seek_t == LTTNG_VIEWER_SEEK_LAST && vstream->index_file) {
173 off_t lseek_ret;
174
175 lseek_ret = fs_handle_seek(
176 vstream->index_file->file, 0, SEEK_END);
177 if (lseek_ret < 0) {
178 goto error_unlock;
179 }
180 }
181 if (stream->is_metadata) {
182 rcu_assign_pointer(stream->trace->viewer_metadata_stream,
183 vstream);
184 }
185
186 /* Globally visible after the add unique. */
187 lttng_ht_node_init_u64(&vstream->stream_n, stream->stream_handle);
188 urcu_ref_init(&vstream->ref);
189 lttng_ht_add_unique_u64(viewer_streams_ht, &vstream->stream_n);
190
191 return vstream;
192
193 error_unlock:
194 pthread_mutex_unlock(&stream->lock);
195 error:
196 if (vstream) {
197 viewer_stream_destroy(vstream);
198 }
199 if (trace_chunk && acquired_reference) {
200 lttng_trace_chunk_put(trace_chunk);
201 }
202 return NULL;
203 }
204
205 static void viewer_stream_unpublish(struct relay_viewer_stream *vstream)
206 {
207 int ret;
208 struct lttng_ht_iter iter;
209
210 iter.iter.node = &vstream->stream_n.node;
211 ret = lttng_ht_del(viewer_streams_ht, &iter);
212 assert(!ret);
213 }
214
215 static void viewer_stream_release(struct urcu_ref *ref)
216 {
217 struct relay_viewer_stream *vstream = caa_container_of(ref,
218 struct relay_viewer_stream, ref);
219
220 if (vstream->stream->is_metadata) {
221 rcu_assign_pointer(vstream->stream->trace->viewer_metadata_stream, NULL);
222 }
223
224 viewer_stream_unpublish(vstream);
225
226 if (vstream->stream_file.handle) {
227 fs_handle_close(vstream->stream_file.handle);
228 vstream->stream_file.handle = NULL;
229 }
230 if (vstream->index_file) {
231 lttng_index_file_put(vstream->index_file);
232 vstream->index_file = NULL;
233 }
234 if (vstream->stream) {
235 stream_put(vstream->stream);
236 vstream->stream = NULL;
237 }
238 lttng_trace_chunk_put(vstream->stream_file.trace_chunk);
239 vstream->stream_file.trace_chunk = NULL;
240 call_rcu(&vstream->rcu_node, viewer_stream_destroy_rcu);
241 }
242
243 /* Must be called with RCU read-side lock held. */
244 bool viewer_stream_get(struct relay_viewer_stream *vstream)
245 {
246 return urcu_ref_get_unless_zero(&vstream->ref);
247 }
248
249 /*
250 * Get viewer stream by id.
251 *
252 * Return viewer stream if found else NULL.
253 */
254 struct relay_viewer_stream *viewer_stream_get_by_id(uint64_t id)
255 {
256 struct lttng_ht_node_u64 *node;
257 struct lttng_ht_iter iter;
258 struct relay_viewer_stream *vstream = NULL;
259
260 rcu_read_lock();
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 }
267 vstream = caa_container_of(node, struct relay_viewer_stream, stream_n);
268 if (!viewer_stream_get(vstream)) {
269 vstream = NULL;
270 }
271 end:
272 rcu_read_unlock();
273 return vstream;
274 }
275
276 void viewer_stream_put(struct relay_viewer_stream *vstream)
277 {
278 rcu_read_lock();
279 urcu_ref_put(&vstream->ref, viewer_stream_release);
280 rcu_read_unlock();
281 }
282
283 void 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 }
289 if (vstream->stream_file.handle) {
290 fs_handle_close(vstream->stream_file.handle);
291 vstream->stream_file.handle = NULL;
292 }
293 }
294
295 void 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
308 /*
309 * Rotate a stream to the next tracefile.
310 *
311 * Must be called with the rstream lock held.
312 * Returns 0 on success, 1 on EOF.
313 */
314 int viewer_stream_rotate(struct relay_viewer_stream *vstream)
315 {
316 int ret;
317 uint64_t new_id;
318 const struct relay_stream *stream = vstream->stream;
319
320 /* Detect the last tracefile to open. */
321 if (stream->index_received_seqcount
322 == vstream->index_sent_seqcount
323 && stream->trace->session->connection_closed) {
324 ret = 1;
325 goto end;
326 }
327
328 if (stream->tracefile_count == 0) {
329 /* Ignore rotation, there is none to do. */
330 ret = 0;
331 goto end;
332 }
333
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;
342 } else {
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 */
354 vstream->current_tracefile_id =
355 tracefile_array_get_file_index_tail(stream->tfa);
356 vstream->index_sent_seqcount = seq_tail;
357 }
358 viewer_stream_close_files(vstream);
359 ret = 0;
360 end:
361 return ret;
362 }
363
364 void print_viewer_streams(void)
365 {
366 struct lttng_ht_iter iter;
367 struct relay_viewer_stream *vstream;
368
369 if (!viewer_streams_ht) {
370 return;
371 }
372
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.057485 seconds and 5 git commands to generate.