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