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