Tests: Add runner script for tests
[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 modify
6 * it under the terms of the GNU General Public License, version 2 only,
7 * as published by the Free Software Foundation.
8 *
9 * This program is distributed in the hope that it will be useful,
10 * but WITHOUT ANY WARRANTY; without even the implied warranty of
11 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
12 * GNU General Public License for more details.
13 *
14 * You should have received a copy of the GNU General Public License along
15 * with this program; if not, write to the Free Software Foundation, Inc.,
16 * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
17 */
18
19 #define _GNU_SOURCE
20 #include <assert.h>
21 #include <poll.h>
22 #include <pthread.h>
23 #include <stdlib.h>
24 #include <string.h>
25 #include <sys/mman.h>
26 #include <sys/socket.h>
27 #include <sys/types.h>
28 #include <inttypes.h>
29 #include <unistd.h>
30 #include <sys/stat.h>
31
32 #include <common/common.h>
33 #include <common/kernel-ctl/kernel-ctl.h>
34 #include <common/sessiond-comm/sessiond-comm.h>
35 #include <common/sessiond-comm/relayd.h>
36 #include <common/compat/fcntl.h>
37 #include <common/relayd/relayd.h>
38
39 #include "kernel-consumer.h"
40
41 extern struct lttng_consumer_global_data consumer_data;
42 extern int consumer_poll_timeout;
43 extern volatile int consumer_quit;
44
45 /*
46 * Take a snapshot for a specific fd
47 *
48 * Returns 0 on success, < 0 on error
49 */
50 int lttng_kconsumer_take_snapshot(struct lttng_consumer_stream *stream)
51 {
52 int ret = 0;
53 int infd = stream->wait_fd;
54
55 ret = kernctl_snapshot(infd);
56 if (ret != 0) {
57 errno = -ret;
58 perror("Getting sub-buffer snapshot.");
59 }
60
61 return ret;
62 }
63
64 /*
65 * Get the produced position
66 *
67 * Returns 0 on success, < 0 on error
68 */
69 int lttng_kconsumer_get_produced_snapshot(struct lttng_consumer_stream *stream,
70 unsigned long *pos)
71 {
72 int ret;
73 int infd = stream->wait_fd;
74
75 ret = kernctl_snapshot_get_produced(infd, pos);
76 if (ret != 0) {
77 errno = -ret;
78 perror("kernctl_snapshot_get_produced");
79 }
80
81 return ret;
82 }
83
84 int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
85 int sock, struct pollfd *consumer_sockpoll)
86 {
87 ssize_t ret;
88 enum lttng_error_code ret_code = LTTNG_OK;
89 struct lttcomm_consumer_msg msg;
90
91 ret = lttcomm_recv_unix_sock(sock, &msg, sizeof(msg));
92 if (ret != sizeof(msg)) {
93 lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_ERROR_RECV_CMD);
94 return ret;
95 }
96 if (msg.cmd_type == LTTNG_CONSUMER_STOP) {
97 /*
98 * Notify the session daemon that the command is completed.
99 *
100 * On transport layer error, the function call will print an error
101 * message so handling the returned code is a bit useless since we
102 * return an error code anyway.
103 */
104 (void) consumer_send_status_msg(sock, ret_code);
105 return -ENOENT;
106 }
107
108 /* relayd needs RCU read-side protection */
109 rcu_read_lock();
110
111 switch (msg.cmd_type) {
112 case LTTNG_CONSUMER_ADD_RELAYD_SOCKET:
113 {
114 /* Session daemon status message are handled in the following call. */
115 ret = consumer_add_relayd_socket(msg.u.relayd_sock.net_index,
116 msg.u.relayd_sock.type, ctx, sock, consumer_sockpoll,
117 &msg.u.relayd_sock.sock, msg.u.relayd_sock.session_id);
118 goto end_nosignal;
119 }
120 case LTTNG_CONSUMER_ADD_CHANNEL:
121 {
122 struct lttng_consumer_channel *new_channel;
123
124 /* First send a status message before receiving the fds. */
125 ret = consumer_send_status_msg(sock, ret_code);
126 if (ret < 0) {
127 /* Somehow, the session daemon is not responding anymore. */
128 goto end_nosignal;
129 }
130
131 DBG("consumer_add_channel %d", msg.u.channel.channel_key);
132 new_channel = consumer_allocate_channel(msg.u.channel.channel_key,
133 msg.u.channel.session_id, msg.u.channel.pathname,
134 msg.u.channel.name, msg.u.channel.uid, msg.u.channel.gid,
135 msg.u.channel.relayd_id, msg.u.channel.output);
136 if (new_channel == NULL) {
137 lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_OUTFD_ERROR);
138 goto end_nosignal;
139 }
140 new_channel->nb_init_stream_left = msg.u.channel.nb_init_streams;
141
142 /* Translate and save channel type. */
143 switch (msg.u.channel.type) {
144 case CONSUMER_CHANNEL_TYPE_DATA:
145 case CONSUMER_CHANNEL_TYPE_METADATA:
146 new_channel->type = msg.u.channel.type;
147 break;
148 default:
149 assert(0);
150 goto end_nosignal;
151 };
152
153 if (ctx->on_recv_channel != NULL) {
154 ret = ctx->on_recv_channel(new_channel);
155 if (ret == 0) {
156 consumer_add_channel(new_channel);
157 } else if (ret < 0) {
158 goto end_nosignal;
159 }
160 } else {
161 consumer_add_channel(new_channel);
162 }
163 goto end_nosignal;
164 }
165 case LTTNG_CONSUMER_ADD_STREAM:
166 {
167 int fd, stream_pipe;
168 struct consumer_relayd_sock_pair *relayd = NULL;
169 struct lttng_consumer_stream *new_stream;
170 struct lttng_consumer_channel *channel;
171 int alloc_ret = 0;
172
173 /*
174 * Get stream's channel reference. Needed when adding the stream to the
175 * global hash table.
176 */
177 channel = consumer_find_channel(msg.u.stream.channel_key);
178 if (!channel) {
179 /*
180 * We could not find the channel. Can happen if cpu hotplug
181 * happens while tearing down.
182 */
183 ERR("Unable to find channel key %d", msg.u.stream.channel_key);
184 ret_code = LTTNG_ERR_KERN_CHAN_NOT_FOUND;
185 }
186
187 /* First send a status message before receiving the fds. */
188 ret = consumer_send_status_msg(sock, ret_code);
189 if (ret < 0 || ret_code != LTTNG_OK) {
190 /*
191 * Somehow, the session daemon is not responding anymore or the
192 * channel was not found.
193 */
194 goto end_nosignal;
195 }
196
197 /* block */
198 if (lttng_consumer_poll_socket(consumer_sockpoll) < 0) {
199 rcu_read_unlock();
200 return -EINTR;
201 }
202
203 /* Get stream file descriptor from socket */
204 ret = lttcomm_recv_fds_unix_sock(sock, &fd, 1);
205 if (ret != sizeof(fd)) {
206 lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_ERROR_RECV_FD);
207 rcu_read_unlock();
208 return ret;
209 }
210
211 /*
212 * Send status code to session daemon only if the recv works. If the
213 * above recv() failed, the session daemon is notified through the
214 * error socket and the teardown is eventually done.
215 */
216 ret = consumer_send_status_msg(sock, ret_code);
217 if (ret < 0) {
218 /* Somehow, the session daemon is not responding anymore. */
219 goto end_nosignal;
220 }
221
222 new_stream = consumer_allocate_stream(channel->key,
223 fd,
224 LTTNG_CONSUMER_ACTIVE_STREAM,
225 channel->name,
226 channel->uid,
227 channel->gid,
228 channel->relayd_id,
229 channel->session_id,
230 msg.u.stream.cpu,
231 &alloc_ret,
232 channel->type);
233 if (new_stream == NULL) {
234 switch (alloc_ret) {
235 case -ENOMEM:
236 case -EINVAL:
237 default:
238 lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_OUTFD_ERROR);
239 break;
240 }
241 goto end_nosignal;
242 }
243 new_stream->chan = channel;
244 new_stream->wait_fd = fd;
245
246 /*
247 * The buffer flush is done on the session daemon side for the kernel
248 * so no need for the stream "hangup_flush_done" variable to be
249 * tracked. This is important for a kernel stream since we don't rely
250 * on the flush state of the stream to read data. It's not the case for
251 * user space tracing.
252 */
253 new_stream->hangup_flush_done = 0;
254
255 /* The stream is not metadata. Get relayd reference if exists. */
256 relayd = consumer_find_relayd(new_stream->net_seq_idx);
257 if (relayd != NULL) {
258 /* Add stream on the relayd */
259 pthread_mutex_lock(&relayd->ctrl_sock_mutex);
260 ret = relayd_add_stream(&relayd->control_sock,
261 new_stream->name, new_stream->chan->pathname,
262 &new_stream->relayd_stream_id);
263 pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
264 if (ret < 0) {
265 consumer_del_stream(new_stream, NULL);
266 goto end_nosignal;
267 }
268 } else if (new_stream->net_seq_idx != -1) {
269 ERR("Network sequence index %d unknown. Not adding stream.",
270 new_stream->net_seq_idx);
271 consumer_del_stream(new_stream, NULL);
272 goto end_nosignal;
273 }
274
275 if (ctx->on_recv_stream) {
276 ret = ctx->on_recv_stream(new_stream);
277 if (ret < 0) {
278 consumer_del_stream(new_stream, NULL);
279 goto end_nosignal;
280 }
281 }
282
283 /* Get the right pipe where the stream will be sent. */
284 if (new_stream->metadata_flag) {
285 stream_pipe = ctx->consumer_metadata_pipe[1];
286 } else {
287 stream_pipe = ctx->consumer_data_pipe[1];
288 }
289
290 do {
291 ret = write(stream_pipe, &new_stream, sizeof(new_stream));
292 } while (ret < 0 && errno == EINTR);
293 if (ret < 0) {
294 PERROR("Consumer write %s stream to pipe %d",
295 new_stream->metadata_flag ? "metadata" : "data",
296 stream_pipe);
297 consumer_del_stream(new_stream, NULL);
298 goto end_nosignal;
299 }
300
301 DBG("Kernel consumer ADD_STREAM %s (fd: %d) with relayd id %" PRIu64,
302 new_stream->name, fd, new_stream->relayd_stream_id);
303 break;
304 }
305 case LTTNG_CONSUMER_UPDATE_STREAM:
306 {
307 rcu_read_unlock();
308 return -ENOSYS;
309 }
310 case LTTNG_CONSUMER_DESTROY_RELAYD:
311 {
312 uint64_t index = msg.u.destroy_relayd.net_seq_idx;
313 struct consumer_relayd_sock_pair *relayd;
314
315 DBG("Kernel consumer destroying relayd %" PRIu64, index);
316
317 /* Get relayd reference if exists. */
318 relayd = consumer_find_relayd(index);
319 if (relayd == NULL) {
320 DBG("Unable to find relayd %" PRIu64, index);
321 ret_code = LTTNG_ERR_NO_CONSUMER;
322 }
323
324 /*
325 * Each relayd socket pair has a refcount of stream attached to it
326 * which tells if the relayd is still active or not depending on the
327 * refcount value.
328 *
329 * This will set the destroy flag of the relayd object and destroy it
330 * if the refcount reaches zero when called.
331 *
332 * The destroy can happen either here or when a stream fd hangs up.
333 */
334 if (relayd) {
335 consumer_flag_relayd_for_destroy(relayd);
336 }
337
338 ret = consumer_send_status_msg(sock, ret_code);
339 if (ret < 0) {
340 /* Somehow, the session daemon is not responding anymore. */
341 goto end_nosignal;
342 }
343
344 goto end_nosignal;
345 }
346 case LTTNG_CONSUMER_DATA_PENDING:
347 {
348 int32_t ret;
349 uint64_t id = msg.u.data_pending.session_id;
350
351 DBG("Kernel consumer data pending command for id %" PRIu64, id);
352
353 ret = consumer_data_pending(id);
354
355 /* Send back returned value to session daemon */
356 ret = lttcomm_send_unix_sock(sock, &ret, sizeof(ret));
357 if (ret < 0) {
358 PERROR("send data pending ret code");
359 }
360
361 /*
362 * No need to send back a status message since the data pending
363 * returned value is the response.
364 */
365 break;
366 }
367 default:
368 goto end_nosignal;
369 }
370
371 end_nosignal:
372 rcu_read_unlock();
373
374 /*
375 * Return 1 to indicate success since the 0 value can be a socket
376 * shutdown during the recv() or send() call.
377 */
378 return 1;
379 }
380
381 /*
382 * Consume data on a file descriptor and write it on a trace file.
383 */
384 ssize_t lttng_kconsumer_read_subbuffer(struct lttng_consumer_stream *stream,
385 struct lttng_consumer_local_data *ctx)
386 {
387 unsigned long len, subbuf_size, padding;
388 int err;
389 ssize_t ret = 0;
390 int infd = stream->wait_fd;
391
392 DBG("In read_subbuffer (infd : %d)", infd);
393 /* Get the next subbuffer */
394 err = kernctl_get_next_subbuf(infd);
395 if (err != 0) {
396 ret = err;
397 /*
398 * This is a debug message even for single-threaded consumer,
399 * because poll() have more relaxed criterions than get subbuf,
400 * so get_subbuf may fail for short race windows where poll()
401 * would issue wakeups.
402 */
403 DBG("Reserving sub buffer failed (everything is normal, "
404 "it is due to concurrency)");
405 goto end;
406 }
407
408 /* Get the full subbuffer size including padding */
409 err = kernctl_get_padded_subbuf_size(infd, &len);
410 if (err != 0) {
411 errno = -err;
412 perror("Getting sub-buffer len failed.");
413 ret = err;
414 goto end;
415 }
416
417 switch (stream->chan->output) {
418 case LTTNG_EVENT_SPLICE:
419
420 /*
421 * XXX: The lttng-modules splice "actor" does not handle copying
422 * partial pages hence only using the subbuffer size without the
423 * padding makes the splice fail.
424 */
425 subbuf_size = len;
426 padding = 0;
427
428 /* splice the subbuffer to the tracefile */
429 ret = lttng_consumer_on_read_subbuffer_splice(ctx, stream, subbuf_size,
430 padding);
431 /*
432 * XXX: Splice does not support network streaming so the return value
433 * is simply checked against subbuf_size and not like the mmap() op.
434 */
435 if (ret != subbuf_size) {
436 /*
437 * display the error but continue processing to try
438 * to release the subbuffer
439 */
440 ERR("Error splicing to tracefile (ret: %zd != len: %lu)",
441 ret, subbuf_size);
442 }
443 break;
444 case LTTNG_EVENT_MMAP:
445 /* Get subbuffer size without padding */
446 err = kernctl_get_subbuf_size(infd, &subbuf_size);
447 if (err != 0) {
448 errno = -err;
449 perror("Getting sub-buffer len failed.");
450 ret = err;
451 goto end;
452 }
453
454 /* Make sure the tracer is not gone mad on us! */
455 assert(len >= subbuf_size);
456
457 padding = len - subbuf_size;
458
459 /* write the subbuffer to the tracefile */
460 ret = lttng_consumer_on_read_subbuffer_mmap(ctx, stream, subbuf_size,
461 padding);
462 /*
463 * The mmap operation should write subbuf_size amount of data when
464 * network streaming or the full padding (len) size when we are _not_
465 * streaming.
466 */
467 if ((ret != subbuf_size && stream->net_seq_idx != -1) ||
468 (ret != len && stream->net_seq_idx == -1)) {
469 /*
470 * Display the error but continue processing to try to release the
471 * subbuffer
472 */
473 ERR("Error writing to tracefile "
474 "(ret: %zd != len: %lu != subbuf_size: %lu)",
475 ret, len, subbuf_size);
476 }
477 break;
478 default:
479 ERR("Unknown output method");
480 ret = -1;
481 }
482
483 err = kernctl_put_next_subbuf(infd);
484 if (err != 0) {
485 errno = -err;
486 if (errno == EFAULT) {
487 perror("Error in unreserving sub buffer\n");
488 } else if (errno == EIO) {
489 /* Should never happen with newer LTTng versions */
490 perror("Reader has been pushed by the writer, last sub-buffer corrupted.");
491 }
492
493 ret = -err;
494 goto end;
495 }
496
497 end:
498 return ret;
499 }
500
501 int lttng_kconsumer_on_recv_stream(struct lttng_consumer_stream *stream)
502 {
503 int ret;
504 char full_path[PATH_MAX];
505
506 assert(stream);
507
508 ret = snprintf(full_path, sizeof(full_path), "%s/%s",
509 stream->chan->pathname, stream->name);
510 if (ret < 0) {
511 PERROR("snprintf on_recv_stream");
512 goto error;
513 }
514
515 /* Opening the tracefile in write mode */
516 if (stream->net_seq_idx == -1) {
517 ret = run_as_open(full_path, O_WRONLY | O_CREAT | O_TRUNC,
518 S_IRWXU|S_IRWXG|S_IRWXO, stream->uid, stream->gid);
519 if (ret < 0) {
520 PERROR("open kernel stream path %s", full_path);
521 goto error;
522 }
523 stream->out_fd = ret;
524 }
525
526 if (stream->output == LTTNG_EVENT_MMAP) {
527 /* get the len of the mmap region */
528 unsigned long mmap_len;
529
530 ret = kernctl_get_mmap_len(stream->wait_fd, &mmap_len);
531 if (ret != 0) {
532 errno = -ret;
533 PERROR("kernctl_get_mmap_len");
534 goto error_close_fd;
535 }
536 stream->mmap_len = (size_t) mmap_len;
537
538 stream->mmap_base = mmap(NULL, stream->mmap_len, PROT_READ,
539 MAP_PRIVATE, stream->wait_fd, 0);
540 if (stream->mmap_base == MAP_FAILED) {
541 PERROR("Error mmaping");
542 ret = -1;
543 goto error_close_fd;
544 }
545 }
546
547 /* we return 0 to let the library handle the FD internally */
548 return 0;
549
550 error_close_fd:
551 {
552 int err;
553
554 err = close(stream->out_fd);
555 assert(!err);
556 }
557 error:
558 return ret;
559 }
560
561 /*
562 * Check if data is still being extracted from the buffers for a specific
563 * stream. Consumer data lock MUST be acquired before calling this function
564 * and the stream lock.
565 *
566 * Return 1 if the traced data are still getting read else 0 meaning that the
567 * data is available for trace viewer reading.
568 */
569 int lttng_kconsumer_data_pending(struct lttng_consumer_stream *stream)
570 {
571 int ret;
572
573 assert(stream);
574
575 ret = kernctl_get_next_subbuf(stream->wait_fd);
576 if (ret == 0) {
577 /* There is still data so let's put back this subbuffer. */
578 ret = kernctl_put_subbuf(stream->wait_fd);
579 assert(ret == 0);
580 ret = 1; /* Data is pending */
581 goto end;
582 }
583
584 /* Data is NOT pending and ready to be read. */
585 ret = 0;
586
587 end:
588 return ret;
589 }
This page took 0.042634 seconds and 4 git commands to generate.