Cleanup: keep the number of pipes used by poll in a variable
[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 * Copyright (C) 2017 - Jérémie Galarneau <jeremie.galarneau@efficios.com>
5 *
6 * This program is free software; you can redistribute it and/or modify
7 * it under the terms of the GNU General Public License, version 2 only,
8 * as published by the Free Software Foundation.
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 along
16 * with this program; if not, write to the Free Software Foundation, Inc.,
17 * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
18 */
19
20 #define _LGPL_SOURCE
21 #include <assert.h>
22 #include <poll.h>
23 #include <pthread.h>
24 #include <stdlib.h>
25 #include <string.h>
26 #include <sys/mman.h>
27 #include <sys/socket.h>
28 #include <sys/types.h>
29 #include <inttypes.h>
30 #include <unistd.h>
31 #include <sys/stat.h>
32
33 #include <bin/lttng-consumerd/health-consumerd.h>
34 #include <common/common.h>
35 #include <common/kernel-ctl/kernel-ctl.h>
36 #include <common/sessiond-comm/sessiond-comm.h>
37 #include <common/sessiond-comm/relayd.h>
38 #include <common/compat/fcntl.h>
39 #include <common/compat/endian.h>
40 #include <common/pipe.h>
41 #include <common/relayd/relayd.h>
42 #include <common/utils.h>
43 #include <common/consumer/consumer-stream.h>
44 #include <common/index/index.h>
45 #include <common/consumer/consumer-timer.h>
46
47 #include "kernel-consumer.h"
48
49 extern struct lttng_consumer_global_data consumer_data;
50 extern int consumer_poll_timeout;
51
52 /*
53 * Take a snapshot for a specific fd
54 *
55 * Returns 0 on success, < 0 on error
56 */
57 int lttng_kconsumer_take_snapshot(struct lttng_consumer_stream *stream)
58 {
59 int ret = 0;
60 int infd = stream->wait_fd;
61
62 ret = kernctl_snapshot(infd);
63 /*
64 * -EAGAIN is not an error, it just means that there is no data to
65 * be read.
66 */
67 if (ret != 0 && ret != -EAGAIN) {
68 PERROR("Getting sub-buffer snapshot.");
69 }
70
71 return ret;
72 }
73
74 /*
75 * Sample consumed and produced positions for a specific fd.
76 *
77 * Returns 0 on success, < 0 on error.
78 */
79 int lttng_kconsumer_sample_snapshot_positions(
80 struct lttng_consumer_stream *stream)
81 {
82 assert(stream);
83
84 return kernctl_snapshot_sample_positions(stream->wait_fd);
85 }
86
87 /*
88 * Get the produced position
89 *
90 * Returns 0 on success, < 0 on error
91 */
92 int lttng_kconsumer_get_produced_snapshot(struct lttng_consumer_stream *stream,
93 unsigned long *pos)
94 {
95 int ret;
96 int infd = stream->wait_fd;
97
98 ret = kernctl_snapshot_get_produced(infd, pos);
99 if (ret != 0) {
100 PERROR("kernctl_snapshot_get_produced");
101 }
102
103 return ret;
104 }
105
106 /*
107 * Get the consumerd position
108 *
109 * Returns 0 on success, < 0 on error
110 */
111 int lttng_kconsumer_get_consumed_snapshot(struct lttng_consumer_stream *stream,
112 unsigned long *pos)
113 {
114 int ret;
115 int infd = stream->wait_fd;
116
117 ret = kernctl_snapshot_get_consumed(infd, pos);
118 if (ret != 0) {
119 PERROR("kernctl_snapshot_get_consumed");
120 }
121
122 return ret;
123 }
124
125 /*
126 * Take a snapshot of all the stream of a channel
127 *
128 * Returns 0 on success, < 0 on error
129 */
130 int lttng_kconsumer_snapshot_channel(uint64_t key, char *path,
131 uint64_t relayd_id, uint64_t nb_packets_per_stream,
132 struct lttng_consumer_local_data *ctx)
133 {
134 int ret;
135 struct lttng_consumer_channel *channel;
136 struct lttng_consumer_stream *stream;
137
138 DBG("Kernel consumer snapshot channel %" PRIu64, key);
139
140 rcu_read_lock();
141
142 channel = consumer_find_channel(key);
143 if (!channel) {
144 ERR("No channel found for key %" PRIu64, key);
145 ret = -1;
146 goto end;
147 }
148
149 /* Splice is not supported yet for channel snapshot. */
150 if (channel->output != CONSUMER_CHANNEL_MMAP) {
151 ERR("Unsupported output %d", channel->output);
152 ret = -1;
153 goto end;
154 }
155
156 cds_list_for_each_entry(stream, &channel->streams.head, send_node) {
157 unsigned long consumed_pos, produced_pos;
158
159 health_code_update();
160
161 /*
162 * Lock stream because we are about to change its state.
163 */
164 pthread_mutex_lock(&stream->lock);
165
166 /*
167 * Assign the received relayd ID so we can use it for streaming. The streams
168 * are not visible to anyone so this is OK to change it.
169 */
170 stream->net_seq_idx = relayd_id;
171 channel->relayd_id = relayd_id;
172 if (relayd_id != (uint64_t) -1ULL) {
173 ret = consumer_send_relayd_stream(stream, path);
174 if (ret < 0) {
175 ERR("sending stream to relayd");
176 goto end_unlock;
177 }
178 } else {
179 ret = utils_create_stream_file(path, stream->name,
180 stream->chan->tracefile_size,
181 stream->tracefile_count_current,
182 stream->uid, stream->gid, NULL);
183 if (ret < 0) {
184 ERR("utils_create_stream_file");
185 goto end_unlock;
186 }
187
188 stream->out_fd = ret;
189 stream->tracefile_size_current = 0;
190
191 DBG("Kernel consumer snapshot stream %s/%s (%" PRIu64 ")",
192 path, stream->name, stream->key);
193 }
194
195 ret = kernctl_buffer_flush_empty(stream->wait_fd);
196 if (ret < 0) {
197 /*
198 * Doing a buffer flush which does not take into
199 * account empty packets. This is not perfect
200 * for stream intersection, but required as a
201 * fall-back when "flush_empty" is not
202 * implemented by lttng-modules.
203 */
204 ret = kernctl_buffer_flush(stream->wait_fd);
205 if (ret < 0) {
206 ERR("Failed to flush kernel stream");
207 goto end_unlock;
208 }
209 goto end_unlock;
210 }
211
212 ret = lttng_kconsumer_take_snapshot(stream);
213 if (ret < 0) {
214 ERR("Taking kernel snapshot");
215 goto end_unlock;
216 }
217
218 ret = lttng_kconsumer_get_produced_snapshot(stream, &produced_pos);
219 if (ret < 0) {
220 ERR("Produced kernel snapshot position");
221 goto end_unlock;
222 }
223
224 ret = lttng_kconsumer_get_consumed_snapshot(stream, &consumed_pos);
225 if (ret < 0) {
226 ERR("Consumerd kernel snapshot position");
227 goto end_unlock;
228 }
229
230 if (stream->max_sb_size == 0) {
231 ret = kernctl_get_max_subbuf_size(stream->wait_fd,
232 &stream->max_sb_size);
233 if (ret < 0) {
234 ERR("Getting kernel max_sb_size");
235 goto end_unlock;
236 }
237 }
238
239 consumed_pos = consumer_get_consume_start_pos(consumed_pos,
240 produced_pos, nb_packets_per_stream,
241 stream->max_sb_size);
242
243 while (consumed_pos < produced_pos) {
244 ssize_t read_len;
245 unsigned long len, padded_len;
246
247 health_code_update();
248
249 DBG("Kernel consumer taking snapshot at pos %lu", consumed_pos);
250
251 ret = kernctl_get_subbuf(stream->wait_fd, &consumed_pos);
252 if (ret < 0) {
253 if (ret != -EAGAIN) {
254 PERROR("kernctl_get_subbuf snapshot");
255 goto end_unlock;
256 }
257 DBG("Kernel consumer get subbuf failed. Skipping it.");
258 consumed_pos += stream->max_sb_size;
259 stream->chan->lost_packets++;
260 continue;
261 }
262
263 ret = kernctl_get_subbuf_size(stream->wait_fd, &len);
264 if (ret < 0) {
265 ERR("Snapshot kernctl_get_subbuf_size");
266 goto error_put_subbuf;
267 }
268
269 ret = kernctl_get_padded_subbuf_size(stream->wait_fd, &padded_len);
270 if (ret < 0) {
271 ERR("Snapshot kernctl_get_padded_subbuf_size");
272 goto error_put_subbuf;
273 }
274
275 read_len = lttng_consumer_on_read_subbuffer_mmap(ctx, stream, len,
276 padded_len - len, NULL);
277 /*
278 * We write the padded len in local tracefiles but the data len
279 * when using a relay. Display the error but continue processing
280 * to try to release the subbuffer.
281 */
282 if (relayd_id != (uint64_t) -1ULL) {
283 if (read_len != len) {
284 ERR("Error sending to the relay (ret: %zd != len: %lu)",
285 read_len, len);
286 }
287 } else {
288 if (read_len != padded_len) {
289 ERR("Error writing to tracefile (ret: %zd != len: %lu)",
290 read_len, padded_len);
291 }
292 }
293
294 ret = kernctl_put_subbuf(stream->wait_fd);
295 if (ret < 0) {
296 ERR("Snapshot kernctl_put_subbuf");
297 goto end_unlock;
298 }
299 consumed_pos += stream->max_sb_size;
300 }
301
302 if (relayd_id == (uint64_t) -1ULL) {
303 if (stream->out_fd >= 0) {
304 ret = close(stream->out_fd);
305 if (ret < 0) {
306 PERROR("Kernel consumer snapshot close out_fd");
307 goto end_unlock;
308 }
309 stream->out_fd = -1;
310 }
311 } else {
312 close_relayd_stream(stream);
313 stream->net_seq_idx = (uint64_t) -1ULL;
314 }
315 pthread_mutex_unlock(&stream->lock);
316 }
317
318 /* All good! */
319 ret = 0;
320 goto end;
321
322 error_put_subbuf:
323 ret = kernctl_put_subbuf(stream->wait_fd);
324 if (ret < 0) {
325 ERR("Snapshot kernctl_put_subbuf error path");
326 }
327 end_unlock:
328 pthread_mutex_unlock(&stream->lock);
329 end:
330 rcu_read_unlock();
331 return ret;
332 }
333
334 /*
335 * Read the whole metadata available for a snapshot.
336 *
337 * Returns 0 on success, < 0 on error
338 */
339 int lttng_kconsumer_snapshot_metadata(uint64_t key, char *path,
340 uint64_t relayd_id, struct lttng_consumer_local_data *ctx)
341 {
342 int ret, use_relayd = 0;
343 ssize_t ret_read;
344 struct lttng_consumer_channel *metadata_channel;
345 struct lttng_consumer_stream *metadata_stream;
346
347 assert(ctx);
348
349 DBG("Kernel consumer snapshot metadata with key %" PRIu64 " at path %s",
350 key, path);
351
352 rcu_read_lock();
353
354 metadata_channel = consumer_find_channel(key);
355 if (!metadata_channel) {
356 ERR("Kernel snapshot metadata not found for key %" PRIu64, key);
357 ret = -1;
358 goto error;
359 }
360
361 metadata_stream = metadata_channel->metadata_stream;
362 assert(metadata_stream);
363
364 /* Flag once that we have a valid relayd for the stream. */
365 if (relayd_id != (uint64_t) -1ULL) {
366 use_relayd = 1;
367 }
368
369 if (use_relayd) {
370 ret = consumer_send_relayd_stream(metadata_stream, path);
371 if (ret < 0) {
372 goto error;
373 }
374 } else {
375 ret = utils_create_stream_file(path, metadata_stream->name,
376 metadata_stream->chan->tracefile_size,
377 metadata_stream->tracefile_count_current,
378 metadata_stream->uid, metadata_stream->gid, NULL);
379 if (ret < 0) {
380 goto error;
381 }
382 metadata_stream->out_fd = ret;
383 }
384
385 do {
386 health_code_update();
387
388 ret_read = lttng_kconsumer_read_subbuffer(metadata_stream, ctx);
389 if (ret_read < 0) {
390 if (ret_read != -EAGAIN) {
391 ERR("Kernel snapshot reading metadata subbuffer (ret: %zd)",
392 ret_read);
393 goto error;
394 }
395 /* ret_read is negative at this point so we will exit the loop. */
396 continue;
397 }
398 } while (ret_read >= 0);
399
400 if (use_relayd) {
401 close_relayd_stream(metadata_stream);
402 metadata_stream->net_seq_idx = (uint64_t) -1ULL;
403 } else {
404 if (metadata_stream->out_fd >= 0) {
405 ret = close(metadata_stream->out_fd);
406 if (ret < 0) {
407 PERROR("Kernel consumer snapshot metadata close out_fd");
408 /*
409 * Don't go on error here since the snapshot was successful at this
410 * point but somehow the close failed.
411 */
412 }
413 metadata_stream->out_fd = -1;
414 }
415 }
416
417 ret = 0;
418
419 cds_list_del(&metadata_stream->send_node);
420 consumer_stream_destroy(metadata_stream, NULL);
421 metadata_channel->metadata_stream = NULL;
422 error:
423 rcu_read_unlock();
424 return ret;
425 }
426
427 /*
428 * Receive command from session daemon and process it.
429 *
430 * Return 1 on success else a negative value or 0.
431 */
432 int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
433 int sock, struct pollfd *consumer_sockpoll)
434 {
435 ssize_t ret;
436 enum lttcomm_return_code ret_code = LTTCOMM_CONSUMERD_SUCCESS;
437 struct lttcomm_consumer_msg msg;
438
439 health_code_update();
440
441 ret = lttcomm_recv_unix_sock(sock, &msg, sizeof(msg));
442 if (ret != sizeof(msg)) {
443 if (ret > 0) {
444 lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_ERROR_RECV_CMD);
445 ret = -1;
446 }
447 return ret;
448 }
449
450 health_code_update();
451
452 /* Deprecated command */
453 assert(msg.cmd_type != LTTNG_CONSUMER_STOP);
454
455 health_code_update();
456
457 /* relayd needs RCU read-side protection */
458 rcu_read_lock();
459
460 switch (msg.cmd_type) {
461 case LTTNG_CONSUMER_ADD_RELAYD_SOCKET:
462 {
463 /* Session daemon status message are handled in the following call. */
464 consumer_add_relayd_socket(msg.u.relayd_sock.net_index,
465 msg.u.relayd_sock.type, ctx, sock, consumer_sockpoll,
466 &msg.u.relayd_sock.sock, msg.u.relayd_sock.session_id,
467 msg.u.relayd_sock.relayd_session_id);
468 goto end_nosignal;
469 }
470 case LTTNG_CONSUMER_ADD_CHANNEL:
471 {
472 struct lttng_consumer_channel *new_channel;
473 int ret_recv;
474
475 health_code_update();
476
477 /* First send a status message before receiving the fds. */
478 ret = consumer_send_status_msg(sock, ret_code);
479 if (ret < 0) {
480 /* Somehow, the session daemon is not responding anymore. */
481 goto error_fatal;
482 }
483
484 health_code_update();
485
486 DBG("consumer_add_channel %" PRIu64, msg.u.channel.channel_key);
487 new_channel = consumer_allocate_channel(msg.u.channel.channel_key,
488 msg.u.channel.session_id, msg.u.channel.pathname,
489 msg.u.channel.name, msg.u.channel.uid, msg.u.channel.gid,
490 msg.u.channel.relayd_id, msg.u.channel.output,
491 msg.u.channel.tracefile_size,
492 msg.u.channel.tracefile_count, 0,
493 msg.u.channel.monitor,
494 msg.u.channel.live_timer_interval,
495 NULL, NULL);
496 if (new_channel == NULL) {
497 lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_OUTFD_ERROR);
498 goto end_nosignal;
499 }
500 new_channel->nb_init_stream_left = msg.u.channel.nb_init_streams;
501 switch (msg.u.channel.output) {
502 case LTTNG_EVENT_SPLICE:
503 new_channel->output = CONSUMER_CHANNEL_SPLICE;
504 break;
505 case LTTNG_EVENT_MMAP:
506 new_channel->output = CONSUMER_CHANNEL_MMAP;
507 break;
508 default:
509 ERR("Channel output unknown %d", msg.u.channel.output);
510 goto end_nosignal;
511 }
512
513 /* Translate and save channel type. */
514 switch (msg.u.channel.type) {
515 case CONSUMER_CHANNEL_TYPE_DATA:
516 case CONSUMER_CHANNEL_TYPE_METADATA:
517 new_channel->type = msg.u.channel.type;
518 break;
519 default:
520 assert(0);
521 goto end_nosignal;
522 };
523
524 health_code_update();
525
526 if (ctx->on_recv_channel != NULL) {
527 ret_recv = ctx->on_recv_channel(new_channel);
528 if (ret_recv == 0) {
529 ret = consumer_add_channel(new_channel, ctx);
530 } else if (ret_recv < 0) {
531 goto end_nosignal;
532 }
533 } else {
534 ret = consumer_add_channel(new_channel, ctx);
535 }
536 if (msg.u.channel.type == CONSUMER_CHANNEL_TYPE_DATA && !ret) {
537 int monitor_start_ret;
538
539 DBG("Consumer starting monitor timer");
540 consumer_timer_live_start(new_channel,
541 msg.u.channel.live_timer_interval);
542 monitor_start_ret = consumer_timer_monitor_start(
543 new_channel,
544 msg.u.channel.monitor_timer_interval);
545 if (monitor_start_ret < 0) {
546 ERR("Starting channel monitoring timer failed");
547 goto end_nosignal;
548 }
549
550 }
551
552 health_code_update();
553
554 /* If we received an error in add_channel, we need to report it. */
555 if (ret < 0) {
556 ret = consumer_send_status_msg(sock, ret);
557 if (ret < 0) {
558 goto error_fatal;
559 }
560 goto end_nosignal;
561 }
562
563 goto end_nosignal;
564 }
565 case LTTNG_CONSUMER_ADD_STREAM:
566 {
567 int fd;
568 struct lttng_pipe *stream_pipe;
569 struct lttng_consumer_stream *new_stream;
570 struct lttng_consumer_channel *channel;
571 int alloc_ret = 0;
572
573 /*
574 * Get stream's channel reference. Needed when adding the stream to the
575 * global hash table.
576 */
577 channel = consumer_find_channel(msg.u.stream.channel_key);
578 if (!channel) {
579 /*
580 * We could not find the channel. Can happen if cpu hotplug
581 * happens while tearing down.
582 */
583 ERR("Unable to find channel key %" PRIu64, msg.u.stream.channel_key);
584 ret_code = LTTCOMM_CONSUMERD_CHAN_NOT_FOUND;
585 }
586
587 health_code_update();
588
589 /* First send a status message before receiving the fds. */
590 ret = consumer_send_status_msg(sock, ret_code);
591 if (ret < 0) {
592 /* Somehow, the session daemon is not responding anymore. */
593 goto error_fatal;
594 }
595
596 health_code_update();
597
598 if (ret_code != LTTCOMM_CONSUMERD_SUCCESS) {
599 /* Channel was not found. */
600 goto end_nosignal;
601 }
602
603 /* Blocking call */
604 health_poll_entry();
605 ret = lttng_consumer_poll_socket(consumer_sockpoll);
606 health_poll_exit();
607 if (ret) {
608 goto error_fatal;
609 }
610
611 health_code_update();
612
613 /* Get stream file descriptor from socket */
614 ret = lttcomm_recv_fds_unix_sock(sock, &fd, 1);
615 if (ret != sizeof(fd)) {
616 lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_ERROR_RECV_FD);
617 rcu_read_unlock();
618 return ret;
619 }
620
621 health_code_update();
622
623 /*
624 * Send status code to session daemon only if the recv works. If the
625 * above recv() failed, the session daemon is notified through the
626 * error socket and the teardown is eventually done.
627 */
628 ret = consumer_send_status_msg(sock, ret_code);
629 if (ret < 0) {
630 /* Somehow, the session daemon is not responding anymore. */
631 goto end_nosignal;
632 }
633
634 health_code_update();
635
636 new_stream = consumer_allocate_stream(channel->key,
637 fd,
638 LTTNG_CONSUMER_ACTIVE_STREAM,
639 channel->name,
640 channel->uid,
641 channel->gid,
642 channel->relayd_id,
643 channel->session_id,
644 msg.u.stream.cpu,
645 &alloc_ret,
646 channel->type,
647 channel->monitor);
648 if (new_stream == NULL) {
649 switch (alloc_ret) {
650 case -ENOMEM:
651 case -EINVAL:
652 default:
653 lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_OUTFD_ERROR);
654 break;
655 }
656 goto end_nosignal;
657 }
658
659 new_stream->chan = channel;
660 new_stream->wait_fd = fd;
661 switch (channel->output) {
662 case CONSUMER_CHANNEL_SPLICE:
663 new_stream->output = LTTNG_EVENT_SPLICE;
664 ret = utils_create_pipe(new_stream->splice_pipe);
665 if (ret < 0) {
666 goto end_nosignal;
667 }
668 break;
669 case CONSUMER_CHANNEL_MMAP:
670 new_stream->output = LTTNG_EVENT_MMAP;
671 break;
672 default:
673 ERR("Stream output unknown %d", channel->output);
674 goto end_nosignal;
675 }
676
677 /*
678 * We've just assigned the channel to the stream so increment the
679 * refcount right now. We don't need to increment the refcount for
680 * streams in no monitor because we handle manually the cleanup of
681 * those. It is very important to make sure there is NO prior
682 * consumer_del_stream() calls or else the refcount will be unbalanced.
683 */
684 if (channel->monitor) {
685 uatomic_inc(&new_stream->chan->refcount);
686 }
687
688 /*
689 * The buffer flush is done on the session daemon side for the kernel
690 * so no need for the stream "hangup_flush_done" variable to be
691 * tracked. This is important for a kernel stream since we don't rely
692 * on the flush state of the stream to read data. It's not the case for
693 * user space tracing.
694 */
695 new_stream->hangup_flush_done = 0;
696
697 health_code_update();
698
699 if (ctx->on_recv_stream) {
700 ret = ctx->on_recv_stream(new_stream);
701 if (ret < 0) {
702 consumer_stream_free(new_stream);
703 goto end_nosignal;
704 }
705 }
706
707 health_code_update();
708
709 if (new_stream->metadata_flag) {
710 channel->metadata_stream = new_stream;
711 }
712
713 /* Do not monitor this stream. */
714 if (!channel->monitor) {
715 DBG("Kernel consumer add stream %s in no monitor mode with "
716 "relayd id %" PRIu64, new_stream->name,
717 new_stream->net_seq_idx);
718 cds_list_add(&new_stream->send_node, &channel->streams.head);
719 break;
720 }
721
722 /* Send stream to relayd if the stream has an ID. */
723 if (new_stream->net_seq_idx != (uint64_t) -1ULL) {
724 ret = consumer_send_relayd_stream(new_stream,
725 new_stream->chan->pathname);
726 if (ret < 0) {
727 consumer_stream_free(new_stream);
728 goto end_nosignal;
729 }
730
731 /*
732 * If adding an extra stream to an already
733 * existing channel (e.g. cpu hotplug), we need
734 * to send the "streams_sent" command to relayd.
735 */
736 if (channel->streams_sent_to_relayd) {
737 ret = consumer_send_relayd_streams_sent(
738 new_stream->net_seq_idx);
739 if (ret < 0) {
740 goto end_nosignal;
741 }
742 }
743 }
744
745 /* Get the right pipe where the stream will be sent. */
746 if (new_stream->metadata_flag) {
747 consumer_add_metadata_stream(new_stream);
748 stream_pipe = ctx->consumer_metadata_pipe;
749 } else {
750 consumer_add_data_stream(new_stream);
751 stream_pipe = ctx->consumer_data_pipe;
752 }
753
754 /* Visible to other threads */
755 new_stream->globally_visible = 1;
756
757 health_code_update();
758
759 ret = lttng_pipe_write(stream_pipe, &new_stream, sizeof(new_stream));
760 if (ret < 0) {
761 ERR("Consumer write %s stream to pipe %d",
762 new_stream->metadata_flag ? "metadata" : "data",
763 lttng_pipe_get_writefd(stream_pipe));
764 if (new_stream->metadata_flag) {
765 consumer_del_stream_for_metadata(new_stream);
766 } else {
767 consumer_del_stream_for_data(new_stream);
768 }
769 goto end_nosignal;
770 }
771
772 DBG("Kernel consumer ADD_STREAM %s (fd: %d) with relayd id %" PRIu64,
773 new_stream->name, fd, new_stream->relayd_stream_id);
774 break;
775 }
776 case LTTNG_CONSUMER_STREAMS_SENT:
777 {
778 struct lttng_consumer_channel *channel;
779
780 /*
781 * Get stream's channel reference. Needed when adding the stream to the
782 * global hash table.
783 */
784 channel = consumer_find_channel(msg.u.sent_streams.channel_key);
785 if (!channel) {
786 /*
787 * We could not find the channel. Can happen if cpu hotplug
788 * happens while tearing down.
789 */
790 ERR("Unable to find channel key %" PRIu64,
791 msg.u.sent_streams.channel_key);
792 ret_code = LTTCOMM_CONSUMERD_CHAN_NOT_FOUND;
793 }
794
795 health_code_update();
796
797 /*
798 * Send status code to session daemon.
799 */
800 ret = consumer_send_status_msg(sock, ret_code);
801 if (ret < 0 || ret_code != LTTCOMM_CONSUMERD_SUCCESS) {
802 /* Somehow, the session daemon is not responding anymore. */
803 goto end_nosignal;
804 }
805
806 health_code_update();
807
808 /*
809 * We should not send this message if we don't monitor the
810 * streams in this channel.
811 */
812 if (!channel->monitor) {
813 break;
814 }
815
816 health_code_update();
817 /* Send stream to relayd if the stream has an ID. */
818 if (msg.u.sent_streams.net_seq_idx != (uint64_t) -1ULL) {
819 ret = consumer_send_relayd_streams_sent(
820 msg.u.sent_streams.net_seq_idx);
821 if (ret < 0) {
822 goto end_nosignal;
823 }
824 channel->streams_sent_to_relayd = true;
825 }
826 break;
827 }
828 case LTTNG_CONSUMER_UPDATE_STREAM:
829 {
830 rcu_read_unlock();
831 return -ENOSYS;
832 }
833 case LTTNG_CONSUMER_DESTROY_RELAYD:
834 {
835 uint64_t index = msg.u.destroy_relayd.net_seq_idx;
836 struct consumer_relayd_sock_pair *relayd;
837
838 DBG("Kernel consumer destroying relayd %" PRIu64, index);
839
840 /* Get relayd reference if exists. */
841 relayd = consumer_find_relayd(index);
842 if (relayd == NULL) {
843 DBG("Unable to find relayd %" PRIu64, index);
844 ret_code = LTTCOMM_CONSUMERD_RELAYD_FAIL;
845 }
846
847 /*
848 * Each relayd socket pair has a refcount of stream attached to it
849 * which tells if the relayd is still active or not depending on the
850 * refcount value.
851 *
852 * This will set the destroy flag of the relayd object and destroy it
853 * if the refcount reaches zero when called.
854 *
855 * The destroy can happen either here or when a stream fd hangs up.
856 */
857 if (relayd) {
858 consumer_flag_relayd_for_destroy(relayd);
859 }
860
861 health_code_update();
862
863 ret = consumer_send_status_msg(sock, ret_code);
864 if (ret < 0) {
865 /* Somehow, the session daemon is not responding anymore. */
866 goto error_fatal;
867 }
868
869 goto end_nosignal;
870 }
871 case LTTNG_CONSUMER_DATA_PENDING:
872 {
873 int32_t ret;
874 uint64_t id = msg.u.data_pending.session_id;
875
876 DBG("Kernel consumer data pending command for id %" PRIu64, id);
877
878 ret = consumer_data_pending(id);
879
880 health_code_update();
881
882 /* Send back returned value to session daemon */
883 ret = lttcomm_send_unix_sock(sock, &ret, sizeof(ret));
884 if (ret < 0) {
885 PERROR("send data pending ret code");
886 goto error_fatal;
887 }
888
889 /*
890 * No need to send back a status message since the data pending
891 * returned value is the response.
892 */
893 break;
894 }
895 case LTTNG_CONSUMER_SNAPSHOT_CHANNEL:
896 {
897 if (msg.u.snapshot_channel.metadata == 1) {
898 ret = lttng_kconsumer_snapshot_metadata(msg.u.snapshot_channel.key,
899 msg.u.snapshot_channel.pathname,
900 msg.u.snapshot_channel.relayd_id, ctx);
901 if (ret < 0) {
902 ERR("Snapshot metadata failed");
903 ret_code = LTTCOMM_CONSUMERD_ERROR_METADATA;
904 }
905 } else {
906 ret = lttng_kconsumer_snapshot_channel(msg.u.snapshot_channel.key,
907 msg.u.snapshot_channel.pathname,
908 msg.u.snapshot_channel.relayd_id,
909 msg.u.snapshot_channel.nb_packets_per_stream,
910 ctx);
911 if (ret < 0) {
912 ERR("Snapshot channel failed");
913 ret_code = LTTCOMM_CONSUMERD_CHAN_NOT_FOUND;
914 }
915 }
916
917 health_code_update();
918
919 ret = consumer_send_status_msg(sock, ret_code);
920 if (ret < 0) {
921 /* Somehow, the session daemon is not responding anymore. */
922 goto end_nosignal;
923 }
924 break;
925 }
926 case LTTNG_CONSUMER_DESTROY_CHANNEL:
927 {
928 uint64_t key = msg.u.destroy_channel.key;
929 struct lttng_consumer_channel *channel;
930
931 channel = consumer_find_channel(key);
932 if (!channel) {
933 ERR("Kernel consumer destroy channel %" PRIu64 " not found", key);
934 ret_code = LTTCOMM_CONSUMERD_CHAN_NOT_FOUND;
935 }
936
937 health_code_update();
938
939 ret = consumer_send_status_msg(sock, ret_code);
940 if (ret < 0) {
941 /* Somehow, the session daemon is not responding anymore. */
942 goto end_nosignal;
943 }
944
945 health_code_update();
946
947 /* Stop right now if no channel was found. */
948 if (!channel) {
949 goto end_nosignal;
950 }
951
952 /*
953 * This command should ONLY be issued for channel with streams set in
954 * no monitor mode.
955 */
956 assert(!channel->monitor);
957
958 /*
959 * The refcount should ALWAYS be 0 in the case of a channel in no
960 * monitor mode.
961 */
962 assert(!uatomic_sub_return(&channel->refcount, 1));
963
964 consumer_del_channel(channel);
965
966 goto end_nosignal;
967 }
968 case LTTNG_CONSUMER_DISCARDED_EVENTS:
969 {
970 ssize_t ret;
971 uint64_t count;
972 struct lttng_consumer_channel *channel;
973 uint64_t id = msg.u.discarded_events.session_id;
974 uint64_t key = msg.u.discarded_events.channel_key;
975
976 DBG("Kernel consumer discarded events command for session id %"
977 PRIu64 ", channel key %" PRIu64, id, key);
978
979 channel = consumer_find_channel(key);
980 if (!channel) {
981 ERR("Kernel consumer discarded events channel %"
982 PRIu64 " not found", key);
983 count = 0;
984 } else {
985 count = channel->discarded_events;
986 }
987
988 health_code_update();
989
990 /* Send back returned value to session daemon */
991 ret = lttcomm_send_unix_sock(sock, &count, sizeof(count));
992 if (ret < 0) {
993 PERROR("send discarded events");
994 goto error_fatal;
995 }
996
997 break;
998 }
999 case LTTNG_CONSUMER_LOST_PACKETS:
1000 {
1001 ssize_t ret;
1002 uint64_t count;
1003 struct lttng_consumer_channel *channel;
1004 uint64_t id = msg.u.lost_packets.session_id;
1005 uint64_t key = msg.u.lost_packets.channel_key;
1006
1007 DBG("Kernel consumer lost packets command for session id %"
1008 PRIu64 ", channel key %" PRIu64, id, key);
1009
1010 channel = consumer_find_channel(key);
1011 if (!channel) {
1012 ERR("Kernel consumer lost packets channel %"
1013 PRIu64 " not found", key);
1014 count = 0;
1015 } else {
1016 count = channel->lost_packets;
1017 }
1018
1019 health_code_update();
1020
1021 /* Send back returned value to session daemon */
1022 ret = lttcomm_send_unix_sock(sock, &count, sizeof(count));
1023 if (ret < 0) {
1024 PERROR("send lost packets");
1025 goto error_fatal;
1026 }
1027
1028 break;
1029 }
1030 case LTTNG_CONSUMER_SET_CHANNEL_MONITOR_PIPE:
1031 {
1032 int channel_monitor_pipe;
1033
1034 ret_code = LTTCOMM_CONSUMERD_SUCCESS;
1035 /* Successfully received the command's type. */
1036 ret = consumer_send_status_msg(sock, ret_code);
1037 if (ret < 0) {
1038 goto error_fatal;
1039 }
1040
1041 ret = lttcomm_recv_fds_unix_sock(sock, &channel_monitor_pipe,
1042 1);
1043 if (ret != sizeof(channel_monitor_pipe)) {
1044 ERR("Failed to receive channel monitor pipe");
1045 goto error_fatal;
1046 }
1047
1048 DBG("Received channel monitor pipe (%d)", channel_monitor_pipe);
1049 ret = consumer_timer_thread_set_channel_monitor_pipe(
1050 channel_monitor_pipe);
1051 if (!ret) {
1052 int flags;
1053
1054 ret_code = LTTCOMM_CONSUMERD_SUCCESS;
1055 /* Set the pipe as non-blocking. */
1056 ret = fcntl(channel_monitor_pipe, F_GETFL, 0);
1057 if (ret == -1) {
1058 PERROR("fcntl get flags of the channel monitoring pipe");
1059 goto error_fatal;
1060 }
1061 flags = ret;
1062
1063 ret = fcntl(channel_monitor_pipe, F_SETFL,
1064 flags | O_NONBLOCK);
1065 if (ret == -1) {
1066 PERROR("fcntl set O_NONBLOCK flag of the channel monitoring pipe");
1067 goto error_fatal;
1068 }
1069 DBG("Channel monitor pipe set as non-blocking");
1070 } else {
1071 ret_code = LTTCOMM_CONSUMERD_ALREADY_SET;
1072 }
1073 ret = consumer_send_status_msg(sock, ret_code);
1074 if (ret < 0) {
1075 goto error_fatal;
1076 }
1077 break;
1078 }
1079 case LTTNG_CONSUMER_ROTATE_RENAME:
1080 {
1081 DBG("Consumer rename session %" PRIu64 " after rotation, old path = \"%s\", new path = \"%s\"",
1082 msg.u.rotate_rename.session_id,
1083 msg.u.rotate_rename.old_path,
1084 msg.u.rotate_rename.new_path);
1085 ret = lttng_consumer_rotate_rename(msg.u.rotate_rename.old_path,
1086 msg.u.rotate_rename.new_path,
1087 msg.u.rotate_rename.uid,
1088 msg.u.rotate_rename.gid,
1089 msg.u.rotate_rename.relayd_id);
1090 if (ret < 0) {
1091 ERR("Rotate rename failed");
1092 ret_code = LTTCOMM_CONSUMERD_CHAN_NOT_FOUND;
1093 }
1094
1095 health_code_update();
1096
1097 ret = consumer_send_status_msg(sock, ret_code);
1098 if (ret < 0) {
1099 /* Somehow, the session daemon is not responding anymore. */
1100 goto end_nosignal;
1101 }
1102 break;
1103 }
1104 case LTTNG_CONSUMER_MKDIR:
1105 {
1106 DBG("Consumer mkdir %s in session %" PRIu64,
1107 msg.u.mkdir.path,
1108 msg.u.mkdir.session_id);
1109 ret = lttng_consumer_mkdir(msg.u.mkdir.path,
1110 msg.u.mkdir.uid,
1111 msg.u.mkdir.gid,
1112 msg.u.mkdir.relayd_id);
1113 if (ret < 0) {
1114 ERR("consumer mkdir failed");
1115 ret_code = LTTCOMM_CONSUMERD_CHAN_NOT_FOUND;
1116 }
1117
1118 health_code_update();
1119
1120 ret = consumer_send_status_msg(sock, ret_code);
1121 if (ret < 0) {
1122 /* Somehow, the session daemon is not responding anymore. */
1123 goto end_nosignal;
1124 }
1125 break;
1126 }
1127 default:
1128 goto end_nosignal;
1129 }
1130
1131 end_nosignal:
1132 rcu_read_unlock();
1133
1134 /*
1135 * Return 1 to indicate success since the 0 value can be a socket
1136 * shutdown during the recv() or send() call.
1137 */
1138 health_code_update();
1139 return 1;
1140
1141 error_fatal:
1142 rcu_read_unlock();
1143 /* This will issue a consumer stop. */
1144 return -1;
1145 }
1146
1147 /*
1148 * Populate index values of a kernel stream. Values are set in big endian order.
1149 *
1150 * Return 0 on success or else a negative value.
1151 */
1152 static int get_index_values(struct ctf_packet_index *index, int infd)
1153 {
1154 int ret;
1155
1156 ret = kernctl_get_timestamp_begin(infd, &index->timestamp_begin);
1157 if (ret < 0) {
1158 PERROR("kernctl_get_timestamp_begin");
1159 goto error;
1160 }
1161 index->timestamp_begin = htobe64(index->timestamp_begin);
1162
1163 ret = kernctl_get_timestamp_end(infd, &index->timestamp_end);
1164 if (ret < 0) {
1165 PERROR("kernctl_get_timestamp_end");
1166 goto error;
1167 }
1168 index->timestamp_end = htobe64(index->timestamp_end);
1169
1170 ret = kernctl_get_events_discarded(infd, &index->events_discarded);
1171 if (ret < 0) {
1172 PERROR("kernctl_get_events_discarded");
1173 goto error;
1174 }
1175 index->events_discarded = htobe64(index->events_discarded);
1176
1177 ret = kernctl_get_content_size(infd, &index->content_size);
1178 if (ret < 0) {
1179 PERROR("kernctl_get_content_size");
1180 goto error;
1181 }
1182 index->content_size = htobe64(index->content_size);
1183
1184 ret = kernctl_get_packet_size(infd, &index->packet_size);
1185 if (ret < 0) {
1186 PERROR("kernctl_get_packet_size");
1187 goto error;
1188 }
1189 index->packet_size = htobe64(index->packet_size);
1190
1191 ret = kernctl_get_stream_id(infd, &index->stream_id);
1192 if (ret < 0) {
1193 PERROR("kernctl_get_stream_id");
1194 goto error;
1195 }
1196 index->stream_id = htobe64(index->stream_id);
1197
1198 ret = kernctl_get_instance_id(infd, &index->stream_instance_id);
1199 if (ret < 0) {
1200 if (ret == -ENOTTY) {
1201 /* Command not implemented by lttng-modules. */
1202 index->stream_instance_id = -1ULL;
1203 } else {
1204 PERROR("kernctl_get_instance_id");
1205 goto error;
1206 }
1207 }
1208 index->stream_instance_id = htobe64(index->stream_instance_id);
1209
1210 ret = kernctl_get_sequence_number(infd, &index->packet_seq_num);
1211 if (ret < 0) {
1212 if (ret == -ENOTTY) {
1213 /* Command not implemented by lttng-modules. */
1214 index->packet_seq_num = -1ULL;
1215 ret = 0;
1216 } else {
1217 PERROR("kernctl_get_sequence_number");
1218 goto error;
1219 }
1220 }
1221 index->packet_seq_num = htobe64(index->packet_seq_num);
1222
1223 error:
1224 return ret;
1225 }
1226 /*
1227 * Sync metadata meaning request them to the session daemon and snapshot to the
1228 * metadata thread can consumer them.
1229 *
1230 * Metadata stream lock MUST be acquired.
1231 *
1232 * Return 0 if new metadatda is available, EAGAIN if the metadata stream
1233 * is empty or a negative value on error.
1234 */
1235 int lttng_kconsumer_sync_metadata(struct lttng_consumer_stream *metadata)
1236 {
1237 int ret;
1238
1239 assert(metadata);
1240
1241 ret = kernctl_buffer_flush(metadata->wait_fd);
1242 if (ret < 0) {
1243 ERR("Failed to flush kernel stream");
1244 goto end;
1245 }
1246
1247 ret = kernctl_snapshot(metadata->wait_fd);
1248 if (ret < 0) {
1249 if (ret != -EAGAIN) {
1250 ERR("Sync metadata, taking kernel snapshot failed.");
1251 goto end;
1252 }
1253 DBG("Sync metadata, no new kernel metadata");
1254 /* No new metadata, exit. */
1255 ret = ENODATA;
1256 goto end;
1257 }
1258
1259 end:
1260 return ret;
1261 }
1262
1263 static
1264 int update_stream_stats(struct lttng_consumer_stream *stream)
1265 {
1266 int ret;
1267 uint64_t seq, discarded;
1268
1269 ret = kernctl_get_sequence_number(stream->wait_fd, &seq);
1270 if (ret < 0) {
1271 if (ret == -ENOTTY) {
1272 /* Command not implemented by lttng-modules. */
1273 seq = -1ULL;
1274 } else {
1275 PERROR("kernctl_get_sequence_number");
1276 goto end;
1277 }
1278 }
1279
1280 /*
1281 * Start the sequence when we extract the first packet in case we don't
1282 * start at 0 (for example if a consumer is not connected to the
1283 * session immediately after the beginning).
1284 */
1285 if (stream->last_sequence_number == -1ULL) {
1286 stream->last_sequence_number = seq;
1287 } else if (seq > stream->last_sequence_number) {
1288 stream->chan->lost_packets += seq -
1289 stream->last_sequence_number - 1;
1290 } else {
1291 /* seq <= last_sequence_number */
1292 ERR("Sequence number inconsistent : prev = %" PRIu64
1293 ", current = %" PRIu64,
1294 stream->last_sequence_number, seq);
1295 ret = -1;
1296 goto end;
1297 }
1298 stream->last_sequence_number = seq;
1299
1300 ret = kernctl_get_events_discarded(stream->wait_fd, &discarded);
1301 if (ret < 0) {
1302 PERROR("kernctl_get_events_discarded");
1303 goto end;
1304 }
1305 if (discarded < stream->last_discarded_events) {
1306 /*
1307 * Overflow has occurred. We assume only one wrap-around
1308 * has occurred.
1309 */
1310 stream->chan->discarded_events += (1ULL << (CAA_BITS_PER_LONG - 1)) -
1311 stream->last_discarded_events + discarded;
1312 } else {
1313 stream->chan->discarded_events += discarded -
1314 stream->last_discarded_events;
1315 }
1316 stream->last_discarded_events = discarded;
1317 ret = 0;
1318
1319 end:
1320 return ret;
1321 }
1322
1323 /*
1324 * Check if the local version of the metadata stream matches with the version
1325 * of the metadata stream in the kernel. If it was updated, set the reset flag
1326 * on the stream.
1327 */
1328 static
1329 int metadata_stream_check_version(int infd, struct lttng_consumer_stream *stream)
1330 {
1331 int ret;
1332 uint64_t cur_version;
1333
1334 ret = kernctl_get_metadata_version(infd, &cur_version);
1335 if (ret < 0) {
1336 if (ret == -ENOTTY) {
1337 /*
1338 * LTTng-modules does not implement this
1339 * command.
1340 */
1341 ret = 0;
1342 goto end;
1343 }
1344 ERR("Failed to get the metadata version");
1345 goto end;
1346 }
1347
1348 if (stream->metadata_version == cur_version) {
1349 ret = 0;
1350 goto end;
1351 }
1352
1353 DBG("New metadata version detected");
1354 stream->metadata_version = cur_version;
1355 stream->reset_metadata_flag = 1;
1356 ret = 0;
1357
1358 end:
1359 return ret;
1360 }
1361
1362 /*
1363 * Consume data on a file descriptor and write it on a trace file.
1364 */
1365 ssize_t lttng_kconsumer_read_subbuffer(struct lttng_consumer_stream *stream,
1366 struct lttng_consumer_local_data *ctx)
1367 {
1368 unsigned long len, subbuf_size, padding;
1369 int err, write_index = 1;
1370 ssize_t ret = 0;
1371 int infd = stream->wait_fd;
1372 struct ctf_packet_index index;
1373
1374 DBG("In read_subbuffer (infd : %d)", infd);
1375
1376 /* Get the next subbuffer */
1377 err = kernctl_get_next_subbuf(infd);
1378 if (err != 0) {
1379 /*
1380 * This is a debug message even for single-threaded consumer,
1381 * because poll() have more relaxed criterions than get subbuf,
1382 * so get_subbuf may fail for short race windows where poll()
1383 * would issue wakeups.
1384 */
1385 DBG("Reserving sub buffer failed (everything is normal, "
1386 "it is due to concurrency)");
1387 ret = err;
1388 goto end;
1389 }
1390
1391 /* Get the full subbuffer size including padding */
1392 err = kernctl_get_padded_subbuf_size(infd, &len);
1393 if (err != 0) {
1394 PERROR("Getting sub-buffer len failed.");
1395 err = kernctl_put_subbuf(infd);
1396 if (err != 0) {
1397 if (err == -EFAULT) {
1398 PERROR("Error in unreserving sub buffer\n");
1399 } else if (err == -EIO) {
1400 /* Should never happen with newer LTTng versions */
1401 PERROR("Reader has been pushed by the writer, last sub-buffer corrupted.");
1402 }
1403 ret = err;
1404 goto end;
1405 }
1406 ret = err;
1407 goto end;
1408 }
1409
1410 if (!stream->metadata_flag) {
1411 ret = get_index_values(&index, infd);
1412 if (ret < 0) {
1413 err = kernctl_put_subbuf(infd);
1414 if (err != 0) {
1415 if (err == -EFAULT) {
1416 PERROR("Error in unreserving sub buffer\n");
1417 } else if (err == -EIO) {
1418 /* Should never happen with newer LTTng versions */
1419 PERROR("Reader has been pushed by the writer, last sub-buffer corrupted.");
1420 }
1421 ret = err;
1422 goto end;
1423 }
1424 goto end;
1425 }
1426 ret = update_stream_stats(stream);
1427 if (ret < 0) {
1428 err = kernctl_put_subbuf(infd);
1429 if (err != 0) {
1430 if (err == -EFAULT) {
1431 PERROR("Error in unreserving sub buffer\n");
1432 } else if (err == -EIO) {
1433 /* Should never happen with newer LTTng versions */
1434 PERROR("Reader has been pushed by the writer, last sub-buffer corrupted.");
1435 }
1436 ret = err;
1437 goto end;
1438 }
1439 goto end;
1440 }
1441 } else {
1442 write_index = 0;
1443 ret = metadata_stream_check_version(infd, stream);
1444 if (ret < 0) {
1445 err = kernctl_put_subbuf(infd);
1446 if (err != 0) {
1447 if (err == -EFAULT) {
1448 PERROR("Error in unreserving sub buffer\n");
1449 } else if (err == -EIO) {
1450 /* Should never happen with newer LTTng versions */
1451 PERROR("Reader has been pushed by the writer, last sub-buffer corrupted.");
1452 }
1453 ret = err;
1454 goto end;
1455 }
1456 goto end;
1457 }
1458 }
1459
1460 switch (stream->chan->output) {
1461 case CONSUMER_CHANNEL_SPLICE:
1462 /*
1463 * XXX: The lttng-modules splice "actor" does not handle copying
1464 * partial pages hence only using the subbuffer size without the
1465 * padding makes the splice fail.
1466 */
1467 subbuf_size = len;
1468 padding = 0;
1469
1470 /* splice the subbuffer to the tracefile */
1471 ret = lttng_consumer_on_read_subbuffer_splice(ctx, stream, subbuf_size,
1472 padding, &index);
1473 /*
1474 * XXX: Splice does not support network streaming so the return value
1475 * is simply checked against subbuf_size and not like the mmap() op.
1476 */
1477 if (ret != subbuf_size) {
1478 /*
1479 * display the error but continue processing to try
1480 * to release the subbuffer
1481 */
1482 ERR("Error splicing to tracefile (ret: %zd != len: %lu)",
1483 ret, subbuf_size);
1484 write_index = 0;
1485 }
1486 break;
1487 case CONSUMER_CHANNEL_MMAP:
1488 /* Get subbuffer size without padding */
1489 err = kernctl_get_subbuf_size(infd, &subbuf_size);
1490 if (err != 0) {
1491 PERROR("Getting sub-buffer len failed.");
1492 err = kernctl_put_subbuf(infd);
1493 if (err != 0) {
1494 if (err == -EFAULT) {
1495 PERROR("Error in unreserving sub buffer\n");
1496 } else if (err == -EIO) {
1497 /* Should never happen with newer LTTng versions */
1498 PERROR("Reader has been pushed by the writer, last sub-buffer corrupted.");
1499 }
1500 ret = err;
1501 goto end;
1502 }
1503 ret = err;
1504 goto end;
1505 }
1506
1507 /* Make sure the tracer is not gone mad on us! */
1508 assert(len >= subbuf_size);
1509
1510 padding = len - subbuf_size;
1511
1512 /* write the subbuffer to the tracefile */
1513 ret = lttng_consumer_on_read_subbuffer_mmap(ctx, stream, subbuf_size,
1514 padding, &index);
1515 /*
1516 * The mmap operation should write subbuf_size amount of data when
1517 * network streaming or the full padding (len) size when we are _not_
1518 * streaming.
1519 */
1520 if ((ret != subbuf_size && stream->net_seq_idx != (uint64_t) -1ULL) ||
1521 (ret != len && stream->net_seq_idx == (uint64_t) -1ULL)) {
1522 /*
1523 * Display the error but continue processing to try to release the
1524 * subbuffer. This is a DBG statement since this is possible to
1525 * happen without being a critical error.
1526 */
1527 DBG("Error writing to tracefile "
1528 "(ret: %zd != len: %lu != subbuf_size: %lu)",
1529 ret, len, subbuf_size);
1530 write_index = 0;
1531 }
1532 break;
1533 default:
1534 ERR("Unknown output method");
1535 ret = -EPERM;
1536 }
1537
1538 err = kernctl_put_next_subbuf(infd);
1539 if (err != 0) {
1540 if (err == -EFAULT) {
1541 PERROR("Error in unreserving sub buffer\n");
1542 } else if (err == -EIO) {
1543 /* Should never happen with newer LTTng versions */
1544 PERROR("Reader has been pushed by the writer, last sub-buffer corrupted.");
1545 }
1546 ret = err;
1547 goto end;
1548 }
1549
1550 /* Write index if needed. */
1551 if (!write_index) {
1552 goto end;
1553 }
1554
1555 if (stream->chan->live_timer_interval && !stream->metadata_flag) {
1556 /*
1557 * In live, block until all the metadata is sent.
1558 */
1559 pthread_mutex_lock(&stream->metadata_timer_lock);
1560 assert(!stream->missed_metadata_flush);
1561 stream->waiting_on_metadata = true;
1562 pthread_mutex_unlock(&stream->metadata_timer_lock);
1563
1564 err = consumer_stream_sync_metadata(ctx, stream->session_id);
1565
1566 pthread_mutex_lock(&stream->metadata_timer_lock);
1567 stream->waiting_on_metadata = false;
1568 if (stream->missed_metadata_flush) {
1569 stream->missed_metadata_flush = false;
1570 pthread_mutex_unlock(&stream->metadata_timer_lock);
1571 (void) consumer_flush_kernel_index(stream);
1572 } else {
1573 pthread_mutex_unlock(&stream->metadata_timer_lock);
1574 }
1575 if (err < 0) {
1576 goto end;
1577 }
1578 }
1579
1580 err = consumer_stream_write_index(stream, &index);
1581 if (err < 0) {
1582 goto end;
1583 }
1584
1585 end:
1586 return ret;
1587 }
1588
1589 int lttng_kconsumer_on_recv_stream(struct lttng_consumer_stream *stream)
1590 {
1591 int ret;
1592
1593 assert(stream);
1594
1595 /*
1596 * Don't create anything if this is set for streaming or should not be
1597 * monitored.
1598 */
1599 if (stream->net_seq_idx == (uint64_t) -1ULL && stream->chan->monitor) {
1600 ret = utils_create_stream_file(stream->chan->pathname, stream->name,
1601 stream->chan->tracefile_size, stream->tracefile_count_current,
1602 stream->uid, stream->gid, NULL);
1603 if (ret < 0) {
1604 goto error;
1605 }
1606 stream->out_fd = ret;
1607 stream->tracefile_size_current = 0;
1608
1609 if (!stream->metadata_flag) {
1610 struct lttng_index_file *index_file;
1611
1612 index_file = lttng_index_file_create(stream->chan->pathname,
1613 stream->name, stream->uid, stream->gid,
1614 stream->chan->tracefile_size,
1615 stream->tracefile_count_current,
1616 CTF_INDEX_MAJOR, CTF_INDEX_MINOR);
1617 if (!index_file) {
1618 goto error;
1619 }
1620 assert(!stream->index_file);
1621 stream->index_file = index_file;
1622 }
1623 }
1624
1625 if (stream->output == LTTNG_EVENT_MMAP) {
1626 /* get the len of the mmap region */
1627 unsigned long mmap_len;
1628
1629 ret = kernctl_get_mmap_len(stream->wait_fd, &mmap_len);
1630 if (ret != 0) {
1631 PERROR("kernctl_get_mmap_len");
1632 goto error_close_fd;
1633 }
1634 stream->mmap_len = (size_t) mmap_len;
1635
1636 stream->mmap_base = mmap(NULL, stream->mmap_len, PROT_READ,
1637 MAP_PRIVATE, stream->wait_fd, 0);
1638 if (stream->mmap_base == MAP_FAILED) {
1639 PERROR("Error mmaping");
1640 ret = -1;
1641 goto error_close_fd;
1642 }
1643 }
1644
1645 /* we return 0 to let the library handle the FD internally */
1646 return 0;
1647
1648 error_close_fd:
1649 if (stream->out_fd >= 0) {
1650 int err;
1651
1652 err = close(stream->out_fd);
1653 assert(!err);
1654 stream->out_fd = -1;
1655 }
1656 error:
1657 return ret;
1658 }
1659
1660 /*
1661 * Check if data is still being extracted from the buffers for a specific
1662 * stream. Consumer data lock MUST be acquired before calling this function
1663 * and the stream lock.
1664 *
1665 * Return 1 if the traced data are still getting read else 0 meaning that the
1666 * data is available for trace viewer reading.
1667 */
1668 int lttng_kconsumer_data_pending(struct lttng_consumer_stream *stream)
1669 {
1670 int ret;
1671
1672 assert(stream);
1673
1674 if (stream->endpoint_status != CONSUMER_ENDPOINT_ACTIVE) {
1675 ret = 0;
1676 goto end;
1677 }
1678
1679 ret = kernctl_get_next_subbuf(stream->wait_fd);
1680 if (ret == 0) {
1681 /* There is still data so let's put back this subbuffer. */
1682 ret = kernctl_put_subbuf(stream->wait_fd);
1683 assert(ret == 0);
1684 ret = 1; /* Data is pending */
1685 goto end;
1686 }
1687
1688 /* Data is NOT pending and ready to be read. */
1689 ret = 0;
1690
1691 end:
1692 return ret;
1693 }
This page took 0.083083 seconds and 5 git commands to generate.