projects / lttng-tools.git / blob
? search:


eeec3a65e7d877291a82ba9d63bf2fb6dc8f30a3
[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 = msg.u.create_trace_chunk.credentials.value.uid,
1211 .gid = 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 default:
1315 goto end_nosignal;
1316 }
1317
1318 end_nosignal:
1319 /*
1320 * Return 1 to indicate success since the 0 value can be a socket
1321 * shutdown during the recv() or send() call.
1322 */
1323 ret = 1;
1324 goto end;
1325 error_fatal:
1326 /* This will issue a consumer stop. */
1327 ret = -1;
1328 goto end;
1329 end_msg_sessiond:
1330 /*
1331 * The returned value here is not useful since either way we'll return 1 to
1332 * the caller because the session daemon socket management is done
1333 * elsewhere. Returning a negative code or 0 will shutdown the consumer.
1334 */
1335 ret = consumer_send_status_msg(sock, ret_code);
1336 if (ret < 0) {
1337 goto error_fatal;
1338 }
1339 ret = 1;
1340 end:
1341 health_code_update();
1342 rcu_read_unlock();
1343 return ret;
1344 }
1345
1346 /*
1347 * Sync metadata meaning request them to the session daemon and snapshot to the
1348 * metadata thread can consumer them.
1349 *
1350 * Metadata stream lock MUST be acquired.
1351 */
1352 enum sync_metadata_status lttng_kconsumer_sync_metadata(
1353 struct lttng_consumer_stream *metadata)
1354 {
1355 int ret;
1356 enum sync_metadata_status status;
1357
1358 assert(metadata);
1359
1360 ret = kernctl_buffer_flush(metadata->wait_fd);
1361 if (ret < 0) {
1362 ERR("Failed to flush kernel stream");
1363 status = SYNC_METADATA_STATUS_ERROR;
1364 goto end;
1365 }
1366
1367 ret = kernctl_snapshot(metadata->wait_fd);
1368 if (ret < 0) {
1369 if (errno == EAGAIN) {
1370 /* No new metadata, exit. */
1371 DBG("Sync metadata, no new kernel metadata");
1372 status = SYNC_METADATA_STATUS_NO_DATA;
1373 } else {
1374 ERR("Sync metadata, taking kernel snapshot failed.");
1375 status = SYNC_METADATA_STATUS_ERROR;
1376 }
1377 } else {
1378 status = SYNC_METADATA_STATUS_NEW_DATA;
1379 }
1380
1381 end:
1382 return status;
1383 }
1384
1385 static
1386 int extract_common_subbuffer_info(struct lttng_consumer_stream *stream,
1387 struct stream_subbuffer *subbuf)
1388 {
1389 int ret;
1390
1391 ret = kernctl_get_subbuf_size(
1392 stream->wait_fd, &subbuf->info.data.subbuf_size);
1393 if (ret) {
1394 goto end;
1395 }
1396
1397 ret = kernctl_get_padded_subbuf_size(
1398 stream->wait_fd, &subbuf->info.data.padded_subbuf_size);
1399 if (ret) {
1400 goto end;
1401 }
1402
1403 end:
1404 return ret;
1405 }
1406
1407 static
1408 int extract_metadata_subbuffer_info(struct lttng_consumer_stream *stream,
1409 struct stream_subbuffer *subbuf)
1410 {
1411 int ret;
1412
1413 ret = extract_common_subbuffer_info(stream, subbuf);
1414 if (ret) {
1415 goto end;
1416 }
1417
1418 ret = kernctl_get_metadata_version(
1419 stream->wait_fd, &subbuf->info.metadata.version);
1420 if (ret) {
1421 goto end;
1422 }
1423
1424 end:
1425 return ret;
1426 }
1427
1428 static
1429 int extract_data_subbuffer_info(struct lttng_consumer_stream *stream,
1430 struct stream_subbuffer *subbuf)
1431 {
1432 int ret;
1433
1434 ret = extract_common_subbuffer_info(stream, subbuf);
1435 if (ret) {
1436 goto end;
1437 }
1438
1439 ret = kernctl_get_packet_size(
1440 stream->wait_fd, &subbuf->info.data.packet_size);
1441 if (ret < 0) {
1442 PERROR("Failed to get sub-buffer packet size");
1443 goto end;
1444 }
1445
1446 ret = kernctl_get_content_size(
1447 stream->wait_fd, &subbuf->info.data.content_size);
1448 if (ret < 0) {
1449 PERROR("Failed to get sub-buffer content size");
1450 goto end;
1451 }
1452
1453 ret = kernctl_get_timestamp_begin(
1454 stream->wait_fd, &subbuf->info.data.timestamp_begin);
1455 if (ret < 0) {
1456 PERROR("Failed to get sub-buffer begin timestamp");
1457 goto end;
1458 }
1459
1460 ret = kernctl_get_timestamp_end(
1461 stream->wait_fd, &subbuf->info.data.timestamp_end);
1462 if (ret < 0) {
1463 PERROR("Failed to get sub-buffer end timestamp");
1464 goto end;
1465 }
1466
1467 ret = kernctl_get_events_discarded(
1468 stream->wait_fd, &subbuf->info.data.events_discarded);
1469 if (ret) {
1470 PERROR("Failed to get sub-buffer events discarded count");
1471 goto end;
1472 }
1473
1474 ret = kernctl_get_sequence_number(stream->wait_fd,
1475 &subbuf->info.data.sequence_number.value);
1476 if (ret) {
1477 /* May not be supported by older LTTng-modules. */
1478 if (ret != -ENOTTY) {
1479 PERROR("Failed to get sub-buffer sequence number");
1480 goto end;
1481 }
1482 } else {
1483 subbuf->info.data.sequence_number.is_set = true;
1484 }
1485
1486 ret = kernctl_get_stream_id(
1487 stream->wait_fd, &subbuf->info.data.stream_id);
1488 if (ret < 0) {
1489 PERROR("Failed to get stream id");
1490 goto end;
1491 }
1492
1493 ret = kernctl_get_instance_id(stream->wait_fd,
1494 &subbuf->info.data.stream_instance_id.value);
1495 if (ret) {
1496 /* May not be supported by older LTTng-modules. */
1497 if (ret != -ENOTTY) {
1498 PERROR("Failed to get stream instance id");
1499 goto end;
1500 }
1501 } else {
1502 subbuf->info.data.stream_instance_id.is_set = true;
1503 }
1504 end:
1505 return ret;
1506 }
1507
1508 static
1509 int get_subbuffer_common(struct lttng_consumer_stream *stream,
1510 struct stream_subbuffer *subbuffer)
1511 {
1512 int ret;
1513
1514 ret = kernctl_get_next_subbuf(stream->wait_fd);
1515 if (ret) {
1516 goto end;
1517 }
1518
1519 ret = stream->read_subbuffer_ops.extract_subbuffer_info(
1520 stream, subbuffer);
1521 end:
1522 return ret;
1523 }
1524
1525 static
1526 int get_next_subbuffer_splice(struct lttng_consumer_stream *stream,
1527 struct stream_subbuffer *subbuffer)
1528 {
1529 int ret;
1530
1531 ret = get_subbuffer_common(stream, subbuffer);
1532 if (ret) {
1533 goto end;
1534 }
1535
1536 subbuffer->buffer.fd = stream->wait_fd;
1537 end:
1538 return ret;
1539 }
1540
1541 static
1542 int get_next_subbuffer_mmap(struct lttng_consumer_stream *stream,
1543 struct stream_subbuffer *subbuffer)
1544 {
1545 int ret;
1546 const char *addr;
1547
1548 ret = get_subbuffer_common(stream, subbuffer);
1549 if (ret) {
1550 goto end;
1551 }
1552
1553 ret = get_current_subbuf_addr(stream, &addr);
1554 if (ret) {
1555 goto end;
1556 }
1557
1558 subbuffer->buffer.buffer = lttng_buffer_view_init(
1559 addr, 0, subbuffer->info.data.padded_subbuf_size);
1560 end:
1561 return ret;
1562 }
1563
1564 static
1565 int get_next_subbuffer_metadata_check(struct lttng_consumer_stream *stream,
1566 struct stream_subbuffer *subbuffer)
1567 {
1568 int ret;
1569 const char *addr;
1570 bool coherent;
1571
1572 ret = kernctl_get_next_subbuf_metadata_check(stream->wait_fd,
1573 &coherent);
1574 if (ret) {
1575 goto end;
1576 }
1577
1578 ret = stream->read_subbuffer_ops.extract_subbuffer_info(
1579 stream, subbuffer);
1580 if (ret) {
1581 goto end;
1582 }
1583
1584 LTTNG_OPTIONAL_SET(&subbuffer->info.metadata.coherent, coherent);
1585
1586 ret = get_current_subbuf_addr(stream, &addr);
1587 if (ret) {
1588 goto end;
1589 }
1590
1591 subbuffer->buffer.buffer = lttng_buffer_view_init(
1592 addr, 0, subbuffer->info.data.padded_subbuf_size);
1593 DBG("Got metadata packet with padded_subbuf_size = %lu, coherent = %s",
1594 subbuffer->info.metadata.padded_subbuf_size,
1595 coherent ? "true" : "false");
1596 end:
1597 return ret;
1598 }
1599
1600 static
1601 int put_next_subbuffer(struct lttng_consumer_stream *stream,
1602 struct stream_subbuffer *subbuffer)
1603 {
1604 const int ret = kernctl_put_next_subbuf(stream->wait_fd);
1605
1606 if (ret) {
1607 if (ret == -EFAULT) {
1608 PERROR("Error in unreserving sub buffer");
1609 } else if (ret == -EIO) {
1610 /* Should never happen with newer LTTng versions */
1611 PERROR("Reader has been pushed by the writer, last sub-buffer corrupted");
1612 }
1613 }
1614
1615 return ret;
1616 }
1617
1618 static
1619 bool is_get_next_check_metadata_available(int tracer_fd)
1620 {
1621 return kernctl_get_next_subbuf_metadata_check(tracer_fd, NULL) !=
1622 -ENOTTY;
1623 }
1624
1625 static
1626 int lttng_kconsumer_set_stream_ops(
1627 struct lttng_consumer_stream *stream)
1628 {
1629 int ret = 0;
1630
1631 if (stream->metadata_flag && stream->chan->is_live) {
1632 DBG("Attempting to enable metadata bucketization for live consumers");
1633 if (is_get_next_check_metadata_available(stream->wait_fd)) {
1634 DBG("Kernel tracer supports get_next_subbuffer_metadata_check, metadata will be accumulated until a coherent state is reached");
1635 stream->read_subbuffer_ops.get_next_subbuffer =
1636 get_next_subbuffer_metadata_check;
1637 ret = consumer_stream_enable_metadata_bucketization(
1638 stream);
1639 if (ret) {
1640 goto end;
1641 }
1642 } else {
1643 /*
1644 * The kernel tracer version is too old to indicate
1645 * when the metadata stream has reached a "coherent"
1646 * (parseable) point.
1647 *
1648 * This means that a live viewer may see an incoherent
1649 * sequence of metadata and fail to parse it.
1650 */
1651 WARN("Kernel tracer does not support get_next_subbuffer_metadata_check which may cause live clients to fail to parse the metadata stream");
1652 metadata_bucket_destroy(stream->metadata_bucket);
1653 stream->metadata_bucket = NULL;
1654 }
1655 }
1656
1657 if (!stream->read_subbuffer_ops.get_next_subbuffer) {
1658 if (stream->chan->output == CONSUMER_CHANNEL_MMAP) {
1659 stream->read_subbuffer_ops.get_next_subbuffer =
1660 get_next_subbuffer_mmap;
1661 } else {
1662 stream->read_subbuffer_ops.get_next_subbuffer =
1663 get_next_subbuffer_splice;
1664 }
1665 }
1666
1667 if (stream->metadata_flag) {
1668 stream->read_subbuffer_ops.extract_subbuffer_info =
1669 extract_metadata_subbuffer_info;
1670 } else {
1671 stream->read_subbuffer_ops.extract_subbuffer_info =
1672 extract_data_subbuffer_info;
1673 if (stream->chan->is_live) {
1674 stream->read_subbuffer_ops.send_live_beacon =
1675 consumer_flush_kernel_index;
1676 }
1677 }
1678
1679 stream->read_subbuffer_ops.put_next_subbuffer = put_next_subbuffer;
1680 end:
1681 return ret;
1682 }
1683
1684 int lttng_kconsumer_on_recv_stream(struct lttng_consumer_stream *stream)
1685 {
1686 int ret;
1687
1688 assert(stream);
1689
1690 /*
1691 * Don't create anything if this is set for streaming or if there is
1692 * no current trace chunk on the parent channel.
1693 */
1694 if (stream->net_seq_idx == (uint64_t) -1ULL && stream->chan->monitor &&
1695 stream->chan->trace_chunk) {
1696 ret = consumer_stream_create_output_files(stream, true);
1697 if (ret) {
1698 goto error;
1699 }
1700 }
1701
1702 if (stream->output == LTTNG_EVENT_MMAP) {
1703 /* get the len of the mmap region */
1704 unsigned long mmap_len;
1705
1706 ret = kernctl_get_mmap_len(stream->wait_fd, &mmap_len);
1707 if (ret != 0) {
1708 PERROR("kernctl_get_mmap_len");
1709 goto error_close_fd;
1710 }
1711 stream->mmap_len = (size_t) mmap_len;
1712
1713 stream->mmap_base = mmap(NULL, stream->mmap_len, PROT_READ,
1714 MAP_PRIVATE, stream->wait_fd, 0);
1715 if (stream->mmap_base == MAP_FAILED) {
1716 PERROR("Error mmaping");
1717 ret = -1;
1718 goto error_close_fd;
1719 }
1720 }
1721
1722 ret = lttng_kconsumer_set_stream_ops(stream);
1723 if (ret) {
1724 goto error_close_fd;
1725 }
1726
1727 /* we return 0 to let the library handle the FD internally */
1728 return 0;
1729
1730 error_close_fd:
1731 if (stream->out_fd >= 0) {
1732 int err;
1733
1734 err = close(stream->out_fd);
1735 assert(!err);
1736 stream->out_fd = -1;
1737 }
1738 error:
1739 return ret;
1740 }
1741
1742 /*
1743 * Check if data is still being extracted from the buffers for a specific
1744 * stream. Consumer data lock MUST be acquired before calling this function
1745 * and the stream lock.
1746 *
1747 * Return 1 if the traced data are still getting read else 0 meaning that the
1748 * data is available for trace viewer reading.
1749 */
1750 int lttng_kconsumer_data_pending(struct lttng_consumer_stream *stream)
1751 {
1752 int ret;
1753
1754 assert(stream);
1755
1756 if (stream->endpoint_status != CONSUMER_ENDPOINT_ACTIVE) {
1757 ret = 0;
1758 goto end;
1759 }
1760
1761 ret = kernctl_get_next_subbuf(stream->wait_fd);
1762 if (ret == 0) {
1763 /* There is still data so let's put back this subbuffer. */
1764 ret = kernctl_put_subbuf(stream->wait_fd);
1765 assert(ret == 0);
1766 ret = 1; /* Data is pending */
1767 goto end;
1768 }
1769
1770 /* Data is NOT pending and ready to be read. */
1771 ret = 0;
1772
1773 end:
1774 return ret;
1775 }
LTTng 2.0 tools and control repository
This page took 0.083954 seconds and 3 git commands to generate.