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