Build fix: Apps defining _LGPL_SOURCE must link to urcu-bp
[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);
61 vstream->channel_name = strndup(stream->channel_name,
62 LTTNG_VIEWER_NAME_MAX);
63 vstream->tracefile_count = stream->tracefile_count;
64 vstream->metadata_flag = stream->metadata_flag;
65 vstream->tracefile_count_last = -1ULL;
66
67 switch (seek_t) {
c4e361a4 68 case LTTNG_VIEWER_SEEK_BEGINNING:
2f8f53af
DG
69 vstream->tracefile_count_current = stream->oldest_tracefile_id;
70 break;
c4e361a4 71 case LTTNG_VIEWER_SEEK_LAST:
2f8f53af
DG
72 vstream->tracefile_count_current = stream->tracefile_count_current;
73 break;
74 default:
75 assert(0);
76 goto error;
77 }
78
2f8f53af 79 if (vstream->metadata_flag) {
2a174661 80 ctf_trace->viewer_metadata_stream = vstream;
2f8f53af 81 }
2f8f53af
DG
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
c4e361a4 116 if (seek_t == LTTNG_VIEWER_SEEK_LAST && vstream->index_read_fd >= 0) {
2f8f53af
DG
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
2a174661
DG
147void viewer_stream_destroy(struct ctf_trace *ctf_trace,
148 struct relay_viewer_stream *stream)
2f8f53af
DG
149{
150 int ret;
2f8f53af
DG
151
152 assert(stream);
2a174661
DG
153
154 if (ctf_trace) {
155 ctf_trace_put_ref(ctf_trace);
156 }
2f8f53af
DG
157
158 if (stream->read_fd >= 0) {
159 ret = close(stream->read_fd);
160 if (ret < 0) {
161 PERROR("close read_fd");
162 }
163 }
164 if (stream->index_read_fd >= 0) {
165 ret = close(stream->index_read_fd);
166 if (ret < 0) {
167 PERROR("close index_read_fd");
168 }
169 }
170
2f8f53af
DG
171 call_rcu(&stream->rcu_node, deferred_free_viewer_stream);
172}
173
174/*
175 * Find viewer stream by id. RCU read side lock MUST be acquired.
176 *
177 * Return stream if found else NULL.
178 */
179struct relay_viewer_stream *viewer_stream_find_by_id(uint64_t id)
180{
181 struct lttng_ht_node_u64 *node;
182 struct lttng_ht_iter iter;
183 struct relay_viewer_stream *stream = NULL;
184
185 lttng_ht_lookup(viewer_streams_ht, &id, &iter);
186 node = lttng_ht_iter_get_node_u64(&iter);
187 if (!node) {
188 DBG("Relay viewer stream %" PRIu64 " not found", id);
189 goto end;
190 }
191 stream = caa_container_of(node, struct relay_viewer_stream, stream_n);
192
193end:
194 return stream;
195}
196
197/*
198 * Rotate a stream to the next tracefile.
199 *
56611b06 200 * Must be called with viewer_stream_rotation_lock held.
2f8f53af
DG
201 * Returns 0 on success, 1 on EOF, a negative value on error.
202 */
203int viewer_stream_rotate(struct relay_viewer_stream *vstream,
204 struct relay_stream *stream)
205{
206 int ret;
207 uint64_t tracefile_id;
208
209 assert(vstream);
2a174661
DG
210 assert(stream);
211
212 if (vstream->tracefile_count == 0) {
213 /* Ignore rotation, there is none to do. */
214 ret = 0;
215 goto end;
216 }
2f8f53af
DG
217
218 tracefile_id = (vstream->tracefile_count_current + 1) %
219 vstream->tracefile_count;
220
221 /* Detect the last tracefile to open. */
222 if (vstream->tracefile_count_last != -1ULL &&
223 vstream->tracefile_count_last ==
224 vstream->tracefile_count_current) {
225 ret = 1;
226 goto end;
227 }
228
2f8f53af
DG
229 /*
230 * The writer and the reader are not working in the same tracefile, we can
231 * read up to EOF, we don't care about the total_index_received.
232 */
2a174661 233 if (stream->close_flag || (stream->tracefile_count_current != tracefile_id)) {
2f8f53af
DG
234 vstream->close_write_flag = 1;
235 } else {
236 /*
237 * We are opening a file that is still open in write, make sure we
238 * limit our reading to the number of indexes received.
239 */
240 vstream->close_write_flag = 0;
2a174661 241 if (stream->close_flag) {
2f8f53af
DG
242 vstream->total_index_received = stream->total_index_received;
243 }
244 }
245 vstream->tracefile_count_current = tracefile_id;
246
247 ret = close(vstream->index_read_fd);
248 if (ret < 0) {
249 PERROR("close index file %d", vstream->index_read_fd);
250 }
251 vstream->index_read_fd = -1;
252
253 ret = close(vstream->read_fd);
254 if (ret < 0) {
255 PERROR("close tracefile %d", vstream->read_fd);
256 }
257 vstream->read_fd = -1;
258
259 pthread_mutex_lock(&vstream->overwrite_lock);
260 vstream->abort_flag = 0;
261 pthread_mutex_unlock(&vstream->overwrite_lock);
262
2f8f53af
DG
263 ret = index_open(vstream->path_name, vstream->channel_name,
264 vstream->tracefile_count, vstream->tracefile_count_current);
265 if (ret < 0) {
266 goto error;
267 }
268 vstream->index_read_fd = ret;
269
270 ret = 0;
271
272end:
273error:
274 return ret;
275}
This page took 0.053382 seconds and 4 git commands to generate.