Support flight recorder mode for a session
[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
990570ed 32#include <common/common.h>
10a8a223 33#include <common/kernel-ctl/kernel-ctl.h>
10a8a223 34#include <common/sessiond-comm/sessiond-comm.h>
00e2e675 35#include <common/sessiond-comm/relayd.h>
dbb5dfe6 36#include <common/compat/fcntl.h>
acdb9057 37#include <common/pipe.h>
00e2e675 38#include <common/relayd/relayd.h>
fe4477ee 39#include <common/utils.h>
0857097f 40
10a8a223 41#include "kernel-consumer.h"
3bd1e081
MD
42
43extern struct lttng_consumer_global_data consumer_data;
44extern int consumer_poll_timeout;
45extern volatile int consumer_quit;
46
3bd1e081
MD
47/*
48 * Take a snapshot for a specific fd
49 *
50 * Returns 0 on success, < 0 on error
51 */
ffe60014 52int lttng_kconsumer_take_snapshot(struct lttng_consumer_stream *stream)
3bd1e081
MD
53{
54 int ret = 0;
55 int infd = stream->wait_fd;
56
57 ret = kernctl_snapshot(infd);
58 if (ret != 0) {
87dc6a9c 59 errno = -ret;
3bd1e081
MD
60 perror("Getting sub-buffer snapshot.");
61 }
62
63 return ret;
64}
65
66/*
67 * Get the produced position
68 *
69 * Returns 0 on success, < 0 on error
70 */
ffe60014 71int lttng_kconsumer_get_produced_snapshot(struct lttng_consumer_stream *stream,
3bd1e081
MD
72 unsigned long *pos)
73{
74 int ret;
75 int infd = stream->wait_fd;
76
77 ret = kernctl_snapshot_get_produced(infd, pos);
78 if (ret != 0) {
87dc6a9c 79 errno = -ret;
3bd1e081
MD
80 perror("kernctl_snapshot_get_produced");
81 }
82
83 return ret;
84}
85
1803a064
MD
86/*
87 * Receive command from session daemon and process it.
88 *
89 * Return 1 on success else a negative value or 0.
90 */
3bd1e081
MD
91int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
92 int sock, struct pollfd *consumer_sockpoll)
93{
94 ssize_t ret;
f50f23d9 95 enum lttng_error_code ret_code = LTTNG_OK;
3bd1e081
MD
96 struct lttcomm_consumer_msg msg;
97
98 ret = lttcomm_recv_unix_sock(sock, &msg, sizeof(msg));
99 if (ret != sizeof(msg)) {
f73fabfd 100 lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_ERROR_RECV_CMD);
1803a064
MD
101 if (ret > 0) {
102 ret = -1;
103 }
3bd1e081
MD
104 return ret;
105 }
106 if (msg.cmd_type == LTTNG_CONSUMER_STOP) {
f50f23d9
DG
107 /*
108 * Notify the session daemon that the command is completed.
109 *
110 * On transport layer error, the function call will print an error
111 * message so handling the returned code is a bit useless since we
112 * return an error code anyway.
113 */
114 (void) consumer_send_status_msg(sock, ret_code);
3bd1e081
MD
115 return -ENOENT;
116 }
117
b0b335c8
MD
118 /* relayd needs RCU read-side protection */
119 rcu_read_lock();
120
3bd1e081 121 switch (msg.cmd_type) {
00e2e675
DG
122 case LTTNG_CONSUMER_ADD_RELAYD_SOCKET:
123 {
f50f23d9 124 /* Session daemon status message are handled in the following call. */
7735ef9e
DG
125 ret = consumer_add_relayd_socket(msg.u.relayd_sock.net_index,
126 msg.u.relayd_sock.type, ctx, sock, consumer_sockpoll,
46e6455f 127 &msg.u.relayd_sock.sock, msg.u.relayd_sock.session_id);
00e2e675
DG
128 goto end_nosignal;
129 }
3bd1e081
MD
130 case LTTNG_CONSUMER_ADD_CHANNEL:
131 {
132 struct lttng_consumer_channel *new_channel;
e43c41c5 133 int ret_recv;
3bd1e081 134
f50f23d9
DG
135 /* First send a status message before receiving the fds. */
136 ret = consumer_send_status_msg(sock, ret_code);
137 if (ret < 0) {
138 /* Somehow, the session daemon is not responding anymore. */
1803a064 139 goto error_fatal;
f50f23d9 140 }
d88aee68 141 DBG("consumer_add_channel %" PRIu64, msg.u.channel.channel_key);
3bd1e081 142 new_channel = consumer_allocate_channel(msg.u.channel.channel_key,
ffe60014
DG
143 msg.u.channel.session_id, msg.u.channel.pathname,
144 msg.u.channel.name, msg.u.channel.uid, msg.u.channel.gid,
1624d5b7
JD
145 msg.u.channel.relayd_id, msg.u.channel.output,
146 msg.u.channel.tracefile_size,
2bba9e53
DG
147 msg.u.channel.tracefile_count,
148 msg.u.channel.monitor);
3bd1e081 149 if (new_channel == NULL) {
f73fabfd 150 lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_OUTFD_ERROR);
3bd1e081
MD
151 goto end_nosignal;
152 }
ffe60014
DG
153 new_channel->nb_init_stream_left = msg.u.channel.nb_init_streams;
154
155 /* Translate and save channel type. */
156 switch (msg.u.channel.type) {
157 case CONSUMER_CHANNEL_TYPE_DATA:
158 case CONSUMER_CHANNEL_TYPE_METADATA:
159 new_channel->type = msg.u.channel.type;
160 break;
161 default:
162 assert(0);
163 goto end_nosignal;
164 };
165
3bd1e081 166 if (ctx->on_recv_channel != NULL) {
e43c41c5
JD
167 ret_recv = ctx->on_recv_channel(new_channel);
168 if (ret_recv == 0) {
169 ret = consumer_add_channel(new_channel, ctx);
170 } else if (ret_recv < 0) {
3bd1e081
MD
171 goto end_nosignal;
172 }
173 } else {
e43c41c5 174 ret = consumer_add_channel(new_channel, ctx);
3bd1e081 175 }
e43c41c5
JD
176
177 /* If we received an error in add_channel, we need to report it. */
821fffb2 178 if (ret < 0) {
1803a064
MD
179 ret = consumer_send_status_msg(sock, ret);
180 if (ret < 0) {
181 goto error_fatal;
182 }
e43c41c5
JD
183 goto end_nosignal;
184 }
185
3bd1e081
MD
186 goto end_nosignal;
187 }
188 case LTTNG_CONSUMER_ADD_STREAM:
189 {
dae10966
DG
190 int fd;
191 struct lttng_pipe *stream_pipe;
00e2e675
DG
192 struct consumer_relayd_sock_pair *relayd = NULL;
193 struct lttng_consumer_stream *new_stream;
ffe60014 194 struct lttng_consumer_channel *channel;
c80048c6 195 int alloc_ret = 0;
3bd1e081 196
ffe60014
DG
197 /*
198 * Get stream's channel reference. Needed when adding the stream to the
199 * global hash table.
200 */
201 channel = consumer_find_channel(msg.u.stream.channel_key);
202 if (!channel) {
203 /*
204 * We could not find the channel. Can happen if cpu hotplug
205 * happens while tearing down.
206 */
d88aee68 207 ERR("Unable to find channel key %" PRIu64, msg.u.stream.channel_key);
ffe60014
DG
208 ret_code = LTTNG_ERR_KERN_CHAN_NOT_FOUND;
209 }
210
f50f23d9
DG
211 /* First send a status message before receiving the fds. */
212 ret = consumer_send_status_msg(sock, ret_code);
1803a064
MD
213 if (ret < 0) {
214 /*
215 * Somehow, the session daemon is not responding
216 * anymore.
217 */
218 goto error_fatal;
219 }
220 if (ret_code != LTTNG_OK) {
ffe60014 221 /*
1803a064 222 * Channel was not found.
ffe60014 223 */
f50f23d9
DG
224 goto end_nosignal;
225 }
226
3bd1e081
MD
227 /* block */
228 if (lttng_consumer_poll_socket(consumer_sockpoll) < 0) {
3f8e211f 229 rcu_read_unlock();
3bd1e081
MD
230 return -EINTR;
231 }
00e2e675
DG
232
233 /* Get stream file descriptor from socket */
f2fc6720
MD
234 ret = lttcomm_recv_fds_unix_sock(sock, &fd, 1);
235 if (ret != sizeof(fd)) {
f73fabfd 236 lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_ERROR_RECV_FD);
3f8e211f 237 rcu_read_unlock();
3bd1e081
MD
238 return ret;
239 }
3bd1e081 240
f50f23d9
DG
241 /*
242 * Send status code to session daemon only if the recv works. If the
243 * above recv() failed, the session daemon is notified through the
244 * error socket and the teardown is eventually done.
245 */
246 ret = consumer_send_status_msg(sock, ret_code);
247 if (ret < 0) {
248 /* Somehow, the session daemon is not responding anymore. */
249 goto end_nosignal;
250 }
251
ffe60014
DG
252 new_stream = consumer_allocate_stream(channel->key,
253 fd,
254 LTTNG_CONSUMER_ACTIVE_STREAM,
255 channel->name,
256 channel->uid,
257 channel->gid,
258 channel->relayd_id,
259 channel->session_id,
260 msg.u.stream.cpu,
261 &alloc_ret,
262 channel->type);
3bd1e081 263 if (new_stream == NULL) {
c80048c6
MD
264 switch (alloc_ret) {
265 case -ENOMEM:
266 case -EINVAL:
267 default:
268 lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_OUTFD_ERROR);
269 break;
c80048c6 270 }
3f8e211f 271 goto end_nosignal;
3bd1e081 272 }
ffe60014
DG
273 new_stream->chan = channel;
274 new_stream->wait_fd = fd;
00e2e675 275
a0c83db9
DG
276 /*
277 * We've just assigned the channel to the stream so increment the
278 * refcount right now.
279 */
280 uatomic_inc(&new_stream->chan->refcount);
9d9353f9 281
fb3a43a9
DG
282 /*
283 * The buffer flush is done on the session daemon side for the kernel
284 * so no need for the stream "hangup_flush_done" variable to be
285 * tracked. This is important for a kernel stream since we don't rely
286 * on the flush state of the stream to read data. It's not the case for
287 * user space tracing.
288 */
289 new_stream->hangup_flush_done = 0;
290
00e2e675 291 /* The stream is not metadata. Get relayd reference if exists. */
ffe60014 292 relayd = consumer_find_relayd(new_stream->net_seq_idx);
00e2e675
DG
293 if (relayd != NULL) {
294 /* Add stream on the relayd */
295 pthread_mutex_lock(&relayd->ctrl_sock_mutex);
296 ret = relayd_add_stream(&relayd->control_sock,
ffe60014 297 new_stream->name, new_stream->chan->pathname,
0f907de1
JD
298 &new_stream->relayd_stream_id,
299 new_stream->chan->tracefile_size,
300 new_stream->chan->tracefile_count);
00e2e675
DG
301 pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
302 if (ret < 0) {
e316aad5 303 consumer_del_stream(new_stream, NULL);
3f8e211f 304 goto end_nosignal;
00e2e675 305 }
d88aee68
DG
306 } else if (new_stream->net_seq_idx != (uint64_t) -1ULL) {
307 ERR("Network sequence index %" PRIu64 " unknown. Not adding stream.",
ffe60014 308 new_stream->net_seq_idx);
e316aad5 309 consumer_del_stream(new_stream, NULL);
3f8e211f 310 goto end_nosignal;
00e2e675
DG
311 }
312
633d0084
DG
313 if (ctx->on_recv_stream) {
314 ret = ctx->on_recv_stream(new_stream);
315 if (ret < 0) {
e316aad5 316 consumer_del_stream(new_stream, NULL);
633d0084 317 goto end_nosignal;
fb3a43a9 318 }
633d0084 319 }
fb3a43a9 320
2bba9e53
DG
321 /* Do not monitor this stream. */
322 if (!channel->monitor) {
6dc3064a
DG
323 DBG("Kernel consumer add stream %s in no monitor mode with"
324 "relayd id %" PRIu64, new_stream->name,
325 new_stream->relayd_stream_id);
326 break;
327 }
328
50f8ae69 329 /* Get the right pipe where the stream will be sent. */
633d0084 330 if (new_stream->metadata_flag) {
dae10966 331 stream_pipe = ctx->consumer_metadata_pipe;
3bd1e081 332 } else {
dae10966 333 stream_pipe = ctx->consumer_data_pipe;
50f8ae69
DG
334 }
335
dae10966 336 ret = lttng_pipe_write(stream_pipe, &new_stream, sizeof(new_stream));
50f8ae69 337 if (ret < 0) {
dae10966 338 ERR("Consumer write %s stream to pipe %d",
50f8ae69 339 new_stream->metadata_flag ? "metadata" : "data",
dae10966 340 lttng_pipe_get_writefd(stream_pipe));
50f8ae69
DG
341 consumer_del_stream(new_stream, NULL);
342 goto end_nosignal;
3bd1e081 343 }
00e2e675 344
50f8ae69 345 DBG("Kernel consumer ADD_STREAM %s (fd: %d) with relayd id %" PRIu64,
ffe60014 346 new_stream->name, fd, new_stream->relayd_stream_id);
3bd1e081
MD
347 break;
348 }
349 case LTTNG_CONSUMER_UPDATE_STREAM:
350 {
3f8e211f
DG
351 rcu_read_unlock();
352 return -ENOSYS;
353 }
354 case LTTNG_CONSUMER_DESTROY_RELAYD:
355 {
a6ba4fe1 356 uint64_t index = msg.u.destroy_relayd.net_seq_idx;
3f8e211f
DG
357 struct consumer_relayd_sock_pair *relayd;
358
a6ba4fe1 359 DBG("Kernel consumer destroying relayd %" PRIu64, index);
3f8e211f
DG
360
361 /* Get relayd reference if exists. */
a6ba4fe1 362 relayd = consumer_find_relayd(index);
3f8e211f 363 if (relayd == NULL) {
3448e266 364 DBG("Unable to find relayd %" PRIu64, index);
f50f23d9 365 ret_code = LTTNG_ERR_NO_CONSUMER;
3bd1e081 366 }
3f8e211f 367
a6ba4fe1
DG
368 /*
369 * Each relayd socket pair has a refcount of stream attached to it
370 * which tells if the relayd is still active or not depending on the
371 * refcount value.
372 *
373 * This will set the destroy flag of the relayd object and destroy it
374 * if the refcount reaches zero when called.
375 *
376 * The destroy can happen either here or when a stream fd hangs up.
377 */
f50f23d9
DG
378 if (relayd) {
379 consumer_flag_relayd_for_destroy(relayd);
380 }
381
382 ret = consumer_send_status_msg(sock, ret_code);
383 if (ret < 0) {
384 /* Somehow, the session daemon is not responding anymore. */
1803a064 385 goto error_fatal;
f50f23d9 386 }
3f8e211f 387
3f8e211f 388 goto end_nosignal;
3bd1e081 389 }
6d805429 390 case LTTNG_CONSUMER_DATA_PENDING:
53632229 391 {
c8f59ee5 392 int32_t ret;
6d805429 393 uint64_t id = msg.u.data_pending.session_id;
c8f59ee5 394
6d805429 395 DBG("Kernel consumer data pending command for id %" PRIu64, id);
c8f59ee5 396
6d805429 397 ret = consumer_data_pending(id);
c8f59ee5
DG
398
399 /* Send back returned value to session daemon */
400 ret = lttcomm_send_unix_sock(sock, &ret, sizeof(ret));
401 if (ret < 0) {
6d805429 402 PERROR("send data pending ret code");
1803a064 403 goto error_fatal;
c8f59ee5 404 }
f50f23d9
DG
405
406 /*
407 * No need to send back a status message since the data pending
408 * returned value is the response.
409 */
c8f59ee5 410 break;
53632229 411 }
6dc3064a
DG
412 case LTTNG_CONSUMER_SNAPSHOT_CHANNEL:
413 {
414 ret = consumer_send_status_msg(sock, ret_code);
415 if (ret < 0) {
416 /* Somehow, the session daemon is not responding anymore. */
417 goto end_nosignal;
418 }
419 break;
420 }
3bd1e081 421 default:
3f8e211f 422 goto end_nosignal;
3bd1e081 423 }
3f8e211f 424
3bd1e081 425end_nosignal:
b0b335c8 426 rcu_read_unlock();
4cbc1a04
DG
427
428 /*
429 * Return 1 to indicate success since the 0 value can be a socket
430 * shutdown during the recv() or send() call.
431 */
432 return 1;
1803a064
MD
433
434error_fatal:
435 rcu_read_unlock();
436 /* This will issue a consumer stop. */
437 return -1;
3bd1e081 438}
d41f73b7
MD
439
440/*
441 * Consume data on a file descriptor and write it on a trace file.
442 */
4078b776 443ssize_t lttng_kconsumer_read_subbuffer(struct lttng_consumer_stream *stream,
d41f73b7
MD
444 struct lttng_consumer_local_data *ctx)
445{
1d4dfdef 446 unsigned long len, subbuf_size, padding;
d41f73b7 447 int err;
4078b776 448 ssize_t ret = 0;
d41f73b7
MD
449 int infd = stream->wait_fd;
450
451 DBG("In read_subbuffer (infd : %d)", infd);
452 /* Get the next subbuffer */
453 err = kernctl_get_next_subbuf(infd);
454 if (err != 0) {
1d4dfdef 455 ret = err;
d41f73b7
MD
456 /*
457 * This is a debug message even for single-threaded consumer,
458 * because poll() have more relaxed criterions than get subbuf,
459 * so get_subbuf may fail for short race windows where poll()
460 * would issue wakeups.
461 */
462 DBG("Reserving sub buffer failed (everything is normal, "
463 "it is due to concurrency)");
464 goto end;
465 }
466
1d4dfdef
DG
467 /* Get the full subbuffer size including padding */
468 err = kernctl_get_padded_subbuf_size(infd, &len);
469 if (err != 0) {
470 errno = -err;
471 perror("Getting sub-buffer len failed.");
472 ret = err;
473 goto end;
474 }
475
ffe60014 476 switch (stream->chan->output) {
1d4dfdef 477 case LTTNG_EVENT_SPLICE:
d41f73b7 478
1d4dfdef
DG
479 /*
480 * XXX: The lttng-modules splice "actor" does not handle copying
481 * partial pages hence only using the subbuffer size without the
482 * padding makes the splice fail.
483 */
484 subbuf_size = len;
485 padding = 0;
486
487 /* splice the subbuffer to the tracefile */
91dfef6e
DG
488 ret = lttng_consumer_on_read_subbuffer_splice(ctx, stream, subbuf_size,
489 padding);
490 /*
491 * XXX: Splice does not support network streaming so the return value
492 * is simply checked against subbuf_size and not like the mmap() op.
493 */
1d4dfdef
DG
494 if (ret != subbuf_size) {
495 /*
496 * display the error but continue processing to try
497 * to release the subbuffer
498 */
499 ERR("Error splicing to tracefile (ret: %zd != len: %lu)",
500 ret, subbuf_size);
501 }
502 break;
503 case LTTNG_EVENT_MMAP:
504 /* Get subbuffer size without padding */
505 err = kernctl_get_subbuf_size(infd, &subbuf_size);
506 if (err != 0) {
507 errno = -err;
508 perror("Getting sub-buffer len failed.");
509 ret = err;
510 goto end;
511 }
47e81c02 512
1d4dfdef
DG
513 /* Make sure the tracer is not gone mad on us! */
514 assert(len >= subbuf_size);
515
516 padding = len - subbuf_size;
517
518 /* write the subbuffer to the tracefile */
91dfef6e
DG
519 ret = lttng_consumer_on_read_subbuffer_mmap(ctx, stream, subbuf_size,
520 padding);
521 /*
522 * The mmap operation should write subbuf_size amount of data when
523 * network streaming or the full padding (len) size when we are _not_
524 * streaming.
525 */
d88aee68
DG
526 if ((ret != subbuf_size && stream->net_seq_idx != (uint64_t) -1ULL) ||
527 (ret != len && stream->net_seq_idx == (uint64_t) -1ULL)) {
1d4dfdef 528 /*
91dfef6e
DG
529 * Display the error but continue processing to try to release the
530 * subbuffer
1d4dfdef 531 */
91dfef6e
DG
532 ERR("Error writing to tracefile "
533 "(ret: %zd != len: %lu != subbuf_size: %lu)",
534 ret, len, subbuf_size);
1d4dfdef
DG
535 }
536 break;
537 default:
538 ERR("Unknown output method");
539 ret = -1;
d41f73b7
MD
540 }
541
542 err = kernctl_put_next_subbuf(infd);
543 if (err != 0) {
21073eaa 544 errno = -err;
d41f73b7
MD
545 if (errno == EFAULT) {
546 perror("Error in unreserving sub buffer\n");
547 } else if (errno == EIO) {
548 /* Should never happen with newer LTTng versions */
549 perror("Reader has been pushed by the writer, last sub-buffer corrupted.");
550 }
21073eaa
DG
551
552 ret = -err;
d41f73b7
MD
553 goto end;
554 }
555
556end:
557 return ret;
558}
559
560int lttng_kconsumer_on_recv_stream(struct lttng_consumer_stream *stream)
561{
562 int ret;
ffe60014
DG
563
564 assert(stream);
565
2bba9e53
DG
566 /*
567 * Don't create anything if this is set for streaming or should not be
568 * monitored.
569 */
570 if (stream->net_seq_idx == (uint64_t) -1ULL && stream->chan->monitor) {
fe4477ee
JD
571 ret = utils_create_stream_file(stream->chan->pathname, stream->name,
572 stream->chan->tracefile_size, stream->tracefile_count_current,
573 stream->uid, stream->gid);
574 if (ret < 0) {
575 goto error;
576 }
577 stream->out_fd = ret;
578 stream->tracefile_size_current = 0;
ffe60014 579 }
d41f73b7 580
d41f73b7
MD
581 if (stream->output == LTTNG_EVENT_MMAP) {
582 /* get the len of the mmap region */
583 unsigned long mmap_len;
584
585 ret = kernctl_get_mmap_len(stream->wait_fd, &mmap_len);
586 if (ret != 0) {
87dc6a9c 587 errno = -ret;
ffe60014 588 PERROR("kernctl_get_mmap_len");
d41f73b7
MD
589 goto error_close_fd;
590 }
591 stream->mmap_len = (size_t) mmap_len;
592
ffe60014
DG
593 stream->mmap_base = mmap(NULL, stream->mmap_len, PROT_READ,
594 MAP_PRIVATE, stream->wait_fd, 0);
d41f73b7 595 if (stream->mmap_base == MAP_FAILED) {
ffe60014 596 PERROR("Error mmaping");
d41f73b7
MD
597 ret = -1;
598 goto error_close_fd;
599 }
600 }
601
602 /* we return 0 to let the library handle the FD internally */
603 return 0;
604
605error_close_fd:
606 {
607 int err;
608
609 err = close(stream->out_fd);
610 assert(!err);
611 }
612error:
613 return ret;
614}
615
ca22feea
DG
616/*
617 * Check if data is still being extracted from the buffers for a specific
4e9a4686
DG
618 * stream. Consumer data lock MUST be acquired before calling this function
619 * and the stream lock.
ca22feea 620 *
6d805429 621 * Return 1 if the traced data are still getting read else 0 meaning that the
ca22feea
DG
622 * data is available for trace viewer reading.
623 */
6d805429 624int lttng_kconsumer_data_pending(struct lttng_consumer_stream *stream)
ca22feea
DG
625{
626 int ret;
627
628 assert(stream);
629
ca22feea
DG
630 ret = kernctl_get_next_subbuf(stream->wait_fd);
631 if (ret == 0) {
632 /* There is still data so let's put back this subbuffer. */
633 ret = kernctl_put_subbuf(stream->wait_fd);
634 assert(ret == 0);
6d805429 635 ret = 1; /* Data is pending */
4e9a4686 636 goto end;
ca22feea
DG
637 }
638
6d805429
DG
639 /* Data is NOT pending and ready to be read. */
640 ret = 0;
ca22feea 641
6efae65e
DG
642end:
643 return ret;
ca22feea 644}
This page took 0.108791 seconds and 4 git commands to generate.