Fix: code refactoring of viewer streams in relayd
[lttng-tools.git] / src / bin / lttng-relayd / viewer-stream.c
CommitLineData
2f8f53af
DG
1/*
2 * Copyright (C) 2013 - Julien Desfossez <jdesfossez@efficios.com>
3 * David Goulet <dgoulet@efficios.com>
4 *
5 * This program is free software; you can redistribute it and/or modify it
6 * under the terms of the GNU General Public License, version 2 only, as
7 * published by the Free Software Foundation.
8 *
9 * This program is distributed in the hope that it will be useful, but WITHOUT
10 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
11 * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
12 * more details.
13 *
14 * You should have received a copy of the GNU General Public License along with
15 * this program; if not, write to the Free Software Foundation, Inc., 51
16 * Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
17 */
18
19#define _GNU_SOURCE
20#include <common/common.h>
21#include <common/index/index.h>
22
23#include "lttng-relayd.h"
24#include "viewer-stream.h"
25
26static void free_stream(struct relay_viewer_stream *stream)
27{
28 assert(stream);
29
30 free(stream->path_name);
31 free(stream->channel_name);
32 free(stream);
33}
34
35static void deferred_free_viewer_stream(struct rcu_head *head)
36{
37 struct relay_viewer_stream *stream =
38 caa_container_of(head, struct relay_viewer_stream, rcu_node);
39
40 free_stream(stream);
41}
42
43struct relay_viewer_stream *viewer_stream_create(struct relay_stream *stream,
44 enum lttng_viewer_seek seek_t)
45{
46 struct relay_viewer_stream *vstream;
47
48 assert(stream);
49
50 vstream = zmalloc(sizeof(*vstream));
51 if (!vstream) {
52 PERROR("relay viewer stream zmalloc");
53 goto error;
54 }
55
56 vstream->session_id = stream->session->id;
57 vstream->stream_handle = stream->stream_handle;
58 vstream->path_name = strndup(stream->path_name, LTTNG_VIEWER_PATH_MAX);
59 vstream->channel_name = strndup(stream->channel_name,
60 LTTNG_VIEWER_NAME_MAX);
61 vstream->tracefile_count = stream->tracefile_count;
62 vstream->metadata_flag = stream->metadata_flag;
63 vstream->tracefile_count_last = -1ULL;
64
65 switch (seek_t) {
66 case VIEWER_SEEK_BEGINNING:
67 vstream->tracefile_count_current = stream->oldest_tracefile_id;
68 break;
69 case VIEWER_SEEK_LAST:
70 vstream->tracefile_count_current = stream->tracefile_count_current;
71 break;
72 default:
73 assert(0);
74 goto error;
75 }
76
77 vstream->ctf_trace = stream->ctf_trace;
78 if (vstream->metadata_flag) {
79 vstream->ctf_trace->viewer_metadata_stream = vstream;
80 }
81 uatomic_inc(&vstream->ctf_trace->refcount);
82
83 /* Globally visible after the add unique. */
84 lttng_ht_node_init_u64(&vstream->stream_n, stream->stream_handle);
85 lttng_ht_add_unique_u64(viewer_streams_ht, &vstream->stream_n);
86
87 vstream->index_read_fd = -1;
88 vstream->read_fd = -1;
89
90 /*
91 * This is to avoid a race between the initialization of this object and
92 * the close of the given stream. If the stream is unable to find this
93 * viewer stream when closing, this copy will at least take the latest
94 * value. We also need that for the seek_last.
95 */
96 vstream->total_index_received = stream->total_index_received;
97
98 /*
99 * If we never received an index for the current stream, delay the opening
100 * of the index, otherwise open it right now.
101 */
102 if (vstream->tracefile_count_current == stream->tracefile_count_current
103 && vstream->total_index_received == 0) {
104 vstream->index_read_fd = -1;
105 } else {
106 int read_fd;
107
108 read_fd = index_open(vstream->path_name, vstream->channel_name,
109 vstream->tracefile_count, vstream->tracefile_count_current);
110 if (read_fd < 0) {
111 goto error;
112 }
113 vstream->index_read_fd = read_fd;
114 }
115
116 if (seek_t == VIEWER_SEEK_LAST && vstream->index_read_fd >= 0) {
117 off_t lseek_ret;
118
119 lseek_ret = lseek(vstream->index_read_fd,
120 vstream->total_index_received * sizeof(struct ctf_packet_index),
121 SEEK_CUR);
122 if (lseek_ret < 0) {
123 goto error;
124 }
125 vstream->last_sent_index = vstream->total_index_received;
126 }
127
128 return vstream;
129
130error:
131 if (vstream) {
132 free_stream(vstream);
133 }
134 return NULL;
135}
136
137void viewer_stream_delete(struct relay_viewer_stream *stream)
138{
139 int ret;
140 struct lttng_ht_iter iter;
141
142 iter.iter.node = &stream->stream_n.node;
143 ret = lttng_ht_del(viewer_streams_ht, &iter);
144 assert(!ret);
145}
146
147void viewer_stream_destroy(struct relay_viewer_stream *stream)
148{
149 int ret;
150 unsigned long ret_ref;
151
152 assert(stream);
153 ret_ref = uatomic_add_return(&stream->ctf_trace->refcount, -1);
154 assert(ret_ref >= 0);
155
156 if (stream->read_fd >= 0) {
157 ret = close(stream->read_fd);
158 if (ret < 0) {
159 PERROR("close read_fd");
160 }
161 }
162 if (stream->index_read_fd >= 0) {
163 ret = close(stream->index_read_fd);
164 if (ret < 0) {
165 PERROR("close index_read_fd");
166 }
167 }
168
169 /*
170 * If the only stream left in the HT is the metadata stream,
171 * we need to remove it because we won't detect a EOF for this
172 * stream.
173 */
174 if (ret_ref == 1 && stream->ctf_trace->viewer_metadata_stream) {
175 viewer_stream_delete(stream->ctf_trace->viewer_metadata_stream);
176 viewer_stream_destroy(stream->ctf_trace->viewer_metadata_stream);
177 stream->ctf_trace->metadata_stream = NULL;
178 DBG("Freeing ctf_trace %" PRIu64, stream->ctf_trace->id);
179 /*
180 * The streaming-side is already closed and we can't receive a new
181 * stream concurrently at this point (since the session is being
182 * destroyed), so when we detect the refcount equals 0, we are the
183 * only owners of the ctf_trace and we can free it ourself.
184 */
185 free(stream->ctf_trace);
186 }
187
188 call_rcu(&stream->rcu_node, deferred_free_viewer_stream);
189}
190
191/*
192 * Find viewer stream by id. RCU read side lock MUST be acquired.
193 *
194 * Return stream if found else NULL.
195 */
196struct relay_viewer_stream *viewer_stream_find_by_id(uint64_t id)
197{
198 struct lttng_ht_node_u64 *node;
199 struct lttng_ht_iter iter;
200 struct relay_viewer_stream *stream = NULL;
201
202 lttng_ht_lookup(viewer_streams_ht, &id, &iter);
203 node = lttng_ht_iter_get_node_u64(&iter);
204 if (!node) {
205 DBG("Relay viewer stream %" PRIu64 " not found", id);
206 goto end;
207 }
208 stream = caa_container_of(node, struct relay_viewer_stream, stream_n);
209
210end:
211 return stream;
212}
213
214/*
215 * Rotate a stream to the next tracefile.
216 *
217 * Returns 0 on success, 1 on EOF, a negative value on error.
218 */
219int viewer_stream_rotate(struct relay_viewer_stream *vstream,
220 struct relay_stream *stream)
221{
222 int ret;
223 uint64_t tracefile_id;
224
225 assert(vstream);
226
227 tracefile_id = (vstream->tracefile_count_current + 1) %
228 vstream->tracefile_count;
229
230 /* Detect the last tracefile to open. */
231 if (vstream->tracefile_count_last != -1ULL &&
232 vstream->tracefile_count_last ==
233 vstream->tracefile_count_current) {
234 ret = 1;
235 goto end;
236 }
237
238 /*
239 * If the stream on the streaming side still exists, lock to execute
240 * rotation in order to avoid races between a modification on the index
241 * values.
242 */
243 if (stream) {
244 pthread_mutex_lock(&stream->viewer_stream_rotation_lock);
245 }
246
247 /*
248 * The writer and the reader are not working in the same tracefile, we can
249 * read up to EOF, we don't care about the total_index_received.
250 */
251 if (!stream || (stream->tracefile_count_current != tracefile_id)) {
252 vstream->close_write_flag = 1;
253 } else {
254 /*
255 * We are opening a file that is still open in write, make sure we
256 * limit our reading to the number of indexes received.
257 */
258 vstream->close_write_flag = 0;
259 if (stream) {
260 vstream->total_index_received = stream->total_index_received;
261 }
262 }
263 vstream->tracefile_count_current = tracefile_id;
264
265 ret = close(vstream->index_read_fd);
266 if (ret < 0) {
267 PERROR("close index file %d", vstream->index_read_fd);
268 }
269 vstream->index_read_fd = -1;
270
271 ret = close(vstream->read_fd);
272 if (ret < 0) {
273 PERROR("close tracefile %d", vstream->read_fd);
274 }
275 vstream->read_fd = -1;
276
277 pthread_mutex_lock(&vstream->overwrite_lock);
278 vstream->abort_flag = 0;
279 pthread_mutex_unlock(&vstream->overwrite_lock);
280
281 if (stream) {
282 pthread_mutex_unlock(&stream->viewer_stream_rotation_lock);
283 }
284
285 ret = index_open(vstream->path_name, vstream->channel_name,
286 vstream->tracefile_count, vstream->tracefile_count_current);
287 if (ret < 0) {
288 goto error;
289 }
290 vstream->index_read_fd = ret;
291
292 ret = 0;
293
294end:
295error:
296 return ret;
297}
This page took 0.050341 seconds and 4 git commands to generate.