Fix: change ERR statement to DBG in kernel read subbuffer
[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>
4 *
d14d33bf
AM
5 * This program is free software; you can redistribute it and/or modify
6 * it under the terms of the GNU General Public License, version 2 only,
7 * as published by the Free Software Foundation.
3bd1e081
MD
8 *
9 * This program is distributed in the hope that it will be useful,
10 * but WITHOUT ANY WARRANTY; without even the implied warranty of
11 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
12 * GNU General Public License for more details.
13 *
d14d33bf
AM
14 * You should have received a copy of the GNU General Public License along
15 * with this program; if not, write to the Free Software Foundation, Inc.,
16 * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
3bd1e081
MD
17 */
18
19#define _GNU_SOURCE
20#include <assert.h>
3bd1e081
MD
21#include <poll.h>
22#include <pthread.h>
23#include <stdlib.h>
24#include <string.h>
25#include <sys/mman.h>
26#include <sys/socket.h>
27#include <sys/types.h>
77c7c900 28#include <inttypes.h>
3bd1e081 29#include <unistd.h>
dbb5dfe6 30#include <sys/stat.h>
3bd1e081 31
51a9e1c7 32#include <bin/lttng-consumerd/health-consumerd.h>
990570ed 33#include <common/common.h>
10a8a223 34#include <common/kernel-ctl/kernel-ctl.h>
10a8a223 35#include <common/sessiond-comm/sessiond-comm.h>
00e2e675 36#include <common/sessiond-comm/relayd.h>
dbb5dfe6 37#include <common/compat/fcntl.h>
acdb9057 38#include <common/pipe.h>
00e2e675 39#include <common/relayd/relayd.h>
fe4477ee 40#include <common/utils.h>
07b86b52 41#include <common/consumer-stream.h>
309167d2 42#include <common/index/index.h>
d3e2ba59 43#include <common/consumer-timer.h>
0857097f 44
10a8a223 45#include "kernel-consumer.h"
3bd1e081
MD
46
47extern struct lttng_consumer_global_data consumer_data;
48extern int consumer_poll_timeout;
49extern volatile int consumer_quit;
50
3bd1e081
MD
51/*
52 * Take a snapshot for a specific fd
53 *
54 * Returns 0 on success, < 0 on error
55 */
ffe60014 56int lttng_kconsumer_take_snapshot(struct lttng_consumer_stream *stream)
3bd1e081
MD
57{
58 int ret = 0;
59 int infd = stream->wait_fd;
60
61 ret = kernctl_snapshot(infd);
62 if (ret != 0) {
3bd1e081 63 perror("Getting sub-buffer snapshot.");
56591bac 64 ret = -errno;
3bd1e081
MD
65 }
66
67 return ret;
68}
69
70/*
71 * Get the produced position
72 *
73 * Returns 0 on success, < 0 on error
74 */
ffe60014 75int lttng_kconsumer_get_produced_snapshot(struct lttng_consumer_stream *stream,
3bd1e081
MD
76 unsigned long *pos)
77{
78 int ret;
79 int infd = stream->wait_fd;
80
81 ret = kernctl_snapshot_get_produced(infd, pos);
82 if (ret != 0) {
3bd1e081 83 perror("kernctl_snapshot_get_produced");
56591bac 84 ret = -errno;
3bd1e081
MD
85 }
86
87 return ret;
88}
89
07b86b52
JD
90/*
91 * Get the consumerd position
92 *
93 * Returns 0 on success, < 0 on error
94 */
95int lttng_kconsumer_get_consumed_snapshot(struct lttng_consumer_stream *stream,
96 unsigned long *pos)
97{
98 int ret;
99 int infd = stream->wait_fd;
100
101 ret = kernctl_snapshot_get_consumed(infd, pos);
102 if (ret != 0) {
07b86b52 103 perror("kernctl_snapshot_get_consumed");
56591bac 104 ret = -errno;
07b86b52
JD
105 }
106
107 return ret;
108}
109
07b86b52
JD
110/*
111 * Take a snapshot of all the stream of a channel
112 *
113 * Returns 0 on success, < 0 on error
114 */
115int lttng_kconsumer_snapshot_channel(uint64_t key, char *path,
5c786ded
JD
116 uint64_t relayd_id, uint64_t max_stream_size,
117 struct lttng_consumer_local_data *ctx)
07b86b52
JD
118{
119 int ret;
120 unsigned long consumed_pos, produced_pos;
121 struct lttng_consumer_channel *channel;
122 struct lttng_consumer_stream *stream;
123
6a00837f 124 DBG("Kernel consumer snapshot channel %" PRIu64, key);
07b86b52
JD
125
126 rcu_read_lock();
127
128 channel = consumer_find_channel(key);
129 if (!channel) {
6a00837f 130 ERR("No channel found for key %" PRIu64, key);
07b86b52
JD
131 ret = -1;
132 goto end;
133 }
134
135 /* Splice is not supported yet for channel snapshot. */
136 if (channel->output != CONSUMER_CHANNEL_MMAP) {
137 ERR("Unsupported output %d", channel->output);
138 ret = -1;
139 goto end;
140 }
141
10a50311 142 cds_list_for_each_entry(stream, &channel->streams.head, send_node) {
9ce5646a
MD
143
144 health_code_update();
145
07b86b52
JD
146 /*
147 * Lock stream because we are about to change its state.
148 */
149 pthread_mutex_lock(&stream->lock);
150
29decac3
DG
151 /*
152 * Assign the received relayd ID so we can use it for streaming. The streams
153 * are not visible to anyone so this is OK to change it.
154 */
07b86b52
JD
155 stream->net_seq_idx = relayd_id;
156 channel->relayd_id = relayd_id;
157 if (relayd_id != (uint64_t) -1ULL) {
10a50311 158 ret = consumer_send_relayd_stream(stream, path);
07b86b52
JD
159 if (ret < 0) {
160 ERR("sending stream to relayd");
161 goto end_unlock;
162 }
07b86b52
JD
163 } else {
164 ret = utils_create_stream_file(path, stream->name,
10a50311
JD
165 stream->chan->tracefile_size,
166 stream->tracefile_count_current,
309167d2 167 stream->uid, stream->gid, NULL);
07b86b52
JD
168 if (ret < 0) {
169 ERR("utils_create_stream_file");
170 goto end_unlock;
171 }
172
173 stream->out_fd = ret;
174 stream->tracefile_size_current = 0;
175
81ea21bf
MD
176 DBG("Kernel consumer snapshot stream %s/%s (%" PRIu64 ")",
177 path, stream->name, stream->key);
07b86b52 178 }
601262d6
JD
179 if (relayd_id != -1ULL) {
180 ret = consumer_send_relayd_streams_sent(relayd_id);
181 if (ret < 0) {
182 ERR("sending streams sent to relayd");
183 goto end_unlock;
184 }
a4baae1b 185 }
07b86b52
JD
186
187 ret = kernctl_buffer_flush(stream->wait_fd);
188 if (ret < 0) {
10a50311 189 ERR("Failed to flush kernel stream");
56591bac 190 ret = -errno;
07b86b52
JD
191 goto end_unlock;
192 }
193
194 ret = lttng_kconsumer_take_snapshot(stream);
195 if (ret < 0) {
196 ERR("Taking kernel snapshot");
197 goto end_unlock;
198 }
199
200 ret = lttng_kconsumer_get_produced_snapshot(stream, &produced_pos);
201 if (ret < 0) {
202 ERR("Produced kernel snapshot position");
203 goto end_unlock;
204 }
205
206 ret = lttng_kconsumer_get_consumed_snapshot(stream, &consumed_pos);
207 if (ret < 0) {
208 ERR("Consumerd kernel snapshot position");
209 goto end_unlock;
210 }
211
212 if (stream->max_sb_size == 0) {
213 ret = kernctl_get_max_subbuf_size(stream->wait_fd,
214 &stream->max_sb_size);
215 if (ret < 0) {
216 ERR("Getting kernel max_sb_size");
56591bac 217 ret = -errno;
07b86b52
JD
218 goto end_unlock;
219 }
220 }
221
5c786ded
JD
222 /*
223 * The original value is sent back if max stream size is larger than
224 * the possible size of the snapshot. Also, we asume that the session
225 * daemon should never send a maximum stream size that is lower than
226 * subbuffer size.
227 */
228 consumed_pos = consumer_get_consumed_maxsize(consumed_pos,
229 produced_pos, max_stream_size);
230
07b86b52
JD
231 while (consumed_pos < produced_pos) {
232 ssize_t read_len;
233 unsigned long len, padded_len;
234
9ce5646a
MD
235 health_code_update();
236
07b86b52
JD
237 DBG("Kernel consumer taking snapshot at pos %lu", consumed_pos);
238
239 ret = kernctl_get_subbuf(stream->wait_fd, &consumed_pos);
240 if (ret < 0) {
241 if (errno != EAGAIN) {
242 PERROR("kernctl_get_subbuf snapshot");
56591bac 243 ret = -errno;
07b86b52
JD
244 goto end_unlock;
245 }
246 DBG("Kernel consumer get subbuf failed. Skipping it.");
247 consumed_pos += stream->max_sb_size;
248 continue;
249 }
250
251 ret = kernctl_get_subbuf_size(stream->wait_fd, &len);
252 if (ret < 0) {
253 ERR("Snapshot kernctl_get_subbuf_size");
56591bac 254 ret = -errno;
29decac3 255 goto error_put_subbuf;
07b86b52
JD
256 }
257
258 ret = kernctl_get_padded_subbuf_size(stream->wait_fd, &padded_len);
259 if (ret < 0) {
260 ERR("Snapshot kernctl_get_padded_subbuf_size");
56591bac 261 ret = -errno;
29decac3 262 goto error_put_subbuf;
07b86b52
JD
263 }
264
265 read_len = lttng_consumer_on_read_subbuffer_mmap(ctx, stream, len,
309167d2 266 padded_len - len, NULL);
07b86b52 267 /*
29decac3
DG
268 * We write the padded len in local tracefiles but the data len
269 * when using a relay. Display the error but continue processing
270 * to try to release the subbuffer.
07b86b52
JD
271 */
272 if (relayd_id != (uint64_t) -1ULL) {
273 if (read_len != len) {
274 ERR("Error sending to the relay (ret: %zd != len: %lu)",
275 read_len, len);
276 }
277 } else {
278 if (read_len != padded_len) {
279 ERR("Error writing to tracefile (ret: %zd != len: %lu)",
280 read_len, padded_len);
281 }
282 }
283
284 ret = kernctl_put_subbuf(stream->wait_fd);
285 if (ret < 0) {
286 ERR("Snapshot kernctl_put_subbuf");
56591bac 287 ret = -errno;
07b86b52
JD
288 goto end_unlock;
289 }
290 consumed_pos += stream->max_sb_size;
291 }
292
293 if (relayd_id == (uint64_t) -1ULL) {
fdf9986c
MD
294 if (stream->out_fd >= 0) {
295 ret = close(stream->out_fd);
296 if (ret < 0) {
297 PERROR("Kernel consumer snapshot close out_fd");
298 goto end_unlock;
299 }
300 stream->out_fd = -1;
07b86b52 301 }
07b86b52
JD
302 } else {
303 close_relayd_stream(stream);
304 stream->net_seq_idx = (uint64_t) -1ULL;
305 }
306 pthread_mutex_unlock(&stream->lock);
307 }
308
309 /* All good! */
310 ret = 0;
311 goto end;
312
29decac3
DG
313error_put_subbuf:
314 ret = kernctl_put_subbuf(stream->wait_fd);
315 if (ret < 0) {
56591bac 316 ret = -errno;
29decac3
DG
317 ERR("Snapshot kernctl_put_subbuf error path");
318 }
07b86b52
JD
319end_unlock:
320 pthread_mutex_unlock(&stream->lock);
321end:
322 rcu_read_unlock();
323 return ret;
324}
325
326/*
327 * Read the whole metadata available for a snapshot.
328 *
329 * Returns 0 on success, < 0 on error
330 */
331int lttng_kconsumer_snapshot_metadata(uint64_t key, char *path,
e2039c7a 332 uint64_t relayd_id, struct lttng_consumer_local_data *ctx)
07b86b52 333{
d771f832
DG
334 int ret, use_relayd = 0;
335 ssize_t ret_read;
07b86b52
JD
336 struct lttng_consumer_channel *metadata_channel;
337 struct lttng_consumer_stream *metadata_stream;
d771f832
DG
338
339 assert(ctx);
07b86b52
JD
340
341 DBG("Kernel consumer snapshot metadata with key %" PRIu64 " at path %s",
342 key, path);
343
344 rcu_read_lock();
345
346 metadata_channel = consumer_find_channel(key);
347 if (!metadata_channel) {
d771f832 348 ERR("Kernel snapshot metadata not found for key %" PRIu64, key);
07b86b52 349 ret = -1;
d771f832 350 goto error;
07b86b52
JD
351 }
352
353 metadata_stream = metadata_channel->metadata_stream;
354 assert(metadata_stream);
355
d771f832 356 /* Flag once that we have a valid relayd for the stream. */
e2039c7a 357 if (relayd_id != (uint64_t) -1ULL) {
d771f832
DG
358 use_relayd = 1;
359 }
360
361 if (use_relayd) {
10a50311 362 ret = consumer_send_relayd_stream(metadata_stream, path);
e2039c7a 363 if (ret < 0) {
d771f832 364 goto error;
e2039c7a 365 }
e2039c7a
JD
366 } else {
367 ret = utils_create_stream_file(path, metadata_stream->name,
368 metadata_stream->chan->tracefile_size,
369 metadata_stream->tracefile_count_current,
309167d2 370 metadata_stream->uid, metadata_stream->gid, NULL);
e2039c7a 371 if (ret < 0) {
d771f832 372 goto error;
e2039c7a
JD
373 }
374 metadata_stream->out_fd = ret;
07b86b52 375 }
07b86b52 376
d771f832 377 do {
9ce5646a
MD
378 health_code_update();
379
d771f832
DG
380 ret_read = lttng_kconsumer_read_subbuffer(metadata_stream, ctx);
381 if (ret_read < 0) {
56591bac 382 if (ret_read != -EAGAIN) {
6a00837f 383 ERR("Kernel snapshot reading metadata subbuffer (ret: %zd)",
d771f832
DG
384 ret_read);
385 goto error;
07b86b52 386 }
d771f832 387 /* ret_read is negative at this point so we will exit the loop. */
07b86b52
JD
388 continue;
389 }
d771f832 390 } while (ret_read >= 0);
07b86b52 391
d771f832
DG
392 if (use_relayd) {
393 close_relayd_stream(metadata_stream);
394 metadata_stream->net_seq_idx = (uint64_t) -1ULL;
395 } else {
fdf9986c
MD
396 if (metadata_stream->out_fd >= 0) {
397 ret = close(metadata_stream->out_fd);
398 if (ret < 0) {
399 PERROR("Kernel consumer snapshot metadata close out_fd");
400 /*
401 * Don't go on error here since the snapshot was successful at this
402 * point but somehow the close failed.
403 */
404 }
405 metadata_stream->out_fd = -1;
e2039c7a 406 }
e2039c7a
JD
407 }
408
07b86b52 409 ret = 0;
d771f832 410
cf53a8a6
JD
411 cds_list_del(&metadata_stream->send_node);
412 consumer_stream_destroy(metadata_stream, NULL);
413 metadata_channel->metadata_stream = NULL;
d771f832 414error:
07b86b52
JD
415 rcu_read_unlock();
416 return ret;
417}
418
1803a064
MD
419/*
420 * Receive command from session daemon and process it.
421 *
422 * Return 1 on success else a negative value or 0.
423 */
3bd1e081
MD
424int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
425 int sock, struct pollfd *consumer_sockpoll)
426{
427 ssize_t ret;
0c759fc9 428 enum lttcomm_return_code ret_code = LTTCOMM_CONSUMERD_SUCCESS;
3bd1e081
MD
429 struct lttcomm_consumer_msg msg;
430
9ce5646a
MD
431 health_code_update();
432
3bd1e081
MD
433 ret = lttcomm_recv_unix_sock(sock, &msg, sizeof(msg));
434 if (ret != sizeof(msg)) {
1803a064 435 if (ret > 0) {
c6857fcf 436 lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_ERROR_RECV_CMD);
1803a064
MD
437 ret = -1;
438 }
3bd1e081
MD
439 return ret;
440 }
9ce5646a
MD
441
442 health_code_update();
443
3bd1e081 444 if (msg.cmd_type == LTTNG_CONSUMER_STOP) {
f50f23d9
DG
445 /*
446 * Notify the session daemon that the command is completed.
447 *
448 * On transport layer error, the function call will print an error
449 * message so handling the returned code is a bit useless since we
450 * return an error code anyway.
451 */
452 (void) consumer_send_status_msg(sock, ret_code);
3bd1e081
MD
453 return -ENOENT;
454 }
455
9ce5646a
MD
456 health_code_update();
457
b0b335c8
MD
458 /* relayd needs RCU read-side protection */
459 rcu_read_lock();
460
3bd1e081 461 switch (msg.cmd_type) {
00e2e675
DG
462 case LTTNG_CONSUMER_ADD_RELAYD_SOCKET:
463 {
f50f23d9 464 /* Session daemon status message are handled in the following call. */
7735ef9e
DG
465 ret = consumer_add_relayd_socket(msg.u.relayd_sock.net_index,
466 msg.u.relayd_sock.type, ctx, sock, consumer_sockpoll,
d3e2ba59
JD
467 &msg.u.relayd_sock.sock, msg.u.relayd_sock.session_id,
468 msg.u.relayd_sock.relayd_session_id);
00e2e675
DG
469 goto end_nosignal;
470 }
3bd1e081
MD
471 case LTTNG_CONSUMER_ADD_CHANNEL:
472 {
473 struct lttng_consumer_channel *new_channel;
e43c41c5 474 int ret_recv;
3bd1e081 475
9ce5646a
MD
476 health_code_update();
477
f50f23d9
DG
478 /* First send a status message before receiving the fds. */
479 ret = consumer_send_status_msg(sock, ret_code);
480 if (ret < 0) {
481 /* Somehow, the session daemon is not responding anymore. */
1803a064 482 goto error_fatal;
f50f23d9 483 }
9ce5646a
MD
484
485 health_code_update();
486
d88aee68 487 DBG("consumer_add_channel %" PRIu64, msg.u.channel.channel_key);
3bd1e081 488 new_channel = consumer_allocate_channel(msg.u.channel.channel_key,
ffe60014
DG
489 msg.u.channel.session_id, msg.u.channel.pathname,
490 msg.u.channel.name, msg.u.channel.uid, msg.u.channel.gid,
1624d5b7
JD
491 msg.u.channel.relayd_id, msg.u.channel.output,
492 msg.u.channel.tracefile_size,
1950109e 493 msg.u.channel.tracefile_count, 0,
ecc48a90
JD
494 msg.u.channel.monitor,
495 msg.u.channel.live_timer_interval);
3bd1e081 496 if (new_channel == NULL) {
f73fabfd 497 lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_OUTFD_ERROR);
3bd1e081
MD
498 goto end_nosignal;
499 }
ffe60014 500 new_channel->nb_init_stream_left = msg.u.channel.nb_init_streams;
95a1109b
JD
501 switch (msg.u.channel.output) {
502 case LTTNG_EVENT_SPLICE:
503 new_channel->output = CONSUMER_CHANNEL_SPLICE;
504 break;
505 case LTTNG_EVENT_MMAP:
506 new_channel->output = CONSUMER_CHANNEL_MMAP;
507 break;
508 default:
509 ERR("Channel output unknown %d", msg.u.channel.output);
510 goto end_nosignal;
511 }
ffe60014
DG
512
513 /* Translate and save channel type. */
514 switch (msg.u.channel.type) {
515 case CONSUMER_CHANNEL_TYPE_DATA:
516 case CONSUMER_CHANNEL_TYPE_METADATA:
517 new_channel->type = msg.u.channel.type;
518 break;
519 default:
520 assert(0);
521 goto end_nosignal;
522 };
523
9ce5646a
MD
524 health_code_update();
525
3bd1e081 526 if (ctx->on_recv_channel != NULL) {
e43c41c5
JD
527 ret_recv = ctx->on_recv_channel(new_channel);
528 if (ret_recv == 0) {
529 ret = consumer_add_channel(new_channel, ctx);
530 } else if (ret_recv < 0) {
3bd1e081
MD
531 goto end_nosignal;
532 }
533 } else {
e43c41c5 534 ret = consumer_add_channel(new_channel, ctx);
3bd1e081 535 }
94d49140
JD
536 if (CONSUMER_CHANNEL_TYPE_DATA) {
537 consumer_timer_live_start(new_channel,
538 msg.u.channel.live_timer_interval);
539 }
e43c41c5 540
9ce5646a
MD
541 health_code_update();
542
e43c41c5 543 /* If we received an error in add_channel, we need to report it. */
821fffb2 544 if (ret < 0) {
1803a064
MD
545 ret = consumer_send_status_msg(sock, ret);
546 if (ret < 0) {
547 goto error_fatal;
548 }
e43c41c5
JD
549 goto end_nosignal;
550 }
551
3bd1e081
MD
552 goto end_nosignal;
553 }
554 case LTTNG_CONSUMER_ADD_STREAM:
555 {
dae10966
DG
556 int fd;
557 struct lttng_pipe *stream_pipe;
00e2e675 558 struct lttng_consumer_stream *new_stream;
ffe60014 559 struct lttng_consumer_channel *channel;
c80048c6 560 int alloc_ret = 0;
3bd1e081 561
ffe60014
DG
562 /*
563 * Get stream's channel reference. Needed when adding the stream to the
564 * global hash table.
565 */
566 channel = consumer_find_channel(msg.u.stream.channel_key);
567 if (!channel) {
568 /*
569 * We could not find the channel. Can happen if cpu hotplug
570 * happens while tearing down.
571 */
d88aee68 572 ERR("Unable to find channel key %" PRIu64, msg.u.stream.channel_key);
ffe60014
DG
573 ret_code = LTTNG_ERR_KERN_CHAN_NOT_FOUND;
574 }
575
9ce5646a
MD
576 health_code_update();
577
f50f23d9
DG
578 /* First send a status message before receiving the fds. */
579 ret = consumer_send_status_msg(sock, ret_code);
1803a064 580 if (ret < 0) {
d771f832 581 /* Somehow, the session daemon is not responding anymore. */
1803a064
MD
582 goto error_fatal;
583 }
9ce5646a
MD
584
585 health_code_update();
586
0c759fc9 587 if (ret_code != LTTCOMM_CONSUMERD_SUCCESS) {
d771f832 588 /* Channel was not found. */
f50f23d9
DG
589 goto end_nosignal;
590 }
591
d771f832 592 /* Blocking call */
9ce5646a
MD
593 health_poll_entry();
594 ret = lttng_consumer_poll_socket(consumer_sockpoll);
595 health_poll_exit();
596 if (ret < 0) {
3f8e211f 597 rcu_read_unlock();
3bd1e081
MD
598 return -EINTR;
599 }
00e2e675 600
9ce5646a
MD
601 health_code_update();
602
00e2e675 603 /* Get stream file descriptor from socket */
f2fc6720
MD
604 ret = lttcomm_recv_fds_unix_sock(sock, &fd, 1);
605 if (ret != sizeof(fd)) {
f73fabfd 606 lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_ERROR_RECV_FD);
3f8e211f 607 rcu_read_unlock();
3bd1e081
MD
608 return ret;
609 }
3bd1e081 610
9ce5646a
MD
611 health_code_update();
612
f50f23d9
DG
613 /*
614 * Send status code to session daemon only if the recv works. If the
615 * above recv() failed, the session daemon is notified through the
616 * error socket and the teardown is eventually done.
617 */
618 ret = consumer_send_status_msg(sock, ret_code);
619 if (ret < 0) {
620 /* Somehow, the session daemon is not responding anymore. */
621 goto end_nosignal;
622 }
623
9ce5646a
MD
624 health_code_update();
625
ffe60014
DG
626 new_stream = consumer_allocate_stream(channel->key,
627 fd,
628 LTTNG_CONSUMER_ACTIVE_STREAM,
629 channel->name,
630 channel->uid,
631 channel->gid,
632 channel->relayd_id,
633 channel->session_id,
634 msg.u.stream.cpu,
635 &alloc_ret,
4891ece8
DG
636 channel->type,
637 channel->monitor);
3bd1e081 638 if (new_stream == NULL) {
c80048c6
MD
639 switch (alloc_ret) {
640 case -ENOMEM:
641 case -EINVAL:
642 default:
643 lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_OUTFD_ERROR);
644 break;
c80048c6 645 }
3f8e211f 646 goto end_nosignal;
3bd1e081 647 }
d771f832 648
ffe60014
DG
649 new_stream->chan = channel;
650 new_stream->wait_fd = fd;
07b86b52
JD
651 switch (channel->output) {
652 case CONSUMER_CHANNEL_SPLICE:
653 new_stream->output = LTTNG_EVENT_SPLICE;
654 break;
655 case CONSUMER_CHANNEL_MMAP:
656 new_stream->output = LTTNG_EVENT_MMAP;
657 break;
658 default:
659 ERR("Stream output unknown %d", channel->output);
660 goto end_nosignal;
661 }
00e2e675 662
a0c83db9
DG
663 /*
664 * We've just assigned the channel to the stream so increment the
07b86b52
JD
665 * refcount right now. We don't need to increment the refcount for
666 * streams in no monitor because we handle manually the cleanup of
667 * those. It is very important to make sure there is NO prior
668 * consumer_del_stream() calls or else the refcount will be unbalanced.
a0c83db9 669 */
07b86b52
JD
670 if (channel->monitor) {
671 uatomic_inc(&new_stream->chan->refcount);
672 }
9d9353f9 673
fb3a43a9
DG
674 /*
675 * The buffer flush is done on the session daemon side for the kernel
676 * so no need for the stream "hangup_flush_done" variable to be
677 * tracked. This is important for a kernel stream since we don't rely
678 * on the flush state of the stream to read data. It's not the case for
679 * user space tracing.
680 */
681 new_stream->hangup_flush_done = 0;
682
9ce5646a
MD
683 health_code_update();
684
633d0084
DG
685 if (ctx->on_recv_stream) {
686 ret = ctx->on_recv_stream(new_stream);
687 if (ret < 0) {
d771f832 688 consumer_stream_free(new_stream);
633d0084 689 goto end_nosignal;
fb3a43a9 690 }
633d0084 691 }
fb3a43a9 692
9ce5646a
MD
693 health_code_update();
694
07b86b52
JD
695 if (new_stream->metadata_flag) {
696 channel->metadata_stream = new_stream;
697 }
698
2bba9e53
DG
699 /* Do not monitor this stream. */
700 if (!channel->monitor) {
5eecee74 701 DBG("Kernel consumer add stream %s in no monitor mode with "
6dc3064a 702 "relayd id %" PRIu64, new_stream->name,
5eecee74 703 new_stream->net_seq_idx);
10a50311 704 cds_list_add(&new_stream->send_node, &channel->streams.head);
6dc3064a
DG
705 break;
706 }
707
e1b71bdc
DG
708 /* Send stream to relayd if the stream has an ID. */
709 if (new_stream->net_seq_idx != (uint64_t) -1ULL) {
194ee077
DG
710 ret = consumer_send_relayd_stream(new_stream,
711 new_stream->chan->pathname);
e1b71bdc
DG
712 if (ret < 0) {
713 consumer_stream_free(new_stream);
714 goto end_nosignal;
715 }
e2039c7a
JD
716 }
717
50f8ae69 718 /* Get the right pipe where the stream will be sent. */
633d0084 719 if (new_stream->metadata_flag) {
5ab66908
MD
720 ret = consumer_add_metadata_stream(new_stream);
721 if (ret) {
722 ERR("Consumer add metadata stream %" PRIu64 " failed. Continuing",
723 new_stream->key);
724 consumer_stream_free(new_stream);
725 goto end_nosignal;
726 }
dae10966 727 stream_pipe = ctx->consumer_metadata_pipe;
3bd1e081 728 } else {
5ab66908
MD
729 ret = consumer_add_data_stream(new_stream);
730 if (ret) {
731 ERR("Consumer add stream %" PRIu64 " failed. Continuing",
732 new_stream->key);
733 consumer_stream_free(new_stream);
734 goto end_nosignal;
735 }
dae10966 736 stream_pipe = ctx->consumer_data_pipe;
50f8ae69
DG
737 }
738
5ab66908
MD
739 /* Vitible to other threads */
740 new_stream->globally_visible = 1;
741
9ce5646a
MD
742 health_code_update();
743
dae10966 744 ret = lttng_pipe_write(stream_pipe, &new_stream, sizeof(new_stream));
50f8ae69 745 if (ret < 0) {
dae10966 746 ERR("Consumer write %s stream to pipe %d",
50f8ae69 747 new_stream->metadata_flag ? "metadata" : "data",
dae10966 748 lttng_pipe_get_writefd(stream_pipe));
5ab66908
MD
749 if (new_stream->metadata_flag) {
750 consumer_del_stream_for_metadata(new_stream);
751 } else {
752 consumer_del_stream_for_data(new_stream);
753 }
50f8ae69 754 goto end_nosignal;
3bd1e081 755 }
00e2e675 756
50f8ae69 757 DBG("Kernel consumer ADD_STREAM %s (fd: %d) with relayd id %" PRIu64,
ffe60014 758 new_stream->name, fd, new_stream->relayd_stream_id);
3bd1e081
MD
759 break;
760 }
a4baae1b
JD
761 case LTTNG_CONSUMER_STREAMS_SENT:
762 {
763 struct lttng_consumer_channel *channel;
764
765 /*
766 * Get stream's channel reference. Needed when adding the stream to the
767 * global hash table.
768 */
769 channel = consumer_find_channel(msg.u.sent_streams.channel_key);
770 if (!channel) {
771 /*
772 * We could not find the channel. Can happen if cpu hotplug
773 * happens while tearing down.
774 */
775 ERR("Unable to find channel key %" PRIu64,
776 msg.u.sent_streams.channel_key);
777 ret_code = LTTNG_ERR_KERN_CHAN_NOT_FOUND;
778 }
779
780 health_code_update();
781
782 /*
783 * Send status code to session daemon.
784 */
785 ret = consumer_send_status_msg(sock, ret_code);
786 if (ret < 0) {
787 /* Somehow, the session daemon is not responding anymore. */
788 goto end_nosignal;
789 }
790
791 health_code_update();
792
793 /*
794 * We should not send this message if we don't monitor the
795 * streams in this channel.
796 */
797 if (!channel->monitor) {
798 break;
799 }
800
801 health_code_update();
802 /* Send stream to relayd if the stream has an ID. */
803 if (msg.u.sent_streams.net_seq_idx != (uint64_t) -1ULL) {
804 ret = consumer_send_relayd_streams_sent(
805 msg.u.sent_streams.net_seq_idx);
806 if (ret < 0) {
807 goto end_nosignal;
808 }
809 }
810 break;
811 }
3bd1e081
MD
812 case LTTNG_CONSUMER_UPDATE_STREAM:
813 {
3f8e211f
DG
814 rcu_read_unlock();
815 return -ENOSYS;
816 }
817 case LTTNG_CONSUMER_DESTROY_RELAYD:
818 {
a6ba4fe1 819 uint64_t index = msg.u.destroy_relayd.net_seq_idx;
3f8e211f
DG
820 struct consumer_relayd_sock_pair *relayd;
821
a6ba4fe1 822 DBG("Kernel consumer destroying relayd %" PRIu64, index);
3f8e211f
DG
823
824 /* Get relayd reference if exists. */
a6ba4fe1 825 relayd = consumer_find_relayd(index);
3f8e211f 826 if (relayd == NULL) {
3448e266 827 DBG("Unable to find relayd %" PRIu64, index);
f50f23d9 828 ret_code = LTTNG_ERR_NO_CONSUMER;
3bd1e081 829 }
3f8e211f 830
a6ba4fe1
DG
831 /*
832 * Each relayd socket pair has a refcount of stream attached to it
833 * which tells if the relayd is still active or not depending on the
834 * refcount value.
835 *
836 * This will set the destroy flag of the relayd object and destroy it
837 * if the refcount reaches zero when called.
838 *
839 * The destroy can happen either here or when a stream fd hangs up.
840 */
f50f23d9
DG
841 if (relayd) {
842 consumer_flag_relayd_for_destroy(relayd);
843 }
844
9ce5646a
MD
845 health_code_update();
846
f50f23d9
DG
847 ret = consumer_send_status_msg(sock, ret_code);
848 if (ret < 0) {
849 /* Somehow, the session daemon is not responding anymore. */
1803a064 850 goto error_fatal;
f50f23d9 851 }
3f8e211f 852
3f8e211f 853 goto end_nosignal;
3bd1e081 854 }
6d805429 855 case LTTNG_CONSUMER_DATA_PENDING:
53632229 856 {
c8f59ee5 857 int32_t ret;
6d805429 858 uint64_t id = msg.u.data_pending.session_id;
c8f59ee5 859
6d805429 860 DBG("Kernel consumer data pending command for id %" PRIu64, id);
c8f59ee5 861
6d805429 862 ret = consumer_data_pending(id);
c8f59ee5 863
9ce5646a
MD
864 health_code_update();
865
c8f59ee5
DG
866 /* Send back returned value to session daemon */
867 ret = lttcomm_send_unix_sock(sock, &ret, sizeof(ret));
868 if (ret < 0) {
6d805429 869 PERROR("send data pending ret code");
1803a064 870 goto error_fatal;
c8f59ee5 871 }
f50f23d9
DG
872
873 /*
874 * No need to send back a status message since the data pending
875 * returned value is the response.
876 */
c8f59ee5 877 break;
53632229 878 }
6dc3064a
DG
879 case LTTNG_CONSUMER_SNAPSHOT_CHANNEL:
880 {
07b86b52
JD
881 if (msg.u.snapshot_channel.metadata == 1) {
882 ret = lttng_kconsumer_snapshot_metadata(msg.u.snapshot_channel.key,
e2039c7a
JD
883 msg.u.snapshot_channel.pathname,
884 msg.u.snapshot_channel.relayd_id, ctx);
07b86b52
JD
885 if (ret < 0) {
886 ERR("Snapshot metadata failed");
887 ret_code = LTTNG_ERR_KERN_META_FAIL;
888 }
889 } else {
890 ret = lttng_kconsumer_snapshot_channel(msg.u.snapshot_channel.key,
891 msg.u.snapshot_channel.pathname,
5c786ded
JD
892 msg.u.snapshot_channel.relayd_id,
893 msg.u.snapshot_channel.max_stream_size,
894 ctx);
07b86b52
JD
895 if (ret < 0) {
896 ERR("Snapshot channel failed");
897 ret_code = LTTNG_ERR_KERN_CHAN_FAIL;
898 }
899 }
900
9ce5646a
MD
901 health_code_update();
902
6dc3064a
DG
903 ret = consumer_send_status_msg(sock, ret_code);
904 if (ret < 0) {
905 /* Somehow, the session daemon is not responding anymore. */
906 goto end_nosignal;
907 }
908 break;
909 }
07b86b52
JD
910 case LTTNG_CONSUMER_DESTROY_CHANNEL:
911 {
912 uint64_t key = msg.u.destroy_channel.key;
913 struct lttng_consumer_channel *channel;
914
915 channel = consumer_find_channel(key);
916 if (!channel) {
917 ERR("Kernel consumer destroy channel %" PRIu64 " not found", key);
918 ret_code = LTTNG_ERR_KERN_CHAN_NOT_FOUND;
919 }
920
9ce5646a
MD
921 health_code_update();
922
07b86b52
JD
923 ret = consumer_send_status_msg(sock, ret_code);
924 if (ret < 0) {
925 /* Somehow, the session daemon is not responding anymore. */
926 goto end_nosignal;
927 }
928
9ce5646a
MD
929 health_code_update();
930
07b86b52
JD
931 /*
932 * This command should ONLY be issued for channel with streams set in
933 * no monitor mode.
934 */
935 assert(!channel->monitor);
936
937 /*
938 * The refcount should ALWAYS be 0 in the case of a channel in no
939 * monitor mode.
940 */
941 assert(!uatomic_sub_return(&channel->refcount, 1));
942
943 consumer_del_channel(channel);
944
945 goto end_nosignal;
946 }
3bd1e081 947 default:
3f8e211f 948 goto end_nosignal;
3bd1e081 949 }
3f8e211f 950
3bd1e081 951end_nosignal:
b0b335c8 952 rcu_read_unlock();
4cbc1a04
DG
953
954 /*
955 * Return 1 to indicate success since the 0 value can be a socket
956 * shutdown during the recv() or send() call.
957 */
9ce5646a 958 health_code_update();
4cbc1a04 959 return 1;
1803a064
MD
960
961error_fatal:
962 rcu_read_unlock();
963 /* This will issue a consumer stop. */
964 return -1;
3bd1e081 965}
d41f73b7 966
309167d2
JD
967/*
968 * Populate index values of a kernel stream. Values are set in big endian order.
969 *
970 * Return 0 on success or else a negative value.
971 */
50adc264 972static int get_index_values(struct ctf_packet_index *index, int infd)
309167d2
JD
973{
974 int ret;
975
976 ret = kernctl_get_timestamp_begin(infd, &index->timestamp_begin);
977 if (ret < 0) {
978 PERROR("kernctl_get_timestamp_begin");
979 goto error;
980 }
981 index->timestamp_begin = htobe64(index->timestamp_begin);
982
983 ret = kernctl_get_timestamp_end(infd, &index->timestamp_end);
984 if (ret < 0) {
985 PERROR("kernctl_get_timestamp_end");
986 goto error;
987 }
988 index->timestamp_end = htobe64(index->timestamp_end);
989
990 ret = kernctl_get_events_discarded(infd, &index->events_discarded);
991 if (ret < 0) {
992 PERROR("kernctl_get_events_discarded");
993 goto error;
994 }
995 index->events_discarded = htobe64(index->events_discarded);
996
997 ret = kernctl_get_content_size(infd, &index->content_size);
998 if (ret < 0) {
999 PERROR("kernctl_get_content_size");
1000 goto error;
1001 }
1002 index->content_size = htobe64(index->content_size);
1003
1004 ret = kernctl_get_packet_size(infd, &index->packet_size);
1005 if (ret < 0) {
1006 PERROR("kernctl_get_packet_size");
1007 goto error;
1008 }
1009 index->packet_size = htobe64(index->packet_size);
1010
1011 ret = kernctl_get_stream_id(infd, &index->stream_id);
1012 if (ret < 0) {
1013 PERROR("kernctl_get_stream_id");
1014 goto error;
1015 }
1016 index->stream_id = htobe64(index->stream_id);
1017
1018error:
1019 return ret;
1020}
94d49140
JD
1021/*
1022 * Sync metadata meaning request them to the session daemon and snapshot to the
1023 * metadata thread can consumer them.
1024 *
1025 * Metadata stream lock MUST be acquired.
1026 *
1027 * Return 0 if new metadatda is available, EAGAIN if the metadata stream
1028 * is empty or a negative value on error.
1029 */
1030int lttng_kconsumer_sync_metadata(struct lttng_consumer_stream *metadata)
1031{
1032 int ret;
1033
1034 assert(metadata);
1035
1036 ret = kernctl_buffer_flush(metadata->wait_fd);
1037 if (ret < 0) {
1038 ERR("Failed to flush kernel stream");
1039 goto end;
1040 }
1041
1042 ret = kernctl_snapshot(metadata->wait_fd);
1043 if (ret < 0) {
1044 if (errno != EAGAIN) {
1045 ERR("Sync metadata, taking kernel snapshot failed.");
1046 goto end;
1047 }
1048 DBG("Sync metadata, no new kernel metadata");
1049 /* No new metadata, exit. */
1050 ret = ENODATA;
1051 goto end;
1052 }
1053
1054end:
1055 return ret;
1056}
309167d2 1057
d41f73b7
MD
1058/*
1059 * Consume data on a file descriptor and write it on a trace file.
1060 */
4078b776 1061ssize_t lttng_kconsumer_read_subbuffer(struct lttng_consumer_stream *stream,
d41f73b7
MD
1062 struct lttng_consumer_local_data *ctx)
1063{
1d4dfdef 1064 unsigned long len, subbuf_size, padding;
1c20f0e2 1065 int err, write_index = 1;
4078b776 1066 ssize_t ret = 0;
d41f73b7 1067 int infd = stream->wait_fd;
50adc264 1068 struct ctf_packet_index index;
d41f73b7
MD
1069
1070 DBG("In read_subbuffer (infd : %d)", infd);
309167d2 1071
d41f73b7
MD
1072 /* Get the next subbuffer */
1073 err = kernctl_get_next_subbuf(infd);
1074 if (err != 0) {
d41f73b7
MD
1075 /*
1076 * This is a debug message even for single-threaded consumer,
1077 * because poll() have more relaxed criterions than get subbuf,
1078 * so get_subbuf may fail for short race windows where poll()
1079 * would issue wakeups.
1080 */
1081 DBG("Reserving sub buffer failed (everything is normal, "
1082 "it is due to concurrency)");
56591bac 1083 ret = -errno;
d41f73b7
MD
1084 goto end;
1085 }
1086
1d4dfdef
DG
1087 /* Get the full subbuffer size including padding */
1088 err = kernctl_get_padded_subbuf_size(infd, &len);
1089 if (err != 0) {
1d4dfdef 1090 perror("Getting sub-buffer len failed.");
56591bac 1091 ret = -errno;
1d4dfdef
DG
1092 goto end;
1093 }
1094
1c20f0e2 1095 if (!stream->metadata_flag) {
309167d2
JD
1096 ret = get_index_values(&index, infd);
1097 if (ret < 0) {
1098 goto end;
1099 }
1c20f0e2
JD
1100 } else {
1101 write_index = 0;
309167d2
JD
1102 }
1103
ffe60014 1104 switch (stream->chan->output) {
07b86b52 1105 case CONSUMER_CHANNEL_SPLICE:
1d4dfdef
DG
1106 /*
1107 * XXX: The lttng-modules splice "actor" does not handle copying
1108 * partial pages hence only using the subbuffer size without the
1109 * padding makes the splice fail.
1110 */
1111 subbuf_size = len;
1112 padding = 0;
1113
1114 /* splice the subbuffer to the tracefile */
91dfef6e 1115 ret = lttng_consumer_on_read_subbuffer_splice(ctx, stream, subbuf_size,
309167d2 1116 padding, &index);
91dfef6e
DG
1117 /*
1118 * XXX: Splice does not support network streaming so the return value
1119 * is simply checked against subbuf_size and not like the mmap() op.
1120 */
1d4dfdef
DG
1121 if (ret != subbuf_size) {
1122 /*
1123 * display the error but continue processing to try
1124 * to release the subbuffer
1125 */
1126 ERR("Error splicing to tracefile (ret: %zd != len: %lu)",
1127 ret, subbuf_size);
309167d2 1128 write_index = 0;
1d4dfdef
DG
1129 }
1130 break;
07b86b52 1131 case CONSUMER_CHANNEL_MMAP:
1d4dfdef
DG
1132 /* Get subbuffer size without padding */
1133 err = kernctl_get_subbuf_size(infd, &subbuf_size);
1134 if (err != 0) {
1d4dfdef 1135 perror("Getting sub-buffer len failed.");
56591bac 1136 ret = -errno;
1d4dfdef
DG
1137 goto end;
1138 }
47e81c02 1139
1d4dfdef
DG
1140 /* Make sure the tracer is not gone mad on us! */
1141 assert(len >= subbuf_size);
1142
1143 padding = len - subbuf_size;
1144
1145 /* write the subbuffer to the tracefile */
91dfef6e 1146 ret = lttng_consumer_on_read_subbuffer_mmap(ctx, stream, subbuf_size,
309167d2 1147 padding, &index);
91dfef6e
DG
1148 /*
1149 * The mmap operation should write subbuf_size amount of data when
1150 * network streaming or the full padding (len) size when we are _not_
1151 * streaming.
1152 */
d88aee68
DG
1153 if ((ret != subbuf_size && stream->net_seq_idx != (uint64_t) -1ULL) ||
1154 (ret != len && stream->net_seq_idx == (uint64_t) -1ULL)) {
1d4dfdef 1155 /*
91dfef6e 1156 * Display the error but continue processing to try to release the
2336629e
DG
1157 * subbuffer. This is a DBG statement since this is possible to
1158 * happen without being a critical error.
1d4dfdef 1159 */
2336629e 1160 DBG("Error writing to tracefile "
91dfef6e
DG
1161 "(ret: %zd != len: %lu != subbuf_size: %lu)",
1162 ret, len, subbuf_size);
309167d2 1163 write_index = 0;
1d4dfdef
DG
1164 }
1165 break;
1166 default:
1167 ERR("Unknown output method");
56591bac 1168 ret = -EPERM;
d41f73b7
MD
1169 }
1170
1171 err = kernctl_put_next_subbuf(infd);
1172 if (err != 0) {
d41f73b7
MD
1173 if (errno == EFAULT) {
1174 perror("Error in unreserving sub buffer\n");
1175 } else if (errno == EIO) {
1176 /* Should never happen with newer LTTng versions */
1177 perror("Reader has been pushed by the writer, last sub-buffer corrupted.");
1178 }
56591bac 1179 ret = -errno;
d41f73b7
MD
1180 goto end;
1181 }
1182
309167d2 1183 /* Write index if needed. */
1c20f0e2
JD
1184 if (!write_index) {
1185 goto end;
1186 }
1187
94d49140
JD
1188 if (stream->chan->live_timer_interval && !stream->metadata_flag) {
1189 /*
1190 * In live, block until all the metadata is sent.
1191 */
1192 err = consumer_stream_sync_metadata(ctx, stream->session_id);
1193 if (err < 0) {
1194 goto end;
1195 }
1196 }
1197
1c20f0e2
JD
1198 err = consumer_stream_write_index(stream, &index);
1199 if (err < 0) {
1200 goto end;
309167d2
JD
1201 }
1202
d41f73b7
MD
1203end:
1204 return ret;
1205}
1206
1207int lttng_kconsumer_on_recv_stream(struct lttng_consumer_stream *stream)
1208{
1209 int ret;
ffe60014
DG
1210
1211 assert(stream);
1212
2bba9e53
DG
1213 /*
1214 * Don't create anything if this is set for streaming or should not be
1215 * monitored.
1216 */
1217 if (stream->net_seq_idx == (uint64_t) -1ULL && stream->chan->monitor) {
fe4477ee
JD
1218 ret = utils_create_stream_file(stream->chan->pathname, stream->name,
1219 stream->chan->tracefile_size, stream->tracefile_count_current,
309167d2 1220 stream->uid, stream->gid, NULL);
fe4477ee
JD
1221 if (ret < 0) {
1222 goto error;
1223 }
1224 stream->out_fd = ret;
1225 stream->tracefile_size_current = 0;
309167d2
JD
1226
1227 if (!stream->metadata_flag) {
1228 ret = index_create_file(stream->chan->pathname,
1229 stream->name, stream->uid, stream->gid,
1230 stream->chan->tracefile_size,
1231 stream->tracefile_count_current);
1232 if (ret < 0) {
1233 goto error;
1234 }
1235 stream->index_fd = ret;
1236 }
ffe60014 1237 }
d41f73b7 1238
d41f73b7
MD
1239 if (stream->output == LTTNG_EVENT_MMAP) {
1240 /* get the len of the mmap region */
1241 unsigned long mmap_len;
1242
1243 ret = kernctl_get_mmap_len(stream->wait_fd, &mmap_len);
1244 if (ret != 0) {
ffe60014 1245 PERROR("kernctl_get_mmap_len");
56591bac 1246 ret = -errno;
d41f73b7
MD
1247 goto error_close_fd;
1248 }
1249 stream->mmap_len = (size_t) mmap_len;
1250
ffe60014
DG
1251 stream->mmap_base = mmap(NULL, stream->mmap_len, PROT_READ,
1252 MAP_PRIVATE, stream->wait_fd, 0);
d41f73b7 1253 if (stream->mmap_base == MAP_FAILED) {
ffe60014 1254 PERROR("Error mmaping");
d41f73b7
MD
1255 ret = -1;
1256 goto error_close_fd;
1257 }
1258 }
1259
1260 /* we return 0 to let the library handle the FD internally */
1261 return 0;
1262
1263error_close_fd:
2f225ce2 1264 if (stream->out_fd >= 0) {
d41f73b7
MD
1265 int err;
1266
1267 err = close(stream->out_fd);
1268 assert(!err);
2f225ce2 1269 stream->out_fd = -1;
d41f73b7
MD
1270 }
1271error:
1272 return ret;
1273}
1274
ca22feea
DG
1275/*
1276 * Check if data is still being extracted from the buffers for a specific
4e9a4686
DG
1277 * stream. Consumer data lock MUST be acquired before calling this function
1278 * and the stream lock.
ca22feea 1279 *
6d805429 1280 * Return 1 if the traced data are still getting read else 0 meaning that the
ca22feea
DG
1281 * data is available for trace viewer reading.
1282 */
6d805429 1283int lttng_kconsumer_data_pending(struct lttng_consumer_stream *stream)
ca22feea
DG
1284{
1285 int ret;
1286
1287 assert(stream);
1288
873b9e9a
MD
1289 if (stream->endpoint_status != CONSUMER_ENDPOINT_ACTIVE) {
1290 ret = 0;
1291 goto end;
1292 }
1293
ca22feea
DG
1294 ret = kernctl_get_next_subbuf(stream->wait_fd);
1295 if (ret == 0) {
1296 /* There is still data so let's put back this subbuffer. */
1297 ret = kernctl_put_subbuf(stream->wait_fd);
1298 assert(ret == 0);
6d805429 1299 ret = 1; /* Data is pending */
4e9a4686 1300 goto end;
ca22feea
DG
1301 }
1302
6d805429
DG
1303 /* Data is NOT pending and ready to be read. */
1304 ret = 0;
ca22feea 1305
6efae65e
DG
1306end:
1307 return ret;
ca22feea 1308}
This page took 0.121594 seconds and 5 git commands to generate.