511eaad93afbb23a6c2799eeb3a154e867b7d4bf
[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;
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;
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;
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;
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;
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;
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:
194 if (vstream) {
195 viewer_stream_destroy(vstream);
196 }
197 if (trace_chunk && acquired_reference) {
198 lttng_trace_chunk_put(trace_chunk);
199 }
200 return NULL;
201 }
202
203 static void viewer_stream_unpublish(struct relay_viewer_stream *vstream)
204 {
205 int ret;
206 struct lttng_ht_iter iter;
207
208 iter.iter.node = &vstream->stream_n.node;
209 ret = lttng_ht_del(viewer_streams_ht, &iter);
210 assert(!ret);
211 }
212
213 static void viewer_stream_release(struct urcu_ref *ref)
214 {
215 struct relay_viewer_stream *vstream = caa_container_of(ref,
216 struct relay_viewer_stream, ref);
217
218 if (vstream->stream->is_metadata) {
219 rcu_assign_pointer(vstream->stream->trace->viewer_metadata_stream, NULL);
220 }
221
222 viewer_stream_unpublish(vstream);
223
224 if (vstream->stream_file.handle) {
225 fs_handle_close(vstream->stream_file.handle);
226 vstream->stream_file.handle = NULL;
227 }
228 if (vstream->index_file) {
229 lttng_index_file_put(vstream->index_file);
230 vstream->index_file = NULL;
231 }
232 if (vstream->stream) {
233 stream_put(vstream->stream);
234 vstream->stream = NULL;
235 }
236 lttng_trace_chunk_put(vstream->stream_file.trace_chunk);
237 vstream->stream_file.trace_chunk = NULL;
238 call_rcu(&vstream->rcu_node, viewer_stream_destroy_rcu);
239 }
240
241 /* Must be called with RCU read-side lock held. */
242 bool viewer_stream_get(struct relay_viewer_stream *vstream)
243 {
244 return urcu_ref_get_unless_zero(&vstream->ref);
245 }
246
247 /*
248 * Get viewer stream by id.
249 *
250 * Return viewer stream if found else NULL.
251 */
252 struct relay_viewer_stream *viewer_stream_get_by_id(uint64_t id)
253 {
254 struct lttng_ht_node_u64 *node;
255 struct lttng_ht_iter iter;
256 struct relay_viewer_stream *vstream = NULL;
257
258 rcu_read_lock();
259 lttng_ht_lookup(viewer_streams_ht, &id, &iter);
260 node = lttng_ht_iter_get_node_u64(&iter);
261 if (!node) {
262 DBG("Relay viewer stream %" PRIu64 " not found", id);
263 goto end;
264 }
265 vstream = caa_container_of(node, struct relay_viewer_stream, stream_n);
266 if (!viewer_stream_get(vstream)) {
267 vstream = NULL;
268 }
269 end:
270 rcu_read_unlock();
271 return vstream;
272 }
273
274 void viewer_stream_put(struct relay_viewer_stream *vstream)
275 {
276 rcu_read_lock();
277 urcu_ref_put(&vstream->ref, viewer_stream_release);
278 rcu_read_unlock();
279 }
280
281 void viewer_stream_close_files(struct relay_viewer_stream *vstream)
282 {
283 if (vstream->index_file) {
284 lttng_index_file_put(vstream->index_file);
285 vstream->index_file = NULL;
286 }
287 if (vstream->stream_file.handle) {
288 fs_handle_close(vstream->stream_file.handle);
289 vstream->stream_file.handle = NULL;
290 }
291 }
292
293 void viewer_stream_sync_tracefile_array_tail(struct relay_viewer_stream *vstream)
294 {
295 const struct relay_stream *stream = vstream->stream;
296 uint64_t seq_tail;
297
298 vstream->current_tracefile_id = tracefile_array_get_file_index_tail(stream->tfa);
299 seq_tail = tracefile_array_get_seq_tail(stream->tfa);
300 if (seq_tail == -1ULL) {
301 seq_tail = 0;
302 }
303 vstream->index_sent_seqcount = seq_tail;
304 }
305
306 /*
307 * Rotate a stream to the next tracefile.
308 *
309 * Must be called with the rstream lock held.
310 * Returns 0 on success, 1 on EOF.
311 */
312 int viewer_stream_rotate(struct relay_viewer_stream *vstream)
313 {
314 int ret;
315 uint64_t new_id;
316 const struct relay_stream *stream = vstream->stream;
317
318 /* Detect the last tracefile to open. */
319 if (stream->index_received_seqcount
320 == vstream->index_sent_seqcount
321 && stream->trace->session->connection_closed) {
322 ret = 1;
323 goto end;
324 }
325
326 if (stream->tracefile_count == 0) {
327 /* Ignore rotation, there is none to do. */
328 ret = 0;
329 goto end;
330 }
331
332 /*
333 * Try to move to the next file.
334 */
335 new_id = (vstream->current_tracefile_id + 1)
336 % stream->tracefile_count;
337 if (tracefile_array_seq_in_file(stream->tfa, new_id,
338 vstream->index_sent_seqcount)) {
339 vstream->current_tracefile_id = new_id;
340 } else {
341 uint64_t seq_tail = tracefile_array_get_seq_tail(stream->tfa);
342
343 /*
344 * This can only be reached on overwrite, which implies there
345 * has been data written at some point, which will have set the
346 * tail.
347 */
348 assert(seq_tail != -1ULL);
349 /*
350 * We need to resync because we lag behind tail.
351 */
352 vstream->current_tracefile_id =
353 tracefile_array_get_file_index_tail(stream->tfa);
354 vstream->index_sent_seqcount = seq_tail;
355 }
356 viewer_stream_close_files(vstream);
357 ret = 0;
358 end:
359 return ret;
360 }
361
362 void print_viewer_streams(void)
363 {
364 struct lttng_ht_iter iter;
365 struct relay_viewer_stream *vstream;
366
367 if (!viewer_streams_ht) {
368 return;
369 }
370
371 rcu_read_lock();
372 cds_lfht_for_each_entry(viewer_streams_ht->ht, &iter.iter, vstream,
373 stream_n.node) {
374 if (!viewer_stream_get(vstream)) {
375 continue;
376 }
377 DBG("vstream %p refcount %ld stream %" PRIu64 " trace %" PRIu64
378 " session %" PRIu64,
379 vstream,
380 vstream->ref.refcount,
381 vstream->stream->stream_handle,
382 vstream->stream->trace->id,
383 vstream->stream->trace->session->id);
384 viewer_stream_put(vstream);
385 }
386 rcu_read_unlock();
387 }
This page took 0.052202 seconds and 3 git commands to generate.