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