Add missing rcu_read_unlock on error paths
[lttng-tools.git] / liblttng-kconsumer / lttng-kconsumer.c
CommitLineData
3bd1e081
MD
1/*
2 * Copyright (C) 2011 - Julien Desfossez <julien.desfossez@polymtl.ca>
3 * Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
4 *
5 * This program is free software; you can redistribute it and/or
6 * modify it under the terms of the GNU General Public License
7 * as published by the Free Software Foundation; only version 2
8 * of the License.
9 *
10 * This program is distributed in the hope that it will be useful,
11 * but WITHOUT ANY WARRANTY; without even the implied warranty of
12 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
13 * GNU General Public License for more details.
14 *
15 * You should have received a copy of the GNU General Public License
16 * along with this program; if not, write to the Free Software
17 * Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
18 */
19
20#define _GNU_SOURCE
21#include <assert.h>
22#include <fcntl.h>
23#include <poll.h>
24#include <pthread.h>
25#include <stdlib.h>
26#include <string.h>
27#include <sys/mman.h>
28#include <sys/socket.h>
29#include <sys/types.h>
30#include <unistd.h>
31
32#include <lttng-kernel-ctl.h>
33#include <lttng-sessiond-comm.h>
34#include <lttng/lttng-kconsumer.h>
35#include <lttngerr.h>
36
37extern struct lttng_consumer_global_data consumer_data;
38extern int consumer_poll_timeout;
39extern volatile int consumer_quit;
40
41/*
42 * Mmap the ring buffer, read it and write the data to the tracefile.
43 *
44 * Returns the number of bytes written
45 */
46int lttng_kconsumer_on_read_subbuffer_mmap(
47 struct lttng_consumer_local_data *ctx,
48 struct lttng_consumer_stream *stream, unsigned long len)
49{
50 unsigned long mmap_offset;
51 long ret = 0;
52 off_t orig_offset = stream->out_fd_offset;
53 int fd = stream->wait_fd;
54 int outfd = stream->out_fd;
55
56 /* get the offset inside the fd to mmap */
57 ret = kernctl_get_mmap_read_offset(fd, &mmap_offset);
58 if (ret != 0) {
59 ret = -errno;
60 perror("kernctl_get_mmap_read_offset");
61 goto end;
62 }
63
64 while (len > 0) {
65 ret = write(outfd, stream->mmap_base + mmap_offset, len);
66 if (ret >= len) {
67 len = 0;
68 } else if (ret < 0) {
69 ret = -errno;
70 perror("Error in file write");
71 goto end;
72 }
73 /* This won't block, but will start writeout asynchronously */
74 sync_file_range(outfd, stream->out_fd_offset, ret,
75 SYNC_FILE_RANGE_WRITE);
76 stream->out_fd_offset += ret;
77 }
78
79 lttng_consumer_sync_trace_file(stream, orig_offset);
80
81 goto end;
82
83end:
84 return ret;
85}
86
87/*
88 * Splice the data from the ring buffer to the tracefile.
89 *
90 * Returns the number of bytes spliced.
91 */
92int lttng_kconsumer_on_read_subbuffer_splice(
93 struct lttng_consumer_local_data *ctx,
94 struct lttng_consumer_stream *stream, unsigned long len)
95{
96 long ret = 0;
97 loff_t offset = 0;
98 off_t orig_offset = stream->out_fd_offset;
99 int fd = stream->wait_fd;
100 int outfd = stream->out_fd;
101
102 while (len > 0) {
103 DBG("splice chan to pipe offset %lu (fd : %d)",
104 (unsigned long)offset, fd);
105 ret = splice(fd, &offset, ctx->consumer_thread_pipe[1], NULL, len,
106 SPLICE_F_MOVE | SPLICE_F_MORE);
107 DBG("splice chan to pipe ret %ld", ret);
108 if (ret < 0) {
109 ret = errno;
110 perror("Error in relay splice");
111 goto splice_error;
112 }
113
114 ret = splice(ctx->consumer_thread_pipe[0], NULL, outfd, NULL, ret,
115 SPLICE_F_MOVE | SPLICE_F_MORE);
116 DBG("splice pipe to file %ld", ret);
117 if (ret < 0) {
118 ret = errno;
119 perror("Error in file splice");
120 goto splice_error;
121 }
122 len -= ret;
123 /* This won't block, but will start writeout asynchronously */
124 sync_file_range(outfd, stream->out_fd_offset, ret,
125 SYNC_FILE_RANGE_WRITE);
126 stream->out_fd_offset += ret;
127 }
128 lttng_consumer_sync_trace_file(stream, orig_offset);
129
130 goto end;
131
132splice_error:
133 /* send the appropriate error description to sessiond */
134 switch(ret) {
135 case EBADF:
136 lttng_consumer_send_error(ctx, CONSUMERD_SPLICE_EBADF);
137 break;
138 case EINVAL:
139 lttng_consumer_send_error(ctx, CONSUMERD_SPLICE_EINVAL);
140 break;
141 case ENOMEM:
142 lttng_consumer_send_error(ctx, CONSUMERD_SPLICE_ENOMEM);
143 break;
144 case ESPIPE:
145 lttng_consumer_send_error(ctx, CONSUMERD_SPLICE_ESPIPE);
146 break;
147 }
148
149end:
150 return ret;
151}
152
153/*
154 * Take a snapshot for a specific fd
155 *
156 * Returns 0 on success, < 0 on error
157 */
158int lttng_kconsumer_take_snapshot(struct lttng_consumer_local_data *ctx,
159 struct lttng_consumer_stream *stream)
160{
161 int ret = 0;
162 int infd = stream->wait_fd;
163
164 ret = kernctl_snapshot(infd);
165 if (ret != 0) {
166 ret = errno;
167 perror("Getting sub-buffer snapshot.");
168 }
169
170 return ret;
171}
172
173/*
174 * Get the produced position
175 *
176 * Returns 0 on success, < 0 on error
177 */
178int lttng_kconsumer_get_produced_snapshot(
179 struct lttng_consumer_local_data *ctx,
180 struct lttng_consumer_stream *stream,
181 unsigned long *pos)
182{
183 int ret;
184 int infd = stream->wait_fd;
185
186 ret = kernctl_snapshot_get_produced(infd, pos);
187 if (ret != 0) {
188 ret = errno;
189 perror("kernctl_snapshot_get_produced");
190 }
191
192 return ret;
193}
194
195int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
196 int sock, struct pollfd *consumer_sockpoll)
197{
198 ssize_t ret;
199 struct lttcomm_consumer_msg msg;
200
201 ret = lttcomm_recv_unix_sock(sock, &msg, sizeof(msg));
202 if (ret != sizeof(msg)) {
f2fc6720 203 lttng_consumer_send_error(ctx, CONSUMERD_ERROR_RECV_CMD);
3bd1e081
MD
204 return ret;
205 }
206 if (msg.cmd_type == LTTNG_CONSUMER_STOP) {
207 return -ENOENT;
208 }
209
210 switch (msg.cmd_type) {
211 case LTTNG_CONSUMER_ADD_CHANNEL:
212 {
213 struct lttng_consumer_channel *new_channel;
214
215 DBG("consumer_add_channel %d", msg.u.channel.channel_key);
216 new_channel = consumer_allocate_channel(msg.u.channel.channel_key,
217 -1, -1,
218 msg.u.channel.mmap_len,
219 msg.u.channel.max_sb_size);
220 if (new_channel == NULL) {
221 lttng_consumer_send_error(ctx, CONSUMERD_OUTFD_ERROR);
222 goto end_nosignal;
223 }
224 if (ctx->on_recv_channel != NULL) {
225 ret = ctx->on_recv_channel(new_channel);
226 if (ret == 0) {
227 consumer_add_channel(new_channel);
228 } else if (ret < 0) {
229 goto end_nosignal;
230 }
231 } else {
232 consumer_add_channel(new_channel);
233 }
234 goto end_nosignal;
235 }
236 case LTTNG_CONSUMER_ADD_STREAM:
237 {
238 struct lttng_consumer_stream *new_stream;
f2fc6720 239 int fd;
3bd1e081
MD
240
241 /* block */
242 if (lttng_consumer_poll_socket(consumer_sockpoll) < 0) {
243 return -EINTR;
244 }
f2fc6720
MD
245 ret = lttcomm_recv_fds_unix_sock(sock, &fd, 1);
246 if (ret != sizeof(fd)) {
3bd1e081
MD
247 lttng_consumer_send_error(ctx, CONSUMERD_ERROR_RECV_FD);
248 return ret;
249 }
3bd1e081 250
f2fc6720
MD
251 DBG("consumer_add_stream %s (%d)", msg.u.stream.path_name,
252 fd);
3bd1e081
MD
253 new_stream = consumer_allocate_stream(msg.u.stream.channel_key,
254 msg.u.stream.stream_key,
f2fc6720 255 fd, fd,
3bd1e081
MD
256 msg.u.stream.state,
257 msg.u.stream.mmap_len,
258 msg.u.stream.output,
259 msg.u.stream.path_name);
260 if (new_stream == NULL) {
261 lttng_consumer_send_error(ctx, CONSUMERD_OUTFD_ERROR);
262 goto end;
263 }
264 if (ctx->on_recv_stream != NULL) {
265 ret = ctx->on_recv_stream(new_stream);
266 if (ret == 0) {
267 consumer_add_stream(new_stream);
268 } else if (ret < 0) {
269 goto end;
270 }
271 } else {
272 consumer_add_stream(new_stream);
273 }
274 break;
275 }
276 case LTTNG_CONSUMER_UPDATE_STREAM:
277 {
278 if (ctx->on_update_stream != NULL) {
279 ret = ctx->on_update_stream(msg.u.stream.stream_key, msg.u.stream.state);
280 if (ret == 0) {
281 consumer_change_stream_state(msg.u.stream.stream_key, msg.u.stream.state);
282 } else if (ret < 0) {
283 goto end;
284 }
285 } else {
286 consumer_change_stream_state(msg.u.stream.stream_key,
287 msg.u.stream.state);
288 }
289 break;
290 }
291 default:
292 break;
293 }
294end:
295 /* signal the poll thread */
296 ret = write(ctx->consumer_poll_pipe[1], "4", 1);
297 if (ret < 0) {
298 perror("write consumer poll");
299 }
300end_nosignal:
301 return 0;
302}
This page took 0.03546 seconds and 5 git commands to generate.