Revert FreeBSD compatibility layer
[lttng-tools.git] / src / common / kernel-consumer / kernel-consumer.c
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 <common/common.h>
33 #include <common/kernel-ctl/kernel-ctl.h>
34 #include <common/sessiond-comm/sessiond-comm.h>
35
36 #include "kernel-consumer.h"
37
38 extern struct lttng_consumer_global_data consumer_data;
39 extern int consumer_poll_timeout;
40 extern volatile int consumer_quit;
41
42 /*
43 * Mmap the ring buffer, read it and write the data to the tracefile.
44 *
45 * Returns the number of bytes written
46 */
47 int lttng_kconsumer_on_read_subbuffer_mmap(
48 struct lttng_consumer_local_data *ctx,
49 struct lttng_consumer_stream *stream, unsigned long len)
50 {
51 unsigned long mmap_offset;
52 long ret = 0;
53 off_t orig_offset = stream->out_fd_offset;
54 int fd = stream->wait_fd;
55 int outfd = stream->out_fd;
56
57 /* get the offset inside the fd to mmap */
58 ret = kernctl_get_mmap_read_offset(fd, &mmap_offset);
59 if (ret != 0) {
60 ret = -errno;
61 perror("kernctl_get_mmap_read_offset");
62 goto end;
63 }
64
65 while (len > 0) {
66 ret = write(outfd, stream->mmap_base + mmap_offset, len);
67 if (ret >= len) {
68 len = 0;
69 } else if (ret < 0) {
70 ret = -errno;
71 perror("Error in file write");
72 goto end;
73 }
74 /* This won't block, but will start writeout asynchronously */
75 sync_file_range(outfd, stream->out_fd_offset, ret,
76 SYNC_FILE_RANGE_WRITE);
77 stream->out_fd_offset += ret;
78 }
79
80 lttng_consumer_sync_trace_file(stream, orig_offset);
81
82 goto end;
83
84 end:
85 return ret;
86 }
87
88 /*
89 * Splice the data from the ring buffer to the tracefile.
90 *
91 * Returns the number of bytes spliced.
92 */
93 int lttng_kconsumer_on_read_subbuffer_splice(
94 struct lttng_consumer_local_data *ctx,
95 struct lttng_consumer_stream *stream, unsigned long len)
96 {
97 long ret = 0;
98 loff_t offset = 0;
99 off_t orig_offset = stream->out_fd_offset;
100 int fd = stream->wait_fd;
101 int outfd = stream->out_fd;
102
103 while (len > 0) {
104 DBG("splice chan to pipe offset %lu (fd : %d)",
105 (unsigned long)offset, fd);
106 ret = splice(fd, &offset, ctx->consumer_thread_pipe[1], NULL, len,
107 SPLICE_F_MOVE | SPLICE_F_MORE);
108 DBG("splice chan to pipe ret %ld", ret);
109 if (ret < 0) {
110 ret = errno;
111 perror("Error in relay splice");
112 goto splice_error;
113 }
114
115 ret = splice(ctx->consumer_thread_pipe[0], NULL, outfd, NULL, ret,
116 SPLICE_F_MOVE | SPLICE_F_MORE);
117 DBG("splice pipe to file %ld", ret);
118 if (ret < 0) {
119 ret = errno;
120 perror("Error in file splice");
121 goto splice_error;
122 }
123 len -= ret;
124 /* This won't block, but will start writeout asynchronously */
125 sync_file_range(outfd, stream->out_fd_offset, ret,
126 SYNC_FILE_RANGE_WRITE);
127 stream->out_fd_offset += ret;
128 }
129 lttng_consumer_sync_trace_file(stream, orig_offset);
130
131 goto end;
132
133 splice_error:
134 /* send the appropriate error description to sessiond */
135 switch(ret) {
136 case EBADF:
137 lttng_consumer_send_error(ctx, CONSUMERD_SPLICE_EBADF);
138 break;
139 case EINVAL:
140 lttng_consumer_send_error(ctx, CONSUMERD_SPLICE_EINVAL);
141 break;
142 case ENOMEM:
143 lttng_consumer_send_error(ctx, CONSUMERD_SPLICE_ENOMEM);
144 break;
145 case ESPIPE:
146 lttng_consumer_send_error(ctx, CONSUMERD_SPLICE_ESPIPE);
147 break;
148 }
149
150 end:
151 return ret;
152 }
153
154 /*
155 * Take a snapshot for a specific fd
156 *
157 * Returns 0 on success, < 0 on error
158 */
159 int lttng_kconsumer_take_snapshot(struct lttng_consumer_local_data *ctx,
160 struct lttng_consumer_stream *stream)
161 {
162 int ret = 0;
163 int infd = stream->wait_fd;
164
165 ret = kernctl_snapshot(infd);
166 if (ret != 0) {
167 ret = errno;
168 perror("Getting sub-buffer snapshot.");
169 }
170
171 return ret;
172 }
173
174 /*
175 * Get the produced position
176 *
177 * Returns 0 on success, < 0 on error
178 */
179 int lttng_kconsumer_get_produced_snapshot(
180 struct lttng_consumer_local_data *ctx,
181 struct lttng_consumer_stream *stream,
182 unsigned long *pos)
183 {
184 int ret;
185 int infd = stream->wait_fd;
186
187 ret = kernctl_snapshot_get_produced(infd, pos);
188 if (ret != 0) {
189 ret = errno;
190 perror("kernctl_snapshot_get_produced");
191 }
192
193 return ret;
194 }
195
196 int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
197 int sock, struct pollfd *consumer_sockpoll)
198 {
199 ssize_t ret;
200 struct lttcomm_consumer_msg msg;
201
202 ret = lttcomm_recv_unix_sock(sock, &msg, sizeof(msg));
203 if (ret != sizeof(msg)) {
204 lttng_consumer_send_error(ctx, CONSUMERD_ERROR_RECV_CMD);
205 return ret;
206 }
207 if (msg.cmd_type == LTTNG_CONSUMER_STOP) {
208 return -ENOENT;
209 }
210
211 switch (msg.cmd_type) {
212 case LTTNG_CONSUMER_ADD_CHANNEL:
213 {
214 struct lttng_consumer_channel *new_channel;
215
216 DBG("consumer_add_channel %d", msg.u.channel.channel_key);
217 new_channel = consumer_allocate_channel(msg.u.channel.channel_key,
218 -1, -1,
219 msg.u.channel.mmap_len,
220 msg.u.channel.max_sb_size);
221 if (new_channel == NULL) {
222 lttng_consumer_send_error(ctx, CONSUMERD_OUTFD_ERROR);
223 goto end_nosignal;
224 }
225 if (ctx->on_recv_channel != NULL) {
226 ret = ctx->on_recv_channel(new_channel);
227 if (ret == 0) {
228 consumer_add_channel(new_channel);
229 } else if (ret < 0) {
230 goto end_nosignal;
231 }
232 } else {
233 consumer_add_channel(new_channel);
234 }
235 goto end_nosignal;
236 }
237 case LTTNG_CONSUMER_ADD_STREAM:
238 {
239 struct lttng_consumer_stream *new_stream;
240 int fd;
241
242 /* block */
243 if (lttng_consumer_poll_socket(consumer_sockpoll) < 0) {
244 return -EINTR;
245 }
246 ret = lttcomm_recv_fds_unix_sock(sock, &fd, 1);
247 if (ret != sizeof(fd)) {
248 lttng_consumer_send_error(ctx, CONSUMERD_ERROR_RECV_FD);
249 return ret;
250 }
251
252 DBG("consumer_add_stream %s (%d)", msg.u.stream.path_name,
253 fd);
254 new_stream = consumer_allocate_stream(msg.u.stream.channel_key,
255 msg.u.stream.stream_key,
256 fd, fd,
257 msg.u.stream.state,
258 msg.u.stream.mmap_len,
259 msg.u.stream.output,
260 msg.u.stream.path_name,
261 msg.u.stream.uid,
262 msg.u.stream.gid);
263 if (new_stream == NULL) {
264 lttng_consumer_send_error(ctx, CONSUMERD_OUTFD_ERROR);
265 goto end;
266 }
267 if (ctx->on_recv_stream != NULL) {
268 ret = ctx->on_recv_stream(new_stream);
269 if (ret == 0) {
270 consumer_add_stream(new_stream);
271 } else if (ret < 0) {
272 goto end;
273 }
274 } else {
275 consumer_add_stream(new_stream);
276 }
277 break;
278 }
279 case LTTNG_CONSUMER_UPDATE_STREAM:
280 {
281 if (ctx->on_update_stream != NULL) {
282 ret = ctx->on_update_stream(msg.u.stream.stream_key, msg.u.stream.state);
283 if (ret == 0) {
284 consumer_change_stream_state(msg.u.stream.stream_key, msg.u.stream.state);
285 } else if (ret < 0) {
286 goto end;
287 }
288 } else {
289 consumer_change_stream_state(msg.u.stream.stream_key,
290 msg.u.stream.state);
291 }
292 break;
293 }
294 default:
295 break;
296 }
297 end:
298 /* signal the poll thread */
299 ret = write(ctx->consumer_poll_pipe[1], "4", 1);
300 if (ret < 0) {
301 perror("write consumer poll");
302 }
303 end_nosignal:
304 return 0;
305 }
306
307 /*
308 * Consume data on a file descriptor and write it on a trace file.
309 */
310 int lttng_kconsumer_read_subbuffer(struct lttng_consumer_stream *stream,
311 struct lttng_consumer_local_data *ctx)
312 {
313 unsigned long len;
314 int err;
315 long ret = 0;
316 int infd = stream->wait_fd;
317
318 DBG("In read_subbuffer (infd : %d)", infd);
319 /* Get the next subbuffer */
320 err = kernctl_get_next_subbuf(infd);
321 if (err != 0) {
322 ret = errno;
323 /*
324 * This is a debug message even for single-threaded consumer,
325 * because poll() have more relaxed criterions than get subbuf,
326 * so get_subbuf may fail for short race windows where poll()
327 * would issue wakeups.
328 */
329 DBG("Reserving sub buffer failed (everything is normal, "
330 "it is due to concurrency)");
331 goto end;
332 }
333
334 switch (stream->output) {
335 case LTTNG_EVENT_SPLICE:
336 /* read the whole subbuffer */
337 err = kernctl_get_padded_subbuf_size(infd, &len);
338 if (err != 0) {
339 ret = errno;
340 perror("Getting sub-buffer len failed.");
341 goto end;
342 }
343
344 /* splice the subbuffer to the tracefile */
345 ret = lttng_consumer_on_read_subbuffer_splice(ctx, stream, len);
346 if (ret < 0) {
347 /*
348 * display the error but continue processing to try
349 * to release the subbuffer
350 */
351 ERR("Error splicing to tracefile");
352 }
353 break;
354 case LTTNG_EVENT_MMAP:
355 /* read the used subbuffer size */
356 err = kernctl_get_padded_subbuf_size(infd, &len);
357 if (err != 0) {
358 ret = errno;
359 perror("Getting sub-buffer len failed.");
360 goto end;
361 }
362 /* write the subbuffer to the tracefile */
363 ret = lttng_consumer_on_read_subbuffer_mmap(ctx, stream, len);
364 if (ret < 0) {
365 /*
366 * display the error but continue processing to try
367 * to release the subbuffer
368 */
369 ERR("Error writing to tracefile");
370 }
371 break;
372 default:
373 ERR("Unknown output method");
374 ret = -1;
375 }
376
377 err = kernctl_put_next_subbuf(infd);
378 if (err != 0) {
379 ret = errno;
380 if (errno == EFAULT) {
381 perror("Error in unreserving sub buffer\n");
382 } else if (errno == EIO) {
383 /* Should never happen with newer LTTng versions */
384 perror("Reader has been pushed by the writer, last sub-buffer corrupted.");
385 }
386 goto end;
387 }
388
389 end:
390 return ret;
391 }
392
393 int lttng_kconsumer_on_recv_stream(struct lttng_consumer_stream *stream)
394 {
395 int ret;
396
397 /* Opening the tracefile in write mode */
398 if (stream->path_name != NULL) {
399 ret = run_as_open(stream->path_name,
400 O_WRONLY|O_CREAT|O_TRUNC,
401 S_IRWXU|S_IRWXG|S_IRWXO,
402 stream->uid, stream->gid);
403 if (ret < 0) {
404 ERR("Opening %s", stream->path_name);
405 perror("open");
406 goto error;
407 }
408 stream->out_fd = ret;
409 }
410
411 if (stream->output == LTTNG_EVENT_MMAP) {
412 /* get the len of the mmap region */
413 unsigned long mmap_len;
414
415 ret = kernctl_get_mmap_len(stream->wait_fd, &mmap_len);
416 if (ret != 0) {
417 ret = errno;
418 perror("kernctl_get_mmap_len");
419 goto error_close_fd;
420 }
421 stream->mmap_len = (size_t) mmap_len;
422
423 stream->mmap_base = mmap(NULL, stream->mmap_len,
424 PROT_READ, MAP_PRIVATE, stream->wait_fd, 0);
425 if (stream->mmap_base == MAP_FAILED) {
426 perror("Error mmaping");
427 ret = -1;
428 goto error_close_fd;
429 }
430 }
431
432 /* we return 0 to let the library handle the FD internally */
433 return 0;
434
435 error_close_fd:
436 {
437 int err;
438
439 err = close(stream->out_fd);
440 assert(!err);
441 }
442 error:
443 return ret;
444 }
445
This page took 0.057086 seconds and 4 git commands to generate.