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