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