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