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