87ade12fa33a32e71dd52cc685a6f3acb02e967b
[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, NULL);
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 msg.u.stream.trace_archive_id);
649 if (new_stream == NULL) {
650 switch (alloc_ret) {
651 case -ENOMEM:
652 case -EINVAL:
653 default:
654 lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_OUTFD_ERROR);
655 break;
656 }
657 goto end_nosignal;
658 }
659
660 new_stream->chan = channel;
661 new_stream->wait_fd = fd;
662 consumer_stream_update_channel_attributes(new_stream,
663 channel);
664 switch (channel->output) {
665 case CONSUMER_CHANNEL_SPLICE:
666 new_stream->output = LTTNG_EVENT_SPLICE;
667 ret = utils_create_pipe(new_stream->splice_pipe);
668 if (ret < 0) {
669 goto end_nosignal;
670 }
671 break;
672 case CONSUMER_CHANNEL_MMAP:
673 new_stream->output = LTTNG_EVENT_MMAP;
674 break;
675 default:
676 ERR("Stream output unknown %d", channel->output);
677 goto end_nosignal;
678 }
679
680 /*
681 * We've just assigned the channel to the stream so increment the
682 * refcount right now. We don't need to increment the refcount for
683 * streams in no monitor because we handle manually the cleanup of
684 * those. It is very important to make sure there is NO prior
685 * consumer_del_stream() calls or else the refcount will be unbalanced.
686 */
687 if (channel->monitor) {
688 uatomic_inc(&new_stream->chan->refcount);
689 }
690
691 /*
692 * The buffer flush is done on the session daemon side for the kernel
693 * so no need for the stream "hangup_flush_done" variable to be
694 * tracked. This is important for a kernel stream since we don't rely
695 * on the flush state of the stream to read data. It's not the case for
696 * user space tracing.
697 */
698 new_stream->hangup_flush_done = 0;
699
700 health_code_update();
701
702 if (ctx->on_recv_stream) {
703 ret = ctx->on_recv_stream(new_stream);
704 if (ret < 0) {
705 consumer_stream_free(new_stream);
706 goto end_nosignal;
707 }
708 }
709
710 health_code_update();
711
712 if (new_stream->metadata_flag) {
713 channel->metadata_stream = new_stream;
714 }
715
716 /* Do not monitor this stream. */
717 if (!channel->monitor) {
718 DBG("Kernel consumer add stream %s in no monitor mode with "
719 "relayd id %" PRIu64, new_stream->name,
720 new_stream->net_seq_idx);
721 cds_list_add(&new_stream->send_node, &channel->streams.head);
722 break;
723 }
724
725 /* Send stream to relayd if the stream has an ID. */
726 if (new_stream->net_seq_idx != (uint64_t) -1ULL) {
727 ret = consumer_send_relayd_stream(new_stream,
728 new_stream->chan->pathname);
729 if (ret < 0) {
730 consumer_stream_free(new_stream);
731 goto end_nosignal;
732 }
733
734 /*
735 * If adding an extra stream to an already
736 * existing channel (e.g. cpu hotplug), we need
737 * to send the "streams_sent" command to relayd.
738 */
739 if (channel->streams_sent_to_relayd) {
740 ret = consumer_send_relayd_streams_sent(
741 new_stream->net_seq_idx);
742 if (ret < 0) {
743 goto end_nosignal;
744 }
745 }
746 }
747
748 /* Get the right pipe where the stream will be sent. */
749 if (new_stream->metadata_flag) {
750 consumer_add_metadata_stream(new_stream);
751 stream_pipe = ctx->consumer_metadata_pipe;
752 } else {
753 consumer_add_data_stream(new_stream);
754 stream_pipe = ctx->consumer_data_pipe;
755 }
756
757 /* Visible to other threads */
758 new_stream->globally_visible = 1;
759
760 health_code_update();
761
762 ret = lttng_pipe_write(stream_pipe, &new_stream, sizeof(new_stream));
763 if (ret < 0) {
764 ERR("Consumer write %s stream to pipe %d",
765 new_stream->metadata_flag ? "metadata" : "data",
766 lttng_pipe_get_writefd(stream_pipe));
767 if (new_stream->metadata_flag) {
768 consumer_del_stream_for_metadata(new_stream);
769 } else {
770 consumer_del_stream_for_data(new_stream);
771 }
772 goto end_nosignal;
773 }
774
775 DBG("Kernel consumer ADD_STREAM %s (fd: %d) %s with relayd id %" PRIu64,
776 new_stream->name, fd, new_stream->chan->pathname, new_stream->relayd_stream_id);
777 break;
778 }
779 case LTTNG_CONSUMER_STREAMS_SENT:
780 {
781 struct lttng_consumer_channel *channel;
782
783 /*
784 * Get stream's channel reference. Needed when adding the stream to the
785 * global hash table.
786 */
787 channel = consumer_find_channel(msg.u.sent_streams.channel_key);
788 if (!channel) {
789 /*
790 * We could not find the channel. Can happen if cpu hotplug
791 * happens while tearing down.
792 */
793 ERR("Unable to find channel key %" PRIu64,
794 msg.u.sent_streams.channel_key);
795 ret_code = LTTCOMM_CONSUMERD_CHAN_NOT_FOUND;
796 }
797
798 health_code_update();
799
800 /*
801 * Send status code to session daemon.
802 */
803 ret = consumer_send_status_msg(sock, ret_code);
804 if (ret < 0 || ret_code != LTTCOMM_CONSUMERD_SUCCESS) {
805 /* Somehow, the session daemon is not responding anymore. */
806 goto end_nosignal;
807 }
808
809 health_code_update();
810
811 /*
812 * We should not send this message if we don't monitor the
813 * streams in this channel.
814 */
815 if (!channel->monitor) {
816 break;
817 }
818
819 health_code_update();
820 /* Send stream to relayd if the stream has an ID. */
821 if (msg.u.sent_streams.net_seq_idx != (uint64_t) -1ULL) {
822 ret = consumer_send_relayd_streams_sent(
823 msg.u.sent_streams.net_seq_idx);
824 if (ret < 0) {
825 goto end_nosignal;
826 }
827 channel->streams_sent_to_relayd = true;
828 }
829 break;
830 }
831 case LTTNG_CONSUMER_UPDATE_STREAM:
832 {
833 rcu_read_unlock();
834 return -ENOSYS;
835 }
836 case LTTNG_CONSUMER_DESTROY_RELAYD:
837 {
838 uint64_t index = msg.u.destroy_relayd.net_seq_idx;
839 struct consumer_relayd_sock_pair *relayd;
840
841 DBG("Kernel consumer destroying relayd %" PRIu64, index);
842
843 /* Get relayd reference if exists. */
844 relayd = consumer_find_relayd(index);
845 if (relayd == NULL) {
846 DBG("Unable to find relayd %" PRIu64, index);
847 ret_code = LTTCOMM_CONSUMERD_RELAYD_FAIL;
848 }
849
850 /*
851 * Each relayd socket pair has a refcount of stream attached to it
852 * which tells if the relayd is still active or not depending on the
853 * refcount value.
854 *
855 * This will set the destroy flag of the relayd object and destroy it
856 * if the refcount reaches zero when called.
857 *
858 * The destroy can happen either here or when a stream fd hangs up.
859 */
860 if (relayd) {
861 consumer_flag_relayd_for_destroy(relayd);
862 }
863
864 health_code_update();
865
866 ret = consumer_send_status_msg(sock, ret_code);
867 if (ret < 0) {
868 /* Somehow, the session daemon is not responding anymore. */
869 goto error_fatal;
870 }
871
872 goto end_nosignal;
873 }
874 case LTTNG_CONSUMER_DATA_PENDING:
875 {
876 int32_t ret;
877 uint64_t id = msg.u.data_pending.session_id;
878
879 DBG("Kernel consumer data pending command for id %" PRIu64, id);
880
881 ret = consumer_data_pending(id);
882
883 health_code_update();
884
885 /* Send back returned value to session daemon */
886 ret = lttcomm_send_unix_sock(sock, &ret, sizeof(ret));
887 if (ret < 0) {
888 PERROR("send data pending ret code");
889 goto error_fatal;
890 }
891
892 /*
893 * No need to send back a status message since the data pending
894 * returned value is the response.
895 */
896 break;
897 }
898 case LTTNG_CONSUMER_SNAPSHOT_CHANNEL:
899 {
900 if (msg.u.snapshot_channel.metadata == 1) {
901 ret = lttng_kconsumer_snapshot_metadata(msg.u.snapshot_channel.key,
902 msg.u.snapshot_channel.pathname,
903 msg.u.snapshot_channel.relayd_id, ctx);
904 if (ret < 0) {
905 ERR("Snapshot metadata failed");
906 ret_code = LTTCOMM_CONSUMERD_ERROR_METADATA;
907 }
908 } else {
909 ret = lttng_kconsumer_snapshot_channel(msg.u.snapshot_channel.key,
910 msg.u.snapshot_channel.pathname,
911 msg.u.snapshot_channel.relayd_id,
912 msg.u.snapshot_channel.nb_packets_per_stream,
913 ctx);
914 if (ret < 0) {
915 ERR("Snapshot channel failed");
916 ret_code = LTTCOMM_CONSUMERD_CHAN_NOT_FOUND;
917 }
918 }
919
920 health_code_update();
921
922 ret = consumer_send_status_msg(sock, ret_code);
923 if (ret < 0) {
924 /* Somehow, the session daemon is not responding anymore. */
925 goto end_nosignal;
926 }
927 break;
928 }
929 case LTTNG_CONSUMER_DESTROY_CHANNEL:
930 {
931 uint64_t key = msg.u.destroy_channel.key;
932 struct lttng_consumer_channel *channel;
933
934 channel = consumer_find_channel(key);
935 if (!channel) {
936 ERR("Kernel consumer destroy channel %" PRIu64 " not found", key);
937 ret_code = LTTCOMM_CONSUMERD_CHAN_NOT_FOUND;
938 }
939
940 health_code_update();
941
942 ret = consumer_send_status_msg(sock, ret_code);
943 if (ret < 0) {
944 /* Somehow, the session daemon is not responding anymore. */
945 goto end_nosignal;
946 }
947
948 health_code_update();
949
950 /* Stop right now if no channel was found. */
951 if (!channel) {
952 goto end_nosignal;
953 }
954
955 /*
956 * This command should ONLY be issued for channel with streams set in
957 * no monitor mode.
958 */
959 assert(!channel->monitor);
960
961 /*
962 * The refcount should ALWAYS be 0 in the case of a channel in no
963 * monitor mode.
964 */
965 assert(!uatomic_sub_return(&channel->refcount, 1));
966
967 consumer_del_channel(channel);
968
969 goto end_nosignal;
970 }
971 case LTTNG_CONSUMER_DISCARDED_EVENTS:
972 {
973 ssize_t ret;
974 uint64_t count;
975 struct lttng_consumer_channel *channel;
976 uint64_t id = msg.u.discarded_events.session_id;
977 uint64_t key = msg.u.discarded_events.channel_key;
978
979 DBG("Kernel consumer discarded events command for session id %"
980 PRIu64 ", channel key %" PRIu64, id, key);
981
982 channel = consumer_find_channel(key);
983 if (!channel) {
984 ERR("Kernel consumer discarded events channel %"
985 PRIu64 " not found", key);
986 count = 0;
987 } else {
988 count = channel->discarded_events;
989 }
990
991 health_code_update();
992
993 /* Send back returned value to session daemon */
994 ret = lttcomm_send_unix_sock(sock, &count, sizeof(count));
995 if (ret < 0) {
996 PERROR("send discarded events");
997 goto error_fatal;
998 }
999
1000 break;
1001 }
1002 case LTTNG_CONSUMER_LOST_PACKETS:
1003 {
1004 ssize_t ret;
1005 uint64_t count;
1006 struct lttng_consumer_channel *channel;
1007 uint64_t id = msg.u.lost_packets.session_id;
1008 uint64_t key = msg.u.lost_packets.channel_key;
1009
1010 DBG("Kernel consumer lost packets command for session id %"
1011 PRIu64 ", channel key %" PRIu64, id, key);
1012
1013 channel = consumer_find_channel(key);
1014 if (!channel) {
1015 ERR("Kernel consumer lost packets channel %"
1016 PRIu64 " not found", key);
1017 count = 0;
1018 } else {
1019 count = channel->lost_packets;
1020 }
1021
1022 health_code_update();
1023
1024 /* Send back returned value to session daemon */
1025 ret = lttcomm_send_unix_sock(sock, &count, sizeof(count));
1026 if (ret < 0) {
1027 PERROR("send lost packets");
1028 goto error_fatal;
1029 }
1030
1031 break;
1032 }
1033 case LTTNG_CONSUMER_SET_CHANNEL_MONITOR_PIPE:
1034 {
1035 int channel_monitor_pipe;
1036
1037 ret_code = LTTCOMM_CONSUMERD_SUCCESS;
1038 /* Successfully received the command's type. */
1039 ret = consumer_send_status_msg(sock, ret_code);
1040 if (ret < 0) {
1041 goto error_fatal;
1042 }
1043
1044 ret = lttcomm_recv_fds_unix_sock(sock, &channel_monitor_pipe,
1045 1);
1046 if (ret != sizeof(channel_monitor_pipe)) {
1047 ERR("Failed to receive channel monitor pipe");
1048 goto error_fatal;
1049 }
1050
1051 DBG("Received channel monitor pipe (%d)", channel_monitor_pipe);
1052 ret = consumer_timer_thread_set_channel_monitor_pipe(
1053 channel_monitor_pipe);
1054 if (!ret) {
1055 int flags;
1056
1057 ret_code = LTTCOMM_CONSUMERD_SUCCESS;
1058 /* Set the pipe as non-blocking. */
1059 ret = fcntl(channel_monitor_pipe, F_GETFL, 0);
1060 if (ret == -1) {
1061 PERROR("fcntl get flags of the channel monitoring pipe");
1062 goto error_fatal;
1063 }
1064 flags = ret;
1065
1066 ret = fcntl(channel_monitor_pipe, F_SETFL,
1067 flags | O_NONBLOCK);
1068 if (ret == -1) {
1069 PERROR("fcntl set O_NONBLOCK flag of the channel monitoring pipe");
1070 goto error_fatal;
1071 }
1072 DBG("Channel monitor pipe set as non-blocking");
1073 } else {
1074 ret_code = LTTCOMM_CONSUMERD_ALREADY_SET;
1075 }
1076 ret = consumer_send_status_msg(sock, ret_code);
1077 if (ret < 0) {
1078 goto error_fatal;
1079 }
1080 break;
1081 }
1082 case LTTNG_CONSUMER_SET_CHANNEL_ROTATE_PIPE:
1083 {
1084 int channel_rotate_pipe;
1085 int flags;
1086
1087 ret_code = LTTCOMM_CONSUMERD_SUCCESS;
1088 /* Successfully received the command's type. */
1089 ret = consumer_send_status_msg(sock, ret_code);
1090 if (ret < 0) {
1091 goto error_fatal;
1092 }
1093
1094 ret = lttcomm_recv_fds_unix_sock(sock, &channel_rotate_pipe, 1);
1095 if (ret != (ssize_t) sizeof(channel_rotate_pipe)) {
1096 ERR("Failed to receive channel rotate pipe");
1097 goto error_fatal;
1098 }
1099
1100 DBG("Received channel rotate pipe (%d)", channel_rotate_pipe);
1101 ctx->channel_rotate_pipe = channel_rotate_pipe;
1102 /* Set the pipe as non-blocking. */
1103 ret = fcntl(channel_rotate_pipe, F_GETFL, 0);
1104 if (ret == -1) {
1105 PERROR("fcntl get flags of the channel rotate pipe");
1106 goto error_fatal;
1107 }
1108 flags = ret;
1109
1110 ret = fcntl(channel_rotate_pipe, F_SETFL, flags | O_NONBLOCK);
1111 if (ret == -1) {
1112 PERROR("fcntl set O_NONBLOCK flag of the channel rotate pipe");
1113 goto error_fatal;
1114 }
1115 DBG("Channel rotate pipe set as non-blocking");
1116 ret_code = LTTCOMM_CONSUMERD_SUCCESS;
1117 ret = consumer_send_status_msg(sock, ret_code);
1118 if (ret < 0) {
1119 goto error_fatal;
1120 }
1121 break;
1122 }
1123 case LTTNG_CONSUMER_ROTATE_CHANNEL:
1124 {
1125 DBG("Consumer rotate channel %" PRIu64, msg.u.rotate_channel.key);
1126
1127 /*
1128 * Sample the rotate position of all the streams in this channel.
1129 */
1130 ret = lttng_consumer_rotate_channel(msg.u.rotate_channel.key,
1131 msg.u.rotate_channel.pathname,
1132 msg.u.rotate_channel.relayd_id,
1133 msg.u.rotate_channel.metadata,
1134 msg.u.rotate_channel.new_chunk_id,
1135 ctx);
1136 if (ret < 0) {
1137 ERR("Rotate channel failed");
1138 ret_code = LTTCOMM_CONSUMERD_CHAN_NOT_FOUND;
1139 }
1140
1141 health_code_update();
1142
1143 ret = consumer_send_status_msg(sock, ret_code);
1144 if (ret < 0) {
1145 /* Somehow, the session daemon is not responding anymore. */
1146 goto end_nosignal;
1147 }
1148
1149 /* Rotate the streams that are ready right now. */
1150 ret = lttng_consumer_rotate_ready_streams(
1151 msg.u.rotate_channel.key, ctx);
1152 if (ret < 0) {
1153 ERR("Rotate ready streams failed");
1154 }
1155
1156 break;
1157 }
1158 case LTTNG_CONSUMER_ROTATE_RENAME:
1159 {
1160 DBG("Consumer rename session %" PRIu64 " after rotation, old path = \"%s\", new path = \"%s\"",
1161 msg.u.rotate_rename.session_id,
1162 msg.u.rotate_rename.old_path,
1163 msg.u.rotate_rename.new_path);
1164 ret = lttng_consumer_rotate_rename(msg.u.rotate_rename.old_path,
1165 msg.u.rotate_rename.new_path,
1166 msg.u.rotate_rename.uid,
1167 msg.u.rotate_rename.gid,
1168 msg.u.rotate_rename.relayd_id);
1169 if (ret < 0) {
1170 ERR("Rotate rename failed");
1171 ret_code = LTTCOMM_CONSUMERD_CHAN_NOT_FOUND;
1172 }
1173
1174 health_code_update();
1175
1176 ret = consumer_send_status_msg(sock, ret_code);
1177 if (ret < 0) {
1178 /* Somehow, the session daemon is not responding anymore. */
1179 goto end_nosignal;
1180 }
1181 break;
1182 }
1183 case LTTNG_CONSUMER_ROTATE_PENDING_RELAY:
1184 {
1185 int pending;
1186 uint32_t pending_reply;
1187
1188 DBG("Consumer rotate pending on relay for session %" PRIu64,
1189 msg.u.rotate_pending_relay.session_id);
1190 pending = lttng_consumer_rotate_pending_relay(
1191 msg.u.rotate_pending_relay.session_id,
1192 msg.u.rotate_pending_relay.relayd_id,
1193 msg.u.rotate_pending_relay.chunk_id);
1194 if (pending < 0) {
1195 ERR("Rotate pending relay failed");
1196 ret_code = LTTCOMM_CONSUMERD_CHAN_NOT_FOUND;
1197 } else {
1198 pending_reply = !!pending;
1199 }
1200
1201 health_code_update();
1202
1203 ret = consumer_send_status_msg(sock, ret_code);
1204 if (ret < 0) {
1205 /* Somehow, the session daemon is not responding anymore. */
1206 goto end_nosignal;
1207 }
1208
1209 if (pending < 0) {
1210 /*
1211 * An error occured while running the command;
1212 * don't send the 'pending' flag as the sessiond
1213 * will not read it.
1214 */
1215 break;
1216 }
1217
1218 /* Send back returned value to session daemon */
1219 ret = lttcomm_send_unix_sock(sock, &pending_reply,
1220 sizeof(pending_reply));
1221 if (ret < 0) {
1222 PERROR("send data pending ret code");
1223 goto error_fatal;
1224 }
1225 break;
1226 }
1227 case LTTNG_CONSUMER_MKDIR:
1228 {
1229 DBG("Consumer mkdir %s in session %" PRIu64,
1230 msg.u.mkdir.path,
1231 msg.u.mkdir.session_id);
1232 ret = lttng_consumer_mkdir(msg.u.mkdir.path,
1233 msg.u.mkdir.uid,
1234 msg.u.mkdir.gid,
1235 msg.u.mkdir.relayd_id);
1236 if (ret < 0) {
1237 ERR("consumer mkdir failed");
1238 ret_code = LTTCOMM_CONSUMERD_CHAN_NOT_FOUND;
1239 }
1240
1241 health_code_update();
1242
1243 ret = consumer_send_status_msg(sock, ret_code);
1244 if (ret < 0) {
1245 /* Somehow, the session daemon is not responding anymore. */
1246 goto end_nosignal;
1247 }
1248 break;
1249 }
1250 default:
1251 goto end_nosignal;
1252 }
1253
1254 end_nosignal:
1255 rcu_read_unlock();
1256
1257 /*
1258 * Return 1 to indicate success since the 0 value can be a socket
1259 * shutdown during the recv() or send() call.
1260 */
1261 health_code_update();
1262 return 1;
1263
1264 error_fatal:
1265 rcu_read_unlock();
1266 /* This will issue a consumer stop. */
1267 return -1;
1268 }
1269
1270 /*
1271 * Populate index values of a kernel stream. Values are set in big endian order.
1272 *
1273 * Return 0 on success or else a negative value.
1274 */
1275 static int get_index_values(struct ctf_packet_index *index, int infd)
1276 {
1277 int ret;
1278
1279 ret = kernctl_get_timestamp_begin(infd, &index->timestamp_begin);
1280 if (ret < 0) {
1281 PERROR("kernctl_get_timestamp_begin");
1282 goto error;
1283 }
1284 index->timestamp_begin = htobe64(index->timestamp_begin);
1285
1286 ret = kernctl_get_timestamp_end(infd, &index->timestamp_end);
1287 if (ret < 0) {
1288 PERROR("kernctl_get_timestamp_end");
1289 goto error;
1290 }
1291 index->timestamp_end = htobe64(index->timestamp_end);
1292
1293 ret = kernctl_get_events_discarded(infd, &index->events_discarded);
1294 if (ret < 0) {
1295 PERROR("kernctl_get_events_discarded");
1296 goto error;
1297 }
1298 index->events_discarded = htobe64(index->events_discarded);
1299
1300 ret = kernctl_get_content_size(infd, &index->content_size);
1301 if (ret < 0) {
1302 PERROR("kernctl_get_content_size");
1303 goto error;
1304 }
1305 index->content_size = htobe64(index->content_size);
1306
1307 ret = kernctl_get_packet_size(infd, &index->packet_size);
1308 if (ret < 0) {
1309 PERROR("kernctl_get_packet_size");
1310 goto error;
1311 }
1312 index->packet_size = htobe64(index->packet_size);
1313
1314 ret = kernctl_get_stream_id(infd, &index->stream_id);
1315 if (ret < 0) {
1316 PERROR("kernctl_get_stream_id");
1317 goto error;
1318 }
1319 index->stream_id = htobe64(index->stream_id);
1320
1321 ret = kernctl_get_instance_id(infd, &index->stream_instance_id);
1322 if (ret < 0) {
1323 if (ret == -ENOTTY) {
1324 /* Command not implemented by lttng-modules. */
1325 index->stream_instance_id = -1ULL;
1326 } else {
1327 PERROR("kernctl_get_instance_id");
1328 goto error;
1329 }
1330 }
1331 index->stream_instance_id = htobe64(index->stream_instance_id);
1332
1333 ret = kernctl_get_sequence_number(infd, &index->packet_seq_num);
1334 if (ret < 0) {
1335 if (ret == -ENOTTY) {
1336 /* Command not implemented by lttng-modules. */
1337 index->packet_seq_num = -1ULL;
1338 ret = 0;
1339 } else {
1340 PERROR("kernctl_get_sequence_number");
1341 goto error;
1342 }
1343 }
1344 index->packet_seq_num = htobe64(index->packet_seq_num);
1345
1346 error:
1347 return ret;
1348 }
1349 /*
1350 * Sync metadata meaning request them to the session daemon and snapshot to the
1351 * metadata thread can consumer them.
1352 *
1353 * Metadata stream lock MUST be acquired.
1354 *
1355 * Return 0 if new metadatda is available, EAGAIN if the metadata stream
1356 * is empty or a negative value on error.
1357 */
1358 int lttng_kconsumer_sync_metadata(struct lttng_consumer_stream *metadata)
1359 {
1360 int ret;
1361
1362 assert(metadata);
1363
1364 ret = kernctl_buffer_flush(metadata->wait_fd);
1365 if (ret < 0) {
1366 ERR("Failed to flush kernel stream");
1367 goto end;
1368 }
1369
1370 ret = kernctl_snapshot(metadata->wait_fd);
1371 if (ret < 0) {
1372 if (ret != -EAGAIN) {
1373 ERR("Sync metadata, taking kernel snapshot failed.");
1374 goto end;
1375 }
1376 DBG("Sync metadata, no new kernel metadata");
1377 /* No new metadata, exit. */
1378 ret = ENODATA;
1379 goto end;
1380 }
1381
1382 end:
1383 return ret;
1384 }
1385
1386 static
1387 int update_stream_stats(struct lttng_consumer_stream *stream)
1388 {
1389 int ret;
1390 uint64_t seq, discarded;
1391
1392 ret = kernctl_get_sequence_number(stream->wait_fd, &seq);
1393 if (ret < 0) {
1394 if (ret == -ENOTTY) {
1395 /* Command not implemented by lttng-modules. */
1396 seq = -1ULL;
1397 } else {
1398 PERROR("kernctl_get_sequence_number");
1399 goto end;
1400 }
1401 }
1402
1403 /*
1404 * Start the sequence when we extract the first packet in case we don't
1405 * start at 0 (for example if a consumer is not connected to the
1406 * session immediately after the beginning).
1407 */
1408 if (stream->last_sequence_number == -1ULL) {
1409 stream->last_sequence_number = seq;
1410 } else if (seq > stream->last_sequence_number) {
1411 stream->chan->lost_packets += seq -
1412 stream->last_sequence_number - 1;
1413 } else {
1414 /* seq <= last_sequence_number */
1415 ERR("Sequence number inconsistent : prev = %" PRIu64
1416 ", current = %" PRIu64,
1417 stream->last_sequence_number, seq);
1418 ret = -1;
1419 goto end;
1420 }
1421 stream->last_sequence_number = seq;
1422
1423 ret = kernctl_get_events_discarded(stream->wait_fd, &discarded);
1424 if (ret < 0) {
1425 PERROR("kernctl_get_events_discarded");
1426 goto end;
1427 }
1428 if (discarded < stream->last_discarded_events) {
1429 /*
1430 * Overflow has occurred. We assume only one wrap-around
1431 * has occurred.
1432 */
1433 stream->chan->discarded_events += (1ULL << (CAA_BITS_PER_LONG - 1)) -
1434 stream->last_discarded_events + discarded;
1435 } else {
1436 stream->chan->discarded_events += discarded -
1437 stream->last_discarded_events;
1438 }
1439 stream->last_discarded_events = discarded;
1440 ret = 0;
1441
1442 end:
1443 return ret;
1444 }
1445
1446 /*
1447 * Check if the local version of the metadata stream matches with the version
1448 * of the metadata stream in the kernel. If it was updated, set the reset flag
1449 * on the stream.
1450 */
1451 static
1452 int metadata_stream_check_version(int infd, struct lttng_consumer_stream *stream)
1453 {
1454 int ret;
1455 uint64_t cur_version;
1456
1457 ret = kernctl_get_metadata_version(infd, &cur_version);
1458 if (ret < 0) {
1459 if (ret == -ENOTTY) {
1460 /*
1461 * LTTng-modules does not implement this
1462 * command.
1463 */
1464 ret = 0;
1465 goto end;
1466 }
1467 ERR("Failed to get the metadata version");
1468 goto end;
1469 }
1470
1471 if (stream->metadata_version == cur_version) {
1472 ret = 0;
1473 goto end;
1474 }
1475
1476 DBG("New metadata version detected");
1477 stream->metadata_version = cur_version;
1478 stream->reset_metadata_flag = 1;
1479 ret = 0;
1480
1481 end:
1482 return ret;
1483 }
1484
1485 /*
1486 * Consume data on a file descriptor and write it on a trace file.
1487 */
1488 ssize_t lttng_kconsumer_read_subbuffer(struct lttng_consumer_stream *stream,
1489 struct lttng_consumer_local_data *ctx, bool *rotated)
1490 {
1491 unsigned long len, subbuf_size, padding;
1492 int err, write_index = 1, rotation_ret;
1493 ssize_t ret = 0;
1494 int infd = stream->wait_fd;
1495 struct ctf_packet_index index;
1496
1497 DBG("In read_subbuffer (infd : %d)", infd);
1498
1499 /*
1500 * If the stream was flagged to be ready for rotation before we extract the
1501 * next packet, rotate it now.
1502 */
1503 if (stream->rotate_ready) {
1504 DBG("Rotate stream before extracting data");
1505 rotation_ret = lttng_consumer_rotate_stream(ctx, stream, rotated);
1506 if (rotation_ret < 0) {
1507 ERR("Stream rotation error");
1508 ret = -1;
1509 goto error;
1510 }
1511 }
1512
1513 /* Get the next subbuffer */
1514 err = kernctl_get_next_subbuf(infd);
1515 if (err != 0) {
1516 /*
1517 * This is a debug message even for single-threaded consumer,
1518 * because poll() have more relaxed criterions than get subbuf,
1519 * so get_subbuf may fail for short race windows where poll()
1520 * would issue wakeups.
1521 */
1522 DBG("Reserving sub buffer failed (everything is normal, "
1523 "it is due to concurrency)");
1524 ret = err;
1525 goto error;
1526 }
1527
1528 /* Get the full subbuffer size including padding */
1529 err = kernctl_get_padded_subbuf_size(infd, &len);
1530 if (err != 0) {
1531 PERROR("Getting sub-buffer len failed.");
1532 err = kernctl_put_subbuf(infd);
1533 if (err != 0) {
1534 if (err == -EFAULT) {
1535 PERROR("Error in unreserving sub buffer\n");
1536 } else if (err == -EIO) {
1537 /* Should never happen with newer LTTng versions */
1538 PERROR("Reader has been pushed by the writer, last sub-buffer corrupted.");
1539 }
1540 ret = err;
1541 goto error;
1542 }
1543 ret = err;
1544 goto error;
1545 }
1546
1547 if (!stream->metadata_flag) {
1548 ret = get_index_values(&index, infd);
1549 if (ret < 0) {
1550 err = kernctl_put_subbuf(infd);
1551 if (err != 0) {
1552 if (err == -EFAULT) {
1553 PERROR("Error in unreserving sub buffer\n");
1554 } else if (err == -EIO) {
1555 /* Should never happen with newer LTTng versions */
1556 PERROR("Reader has been pushed by the writer, last sub-buffer corrupted.");
1557 }
1558 ret = err;
1559 goto error;
1560 }
1561 goto error;
1562 }
1563 ret = update_stream_stats(stream);
1564 if (ret < 0) {
1565 err = kernctl_put_subbuf(infd);
1566 if (err != 0) {
1567 if (err == -EFAULT) {
1568 PERROR("Error in unreserving sub buffer\n");
1569 } else if (err == -EIO) {
1570 /* Should never happen with newer LTTng versions */
1571 PERROR("Reader has been pushed by the writer, last sub-buffer corrupted.");
1572 }
1573 ret = err;
1574 goto error;
1575 }
1576 goto error;
1577 }
1578 } else {
1579 write_index = 0;
1580 ret = metadata_stream_check_version(infd, stream);
1581 if (ret < 0) {
1582 err = kernctl_put_subbuf(infd);
1583 if (err != 0) {
1584 if (err == -EFAULT) {
1585 PERROR("Error in unreserving sub buffer\n");
1586 } else if (err == -EIO) {
1587 /* Should never happen with newer LTTng versions */
1588 PERROR("Reader has been pushed by the writer, last sub-buffer corrupted.");
1589 }
1590 ret = err;
1591 goto error;
1592 }
1593 goto error;
1594 }
1595 }
1596
1597 switch (stream->chan->output) {
1598 case CONSUMER_CHANNEL_SPLICE:
1599 /*
1600 * XXX: The lttng-modules splice "actor" does not handle copying
1601 * partial pages hence only using the subbuffer size without the
1602 * padding makes the splice fail.
1603 */
1604 subbuf_size = len;
1605 padding = 0;
1606
1607 /* splice the subbuffer to the tracefile */
1608 ret = lttng_consumer_on_read_subbuffer_splice(ctx, stream, subbuf_size,
1609 padding, &index);
1610 /*
1611 * XXX: Splice does not support network streaming so the return value
1612 * is simply checked against subbuf_size and not like the mmap() op.
1613 */
1614 if (ret != subbuf_size) {
1615 /*
1616 * display the error but continue processing to try
1617 * to release the subbuffer
1618 */
1619 ERR("Error splicing to tracefile (ret: %zd != len: %lu)",
1620 ret, subbuf_size);
1621 write_index = 0;
1622 }
1623 break;
1624 case CONSUMER_CHANNEL_MMAP:
1625 /* Get subbuffer size without padding */
1626 err = kernctl_get_subbuf_size(infd, &subbuf_size);
1627 if (err != 0) {
1628 PERROR("Getting sub-buffer len failed.");
1629 err = kernctl_put_subbuf(infd);
1630 if (err != 0) {
1631 if (err == -EFAULT) {
1632 PERROR("Error in unreserving sub buffer\n");
1633 } else if (err == -EIO) {
1634 /* Should never happen with newer LTTng versions */
1635 PERROR("Reader has been pushed by the writer, last sub-buffer corrupted.");
1636 }
1637 ret = err;
1638 goto error;
1639 }
1640 ret = err;
1641 goto error;
1642 }
1643
1644 /* Make sure the tracer is not gone mad on us! */
1645 assert(len >= subbuf_size);
1646
1647 padding = len - subbuf_size;
1648
1649 /* write the subbuffer to the tracefile */
1650 ret = lttng_consumer_on_read_subbuffer_mmap(ctx, stream, subbuf_size,
1651 padding, &index);
1652 /*
1653 * The mmap operation should write subbuf_size amount of data when
1654 * network streaming or the full padding (len) size when we are _not_
1655 * streaming.
1656 */
1657 if ((ret != subbuf_size && stream->net_seq_idx != (uint64_t) -1ULL) ||
1658 (ret != len && stream->net_seq_idx == (uint64_t) -1ULL)) {
1659 /*
1660 * Display the error but continue processing to try to release the
1661 * subbuffer. This is a DBG statement since this is possible to
1662 * happen without being a critical error.
1663 */
1664 DBG("Error writing to tracefile "
1665 "(ret: %zd != len: %lu != subbuf_size: %lu)",
1666 ret, len, subbuf_size);
1667 write_index = 0;
1668 }
1669 break;
1670 default:
1671 ERR("Unknown output method");
1672 ret = -EPERM;
1673 }
1674
1675 err = kernctl_put_next_subbuf(infd);
1676 if (err != 0) {
1677 if (err == -EFAULT) {
1678 PERROR("Error in unreserving sub buffer\n");
1679 } else if (err == -EIO) {
1680 /* Should never happen with newer LTTng versions */
1681 PERROR("Reader has been pushed by the writer, last sub-buffer corrupted.");
1682 }
1683 ret = err;
1684 goto error;
1685 }
1686
1687 /* Write index if needed. */
1688 if (!write_index) {
1689 goto rotate;
1690 }
1691
1692 if (stream->chan->live_timer_interval && !stream->metadata_flag) {
1693 /*
1694 * In live, block until all the metadata is sent.
1695 */
1696 pthread_mutex_lock(&stream->metadata_timer_lock);
1697 assert(!stream->missed_metadata_flush);
1698 stream->waiting_on_metadata = true;
1699 pthread_mutex_unlock(&stream->metadata_timer_lock);
1700
1701 err = consumer_stream_sync_metadata(ctx, stream->session_id);
1702
1703 pthread_mutex_lock(&stream->metadata_timer_lock);
1704 stream->waiting_on_metadata = false;
1705 if (stream->missed_metadata_flush) {
1706 stream->missed_metadata_flush = false;
1707 pthread_mutex_unlock(&stream->metadata_timer_lock);
1708 (void) consumer_flush_kernel_index(stream);
1709 } else {
1710 pthread_mutex_unlock(&stream->metadata_timer_lock);
1711 }
1712 if (err < 0) {
1713 goto error;
1714 }
1715 }
1716
1717 err = consumer_stream_write_index(stream, &index);
1718 if (err < 0) {
1719 goto error;
1720 }
1721
1722 rotate:
1723 /*
1724 * After extracting the packet, we check if the stream is now ready to be
1725 * rotated and perform the action immediately.
1726 */
1727 rotation_ret = lttng_consumer_stream_is_rotate_ready(stream);
1728 if (rotation_ret == 1) {
1729 rotation_ret = lttng_consumer_rotate_stream(ctx, stream, rotated);
1730 if (rotation_ret < 0) {
1731 ERR("Stream rotation error");
1732 ret = -1;
1733 goto error;
1734 }
1735 } else if (rotation_ret < 0) {
1736 ERR("Checking if stream is ready to rotate");
1737 ret = -1;
1738 goto error;
1739 }
1740
1741 error:
1742 return ret;
1743 }
1744
1745 int lttng_kconsumer_on_recv_stream(struct lttng_consumer_stream *stream)
1746 {
1747 int ret;
1748
1749 assert(stream);
1750
1751 /*
1752 * Don't create anything if this is set for streaming or should not be
1753 * monitored.
1754 */
1755 if (stream->net_seq_idx == (uint64_t) -1ULL && stream->chan->monitor) {
1756 ret = utils_create_stream_file(stream->chan->pathname, stream->name,
1757 stream->chan->tracefile_size, stream->tracefile_count_current,
1758 stream->uid, stream->gid, NULL);
1759 if (ret < 0) {
1760 goto error;
1761 }
1762 stream->out_fd = ret;
1763 stream->tracefile_size_current = 0;
1764
1765 if (!stream->metadata_flag) {
1766 struct lttng_index_file *index_file;
1767
1768 index_file = lttng_index_file_create(stream->chan->pathname,
1769 stream->name, stream->uid, stream->gid,
1770 stream->chan->tracefile_size,
1771 stream->tracefile_count_current,
1772 CTF_INDEX_MAJOR, CTF_INDEX_MINOR);
1773 if (!index_file) {
1774 goto error;
1775 }
1776 assert(!stream->index_file);
1777 stream->index_file = index_file;
1778 }
1779 }
1780
1781 if (stream->output == LTTNG_EVENT_MMAP) {
1782 /* get the len of the mmap region */
1783 unsigned long mmap_len;
1784
1785 ret = kernctl_get_mmap_len(stream->wait_fd, &mmap_len);
1786 if (ret != 0) {
1787 PERROR("kernctl_get_mmap_len");
1788 goto error_close_fd;
1789 }
1790 stream->mmap_len = (size_t) mmap_len;
1791
1792 stream->mmap_base = mmap(NULL, stream->mmap_len, PROT_READ,
1793 MAP_PRIVATE, stream->wait_fd, 0);
1794 if (stream->mmap_base == MAP_FAILED) {
1795 PERROR("Error mmaping");
1796 ret = -1;
1797 goto error_close_fd;
1798 }
1799 }
1800
1801 /* we return 0 to let the library handle the FD internally */
1802 return 0;
1803
1804 error_close_fd:
1805 if (stream->out_fd >= 0) {
1806 int err;
1807
1808 err = close(stream->out_fd);
1809 assert(!err);
1810 stream->out_fd = -1;
1811 }
1812 error:
1813 return ret;
1814 }
1815
1816 /*
1817 * Check if data is still being extracted from the buffers for a specific
1818 * stream. Consumer data lock MUST be acquired before calling this function
1819 * and the stream lock.
1820 *
1821 * Return 1 if the traced data are still getting read else 0 meaning that the
1822 * data is available for trace viewer reading.
1823 */
1824 int lttng_kconsumer_data_pending(struct lttng_consumer_stream *stream)
1825 {
1826 int ret;
1827
1828 assert(stream);
1829
1830 if (stream->endpoint_status != CONSUMER_ENDPOINT_ACTIVE) {
1831 ret = 0;
1832 goto end;
1833 }
1834
1835 ret = kernctl_get_next_subbuf(stream->wait_fd);
1836 if (ret == 0) {
1837 /* There is still data so let's put back this subbuffer. */
1838 ret = kernctl_put_subbuf(stream->wait_fd);
1839 assert(ret == 0);
1840 ret = 1; /* Data is pending */
1841 goto end;
1842 }
1843
1844 /* Data is NOT pending and ready to be read. */
1845 ret = 0;
1846
1847 end:
1848 return ret;
1849 }
This page took 0.113472 seconds and 3 git commands to generate.