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