Fix --enable-lttng-ust configure option
[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}
d41f73b7
MD
303
304/*
305 * Consume data on a file descriptor and write it on a trace file.
306 */
307int lttng_kconsumer_read_subbuffer(struct lttng_consumer_stream *stream,
308 struct lttng_consumer_local_data *ctx)
309{
310 unsigned long len;
311 int err;
312 long ret = 0;
313 int infd = stream->wait_fd;
314
315 DBG("In read_subbuffer (infd : %d)", infd);
316 /* Get the next subbuffer */
317 err = kernctl_get_next_subbuf(infd);
318 if (err != 0) {
319 ret = errno;
320 /*
321 * This is a debug message even for single-threaded consumer,
322 * because poll() have more relaxed criterions than get subbuf,
323 * so get_subbuf may fail for short race windows where poll()
324 * would issue wakeups.
325 */
326 DBG("Reserving sub buffer failed (everything is normal, "
327 "it is due to concurrency)");
328 goto end;
329 }
330
331 switch (stream->output) {
332 case LTTNG_EVENT_SPLICE:
333 /* read the whole subbuffer */
334 err = kernctl_get_padded_subbuf_size(infd, &len);
335 if (err != 0) {
336 ret = errno;
337 perror("Getting sub-buffer len failed.");
338 goto end;
339 }
340
341 /* splice the subbuffer to the tracefile */
342 ret = lttng_consumer_on_read_subbuffer_splice(ctx, stream, len);
343 if (ret < 0) {
344 /*
345 * display the error but continue processing to try
346 * to release the subbuffer
347 */
348 ERR("Error splicing to tracefile");
349 }
350 break;
351 case LTTNG_EVENT_MMAP:
352 /* read the used subbuffer size */
353 err = kernctl_get_padded_subbuf_size(infd, &len);
354 if (err != 0) {
355 ret = errno;
356 perror("Getting sub-buffer len failed.");
357 goto end;
358 }
359 /* write the subbuffer to the tracefile */
360 ret = lttng_consumer_on_read_subbuffer_mmap(ctx, stream, len);
361 if (ret < 0) {
362 /*
363 * display the error but continue processing to try
364 * to release the subbuffer
365 */
366 ERR("Error writing to tracefile");
367 }
368 break;
369 default:
370 ERR("Unknown output method");
371 ret = -1;
372 }
373
374 err = kernctl_put_next_subbuf(infd);
375 if (err != 0) {
376 ret = errno;
377 if (errno == EFAULT) {
378 perror("Error in unreserving sub buffer\n");
379 } else if (errno == EIO) {
380 /* Should never happen with newer LTTng versions */
381 perror("Reader has been pushed by the writer, last sub-buffer corrupted.");
382 }
383 goto end;
384 }
385
386end:
387 return ret;
388}
389
390int lttng_kconsumer_on_recv_stream(struct lttng_consumer_stream *stream)
391{
392 int ret;
393
394 /* Opening the tracefile in write mode */
395 if (stream->path_name != NULL) {
396 ret = open(stream->path_name,
397 O_WRONLY|O_CREAT|O_TRUNC, S_IRWXU|S_IRWXG|S_IRWXO);
398 if (ret < 0) {
399 ERR("Opening %s", stream->path_name);
400 perror("open");
401 goto error;
402 }
403 stream->out_fd = ret;
404 }
405
406 if (stream->output == LTTNG_EVENT_MMAP) {
407 /* get the len of the mmap region */
408 unsigned long mmap_len;
409
410 ret = kernctl_get_mmap_len(stream->wait_fd, &mmap_len);
411 if (ret != 0) {
412 ret = errno;
413 perror("kernctl_get_mmap_len");
414 goto error_close_fd;
415 }
416 stream->mmap_len = (size_t) mmap_len;
417
418 stream->mmap_base = mmap(NULL, stream->mmap_len,
419 PROT_READ, MAP_PRIVATE, stream->wait_fd, 0);
420 if (stream->mmap_base == MAP_FAILED) {
421 perror("Error mmaping");
422 ret = -1;
423 goto error_close_fd;
424 }
425 }
426
427 /* we return 0 to let the library handle the FD internally */
428 return 0;
429
430error_close_fd:
431 {
432 int err;
433
434 err = close(stream->out_fd);
435 assert(!err);
436 }
437error:
438 return ret;
439}
440
This page took 0.059793 seconds and 4 git commands to generate.