Fix: relayd connection object refactor
[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,
2a174661 44 enum lttng_viewer_seek seek_t, struct ctf_trace *ctf_trace)
2f8f53af
DG
45{
46 struct relay_viewer_stream *vstream;
47
48 assert(stream);
2a174661 49 assert(ctf_trace);
2f8f53af
DG
50
51 vstream = zmalloc(sizeof(*vstream));
52 if (!vstream) {
53 PERROR("relay viewer stream zmalloc");
54 goto error;
55 }
56
2a174661 57 vstream->session_id = stream->session_id;
2f8f53af
DG
58 vstream->stream_handle = stream->stream_handle;
59 vstream->path_name = strndup(stream->path_name, LTTNG_VIEWER_PATH_MAX);
60 vstream->channel_name = strndup(stream->channel_name,
61 LTTNG_VIEWER_NAME_MAX);
62 vstream->tracefile_count = stream->tracefile_count;
63 vstream->metadata_flag = stream->metadata_flag;
64 vstream->tracefile_count_last = -1ULL;
65
66 switch (seek_t) {
67 case VIEWER_SEEK_BEGINNING:
68 vstream->tracefile_count_current = stream->oldest_tracefile_id;
69 break;
70 case VIEWER_SEEK_LAST:
71 vstream->tracefile_count_current = stream->tracefile_count_current;
72 break;
73 default:
74 assert(0);
75 goto error;
76 }
77
2f8f53af 78 if (vstream->metadata_flag) {
2a174661 79 ctf_trace->viewer_metadata_stream = vstream;
2f8f53af 80 }
2f8f53af
DG
81
82 /* Globally visible after the add unique. */
83 lttng_ht_node_init_u64(&vstream->stream_n, stream->stream_handle);
84 lttng_ht_add_unique_u64(viewer_streams_ht, &vstream->stream_n);
85
86 vstream->index_read_fd = -1;
87 vstream->read_fd = -1;
88
89 /*
90 * This is to avoid a race between the initialization of this object and
91 * the close of the given stream. If the stream is unable to find this
92 * viewer stream when closing, this copy will at least take the latest
93 * value. We also need that for the seek_last.
94 */
95 vstream->total_index_received = stream->total_index_received;
96
97 /*
98 * If we never received an index for the current stream, delay the opening
99 * of the index, otherwise open it right now.
100 */
101 if (vstream->tracefile_count_current == stream->tracefile_count_current
102 && vstream->total_index_received == 0) {
103 vstream->index_read_fd = -1;
104 } else {
105 int read_fd;
106
107 read_fd = index_open(vstream->path_name, vstream->channel_name,
108 vstream->tracefile_count, vstream->tracefile_count_current);
109 if (read_fd < 0) {
110 goto error;
111 }
112 vstream->index_read_fd = read_fd;
113 }
114
115 if (seek_t == VIEWER_SEEK_LAST && vstream->index_read_fd >= 0) {
116 off_t lseek_ret;
117
118 lseek_ret = lseek(vstream->index_read_fd,
119 vstream->total_index_received * sizeof(struct ctf_packet_index),
120 SEEK_CUR);
121 if (lseek_ret < 0) {
122 goto error;
123 }
124 vstream->last_sent_index = vstream->total_index_received;
125 }
126
127 return vstream;
128
129error:
130 if (vstream) {
131 free_stream(vstream);
132 }
133 return NULL;
134}
135
136void viewer_stream_delete(struct relay_viewer_stream *stream)
137{
138 int ret;
139 struct lttng_ht_iter iter;
140
141 iter.iter.node = &stream->stream_n.node;
142 ret = lttng_ht_del(viewer_streams_ht, &iter);
143 assert(!ret);
144}
145
2a174661
DG
146void viewer_stream_destroy(struct ctf_trace *ctf_trace,
147 struct relay_viewer_stream *stream)
2f8f53af
DG
148{
149 int ret;
2f8f53af
DG
150
151 assert(stream);
2a174661
DG
152
153 if (ctf_trace) {
154 ctf_trace_put_ref(ctf_trace);
155 }
2f8f53af
DG
156
157 if (stream->read_fd >= 0) {
158 ret = close(stream->read_fd);
159 if (ret < 0) {
160 PERROR("close read_fd");
161 }
162 }
163 if (stream->index_read_fd >= 0) {
164 ret = close(stream->index_read_fd);
165 if (ret < 0) {
166 PERROR("close index_read_fd");
167 }
168 }
169
2f8f53af
DG
170 call_rcu(&stream->rcu_node, deferred_free_viewer_stream);
171}
172
173/*
174 * Find viewer stream by id. RCU read side lock MUST be acquired.
175 *
176 * Return stream if found else NULL.
177 */
178struct relay_viewer_stream *viewer_stream_find_by_id(uint64_t id)
179{
180 struct lttng_ht_node_u64 *node;
181 struct lttng_ht_iter iter;
182 struct relay_viewer_stream *stream = NULL;
183
184 lttng_ht_lookup(viewer_streams_ht, &id, &iter);
185 node = lttng_ht_iter_get_node_u64(&iter);
186 if (!node) {
187 DBG("Relay viewer stream %" PRIu64 " not found", id);
188 goto end;
189 }
190 stream = caa_container_of(node, struct relay_viewer_stream, stream_n);
191
192end:
193 return stream;
194}
195
196/*
197 * Rotate a stream to the next tracefile.
198 *
199 * Returns 0 on success, 1 on EOF, a negative value on error.
200 */
201int viewer_stream_rotate(struct relay_viewer_stream *vstream,
202 struct relay_stream *stream)
203{
204 int ret;
205 uint64_t tracefile_id;
206
207 assert(vstream);
2a174661
DG
208 assert(stream);
209
210 if (vstream->tracefile_count == 0) {
211 /* Ignore rotation, there is none to do. */
212 ret = 0;
213 goto end;
214 }
2f8f53af
DG
215
216 tracefile_id = (vstream->tracefile_count_current + 1) %
217 vstream->tracefile_count;
218
219 /* Detect the last tracefile to open. */
220 if (vstream->tracefile_count_last != -1ULL &&
221 vstream->tracefile_count_last ==
222 vstream->tracefile_count_current) {
223 ret = 1;
224 goto end;
225 }
226
227 /*
2a174661
DG
228 * Lock to execute rotation in order to avoid races between a modification
229 * on the index values.
2f8f53af 230 */
2a174661 231 pthread_mutex_lock(&stream->viewer_stream_rotation_lock);
2f8f53af
DG
232
233 /*
234 * The writer and the reader are not working in the same tracefile, we can
235 * read up to EOF, we don't care about the total_index_received.
236 */
2a174661 237 if (stream->close_flag || (stream->tracefile_count_current != tracefile_id)) {
2f8f53af
DG
238 vstream->close_write_flag = 1;
239 } else {
240 /*
241 * We are opening a file that is still open in write, make sure we
242 * limit our reading to the number of indexes received.
243 */
244 vstream->close_write_flag = 0;
2a174661 245 if (stream->close_flag) {
2f8f53af
DG
246 vstream->total_index_received = stream->total_index_received;
247 }
248 }
249 vstream->tracefile_count_current = tracefile_id;
250
251 ret = close(vstream->index_read_fd);
252 if (ret < 0) {
253 PERROR("close index file %d", vstream->index_read_fd);
254 }
255 vstream->index_read_fd = -1;
256
257 ret = close(vstream->read_fd);
258 if (ret < 0) {
259 PERROR("close tracefile %d", vstream->read_fd);
260 }
261 vstream->read_fd = -1;
262
263 pthread_mutex_lock(&vstream->overwrite_lock);
264 vstream->abort_flag = 0;
265 pthread_mutex_unlock(&vstream->overwrite_lock);
266
2a174661 267 pthread_mutex_unlock(&stream->viewer_stream_rotation_lock);
2f8f53af
DG
268
269 ret = index_open(vstream->path_name, vstream->channel_name,
270 vstream->tracefile_count, vstream->tracefile_count_current);
271 if (ret < 0) {
272 goto error;
273 }
274 vstream->index_read_fd = ret;
275
276 ret = 0;
277
278end:
279error:
280 return ret;
281}
This page took 0.04932 seconds and 4 git commands to generate.