Implement run_as wrappers for mkdir/mkdir_recursive/open
[lttng-tools.git] / liblttng-kconsumer / lttng-kconsumer.c
... / ...
CommitLineData
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)) {
203 lttng_consumer_send_error(ctx, CONSUMERD_ERROR_RECV_CMD);
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;
239 int fd;
240
241 /* block */
242 if (lttng_consumer_poll_socket(consumer_sockpoll) < 0) {
243 return -EINTR;
244 }
245 ret = lttcomm_recv_fds_unix_sock(sock, &fd, 1);
246 if (ret != sizeof(fd)) {
247 lttng_consumer_send_error(ctx, CONSUMERD_ERROR_RECV_FD);
248 return ret;
249 }
250
251 DBG("consumer_add_stream %s (%d)", msg.u.stream.path_name,
252 fd);
253 new_stream = consumer_allocate_stream(msg.u.stream.channel_key,
254 msg.u.stream.stream_key,
255 fd, fd,
256 msg.u.stream.state,
257 msg.u.stream.mmap_len,
258 msg.u.stream.output,
259 msg.u.stream.path_name,
260 msg.u.stream.uid,
261 msg.u.stream.gid);
262 if (new_stream == NULL) {
263 lttng_consumer_send_error(ctx, CONSUMERD_OUTFD_ERROR);
264 goto end;
265 }
266 if (ctx->on_recv_stream != NULL) {
267 ret = ctx->on_recv_stream(new_stream);
268 if (ret == 0) {
269 consumer_add_stream(new_stream);
270 } else if (ret < 0) {
271 goto end;
272 }
273 } else {
274 consumer_add_stream(new_stream);
275 }
276 break;
277 }
278 case LTTNG_CONSUMER_UPDATE_STREAM:
279 {
280 if (ctx->on_update_stream != NULL) {
281 ret = ctx->on_update_stream(msg.u.stream.stream_key, msg.u.stream.state);
282 if (ret == 0) {
283 consumer_change_stream_state(msg.u.stream.stream_key, msg.u.stream.state);
284 } else if (ret < 0) {
285 goto end;
286 }
287 } else {
288 consumer_change_stream_state(msg.u.stream.stream_key,
289 msg.u.stream.state);
290 }
291 break;
292 }
293 default:
294 break;
295 }
296end:
297 /* signal the poll thread */
298 ret = write(ctx->consumer_poll_pipe[1], "4", 1);
299 if (ret < 0) {
300 perror("write consumer poll");
301 }
302end_nosignal:
303 return 0;
304}
305
306/*
307 * Consume data on a file descriptor and write it on a trace file.
308 */
309int lttng_kconsumer_read_subbuffer(struct lttng_consumer_stream *stream,
310 struct lttng_consumer_local_data *ctx)
311{
312 unsigned long len;
313 int err;
314 long ret = 0;
315 int infd = stream->wait_fd;
316
317 DBG("In read_subbuffer (infd : %d)", infd);
318 /* Get the next subbuffer */
319 err = kernctl_get_next_subbuf(infd);
320 if (err != 0) {
321 ret = errno;
322 /*
323 * This is a debug message even for single-threaded consumer,
324 * because poll() have more relaxed criterions than get subbuf,
325 * so get_subbuf may fail for short race windows where poll()
326 * would issue wakeups.
327 */
328 DBG("Reserving sub buffer failed (everything is normal, "
329 "it is due to concurrency)");
330 goto end;
331 }
332
333 switch (stream->output) {
334 case LTTNG_EVENT_SPLICE:
335 /* read the whole subbuffer */
336 err = kernctl_get_padded_subbuf_size(infd, &len);
337 if (err != 0) {
338 ret = errno;
339 perror("Getting sub-buffer len failed.");
340 goto end;
341 }
342
343 /* splice the subbuffer to the tracefile */
344 ret = lttng_consumer_on_read_subbuffer_splice(ctx, stream, len);
345 if (ret < 0) {
346 /*
347 * display the error but continue processing to try
348 * to release the subbuffer
349 */
350 ERR("Error splicing to tracefile");
351 }
352 break;
353 case LTTNG_EVENT_MMAP:
354 /* read the used subbuffer size */
355 err = kernctl_get_padded_subbuf_size(infd, &len);
356 if (err != 0) {
357 ret = errno;
358 perror("Getting sub-buffer len failed.");
359 goto end;
360 }
361 /* write the subbuffer to the tracefile */
362 ret = lttng_consumer_on_read_subbuffer_mmap(ctx, stream, len);
363 if (ret < 0) {
364 /*
365 * display the error but continue processing to try
366 * to release the subbuffer
367 */
368 ERR("Error writing to tracefile");
369 }
370 break;
371 default:
372 ERR("Unknown output method");
373 ret = -1;
374 }
375
376 err = kernctl_put_next_subbuf(infd);
377 if (err != 0) {
378 ret = errno;
379 if (errno == EFAULT) {
380 perror("Error in unreserving sub buffer\n");
381 } else if (errno == EIO) {
382 /* Should never happen with newer LTTng versions */
383 perror("Reader has been pushed by the writer, last sub-buffer corrupted.");
384 }
385 goto end;
386 }
387
388end:
389 return ret;
390}
391
392int lttng_kconsumer_on_recv_stream(struct lttng_consumer_stream *stream)
393{
394 int ret;
395
396 /* Opening the tracefile in write mode */
397 if (stream->path_name != NULL) {
398 ret = open(stream->path_name,
399 O_WRONLY|O_CREAT|O_TRUNC, S_IRWXU|S_IRWXG|S_IRWXO);
400 if (ret < 0) {
401 ERR("Opening %s", stream->path_name);
402 perror("open");
403 goto error;
404 }
405 stream->out_fd = ret;
406 ret = chown(stream->path_name, stream->uid, stream->gid);
407 if (ret < 0) {
408 ERR("Changing ownership of %s", stream->path_name);
409 perror("chown");
410 }
411 }
412
413 if (stream->output == LTTNG_EVENT_MMAP) {
414 /* get the len of the mmap region */
415 unsigned long mmap_len;
416
417 ret = kernctl_get_mmap_len(stream->wait_fd, &mmap_len);
418 if (ret != 0) {
419 ret = errno;
420 perror("kernctl_get_mmap_len");
421 goto error_close_fd;
422 }
423 stream->mmap_len = (size_t) mmap_len;
424
425 stream->mmap_base = mmap(NULL, stream->mmap_len,
426 PROT_READ, MAP_PRIVATE, stream->wait_fd, 0);
427 if (stream->mmap_base == MAP_FAILED) {
428 perror("Error mmaping");
429 ret = -1;
430 goto error_close_fd;
431 }
432 }
433
434 /* we return 0 to let the library handle the FD internally */
435 return 0;
436
437error_close_fd:
438 {
439 int err;
440
441 err = close(stream->out_fd);
442 assert(!err);
443 }
444error:
445 return ret;
446}
447
This page took 0.037219 seconds and 5 git commands to generate.