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