Implement consumer ring buffer position sampling
[lttng-tools.git] / src / common / ust-consumer / ust-consumer.c
CommitLineData
3bd1e081
MD
1/*
2 * Copyright (C) 2011 - Julien Desfossez <julien.desfossez@polymtl.ca>
3 * Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
b3530820 4 * Copyright (C) 2017 - Jérémie Galarneau <jeremie.galarneau@efficios.com>
3bd1e081 5 *
d14d33bf
AM
6 * This program is free software; you can redistribute it and/or modify
7 * it under the terms of the GNU General Public License, version 2 only,
8 * as published by the Free Software Foundation.
3bd1e081
MD
9 *
10 * This program is distributed in the hope that it will be useful,
11 * but WITHOUT ANY WARRANTY; without even the implied warranty of
12 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
13 * GNU General Public License for more details.
14 *
d14d33bf
AM
15 * You should have received a copy of the GNU General Public License along
16 * with this program; if not, write to the Free Software Foundation, Inc.,
17 * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
3bd1e081
MD
18 */
19
6c1c0768 20#define _LGPL_SOURCE
3bd1e081 21#include <assert.h>
f02e1e8a 22#include <lttng/ust-ctl.h>
3bd1e081
MD
23#include <poll.h>
24#include <pthread.h>
25#include <stdlib.h>
26#include <string.h>
27#include <sys/mman.h>
28#include <sys/socket.h>
dbb5dfe6 29#include <sys/stat.h>
3bd1e081 30#include <sys/types.h>
77c7c900 31#include <inttypes.h>
3bd1e081 32#include <unistd.h>
ffe60014 33#include <urcu/list.h>
331744e3 34#include <signal.h>
0857097f 35
51a9e1c7 36#include <bin/lttng-consumerd/health-consumerd.h>
990570ed 37#include <common/common.h>
10a8a223 38#include <common/sessiond-comm/sessiond-comm.h>
00e2e675 39#include <common/relayd/relayd.h>
dbb5dfe6 40#include <common/compat/fcntl.h>
f263b7fd 41#include <common/compat/endian.h>
c8fea79c
JR
42#include <common/consumer/consumer-metadata-cache.h>
43#include <common/consumer/consumer-stream.h>
44#include <common/consumer/consumer-timer.h>
fe4477ee 45#include <common/utils.h>
309167d2 46#include <common/index/index.h>
10a8a223
DG
47
48#include "ust-consumer.h"
3bd1e081 49
45863397 50#define INT_MAX_STR_LEN 12 /* includes \0 */
4628484a 51
3bd1e081
MD
52extern struct lttng_consumer_global_data consumer_data;
53extern int consumer_poll_timeout;
54extern volatile int consumer_quit;
55
56/*
ffe60014
DG
57 * Free channel object and all streams associated with it. This MUST be used
58 * only and only if the channel has _NEVER_ been added to the global channel
59 * hash table.
3bd1e081 60 */
ffe60014 61static void destroy_channel(struct lttng_consumer_channel *channel)
3bd1e081 62{
ffe60014
DG
63 struct lttng_consumer_stream *stream, *stmp;
64
65 assert(channel);
66
67 DBG("UST consumer cleaning stream list");
68
69 cds_list_for_each_entry_safe(stream, stmp, &channel->streams.head,
70 send_node) {
9ce5646a
MD
71
72 health_code_update();
73
ffe60014
DG
74 cds_list_del(&stream->send_node);
75 ustctl_destroy_stream(stream->ustream);
76 free(stream);
77 }
78
79 /*
80 * If a channel is available meaning that was created before the streams
81 * were, delete it.
82 */
83 if (channel->uchan) {
84 lttng_ustconsumer_del_channel(channel);
b83e03c4 85 lttng_ustconsumer_free_channel(channel);
ffe60014
DG
86 }
87 free(channel);
88}
3bd1e081
MD
89
90/*
ffe60014 91 * Add channel to internal consumer state.
3bd1e081 92 *
ffe60014 93 * Returns 0 on success or else a negative value.
3bd1e081 94 */
ffe60014
DG
95static int add_channel(struct lttng_consumer_channel *channel,
96 struct lttng_consumer_local_data *ctx)
3bd1e081
MD
97{
98 int ret = 0;
99
ffe60014
DG
100 assert(channel);
101 assert(ctx);
102
103 if (ctx->on_recv_channel != NULL) {
104 ret = ctx->on_recv_channel(channel);
105 if (ret == 0) {
d8ef542d 106 ret = consumer_add_channel(channel, ctx);
ffe60014
DG
107 } else if (ret < 0) {
108 /* Most likely an ENOMEM. */
109 lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_OUTFD_ERROR);
110 goto error;
111 }
112 } else {
d8ef542d 113 ret = consumer_add_channel(channel, ctx);
3bd1e081
MD
114 }
115
d88aee68 116 DBG("UST consumer channel added (key: %" PRIu64 ")", channel->key);
ffe60014
DG
117
118error:
3bd1e081
MD
119 return ret;
120}
121
122/*
ffe60014
DG
123 * Allocate and return a consumer channel object.
124 */
125static struct lttng_consumer_channel *allocate_channel(uint64_t session_id,
126 const char *pathname, const char *name, uid_t uid, gid_t gid,
da009f2c 127 uint64_t relayd_id, uint64_t key, enum lttng_event_output output,
2bba9e53 128 uint64_t tracefile_size, uint64_t tracefile_count,
ecc48a90 129 uint64_t session_id_per_pid, unsigned int monitor,
d7ba1388 130 unsigned int live_timer_interval,
3d071855 131 const char *root_shm_path, const char *shm_path)
ffe60014
DG
132{
133 assert(pathname);
134 assert(name);
135
1950109e
JD
136 return consumer_allocate_channel(key, session_id, pathname, name, uid,
137 gid, relayd_id, output, tracefile_size,
d7ba1388 138 tracefile_count, session_id_per_pid, monitor,
3d071855 139 live_timer_interval, root_shm_path, shm_path);
ffe60014
DG
140}
141
142/*
143 * Allocate and return a consumer stream object. If _alloc_ret is not NULL, the
144 * error value if applicable is set in it else it is kept untouched.
3bd1e081 145 *
ffe60014 146 * Return NULL on error else the newly allocated stream object.
3bd1e081 147 */
ffe60014
DG
148static struct lttng_consumer_stream *allocate_stream(int cpu, int key,
149 struct lttng_consumer_channel *channel,
150 struct lttng_consumer_local_data *ctx, int *_alloc_ret)
151{
152 int alloc_ret;
153 struct lttng_consumer_stream *stream = NULL;
154
155 assert(channel);
156 assert(ctx);
157
158 stream = consumer_allocate_stream(channel->key,
159 key,
160 LTTNG_CONSUMER_ACTIVE_STREAM,
161 channel->name,
162 channel->uid,
163 channel->gid,
164 channel->relayd_id,
165 channel->session_id,
166 cpu,
167 &alloc_ret,
4891ece8
DG
168 channel->type,
169 channel->monitor);
ffe60014
DG
170 if (stream == NULL) {
171 switch (alloc_ret) {
172 case -ENOENT:
173 /*
174 * We could not find the channel. Can happen if cpu hotplug
175 * happens while tearing down.
176 */
177 DBG3("Could not find channel");
178 break;
179 case -ENOMEM:
180 case -EINVAL:
181 default:
182 lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_OUTFD_ERROR);
183 break;
184 }
185 goto error;
186 }
187
188 stream->chan = channel;
189
190error:
191 if (_alloc_ret) {
192 *_alloc_ret = alloc_ret;
193 }
194 return stream;
195}
196
197/*
198 * Send the given stream pointer to the corresponding thread.
199 *
200 * Returns 0 on success else a negative value.
201 */
202static int send_stream_to_thread(struct lttng_consumer_stream *stream,
203 struct lttng_consumer_local_data *ctx)
204{
dae10966
DG
205 int ret;
206 struct lttng_pipe *stream_pipe;
ffe60014
DG
207
208 /* Get the right pipe where the stream will be sent. */
209 if (stream->metadata_flag) {
5ab66908
MD
210 ret = consumer_add_metadata_stream(stream);
211 if (ret) {
212 ERR("Consumer add metadata stream %" PRIu64 " failed.",
213 stream->key);
214 goto error;
215 }
dae10966 216 stream_pipe = ctx->consumer_metadata_pipe;
ffe60014 217 } else {
5ab66908
MD
218 ret = consumer_add_data_stream(stream);
219 if (ret) {
220 ERR("Consumer add stream %" PRIu64 " failed.",
221 stream->key);
222 goto error;
223 }
dae10966 224 stream_pipe = ctx->consumer_data_pipe;
ffe60014
DG
225 }
226
5ab66908
MD
227 /*
228 * From this point on, the stream's ownership has been moved away from
229 * the channel and becomes globally visible.
230 */
231 stream->globally_visible = 1;
232
dae10966 233 ret = lttng_pipe_write(stream_pipe, &stream, sizeof(stream));
ffe60014 234 if (ret < 0) {
dae10966
DG
235 ERR("Consumer write %s stream to pipe %d",
236 stream->metadata_flag ? "metadata" : "data",
237 lttng_pipe_get_writefd(stream_pipe));
5ab66908
MD
238 if (stream->metadata_flag) {
239 consumer_del_stream_for_metadata(stream);
240 } else {
241 consumer_del_stream_for_data(stream);
242 }
ffe60014 243 }
5ab66908 244error:
ffe60014
DG
245 return ret;
246}
247
4628484a
MD
248static
249int get_stream_shm_path(char *stream_shm_path, const char *shm_path, int cpu)
250{
45863397 251 char cpu_nr[INT_MAX_STR_LEN]; /* int max len */
4628484a
MD
252 int ret;
253
254 strncpy(stream_shm_path, shm_path, PATH_MAX);
255 stream_shm_path[PATH_MAX - 1] = '\0';
45863397 256 ret = snprintf(cpu_nr, INT_MAX_STR_LEN, "%i", cpu);
67f8cb8d
MD
257 if (ret < 0) {
258 PERROR("snprintf");
4628484a
MD
259 goto end;
260 }
261 strncat(stream_shm_path, cpu_nr,
262 PATH_MAX - strlen(stream_shm_path) - 1);
263 ret = 0;
264end:
265 return ret;
266}
267
d88aee68
DG
268/*
269 * Create streams for the given channel using liblttng-ust-ctl.
270 *
271 * Return 0 on success else a negative value.
272 */
ffe60014
DG
273static int create_ust_streams(struct lttng_consumer_channel *channel,
274 struct lttng_consumer_local_data *ctx)
275{
276 int ret, cpu = 0;
277 struct ustctl_consumer_stream *ustream;
278 struct lttng_consumer_stream *stream;
279
280 assert(channel);
281 assert(ctx);
282
283 /*
284 * While a stream is available from ustctl. When NULL is returned, we've
285 * reached the end of the possible stream for the channel.
286 */
287 while ((ustream = ustctl_create_stream(channel->uchan, cpu))) {
288 int wait_fd;
04ef1097 289 int ust_metadata_pipe[2];
ffe60014 290
9ce5646a
MD
291 health_code_update();
292
04ef1097
MD
293 if (channel->type == CONSUMER_CHANNEL_TYPE_METADATA && channel->monitor) {
294 ret = utils_create_pipe_cloexec_nonblock(ust_metadata_pipe);
295 if (ret < 0) {
296 ERR("Create ust metadata poll pipe");
297 goto error;
298 }
299 wait_fd = ust_metadata_pipe[0];
300 } else {
301 wait_fd = ustctl_stream_get_wait_fd(ustream);
302 }
ffe60014
DG
303
304 /* Allocate consumer stream object. */
305 stream = allocate_stream(cpu, wait_fd, channel, ctx, &ret);
306 if (!stream) {
307 goto error_alloc;
308 }
309 stream->ustream = ustream;
310 /*
311 * Store it so we can save multiple function calls afterwards since
312 * this value is used heavily in the stream threads. This is UST
313 * specific so this is why it's done after allocation.
314 */
315 stream->wait_fd = wait_fd;
316
b31398bb
DG
317 /*
318 * Increment channel refcount since the channel reference has now been
319 * assigned in the allocation process above.
320 */
10a50311
JD
321 if (stream->chan->monitor) {
322 uatomic_inc(&stream->chan->refcount);
323 }
b31398bb 324
ffe60014
DG
325 /*
326 * Order is important this is why a list is used. On error, the caller
327 * should clean this list.
328 */
329 cds_list_add_tail(&stream->send_node, &channel->streams.head);
330
331 ret = ustctl_get_max_subbuf_size(stream->ustream,
332 &stream->max_sb_size);
333 if (ret < 0) {
334 ERR("ustctl_get_max_subbuf_size failed for stream %s",
335 stream->name);
336 goto error;
337 }
338
339 /* Do actions once stream has been received. */
340 if (ctx->on_recv_stream) {
341 ret = ctx->on_recv_stream(stream);
342 if (ret < 0) {
343 goto error;
344 }
345 }
346
d88aee68 347 DBG("UST consumer add stream %s (key: %" PRIu64 ") with relayd id %" PRIu64,
ffe60014
DG
348 stream->name, stream->key, stream->relayd_stream_id);
349
350 /* Set next CPU stream. */
351 channel->streams.count = ++cpu;
d88aee68
DG
352
353 /* Keep stream reference when creating metadata. */
354 if (channel->type == CONSUMER_CHANNEL_TYPE_METADATA) {
355 channel->metadata_stream = stream;
8de4f941
JG
356 if (channel->monitor) {
357 /* Set metadata poll pipe if we created one */
358 memcpy(stream->ust_metadata_poll_pipe,
359 ust_metadata_pipe,
360 sizeof(ust_metadata_pipe));
361 }
d88aee68 362 }
ffe60014
DG
363 }
364
365 return 0;
366
367error:
368error_alloc:
369 return ret;
370}
371
4628484a
MD
372/*
373 * create_posix_shm is never called concurrently within a process.
374 */
375static
376int create_posix_shm(void)
377{
378 char tmp_name[NAME_MAX];
379 int shmfd, ret;
380
381 ret = snprintf(tmp_name, NAME_MAX, "/ust-shm-consumer-%d", getpid());
382 if (ret < 0) {
383 PERROR("snprintf");
384 return -1;
385 }
386 /*
387 * Allocate shm, and immediately unlink its shm oject, keeping
388 * only the file descriptor as a reference to the object.
389 * We specifically do _not_ use the / at the beginning of the
390 * pathname so that some OS implementations can keep it local to
391 * the process (POSIX leaves this implementation-defined).
392 */
393 shmfd = shm_open(tmp_name, O_CREAT | O_EXCL | O_RDWR, 0700);
394 if (shmfd < 0) {
395 PERROR("shm_open");
396 goto error_shm_open;
397 }
398 ret = shm_unlink(tmp_name);
399 if (ret < 0 && errno != ENOENT) {
400 PERROR("shm_unlink");
401 goto error_shm_release;
402 }
403 return shmfd;
404
405error_shm_release:
406 ret = close(shmfd);
407 if (ret) {
408 PERROR("close");
409 }
410error_shm_open:
411 return -1;
412}
413
414static int open_ust_stream_fd(struct lttng_consumer_channel *channel,
415 struct ustctl_consumer_channel_attr *attr,
416 int cpu)
417{
418 char shm_path[PATH_MAX];
419 int ret;
420
421 if (!channel->shm_path[0]) {
422 return create_posix_shm();
423 }
424 ret = get_stream_shm_path(shm_path, channel->shm_path, cpu);
425 if (ret) {
426 goto error_shm_path;
427 }
428 return run_as_open(shm_path,
429 O_RDWR | O_CREAT | O_EXCL, S_IRUSR | S_IWUSR,
430 channel->uid, channel->gid);
431
432error_shm_path:
433 return -1;
434}
435
ffe60014
DG
436/*
437 * Create an UST channel with the given attributes and send it to the session
438 * daemon using the ust ctl API.
439 *
440 * Return 0 on success or else a negative value.
441 */
4628484a
MD
442static int create_ust_channel(struct lttng_consumer_channel *channel,
443 struct ustctl_consumer_channel_attr *attr,
444 struct ustctl_consumer_channel **ust_chanp)
ffe60014 445{
4628484a
MD
446 int ret, nr_stream_fds, i, j;
447 int *stream_fds;
448 struct ustctl_consumer_channel *ust_channel;
ffe60014 449
4628484a 450 assert(channel);
ffe60014 451 assert(attr);
4628484a 452 assert(ust_chanp);
ffe60014
DG
453
454 DBG3("Creating channel to ustctl with attr: [overwrite: %d, "
455 "subbuf_size: %" PRIu64 ", num_subbuf: %" PRIu64 ", "
456 "switch_timer_interval: %u, read_timer_interval: %u, "
457 "output: %d, type: %d", attr->overwrite, attr->subbuf_size,
458 attr->num_subbuf, attr->switch_timer_interval,
459 attr->read_timer_interval, attr->output, attr->type);
460
4628484a
MD
461 if (channel->type == CONSUMER_CHANNEL_TYPE_METADATA)
462 nr_stream_fds = 1;
463 else
464 nr_stream_fds = ustctl_get_nr_stream_per_channel();
465 stream_fds = zmalloc(nr_stream_fds * sizeof(*stream_fds));
466 if (!stream_fds) {
467 ret = -1;
468 goto error_alloc;
469 }
470 for (i = 0; i < nr_stream_fds; i++) {
471 stream_fds[i] = open_ust_stream_fd(channel, attr, i);
472 if (stream_fds[i] < 0) {
473 ret = -1;
474 goto error_open;
475 }
476 }
477 ust_channel = ustctl_create_channel(attr, stream_fds, nr_stream_fds);
478 if (!ust_channel) {
ffe60014
DG
479 ret = -1;
480 goto error_create;
481 }
4628484a
MD
482 channel->nr_stream_fds = nr_stream_fds;
483 channel->stream_fds = stream_fds;
484 *ust_chanp = ust_channel;
ffe60014
DG
485
486 return 0;
487
488error_create:
4628484a
MD
489error_open:
490 for (j = i - 1; j >= 0; j--) {
491 int closeret;
492
493 closeret = close(stream_fds[j]);
494 if (closeret) {
495 PERROR("close");
496 }
497 if (channel->shm_path[0]) {
498 char shm_path[PATH_MAX];
499
500 closeret = get_stream_shm_path(shm_path,
501 channel->shm_path, j);
502 if (closeret) {
503 ERR("Cannot get stream shm path");
504 }
505 closeret = run_as_unlink(shm_path,
506 channel->uid, channel->gid);
507 if (closeret) {
4628484a
MD
508 PERROR("unlink %s", shm_path);
509 }
510 }
511 }
512 /* Try to rmdir all directories under shm_path root. */
513 if (channel->root_shm_path[0]) {
602766ec 514 (void) run_as_rmdir_recursive(channel->root_shm_path,
4628484a
MD
515 channel->uid, channel->gid);
516 }
517 free(stream_fds);
518error_alloc:
ffe60014
DG
519 return ret;
520}
521
d88aee68
DG
522/*
523 * Send a single given stream to the session daemon using the sock.
524 *
525 * Return 0 on success else a negative value.
526 */
ffe60014
DG
527static int send_sessiond_stream(int sock, struct lttng_consumer_stream *stream)
528{
529 int ret;
530
531 assert(stream);
532 assert(sock >= 0);
533
3eb914c0 534 DBG("UST consumer sending stream %" PRIu64 " to sessiond", stream->key);
ffe60014
DG
535
536 /* Send stream to session daemon. */
537 ret = ustctl_send_stream_to_sessiond(sock, stream->ustream);
538 if (ret < 0) {
539 goto error;
540 }
541
ffe60014
DG
542error:
543 return ret;
544}
545
546/*
547 * Send channel to sessiond.
548 *
d88aee68 549 * Return 0 on success or else a negative value.
ffe60014
DG
550 */
551static int send_sessiond_channel(int sock,
552 struct lttng_consumer_channel *channel,
553 struct lttng_consumer_local_data *ctx, int *relayd_error)
554{
0c759fc9 555 int ret, ret_code = LTTCOMM_CONSUMERD_SUCCESS;
ffe60014 556 struct lttng_consumer_stream *stream;
a4baae1b 557 uint64_t net_seq_idx = -1ULL;
ffe60014
DG
558
559 assert(channel);
560 assert(ctx);
561 assert(sock >= 0);
562
563 DBG("UST consumer sending channel %s to sessiond", channel->name);
564
62285ea4
DG
565 if (channel->relayd_id != (uint64_t) -1ULL) {
566 cds_list_for_each_entry(stream, &channel->streams.head, send_node) {
9ce5646a
MD
567
568 health_code_update();
569
62285ea4
DG
570 /* Try to send the stream to the relayd if one is available. */
571 ret = consumer_send_relayd_stream(stream, stream->chan->pathname);
572 if (ret < 0) {
573 /*
574 * Flag that the relayd was the problem here probably due to a
575 * communicaton error on the socket.
576 */
577 if (relayd_error) {
578 *relayd_error = 1;
579 }
725d28b2 580 ret_code = LTTCOMM_CONSUMERD_RELAYD_FAIL;
ffe60014 581 }
a4baae1b
JD
582 if (net_seq_idx == -1ULL) {
583 net_seq_idx = stream->net_seq_idx;
584 }
585 }
f2a444f1 586 }
ffe60014 587
f2a444f1
DG
588 /* Inform sessiond that we are about to send channel and streams. */
589 ret = consumer_send_status_msg(sock, ret_code);
0c759fc9 590 if (ret < 0 || ret_code != LTTCOMM_CONSUMERD_SUCCESS) {
f2a444f1
DG
591 /*
592 * Either the session daemon is not responding or the relayd died so we
593 * stop now.
594 */
595 goto error;
596 }
597
598 /* Send channel to sessiond. */
599 ret = ustctl_send_channel_to_sessiond(sock, channel->uchan);
600 if (ret < 0) {
601 goto error;
602 }
603
604 ret = ustctl_channel_close_wakeup_fd(channel->uchan);
605 if (ret < 0) {
606 goto error;
607 }
608
609 /* The channel was sent successfully to the sessiond at this point. */
610 cds_list_for_each_entry(stream, &channel->streams.head, send_node) {
9ce5646a
MD
611
612 health_code_update();
613
ffe60014
DG
614 /* Send stream to session daemon. */
615 ret = send_sessiond_stream(sock, stream);
616 if (ret < 0) {
617 goto error;
618 }
619 }
620
621 /* Tell sessiond there is no more stream. */
622 ret = ustctl_send_stream_to_sessiond(sock, NULL);
623 if (ret < 0) {
624 goto error;
625 }
626
627 DBG("UST consumer NULL stream sent to sessiond");
628
629 return 0;
630
631error:
0c759fc9 632 if (ret_code != LTTCOMM_CONSUMERD_SUCCESS) {
f2a444f1
DG
633 ret = -1;
634 }
ffe60014
DG
635 return ret;
636}
637
638/*
639 * Creates a channel and streams and add the channel it to the channel internal
640 * state. The created stream must ONLY be sent once the GET_CHANNEL command is
641 * received.
642 *
643 * Return 0 on success or else, a negative value is returned and the channel
644 * MUST be destroyed by consumer_del_channel().
645 */
646static int ask_channel(struct lttng_consumer_local_data *ctx, int sock,
647 struct lttng_consumer_channel *channel,
648 struct ustctl_consumer_channel_attr *attr)
3bd1e081
MD
649{
650 int ret;
651
ffe60014
DG
652 assert(ctx);
653 assert(channel);
654 assert(attr);
655
656 /*
657 * This value is still used by the kernel consumer since for the kernel,
658 * the stream ownership is not IN the consumer so we need to have the
659 * number of left stream that needs to be initialized so we can know when
660 * to delete the channel (see consumer.c).
661 *
662 * As for the user space tracer now, the consumer creates and sends the
663 * stream to the session daemon which only sends them to the application
664 * once every stream of a channel is received making this value useless
665 * because we they will be added to the poll thread before the application
666 * receives them. This ensures that a stream can not hang up during
667 * initilization of a channel.
668 */
669 channel->nb_init_stream_left = 0;
670
671 /* The reply msg status is handled in the following call. */
4628484a 672 ret = create_ust_channel(channel, attr, &channel->uchan);
ffe60014 673 if (ret < 0) {
10a50311 674 goto end;
3bd1e081
MD
675 }
676
d8ef542d
MD
677 channel->wait_fd = ustctl_channel_get_wait_fd(channel->uchan);
678
10a50311
JD
679 /*
680 * For the snapshots (no monitor), we create the metadata streams
681 * on demand, not during the channel creation.
682 */
683 if (channel->type == CONSUMER_CHANNEL_TYPE_METADATA && !channel->monitor) {
684 ret = 0;
685 goto end;
686 }
687
ffe60014
DG
688 /* Open all streams for this channel. */
689 ret = create_ust_streams(channel, ctx);
690 if (ret < 0) {
10a50311 691 goto end;
ffe60014
DG
692 }
693
10a50311 694end:
3bd1e081
MD
695 return ret;
696}
697
d88aee68
DG
698/*
699 * Send all stream of a channel to the right thread handling it.
700 *
701 * On error, return a negative value else 0 on success.
702 */
703static int send_streams_to_thread(struct lttng_consumer_channel *channel,
704 struct lttng_consumer_local_data *ctx)
705{
706 int ret = 0;
707 struct lttng_consumer_stream *stream, *stmp;
708
709 assert(channel);
710 assert(ctx);
711
712 /* Send streams to the corresponding thread. */
713 cds_list_for_each_entry_safe(stream, stmp, &channel->streams.head,
714 send_node) {
9ce5646a
MD
715
716 health_code_update();
717
d88aee68
DG
718 /* Sending the stream to the thread. */
719 ret = send_stream_to_thread(stream, ctx);
720 if (ret < 0) {
721 /*
722 * If we are unable to send the stream to the thread, there is
723 * a big problem so just stop everything.
724 */
5ab66908
MD
725 /* Remove node from the channel stream list. */
726 cds_list_del(&stream->send_node);
d88aee68
DG
727 goto error;
728 }
729
730 /* Remove node from the channel stream list. */
731 cds_list_del(&stream->send_node);
4891ece8 732
d88aee68
DG
733 }
734
735error:
736 return ret;
737}
738
7972aab2
DG
739/*
740 * Flush channel's streams using the given key to retrieve the channel.
741 *
742 * Return 0 on success else an LTTng error code.
743 */
744static int flush_channel(uint64_t chan_key)
745{
746 int ret = 0;
747 struct lttng_consumer_channel *channel;
748 struct lttng_consumer_stream *stream;
749 struct lttng_ht *ht;
750 struct lttng_ht_iter iter;
751
8fd623e0 752 DBG("UST consumer flush channel key %" PRIu64, chan_key);
7972aab2 753
a500c257 754 rcu_read_lock();
7972aab2
DG
755 channel = consumer_find_channel(chan_key);
756 if (!channel) {
8fd623e0 757 ERR("UST consumer flush channel %" PRIu64 " not found", chan_key);
7972aab2
DG
758 ret = LTTNG_ERR_UST_CHAN_NOT_FOUND;
759 goto error;
760 }
761
762 ht = consumer_data.stream_per_chan_id_ht;
763
764 /* For each stream of the channel id, flush it. */
7972aab2
DG
765 cds_lfht_for_each_entry_duplicate(ht->ht,
766 ht->hash_fct(&channel->key, lttng_ht_seed), ht->match_fct,
767 &channel->key, &iter.iter, stream, node_channel_id.node) {
9ce5646a
MD
768
769 health_code_update();
770
0dd01979
MD
771 pthread_mutex_lock(&stream->lock);
772 if (!stream->quiescent) {
773 ustctl_flush_buffer(stream->ustream, 0);
774 stream->quiescent = true;
775 }
776 pthread_mutex_unlock(&stream->lock);
777 }
778error:
779 rcu_read_unlock();
780 return ret;
781}
782
783/*
784 * Clear quiescent state from channel's streams using the given key to
785 * retrieve the channel.
786 *
787 * Return 0 on success else an LTTng error code.
788 */
789static int clear_quiescent_channel(uint64_t chan_key)
790{
791 int ret = 0;
792 struct lttng_consumer_channel *channel;
793 struct lttng_consumer_stream *stream;
794 struct lttng_ht *ht;
795 struct lttng_ht_iter iter;
796
797 DBG("UST consumer clear quiescent channel key %" PRIu64, chan_key);
798
799 rcu_read_lock();
800 channel = consumer_find_channel(chan_key);
801 if (!channel) {
802 ERR("UST consumer clear quiescent channel %" PRIu64 " not found", chan_key);
803 ret = LTTNG_ERR_UST_CHAN_NOT_FOUND;
804 goto error;
805 }
806
807 ht = consumer_data.stream_per_chan_id_ht;
808
809 /* For each stream of the channel id, clear quiescent state. */
810 cds_lfht_for_each_entry_duplicate(ht->ht,
811 ht->hash_fct(&channel->key, lttng_ht_seed), ht->match_fct,
812 &channel->key, &iter.iter, stream, node_channel_id.node) {
813
814 health_code_update();
815
816 pthread_mutex_lock(&stream->lock);
817 stream->quiescent = false;
818 pthread_mutex_unlock(&stream->lock);
7972aab2 819 }
7972aab2 820error:
a500c257 821 rcu_read_unlock();
7972aab2
DG
822 return ret;
823}
824
d88aee68
DG
825/*
826 * Close metadata stream wakeup_fd using the given key to retrieve the channel.
a500c257 827 * RCU read side lock MUST be acquired before calling this function.
d88aee68
DG
828 *
829 * Return 0 on success else an LTTng error code.
830 */
831static int close_metadata(uint64_t chan_key)
832{
ea88ca2a 833 int ret = 0;
d88aee68
DG
834 struct lttng_consumer_channel *channel;
835
8fd623e0 836 DBG("UST consumer close metadata key %" PRIu64, chan_key);
d88aee68
DG
837
838 channel = consumer_find_channel(chan_key);
839 if (!channel) {
84cc9aa0
DG
840 /*
841 * This is possible if the metadata thread has issue a delete because
842 * the endpoint point of the stream hung up. There is no way the
843 * session daemon can know about it thus use a DBG instead of an actual
844 * error.
845 */
846 DBG("UST consumer close metadata %" PRIu64 " not found", chan_key);
d88aee68
DG
847 ret = LTTNG_ERR_UST_CHAN_NOT_FOUND;
848 goto error;
849 }
850
ea88ca2a 851 pthread_mutex_lock(&consumer_data.lock);
a9838785 852 pthread_mutex_lock(&channel->lock);
73811ecc
DG
853
854 if (cds_lfht_is_node_deleted(&channel->node.node)) {
855 goto error_unlock;
856 }
857
6d574024 858 lttng_ustconsumer_close_metadata(channel);
d88aee68 859
ea88ca2a 860error_unlock:
a9838785 861 pthread_mutex_unlock(&channel->lock);
ea88ca2a 862 pthread_mutex_unlock(&consumer_data.lock);
d88aee68
DG
863error:
864 return ret;
865}
866
867/*
868 * RCU read side lock MUST be acquired before calling this function.
869 *
870 * Return 0 on success else an LTTng error code.
871 */
872static int setup_metadata(struct lttng_consumer_local_data *ctx, uint64_t key)
873{
874 int ret;
875 struct lttng_consumer_channel *metadata;
876
8fd623e0 877 DBG("UST consumer setup metadata key %" PRIu64, key);
d88aee68
DG
878
879 metadata = consumer_find_channel(key);
880 if (!metadata) {
881 ERR("UST consumer push metadata %" PRIu64 " not found", key);
882 ret = LTTNG_ERR_UST_CHAN_NOT_FOUND;
10a50311
JD
883 goto end;
884 }
885
886 /*
887 * In no monitor mode, the metadata channel has no stream(s) so skip the
888 * ownership transfer to the metadata thread.
889 */
890 if (!metadata->monitor) {
891 DBG("Metadata channel in no monitor");
892 ret = 0;
893 goto end;
d88aee68
DG
894 }
895
896 /*
897 * Send metadata stream to relayd if one available. Availability is
898 * known if the stream is still in the list of the channel.
899 */
900 if (cds_list_empty(&metadata->streams.head)) {
901 ERR("Metadata channel key %" PRIu64 ", no stream available.", key);
902 ret = LTTCOMM_CONSUMERD_ERROR_METADATA;
f5a0c9cf 903 goto error_no_stream;
d88aee68
DG
904 }
905
906 /* Send metadata stream to relayd if needed. */
62285ea4
DG
907 if (metadata->metadata_stream->net_seq_idx != (uint64_t) -1ULL) {
908 ret = consumer_send_relayd_stream(metadata->metadata_stream,
909 metadata->pathname);
910 if (ret < 0) {
911 ret = LTTCOMM_CONSUMERD_ERROR_METADATA;
912 goto error;
913 }
601262d6
JD
914 ret = consumer_send_relayd_streams_sent(
915 metadata->metadata_stream->net_seq_idx);
916 if (ret < 0) {
917 ret = LTTCOMM_CONSUMERD_RELAYD_FAIL;
918 goto error;
919 }
d88aee68
DG
920 }
921
922 ret = send_streams_to_thread(metadata, ctx);
923 if (ret < 0) {
924 /*
925 * If we are unable to send the stream to the thread, there is
926 * a big problem so just stop everything.
927 */
928 ret = LTTCOMM_CONSUMERD_FATAL;
929 goto error;
930 }
931 /* List MUST be empty after or else it could be reused. */
932 assert(cds_list_empty(&metadata->streams.head));
933
10a50311
JD
934 ret = 0;
935 goto end;
d88aee68
DG
936
937error:
f2a444f1
DG
938 /*
939 * Delete metadata channel on error. At this point, the metadata stream can
940 * NOT be monitored by the metadata thread thus having the guarantee that
941 * the stream is still in the local stream list of the channel. This call
942 * will make sure to clean that list.
943 */
f5a0c9cf 944 consumer_stream_destroy(metadata->metadata_stream, NULL);
212d67a2
DG
945 cds_list_del(&metadata->metadata_stream->send_node);
946 metadata->metadata_stream = NULL;
f5a0c9cf 947error_no_stream:
10a50311
JD
948end:
949 return ret;
950}
951
952/*
953 * Snapshot the whole metadata.
954 *
955 * Returns 0 on success, < 0 on error
956 */
957static int snapshot_metadata(uint64_t key, char *path, uint64_t relayd_id,
958 struct lttng_consumer_local_data *ctx)
959{
960 int ret = 0;
10a50311
JD
961 struct lttng_consumer_channel *metadata_channel;
962 struct lttng_consumer_stream *metadata_stream;
963
964 assert(path);
965 assert(ctx);
966
967 DBG("UST consumer snapshot metadata with key %" PRIu64 " at path %s",
968 key, path);
969
970 rcu_read_lock();
971
972 metadata_channel = consumer_find_channel(key);
973 if (!metadata_channel) {
6a00837f
MD
974 ERR("UST snapshot metadata channel not found for key %" PRIu64,
975 key);
10a50311
JD
976 ret = -1;
977 goto error;
978 }
979 assert(!metadata_channel->monitor);
980
9ce5646a
MD
981 health_code_update();
982
10a50311
JD
983 /*
984 * Ask the sessiond if we have new metadata waiting and update the
985 * consumer metadata cache.
986 */
94d49140 987 ret = lttng_ustconsumer_request_metadata(ctx, metadata_channel, 0, 1);
10a50311
JD
988 if (ret < 0) {
989 goto error;
990 }
991
9ce5646a
MD
992 health_code_update();
993
10a50311
JD
994 /*
995 * The metadata stream is NOT created in no monitor mode when the channel
996 * is created on a sessiond ask channel command.
997 */
998 ret = create_ust_streams(metadata_channel, ctx);
999 if (ret < 0) {
1000 goto error;
1001 }
1002
1003 metadata_stream = metadata_channel->metadata_stream;
1004 assert(metadata_stream);
1005
1006 if (relayd_id != (uint64_t) -1ULL) {
1007 metadata_stream->net_seq_idx = relayd_id;
1008 ret = consumer_send_relayd_stream(metadata_stream, path);
1009 if (ret < 0) {
1010 goto error_stream;
1011 }
1012 } else {
1013 ret = utils_create_stream_file(path, metadata_stream->name,
1014 metadata_stream->chan->tracefile_size,
1015 metadata_stream->tracefile_count_current,
309167d2 1016 metadata_stream->uid, metadata_stream->gid, NULL);
10a50311
JD
1017 if (ret < 0) {
1018 goto error_stream;
1019 }
1020 metadata_stream->out_fd = ret;
1021 metadata_stream->tracefile_size_current = 0;
1022 }
1023
04ef1097 1024 do {
9ce5646a
MD
1025 health_code_update();
1026
10a50311
JD
1027 ret = lttng_consumer_read_subbuffer(metadata_stream, ctx);
1028 if (ret < 0) {
94d49140 1029 goto error_stream;
10a50311 1030 }
04ef1097 1031 } while (ret > 0);
10a50311 1032
10a50311
JD
1033error_stream:
1034 /*
1035 * Clean up the stream completly because the next snapshot will use a new
1036 * metadata stream.
1037 */
10a50311 1038 consumer_stream_destroy(metadata_stream, NULL);
212d67a2 1039 cds_list_del(&metadata_stream->send_node);
10a50311
JD
1040 metadata_channel->metadata_stream = NULL;
1041
1042error:
1043 rcu_read_unlock();
1044 return ret;
1045}
1046
1047/*
1048 * Take a snapshot of all the stream of a channel.
1049 *
1050 * Returns 0 on success, < 0 on error
1051 */
1052static int snapshot_channel(uint64_t key, char *path, uint64_t relayd_id,
d07ceecd 1053 uint64_t nb_packets_per_stream, struct lttng_consumer_local_data *ctx)
10a50311
JD
1054{
1055 int ret;
1056 unsigned use_relayd = 0;
1057 unsigned long consumed_pos, produced_pos;
1058 struct lttng_consumer_channel *channel;
1059 struct lttng_consumer_stream *stream;
1060
1061 assert(path);
1062 assert(ctx);
1063
1064 rcu_read_lock();
1065
1066 if (relayd_id != (uint64_t) -1ULL) {
1067 use_relayd = 1;
1068 }
1069
1070 channel = consumer_find_channel(key);
1071 if (!channel) {
6a00837f 1072 ERR("UST snapshot channel not found for key %" PRIu64, key);
10a50311
JD
1073 ret = -1;
1074 goto error;
1075 }
1076 assert(!channel->monitor);
6a00837f 1077 DBG("UST consumer snapshot channel %" PRIu64, key);
10a50311
JD
1078
1079 cds_list_for_each_entry(stream, &channel->streams.head, send_node) {
c13082a9
JD
1080 /* Are we at a position _before_ the first available packet ? */
1081 bool before_first_packet = true;
9ce5646a
MD
1082
1083 health_code_update();
1084
10a50311
JD
1085 /* Lock stream because we are about to change its state. */
1086 pthread_mutex_lock(&stream->lock);
1087 stream->net_seq_idx = relayd_id;
1088
1089 if (use_relayd) {
1090 ret = consumer_send_relayd_stream(stream, path);
1091 if (ret < 0) {
1092 goto error_unlock;
1093 }
1094 } else {
1095 ret = utils_create_stream_file(path, stream->name,
1096 stream->chan->tracefile_size,
1097 stream->tracefile_count_current,
309167d2 1098 stream->uid, stream->gid, NULL);
10a50311
JD
1099 if (ret < 0) {
1100 goto error_unlock;
1101 }
1102 stream->out_fd = ret;
1103 stream->tracefile_size_current = 0;
1104
1105 DBG("UST consumer snapshot stream %s/%s (%" PRIu64 ")", path,
1106 stream->name, stream->key);
1107 }
a4baae1b
JD
1108 if (relayd_id != -1ULL) {
1109 ret = consumer_send_relayd_streams_sent(relayd_id);
1110 if (ret < 0) {
1111 goto error_unlock;
1112 }
1113 }
10a50311 1114
d4d80f77
MD
1115 /*
1116 * If tracing is active, we want to perform a "full" buffer flush.
1117 * Else, if quiescent, it has already been done by the prior stop.
1118 */
1119 if (!stream->quiescent) {
1120 ustctl_flush_buffer(stream->ustream, 0);
1121 }
10a50311
JD
1122
1123 ret = lttng_ustconsumer_take_snapshot(stream);
1124 if (ret < 0) {
1125 ERR("Taking UST snapshot");
1126 goto error_unlock;
1127 }
1128
1129 ret = lttng_ustconsumer_get_produced_snapshot(stream, &produced_pos);
1130 if (ret < 0) {
1131 ERR("Produced UST snapshot position");
1132 goto error_unlock;
1133 }
1134
1135 ret = lttng_ustconsumer_get_consumed_snapshot(stream, &consumed_pos);
1136 if (ret < 0) {
1137 ERR("Consumerd UST snapshot position");
1138 goto error_unlock;
1139 }
1140
5c786ded
JD
1141 /*
1142 * The original value is sent back if max stream size is larger than
d07ceecd 1143 * the possible size of the snapshot. Also, we assume that the session
5c786ded
JD
1144 * daemon should never send a maximum stream size that is lower than
1145 * subbuffer size.
1146 */
d07ceecd
MD
1147 consumed_pos = consumer_get_consume_start_pos(consumed_pos,
1148 produced_pos, nb_packets_per_stream,
1149 stream->max_sb_size);
5c786ded 1150
10a50311
JD
1151 while (consumed_pos < produced_pos) {
1152 ssize_t read_len;
1153 unsigned long len, padded_len;
c13082a9 1154 int lost_packet = 0;
10a50311 1155
9ce5646a
MD
1156 health_code_update();
1157
10a50311
JD
1158 DBG("UST consumer taking snapshot at pos %lu", consumed_pos);
1159
1160 ret = ustctl_get_subbuf(stream->ustream, &consumed_pos);
1161 if (ret < 0) {
1162 if (ret != -EAGAIN) {
1163 PERROR("ustctl_get_subbuf snapshot");
1164 goto error_close_stream;
1165 }
1166 DBG("UST consumer get subbuf failed. Skipping it.");
1167 consumed_pos += stream->max_sb_size;
c13082a9
JD
1168
1169 /*
1170 * Start accounting lost packets only when we
1171 * already have extracted packets (to match the
1172 * content of the final snapshot).
1173 */
1174 if (!before_first_packet) {
1175 lost_packet = 1;
1176 }
10a50311
JD
1177 continue;
1178 }
1179
1180 ret = ustctl_get_subbuf_size(stream->ustream, &len);
1181 if (ret < 0) {
1182 ERR("Snapshot ustctl_get_subbuf_size");
1183 goto error_put_subbuf;
1184 }
1185
1186 ret = ustctl_get_padded_subbuf_size(stream->ustream, &padded_len);
1187 if (ret < 0) {
1188 ERR("Snapshot ustctl_get_padded_subbuf_size");
1189 goto error_put_subbuf;
1190 }
1191
1192 read_len = lttng_consumer_on_read_subbuffer_mmap(ctx, stream, len,
309167d2 1193 padded_len - len, NULL);
10a50311
JD
1194 if (use_relayd) {
1195 if (read_len != len) {
56591bac 1196 ret = -EPERM;
10a50311
JD
1197 goto error_put_subbuf;
1198 }
1199 } else {
1200 if (read_len != padded_len) {
56591bac 1201 ret = -EPERM;
10a50311
JD
1202 goto error_put_subbuf;
1203 }
1204 }
1205
1206 ret = ustctl_put_subbuf(stream->ustream);
1207 if (ret < 0) {
1208 ERR("Snapshot ustctl_put_subbuf");
1209 goto error_close_stream;
1210 }
1211 consumed_pos += stream->max_sb_size;
c13082a9
JD
1212
1213 /*
1214 * Only account lost packets located between
1215 * succesfully extracted packets (do not account before
1216 * and after since they are not visible in the
1217 * resulting snapshot).
1218 */
1219 stream->chan->lost_packets += lost_packet;
1220 lost_packet = 0;
1221 before_first_packet = false;
10a50311
JD
1222 }
1223
1224 /* Simply close the stream so we can use it on the next snapshot. */
1225 consumer_stream_close(stream);
1226 pthread_mutex_unlock(&stream->lock);
1227 }
1228
1229 rcu_read_unlock();
1230 return 0;
1231
1232error_put_subbuf:
1233 if (ustctl_put_subbuf(stream->ustream) < 0) {
1234 ERR("Snapshot ustctl_put_subbuf");
1235 }
1236error_close_stream:
1237 consumer_stream_close(stream);
1238error_unlock:
1239 pthread_mutex_unlock(&stream->lock);
1240error:
1241 rcu_read_unlock();
d88aee68
DG
1242 return ret;
1243}
1244
331744e3 1245/*
c585821b
MD
1246 * Receive the metadata updates from the sessiond. Supports receiving
1247 * overlapping metadata, but is needs to always belong to a contiguous
1248 * range starting from 0.
1249 * Be careful about the locks held when calling this function: it needs
1250 * the metadata cache flush to concurrently progress in order to
1251 * complete.
331744e3
JD
1252 */
1253int lttng_ustconsumer_recv_metadata(int sock, uint64_t key, uint64_t offset,
93ec662e
JD
1254 uint64_t len, uint64_t version,
1255 struct lttng_consumer_channel *channel, int timer, int wait)
331744e3 1256{
0c759fc9 1257 int ret, ret_code = LTTCOMM_CONSUMERD_SUCCESS;
331744e3
JD
1258 char *metadata_str;
1259
8fd623e0 1260 DBG("UST consumer push metadata key %" PRIu64 " of len %" PRIu64, key, len);
331744e3
JD
1261
1262 metadata_str = zmalloc(len * sizeof(char));
1263 if (!metadata_str) {
1264 PERROR("zmalloc metadata string");
1265 ret_code = LTTCOMM_CONSUMERD_ENOMEM;
1266 goto end;
1267 }
1268
9ce5646a
MD
1269 health_code_update();
1270
331744e3
JD
1271 /* Receive metadata string. */
1272 ret = lttcomm_recv_unix_sock(sock, metadata_str, len);
1273 if (ret < 0) {
1274 /* Session daemon is dead so return gracefully. */
1275 ret_code = ret;
1276 goto end_free;
1277 }
1278
9ce5646a
MD
1279 health_code_update();
1280
331744e3 1281 pthread_mutex_lock(&channel->metadata_cache->lock);
93ec662e
JD
1282 ret = consumer_metadata_cache_write(channel, offset, len, version,
1283 metadata_str);
331744e3
JD
1284 if (ret < 0) {
1285 /* Unable to handle metadata. Notify session daemon. */
1286 ret_code = LTTCOMM_CONSUMERD_ERROR_METADATA;
a32bd775
DG
1287 /*
1288 * Skip metadata flush on write error since the offset and len might
1289 * not have been updated which could create an infinite loop below when
1290 * waiting for the metadata cache to be flushed.
1291 */
1292 pthread_mutex_unlock(&channel->metadata_cache->lock);
a32bd775 1293 goto end_free;
331744e3
JD
1294 }
1295 pthread_mutex_unlock(&channel->metadata_cache->lock);
1296
94d49140
JD
1297 if (!wait) {
1298 goto end_free;
1299 }
5e41ebe1 1300 while (consumer_metadata_cache_flushed(channel, offset + len, timer)) {
331744e3 1301 DBG("Waiting for metadata to be flushed");
9ce5646a
MD
1302
1303 health_code_update();
1304
331744e3
JD
1305 usleep(DEFAULT_METADATA_AVAILABILITY_WAIT_TIME);
1306 }
1307
1308end_free:
1309 free(metadata_str);
1310end:
1311 return ret_code;
1312}
1313
4cbc1a04
DG
1314/*
1315 * Receive command from session daemon and process it.
1316 *
1317 * Return 1 on success else a negative value or 0.
1318 */
3bd1e081
MD
1319int lttng_ustconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
1320 int sock, struct pollfd *consumer_sockpoll)
1321{
1322 ssize_t ret;
0c759fc9 1323 enum lttcomm_return_code ret_code = LTTCOMM_CONSUMERD_SUCCESS;
3bd1e081 1324 struct lttcomm_consumer_msg msg;
ffe60014 1325 struct lttng_consumer_channel *channel = NULL;
3bd1e081 1326
9ce5646a
MD
1327 health_code_update();
1328
3bd1e081
MD
1329 ret = lttcomm_recv_unix_sock(sock, &msg, sizeof(msg));
1330 if (ret != sizeof(msg)) {
173af62f
DG
1331 DBG("Consumer received unexpected message size %zd (expects %zu)",
1332 ret, sizeof(msg));
3be74084
DG
1333 /*
1334 * The ret value might 0 meaning an orderly shutdown but this is ok
1335 * since the caller handles this.
1336 */
489f70e9 1337 if (ret > 0) {
c6857fcf 1338 lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_ERROR_RECV_CMD);
489f70e9
MD
1339 ret = -1;
1340 }
3bd1e081
MD
1341 return ret;
1342 }
9ce5646a
MD
1343
1344 health_code_update();
1345
84382d49
MD
1346 /* deprecated */
1347 assert(msg.cmd_type != LTTNG_CONSUMER_STOP);
3bd1e081 1348
9ce5646a
MD
1349 health_code_update();
1350
3f8e211f 1351 /* relayd needs RCU read-side lock */
b0b335c8
MD
1352 rcu_read_lock();
1353
3bd1e081 1354 switch (msg.cmd_type) {
00e2e675
DG
1355 case LTTNG_CONSUMER_ADD_RELAYD_SOCKET:
1356 {
f50f23d9 1357 /* Session daemon status message are handled in the following call. */
7735ef9e
DG
1358 ret = consumer_add_relayd_socket(msg.u.relayd_sock.net_index,
1359 msg.u.relayd_sock.type, ctx, sock, consumer_sockpoll,
d3e2ba59
JD
1360 &msg.u.relayd_sock.sock, msg.u.relayd_sock.session_id,
1361 msg.u.relayd_sock.relayd_session_id);
00e2e675
DG
1362 goto end_nosignal;
1363 }
173af62f
DG
1364 case LTTNG_CONSUMER_DESTROY_RELAYD:
1365 {
a6ba4fe1 1366 uint64_t index = msg.u.destroy_relayd.net_seq_idx;
173af62f
DG
1367 struct consumer_relayd_sock_pair *relayd;
1368
a6ba4fe1 1369 DBG("UST consumer destroying relayd %" PRIu64, index);
173af62f
DG
1370
1371 /* Get relayd reference if exists. */
a6ba4fe1 1372 relayd = consumer_find_relayd(index);
173af62f 1373 if (relayd == NULL) {
3448e266 1374 DBG("Unable to find relayd %" PRIu64, index);
e462382a 1375 ret_code = LTTCOMM_CONSUMERD_RELAYD_FAIL;
173af62f
DG
1376 }
1377
a6ba4fe1
DG
1378 /*
1379 * Each relayd socket pair has a refcount of stream attached to it
1380 * which tells if the relayd is still active or not depending on the
1381 * refcount value.
1382 *
1383 * This will set the destroy flag of the relayd object and destroy it
1384 * if the refcount reaches zero when called.
1385 *
1386 * The destroy can happen either here or when a stream fd hangs up.
1387 */
f50f23d9
DG
1388 if (relayd) {
1389 consumer_flag_relayd_for_destroy(relayd);
1390 }
1391
d88aee68 1392 goto end_msg_sessiond;
173af62f 1393 }
3bd1e081
MD
1394 case LTTNG_CONSUMER_UPDATE_STREAM:
1395 {
3f8e211f 1396 rcu_read_unlock();
7ad0a0cb 1397 return -ENOSYS;
3bd1e081 1398 }
6d805429 1399 case LTTNG_CONSUMER_DATA_PENDING:
53632229 1400 {
3be74084 1401 int ret, is_data_pending;
6d805429 1402 uint64_t id = msg.u.data_pending.session_id;
ca22feea 1403
6d805429 1404 DBG("UST consumer data pending command for id %" PRIu64, id);
ca22feea 1405
3be74084 1406 is_data_pending = consumer_data_pending(id);
ca22feea
DG
1407
1408 /* Send back returned value to session daemon */
3be74084
DG
1409 ret = lttcomm_send_unix_sock(sock, &is_data_pending,
1410 sizeof(is_data_pending));
ca22feea 1411 if (ret < 0) {
3be74084 1412 DBG("Error when sending the data pending ret code: %d", ret);
489f70e9 1413 goto error_fatal;
ca22feea 1414 }
f50f23d9
DG
1415
1416 /*
1417 * No need to send back a status message since the data pending
1418 * returned value is the response.
1419 */
ca22feea 1420 break;
53632229 1421 }
ffe60014
DG
1422 case LTTNG_CONSUMER_ASK_CHANNEL_CREATION:
1423 {
1424 int ret;
1425 struct ustctl_consumer_channel_attr attr;
1426
1427 /* Create a plain object and reserve a channel key. */
1428 channel = allocate_channel(msg.u.ask_channel.session_id,
1429 msg.u.ask_channel.pathname, msg.u.ask_channel.name,
1430 msg.u.ask_channel.uid, msg.u.ask_channel.gid,
1431 msg.u.ask_channel.relayd_id, msg.u.ask_channel.key,
1624d5b7
JD
1432 (enum lttng_event_output) msg.u.ask_channel.output,
1433 msg.u.ask_channel.tracefile_size,
2bba9e53 1434 msg.u.ask_channel.tracefile_count,
1950109e 1435 msg.u.ask_channel.session_id_per_pid,
ecc48a90 1436 msg.u.ask_channel.monitor,
d7ba1388 1437 msg.u.ask_channel.live_timer_interval,
3d071855 1438 msg.u.ask_channel.root_shm_path,
d7ba1388 1439 msg.u.ask_channel.shm_path);
ffe60014
DG
1440 if (!channel) {
1441 goto end_channel_error;
1442 }
1443
567eb353
DG
1444 /*
1445 * Assign UST application UID to the channel. This value is ignored for
1446 * per PID buffers. This is specific to UST thus setting this after the
1447 * allocation.
1448 */
1449 channel->ust_app_uid = msg.u.ask_channel.ust_app_uid;
1450
ffe60014
DG
1451 /* Build channel attributes from received message. */
1452 attr.subbuf_size = msg.u.ask_channel.subbuf_size;
1453 attr.num_subbuf = msg.u.ask_channel.num_subbuf;
1454 attr.overwrite = msg.u.ask_channel.overwrite;
1455 attr.switch_timer_interval = msg.u.ask_channel.switch_timer_interval;
1456 attr.read_timer_interval = msg.u.ask_channel.read_timer_interval;
7972aab2 1457 attr.chan_id = msg.u.ask_channel.chan_id;
ffe60014
DG
1458 memcpy(attr.uuid, msg.u.ask_channel.uuid, sizeof(attr.uuid));
1459
0c759fc9
DG
1460 /* Match channel buffer type to the UST abi. */
1461 switch (msg.u.ask_channel.output) {
1462 case LTTNG_EVENT_MMAP:
1463 default:
1464 attr.output = LTTNG_UST_MMAP;
1465 break;
1466 }
1467
ffe60014
DG
1468 /* Translate and save channel type. */
1469 switch (msg.u.ask_channel.type) {
1470 case LTTNG_UST_CHAN_PER_CPU:
1471 channel->type = CONSUMER_CHANNEL_TYPE_DATA;
1472 attr.type = LTTNG_UST_CHAN_PER_CPU;
8633d6e3
MD
1473 /*
1474 * Set refcount to 1 for owner. Below, we will
1475 * pass ownership to the
1476 * consumer_thread_channel_poll() thread.
1477 */
1478 channel->refcount = 1;
ffe60014
DG
1479 break;
1480 case LTTNG_UST_CHAN_METADATA:
1481 channel->type = CONSUMER_CHANNEL_TYPE_METADATA;
1482 attr.type = LTTNG_UST_CHAN_METADATA;
1483 break;
1484 default:
1485 assert(0);
1486 goto error_fatal;
1487 };
1488
9ce5646a
MD
1489 health_code_update();
1490
ffe60014
DG
1491 ret = ask_channel(ctx, sock, channel, &attr);
1492 if (ret < 0) {
1493 goto end_channel_error;
1494 }
1495
fc643247
MD
1496 if (msg.u.ask_channel.type == LTTNG_UST_CHAN_METADATA) {
1497 ret = consumer_metadata_cache_allocate(channel);
1498 if (ret < 0) {
1499 ERR("Allocating metadata cache");
1500 goto end_channel_error;
1501 }
1502 consumer_timer_switch_start(channel, attr.switch_timer_interval);
1503 attr.switch_timer_interval = 0;
94d49140 1504 } else {
e9404c27
JG
1505 int monitor_start_ret;
1506
94d49140
JD
1507 consumer_timer_live_start(channel,
1508 msg.u.ask_channel.live_timer_interval);
e9404c27
JG
1509 monitor_start_ret = consumer_timer_monitor_start(
1510 channel,
1511 msg.u.ask_channel.monitor_timer_interval);
1512 if (monitor_start_ret < 0) {
1513 ERR("Starting channel monitoring timer failed");
1514 goto end_channel_error;
1515 }
fc643247
MD
1516 }
1517
9ce5646a
MD
1518 health_code_update();
1519
ffe60014
DG
1520 /*
1521 * Add the channel to the internal state AFTER all streams were created
1522 * and successfully sent to session daemon. This way, all streams must
1523 * be ready before this channel is visible to the threads.
fc643247
MD
1524 * If add_channel succeeds, ownership of the channel is
1525 * passed to consumer_thread_channel_poll().
ffe60014
DG
1526 */
1527 ret = add_channel(channel, ctx);
1528 if (ret < 0) {
ea88ca2a
MD
1529 if (msg.u.ask_channel.type == LTTNG_UST_CHAN_METADATA) {
1530 if (channel->switch_timer_enabled == 1) {
1531 consumer_timer_switch_stop(channel);
1532 }
1533 consumer_metadata_cache_destroy(channel);
1534 }
d3e2ba59
JD
1535 if (channel->live_timer_enabled == 1) {
1536 consumer_timer_live_stop(channel);
1537 }
e9404c27
JG
1538 if (channel->monitor_timer_enabled == 1) {
1539 consumer_timer_monitor_stop(channel);
1540 }
ffe60014
DG
1541 goto end_channel_error;
1542 }
1543
9ce5646a
MD
1544 health_code_update();
1545
ffe60014
DG
1546 /*
1547 * Channel and streams are now created. Inform the session daemon that
1548 * everything went well and should wait to receive the channel and
1549 * streams with ustctl API.
1550 */
1551 ret = consumer_send_status_channel(sock, channel);
1552 if (ret < 0) {
1553 /*
489f70e9 1554 * There is probably a problem on the socket.
ffe60014 1555 */
489f70e9 1556 goto error_fatal;
ffe60014
DG
1557 }
1558
1559 break;
1560 }
1561 case LTTNG_CONSUMER_GET_CHANNEL:
1562 {
1563 int ret, relayd_err = 0;
d88aee68 1564 uint64_t key = msg.u.get_channel.key;
ffe60014 1565 struct lttng_consumer_channel *channel;
ffe60014
DG
1566
1567 channel = consumer_find_channel(key);
1568 if (!channel) {
8fd623e0 1569 ERR("UST consumer get channel key %" PRIu64 " not found", key);
e462382a 1570 ret_code = LTTCOMM_CONSUMERD_CHAN_NOT_FOUND;
ffe60014
DG
1571 goto end_msg_sessiond;
1572 }
1573
9ce5646a
MD
1574 health_code_update();
1575
ffe60014
DG
1576 /* Send everything to sessiond. */
1577 ret = send_sessiond_channel(sock, channel, ctx, &relayd_err);
1578 if (ret < 0) {
1579 if (relayd_err) {
1580 /*
1581 * We were unable to send to the relayd the stream so avoid
1582 * sending back a fatal error to the thread since this is OK
f2a444f1
DG
1583 * and the consumer can continue its work. The above call
1584 * has sent the error status message to the sessiond.
ffe60014 1585 */
f2a444f1 1586 goto end_nosignal;
ffe60014
DG
1587 }
1588 /*
1589 * The communicaton was broken hence there is a bad state between
1590 * the consumer and sessiond so stop everything.
1591 */
1592 goto error_fatal;
1593 }
1594
9ce5646a
MD
1595 health_code_update();
1596
10a50311
JD
1597 /*
1598 * In no monitor mode, the streams ownership is kept inside the channel
1599 * so don't send them to the data thread.
1600 */
1601 if (!channel->monitor) {
1602 goto end_msg_sessiond;
1603 }
1604
d88aee68
DG
1605 ret = send_streams_to_thread(channel, ctx);
1606 if (ret < 0) {
1607 /*
1608 * If we are unable to send the stream to the thread, there is
1609 * a big problem so just stop everything.
1610 */
1611 goto error_fatal;
ffe60014 1612 }
ffe60014
DG
1613 /* List MUST be empty after or else it could be reused. */
1614 assert(cds_list_empty(&channel->streams.head));
d88aee68
DG
1615 goto end_msg_sessiond;
1616 }
1617 case LTTNG_CONSUMER_DESTROY_CHANNEL:
1618 {
1619 uint64_t key = msg.u.destroy_channel.key;
d88aee68 1620
a0cbdd2e
MD
1621 /*
1622 * Only called if streams have not been sent to stream
1623 * manager thread. However, channel has been sent to
1624 * channel manager thread.
1625 */
1626 notify_thread_del_channel(ctx, key);
d88aee68 1627 goto end_msg_sessiond;
ffe60014 1628 }
d88aee68
DG
1629 case LTTNG_CONSUMER_CLOSE_METADATA:
1630 {
1631 int ret;
1632
1633 ret = close_metadata(msg.u.close_metadata.key);
1634 if (ret != 0) {
1635 ret_code = ret;
1636 }
1637
1638 goto end_msg_sessiond;
1639 }
7972aab2
DG
1640 case LTTNG_CONSUMER_FLUSH_CHANNEL:
1641 {
1642 int ret;
1643
1644 ret = flush_channel(msg.u.flush_channel.key);
1645 if (ret != 0) {
1646 ret_code = ret;
1647 }
1648
1649 goto end_msg_sessiond;
1650 }
0dd01979
MD
1651 case LTTNG_CONSUMER_CLEAR_QUIESCENT_CHANNEL:
1652 {
1653 int ret;
1654
1655 ret = clear_quiescent_channel(
1656 msg.u.clear_quiescent_channel.key);
1657 if (ret != 0) {
1658 ret_code = ret;
1659 }
1660
1661 goto end_msg_sessiond;
1662 }
d88aee68 1663 case LTTNG_CONSUMER_PUSH_METADATA:
ffe60014
DG
1664 {
1665 int ret;
d88aee68 1666 uint64_t len = msg.u.push_metadata.len;
d88aee68 1667 uint64_t key = msg.u.push_metadata.key;
331744e3 1668 uint64_t offset = msg.u.push_metadata.target_offset;
93ec662e 1669 uint64_t version = msg.u.push_metadata.version;
ffe60014
DG
1670 struct lttng_consumer_channel *channel;
1671
8fd623e0
DG
1672 DBG("UST consumer push metadata key %" PRIu64 " of len %" PRIu64, key,
1673 len);
ffe60014
DG
1674
1675 channel = consumer_find_channel(key);
1676 if (!channel) {
000baf6a
DG
1677 /*
1678 * This is possible if the metadata creation on the consumer side
1679 * is in flight vis-a-vis a concurrent push metadata from the
1680 * session daemon. Simply return that the channel failed and the
1681 * session daemon will handle that message correctly considering
1682 * that this race is acceptable thus the DBG() statement here.
1683 */
1684 DBG("UST consumer push metadata %" PRIu64 " not found", key);
1685 ret_code = LTTCOMM_CONSUMERD_CHANNEL_FAIL;
4a2eb0ca 1686 goto end_msg_sessiond;
d88aee68
DG
1687 }
1688
9ce5646a
MD
1689 health_code_update();
1690
c585821b
MD
1691 if (!len) {
1692 /*
1693 * There is nothing to receive. We have simply
1694 * checked whether the channel can be found.
1695 */
1696 ret_code = LTTCOMM_CONSUMERD_SUCCESS;
1697 goto end_msg_sessiond;
1698 }
1699
d88aee68 1700 /* Tell session daemon we are ready to receive the metadata. */
0c759fc9 1701 ret = consumer_send_status_msg(sock, LTTCOMM_CONSUMERD_SUCCESS);
ffe60014
DG
1702 if (ret < 0) {
1703 /* Somehow, the session daemon is not responding anymore. */
d88aee68
DG
1704 goto error_fatal;
1705 }
1706
9ce5646a
MD
1707 health_code_update();
1708
d88aee68 1709 /* Wait for more data. */
9ce5646a
MD
1710 health_poll_entry();
1711 ret = lttng_consumer_poll_socket(consumer_sockpoll);
1712 health_poll_exit();
84382d49 1713 if (ret) {
489f70e9 1714 goto error_fatal;
d88aee68
DG
1715 }
1716
9ce5646a
MD
1717 health_code_update();
1718
331744e3 1719 ret = lttng_ustconsumer_recv_metadata(sock, key, offset,
93ec662e 1720 len, version, channel, 0, 1);
d88aee68 1721 if (ret < 0) {
331744e3 1722 /* error receiving from sessiond */
489f70e9 1723 goto error_fatal;
331744e3
JD
1724 } else {
1725 ret_code = ret;
d88aee68
DG
1726 goto end_msg_sessiond;
1727 }
d88aee68
DG
1728 }
1729 case LTTNG_CONSUMER_SETUP_METADATA:
1730 {
1731 int ret;
1732
1733 ret = setup_metadata(ctx, msg.u.setup_metadata.key);
1734 if (ret) {
1735 ret_code = ret;
1736 }
1737 goto end_msg_sessiond;
ffe60014 1738 }
6dc3064a
DG
1739 case LTTNG_CONSUMER_SNAPSHOT_CHANNEL:
1740 {
10a50311
JD
1741 if (msg.u.snapshot_channel.metadata) {
1742 ret = snapshot_metadata(msg.u.snapshot_channel.key,
1743 msg.u.snapshot_channel.pathname,
1744 msg.u.snapshot_channel.relayd_id,
1745 ctx);
1746 if (ret < 0) {
1747 ERR("Snapshot metadata failed");
e462382a 1748 ret_code = LTTCOMM_CONSUMERD_ERROR_METADATA;
10a50311
JD
1749 }
1750 } else {
1751 ret = snapshot_channel(msg.u.snapshot_channel.key,
1752 msg.u.snapshot_channel.pathname,
1753 msg.u.snapshot_channel.relayd_id,
d07ceecd 1754 msg.u.snapshot_channel.nb_packets_per_stream,
10a50311
JD
1755 ctx);
1756 if (ret < 0) {
1757 ERR("Snapshot channel failed");
e462382a 1758 ret_code = LTTCOMM_CONSUMERD_CHANNEL_FAIL;
10a50311
JD
1759 }
1760 }
1761
9ce5646a 1762 health_code_update();
6dc3064a
DG
1763 ret = consumer_send_status_msg(sock, ret_code);
1764 if (ret < 0) {
1765 /* Somehow, the session daemon is not responding anymore. */
1766 goto end_nosignal;
1767 }
9ce5646a 1768 health_code_update();
6dc3064a
DG
1769 break;
1770 }
fb83fe64
JD
1771 case LTTNG_CONSUMER_DISCARDED_EVENTS:
1772 {
beb59458
MJ
1773 int ret = 0;
1774 uint64_t discarded_events;
fb83fe64
JD
1775 struct lttng_ht_iter iter;
1776 struct lttng_ht *ht;
1777 struct lttng_consumer_stream *stream;
1778 uint64_t id = msg.u.discarded_events.session_id;
1779 uint64_t key = msg.u.discarded_events.channel_key;
1780
1781 DBG("UST consumer discarded events command for session id %"
1782 PRIu64, id);
1783 rcu_read_lock();
1784 pthread_mutex_lock(&consumer_data.lock);
1785
1786 ht = consumer_data.stream_list_ht;
1787
1788 /*
1789 * We only need a reference to the channel, but they are not
1790 * directly indexed, so we just use the first matching stream
1791 * to extract the information we need, we default to 0 if not
1792 * found (no events are dropped if the channel is not yet in
1793 * use).
1794 */
beb59458 1795 discarded_events = 0;
fb83fe64
JD
1796 cds_lfht_for_each_entry_duplicate(ht->ht,
1797 ht->hash_fct(&id, lttng_ht_seed),
1798 ht->match_fct, &id,
1799 &iter.iter, stream, node_session_id.node) {
1800 if (stream->chan->key == key) {
beb59458 1801 discarded_events = stream->chan->discarded_events;
fb83fe64
JD
1802 break;
1803 }
1804 }
1805 pthread_mutex_unlock(&consumer_data.lock);
1806 rcu_read_unlock();
1807
1808 DBG("UST consumer discarded events command for session id %"
1809 PRIu64 ", channel key %" PRIu64, id, key);
1810
1811 health_code_update();
1812
1813 /* Send back returned value to session daemon */
beb59458 1814 ret = lttcomm_send_unix_sock(sock, &discarded_events, sizeof(discarded_events));
fb83fe64
JD
1815 if (ret < 0) {
1816 PERROR("send discarded events");
1817 goto error_fatal;
1818 }
1819
1820 break;
1821 }
1822 case LTTNG_CONSUMER_LOST_PACKETS:
1823 {
9a06e8d4
JG
1824 int ret;
1825 uint64_t lost_packets;
fb83fe64
JD
1826 struct lttng_ht_iter iter;
1827 struct lttng_ht *ht;
1828 struct lttng_consumer_stream *stream;
1829 uint64_t id = msg.u.lost_packets.session_id;
1830 uint64_t key = msg.u.lost_packets.channel_key;
1831
1832 DBG("UST consumer lost packets command for session id %"
1833 PRIu64, id);
1834 rcu_read_lock();
1835 pthread_mutex_lock(&consumer_data.lock);
1836
1837 ht = consumer_data.stream_list_ht;
1838
1839 /*
1840 * We only need a reference to the channel, but they are not
1841 * directly indexed, so we just use the first matching stream
1842 * to extract the information we need, we default to 0 if not
1843 * found (no packets lost if the channel is not yet in use).
1844 */
9a06e8d4 1845 lost_packets = 0;
fb83fe64
JD
1846 cds_lfht_for_each_entry_duplicate(ht->ht,
1847 ht->hash_fct(&id, lttng_ht_seed),
1848 ht->match_fct, &id,
1849 &iter.iter, stream, node_session_id.node) {
1850 if (stream->chan->key == key) {
9a06e8d4 1851 lost_packets = stream->chan->lost_packets;
fb83fe64
JD
1852 break;
1853 }
1854 }
1855 pthread_mutex_unlock(&consumer_data.lock);
1856 rcu_read_unlock();
1857
1858 DBG("UST consumer lost packets command for session id %"
1859 PRIu64 ", channel key %" PRIu64, id, key);
1860
1861 health_code_update();
1862
1863 /* Send back returned value to session daemon */
9a06e8d4
JG
1864 ret = lttcomm_send_unix_sock(sock, &lost_packets,
1865 sizeof(lost_packets));
fb83fe64
JD
1866 if (ret < 0) {
1867 PERROR("send lost packets");
1868 goto error_fatal;
1869 }
1870
1871 break;
1872 }
b3530820
JG
1873 case LTTNG_CONSUMER_SET_CHANNEL_MONITOR_PIPE:
1874 {
1875 int channel_monitor_pipe;
1876
1877 ret_code = LTTCOMM_CONSUMERD_SUCCESS;
1878 /* Successfully received the command's type. */
1879 ret = consumer_send_status_msg(sock, ret_code);
1880 if (ret < 0) {
1881 goto error_fatal;
1882 }
1883
1884 ret = lttcomm_recv_fds_unix_sock(sock, &channel_monitor_pipe,
1885 1);
1886 if (ret != sizeof(channel_monitor_pipe)) {
1887 ERR("Failed to receive channel monitor pipe");
1888 goto error_fatal;
1889 }
1890
1891 DBG("Received channel monitor pipe (%d)", channel_monitor_pipe);
1892 ret = consumer_timer_thread_set_channel_monitor_pipe(
1893 channel_monitor_pipe);
1894 if (!ret) {
1895 int flags;
1896
1897 ret_code = LTTCOMM_CONSUMERD_SUCCESS;
1898 /* Set the pipe as non-blocking. */
1899 ret = fcntl(channel_monitor_pipe, F_GETFL, 0);
1900 if (ret == -1) {
1901 PERROR("fcntl get flags of the channel monitoring pipe");
1902 goto error_fatal;
1903 }
1904 flags = ret;
1905
1906 ret = fcntl(channel_monitor_pipe, F_SETFL,
1907 flags | O_NONBLOCK);
1908 if (ret == -1) {
1909 PERROR("fcntl set O_NONBLOCK flag of the channel monitoring pipe");
1910 goto error_fatal;
1911 }
1912 DBG("Channel monitor pipe set as non-blocking");
1913 } else {
1914 ret_code = LTTCOMM_CONSUMERD_ALREADY_SET;
1915 }
1916 goto end_msg_sessiond;
1917 }
3bd1e081
MD
1918 default:
1919 break;
1920 }
3f8e211f 1921
3bd1e081 1922end_nosignal:
b0b335c8 1923 rcu_read_unlock();
4cbc1a04 1924
9ce5646a
MD
1925 health_code_update();
1926
4cbc1a04
DG
1927 /*
1928 * Return 1 to indicate success since the 0 value can be a socket
1929 * shutdown during the recv() or send() call.
1930 */
1931 return 1;
ffe60014
DG
1932
1933end_msg_sessiond:
1934 /*
1935 * The returned value here is not useful since either way we'll return 1 to
1936 * the caller because the session daemon socket management is done
1937 * elsewhere. Returning a negative code or 0 will shutdown the consumer.
1938 */
489f70e9
MD
1939 ret = consumer_send_status_msg(sock, ret_code);
1940 if (ret < 0) {
1941 goto error_fatal;
1942 }
ffe60014 1943 rcu_read_unlock();
9ce5646a
MD
1944
1945 health_code_update();
1946
ffe60014
DG
1947 return 1;
1948end_channel_error:
1949 if (channel) {
1950 /*
1951 * Free channel here since no one has a reference to it. We don't
1952 * free after that because a stream can store this pointer.
1953 */
1954 destroy_channel(channel);
1955 }
1956 /* We have to send a status channel message indicating an error. */
1957 ret = consumer_send_status_channel(sock, NULL);
1958 if (ret < 0) {
1959 /* Stop everything if session daemon can not be notified. */
1960 goto error_fatal;
1961 }
1962 rcu_read_unlock();
9ce5646a
MD
1963
1964 health_code_update();
1965
ffe60014
DG
1966 return 1;
1967error_fatal:
1968 rcu_read_unlock();
1969 /* This will issue a consumer stop. */
1970 return -1;
3bd1e081
MD
1971}
1972
ffe60014
DG
1973/*
1974 * Wrapper over the mmap() read offset from ust-ctl library. Since this can be
1975 * compiled out, we isolate it in this library.
1976 */
1977int lttng_ustctl_get_mmap_read_offset(struct lttng_consumer_stream *stream,
1978 unsigned long *off)
3bd1e081 1979{
ffe60014
DG
1980 assert(stream);
1981 assert(stream->ustream);
b5c5fc29 1982
ffe60014 1983 return ustctl_get_mmap_read_offset(stream->ustream, off);
3bd1e081
MD
1984}
1985
ffe60014
DG
1986/*
1987 * Wrapper over the mmap() read offset from ust-ctl library. Since this can be
1988 * compiled out, we isolate it in this library.
1989 */
1990void *lttng_ustctl_get_mmap_base(struct lttng_consumer_stream *stream)
d056b477 1991{
ffe60014
DG
1992 assert(stream);
1993 assert(stream->ustream);
1994
1995 return ustctl_get_mmap_base(stream->ustream);
d056b477
MD
1996}
1997
ffe60014 1998/*
e9404c27 1999 * Take a snapshot for a specific stream.
ffe60014
DG
2000 *
2001 * Returns 0 on success, < 0 on error
2002 */
2003int lttng_ustconsumer_take_snapshot(struct lttng_consumer_stream *stream)
3bd1e081 2004{
ffe60014
DG
2005 assert(stream);
2006 assert(stream->ustream);
2007
2008 return ustctl_snapshot(stream->ustream);
3bd1e081
MD
2009}
2010
e9404c27
JG
2011/*
2012 * Sample consumed and produced positions for a specific stream.
2013 *
2014 * Returns 0 on success, < 0 on error.
2015 */
2016int lttng_ustconsumer_sample_snapshot_positions(
2017 struct lttng_consumer_stream *stream)
2018{
2019 assert(stream);
2020 assert(stream->ustream);
2021
2022 return ustctl_snapshot_sample_positions(stream->ustream);
2023}
2024
ffe60014
DG
2025/*
2026 * Get the produced position
2027 *
2028 * Returns 0 on success, < 0 on error
2029 */
2030int lttng_ustconsumer_get_produced_snapshot(
2031 struct lttng_consumer_stream *stream, unsigned long *pos)
3bd1e081 2032{
ffe60014
DG
2033 assert(stream);
2034 assert(stream->ustream);
2035 assert(pos);
7a57cf92 2036
ffe60014
DG
2037 return ustctl_snapshot_get_produced(stream->ustream, pos);
2038}
7a57cf92 2039
10a50311
JD
2040/*
2041 * Get the consumed position
2042 *
2043 * Returns 0 on success, < 0 on error
2044 */
2045int lttng_ustconsumer_get_consumed_snapshot(
2046 struct lttng_consumer_stream *stream, unsigned long *pos)
2047{
2048 assert(stream);
2049 assert(stream->ustream);
2050 assert(pos);
2051
2052 return ustctl_snapshot_get_consumed(stream->ustream, pos);
2053}
2054
84a182ce
DG
2055void lttng_ustconsumer_flush_buffer(struct lttng_consumer_stream *stream,
2056 int producer)
2057{
2058 assert(stream);
2059 assert(stream->ustream);
2060
2061 ustctl_flush_buffer(stream->ustream, producer);
2062}
2063
2064int lttng_ustconsumer_get_current_timestamp(
2065 struct lttng_consumer_stream *stream, uint64_t *ts)
2066{
2067 assert(stream);
2068 assert(stream->ustream);
2069 assert(ts);
2070
2071 return ustctl_get_current_timestamp(stream->ustream, ts);
2072}
2073
fb83fe64
JD
2074int lttng_ustconsumer_get_sequence_number(
2075 struct lttng_consumer_stream *stream, uint64_t *seq)
2076{
2077 assert(stream);
2078 assert(stream->ustream);
2079 assert(seq);
2080
2081 return ustctl_get_sequence_number(stream->ustream, seq);
2082}
2083
ffe60014 2084/*
0dd01979 2085 * Called when the stream signals the consumer that it has hung up.
ffe60014
DG
2086 */
2087void lttng_ustconsumer_on_stream_hangup(struct lttng_consumer_stream *stream)
2088{
2089 assert(stream);
2090 assert(stream->ustream);
2c1dd183 2091
0dd01979
MD
2092 pthread_mutex_lock(&stream->lock);
2093 if (!stream->quiescent) {
2094 ustctl_flush_buffer(stream->ustream, 0);
2095 stream->quiescent = true;
2096 }
2097 pthread_mutex_unlock(&stream->lock);
ffe60014
DG
2098 stream->hangup_flush_done = 1;
2099}
ee77a7b0 2100
ffe60014
DG
2101void lttng_ustconsumer_del_channel(struct lttng_consumer_channel *chan)
2102{
4628484a
MD
2103 int i;
2104
ffe60014
DG
2105 assert(chan);
2106 assert(chan->uchan);
e316aad5 2107
ea88ca2a
MD
2108 if (chan->switch_timer_enabled == 1) {
2109 consumer_timer_switch_stop(chan);
2110 }
4628484a
MD
2111 for (i = 0; i < chan->nr_stream_fds; i++) {
2112 int ret;
2113
2114 ret = close(chan->stream_fds[i]);
2115 if (ret) {
2116 PERROR("close");
2117 }
2118 if (chan->shm_path[0]) {
2119 char shm_path[PATH_MAX];
2120
2121 ret = get_stream_shm_path(shm_path, chan->shm_path, i);
2122 if (ret) {
2123 ERR("Cannot get stream shm path");
2124 }
2125 ret = run_as_unlink(shm_path, chan->uid, chan->gid);
2126 if (ret) {
4628484a
MD
2127 PERROR("unlink %s", shm_path);
2128 }
2129 }
2130 }
3bd1e081
MD
2131}
2132
b83e03c4
MD
2133void lttng_ustconsumer_free_channel(struct lttng_consumer_channel *chan)
2134{
2135 assert(chan);
2136 assert(chan->uchan);
2137
2138 consumer_metadata_cache_destroy(chan);
2139 ustctl_destroy_channel(chan->uchan);
ea853771
JR
2140 /* Try to rmdir all directories under shm_path root. */
2141 if (chan->root_shm_path[0]) {
602766ec 2142 (void) run_as_rmdir_recursive(chan->root_shm_path,
ea853771
JR
2143 chan->uid, chan->gid);
2144 }
b83e03c4
MD
2145 free(chan->stream_fds);
2146}
2147
3bd1e081
MD
2148void lttng_ustconsumer_del_stream(struct lttng_consumer_stream *stream)
2149{
ffe60014
DG
2150 assert(stream);
2151 assert(stream->ustream);
d41f73b7 2152
ea88ca2a
MD
2153 if (stream->chan->switch_timer_enabled == 1) {
2154 consumer_timer_switch_stop(stream->chan);
2155 }
ffe60014
DG
2156 ustctl_destroy_stream(stream->ustream);
2157}
d41f73b7 2158
6d574024
DG
2159int lttng_ustconsumer_get_wakeup_fd(struct lttng_consumer_stream *stream)
2160{
2161 assert(stream);
2162 assert(stream->ustream);
2163
2164 return ustctl_stream_get_wakeup_fd(stream->ustream);
2165}
2166
2167int lttng_ustconsumer_close_wakeup_fd(struct lttng_consumer_stream *stream)
2168{
2169 assert(stream);
2170 assert(stream->ustream);
2171
2172 return ustctl_stream_close_wakeup_fd(stream->ustream);
2173}
2174
309167d2
JD
2175/*
2176 * Populate index values of a UST stream. Values are set in big endian order.
2177 *
2178 * Return 0 on success or else a negative value.
2179 */
50adc264 2180static int get_index_values(struct ctf_packet_index *index,
309167d2
JD
2181 struct ustctl_consumer_stream *ustream)
2182{
2183 int ret;
2184
2185 ret = ustctl_get_timestamp_begin(ustream, &index->timestamp_begin);
2186 if (ret < 0) {
2187 PERROR("ustctl_get_timestamp_begin");
2188 goto error;
2189 }
2190 index->timestamp_begin = htobe64(index->timestamp_begin);
2191
2192 ret = ustctl_get_timestamp_end(ustream, &index->timestamp_end);
2193 if (ret < 0) {
2194 PERROR("ustctl_get_timestamp_end");
2195 goto error;
2196 }
2197 index->timestamp_end = htobe64(index->timestamp_end);
2198
2199 ret = ustctl_get_events_discarded(ustream, &index->events_discarded);
2200 if (ret < 0) {
2201 PERROR("ustctl_get_events_discarded");
2202 goto error;
2203 }
2204 index->events_discarded = htobe64(index->events_discarded);
2205
2206 ret = ustctl_get_content_size(ustream, &index->content_size);
2207 if (ret < 0) {
2208 PERROR("ustctl_get_content_size");
2209 goto error;
2210 }
2211 index->content_size = htobe64(index->content_size);
2212
2213 ret = ustctl_get_packet_size(ustream, &index->packet_size);
2214 if (ret < 0) {
2215 PERROR("ustctl_get_packet_size");
2216 goto error;
2217 }
2218 index->packet_size = htobe64(index->packet_size);
2219
2220 ret = ustctl_get_stream_id(ustream, &index->stream_id);
2221 if (ret < 0) {
2222 PERROR("ustctl_get_stream_id");
2223 goto error;
2224 }
2225 index->stream_id = htobe64(index->stream_id);
2226
234cd636
JD
2227 ret = ustctl_get_instance_id(ustream, &index->stream_instance_id);
2228 if (ret < 0) {
2229 PERROR("ustctl_get_instance_id");
2230 goto error;
2231 }
2232 index->stream_instance_id = htobe64(index->stream_instance_id);
2233
2234 ret = ustctl_get_sequence_number(ustream, &index->packet_seq_num);
2235 if (ret < 0) {
2236 PERROR("ustctl_get_sequence_number");
2237 goto error;
2238 }
2239 index->packet_seq_num = htobe64(index->packet_seq_num);
2240
309167d2
JD
2241error:
2242 return ret;
2243}
2244
93ec662e
JD
2245static
2246void metadata_stream_reset_cache(struct lttng_consumer_stream *stream,
2247 struct consumer_metadata_cache *cache)
2248{
2249 DBG("Metadata stream update to version %" PRIu64,
2250 cache->version);
2251 stream->ust_metadata_pushed = 0;
2252 stream->metadata_version = cache->version;
2253 stream->reset_metadata_flag = 1;
2254}
2255
2256/*
2257 * Check if the version of the metadata stream and metadata cache match.
2258 * If the cache got updated, reset the metadata stream.
2259 * The stream lock and metadata cache lock MUST be held.
2260 * Return 0 on success, a negative value on error.
2261 */
2262static
2263int metadata_stream_check_version(struct lttng_consumer_stream *stream)
2264{
2265 int ret = 0;
2266 struct consumer_metadata_cache *cache = stream->chan->metadata_cache;
2267
2268 if (cache->version == stream->metadata_version) {
2269 goto end;
2270 }
2271 metadata_stream_reset_cache(stream, cache);
2272
2273end:
2274 return ret;
2275}
2276
94d49140
JD
2277/*
2278 * Write up to one packet from the metadata cache to the channel.
2279 *
2280 * Returns the number of bytes pushed in the cache, or a negative value
2281 * on error.
2282 */
2283static
2284int commit_one_metadata_packet(struct lttng_consumer_stream *stream)
2285{
2286 ssize_t write_len;
2287 int ret;
2288
2289 pthread_mutex_lock(&stream->chan->metadata_cache->lock);
93ec662e
JD
2290 ret = metadata_stream_check_version(stream);
2291 if (ret < 0) {
2292 goto end;
2293 }
c585821b 2294 if (stream->chan->metadata_cache->max_offset
94d49140
JD
2295 == stream->ust_metadata_pushed) {
2296 ret = 0;
2297 goto end;
2298 }
2299
2300 write_len = ustctl_write_one_packet_to_channel(stream->chan->uchan,
2301 &stream->chan->metadata_cache->data[stream->ust_metadata_pushed],
c585821b 2302 stream->chan->metadata_cache->max_offset
94d49140
JD
2303 - stream->ust_metadata_pushed);
2304 assert(write_len != 0);
2305 if (write_len < 0) {
2306 ERR("Writing one metadata packet");
2307 ret = -1;
2308 goto end;
2309 }
2310 stream->ust_metadata_pushed += write_len;
2311
c585821b 2312 assert(stream->chan->metadata_cache->max_offset >=
94d49140
JD
2313 stream->ust_metadata_pushed);
2314 ret = write_len;
2315
2316end:
2317 pthread_mutex_unlock(&stream->chan->metadata_cache->lock);
2318 return ret;
2319}
2320
309167d2 2321
94d49140
JD
2322/*
2323 * Sync metadata meaning request them to the session daemon and snapshot to the
2324 * metadata thread can consumer them.
2325 *
c585821b
MD
2326 * Metadata stream lock is held here, but we need to release it when
2327 * interacting with sessiond, else we cause a deadlock with live
2328 * awaiting on metadata to be pushed out.
94d49140
JD
2329 *
2330 * Return 0 if new metadatda is available, EAGAIN if the metadata stream
2331 * is empty or a negative value on error.
2332 */
2333int lttng_ustconsumer_sync_metadata(struct lttng_consumer_local_data *ctx,
2334 struct lttng_consumer_stream *metadata)
2335{
2336 int ret;
2337 int retry = 0;
2338
2339 assert(ctx);
2340 assert(metadata);
2341
c585821b 2342 pthread_mutex_unlock(&metadata->lock);
94d49140
JD
2343 /*
2344 * Request metadata from the sessiond, but don't wait for the flush
2345 * because we locked the metadata thread.
2346 */
2347 ret = lttng_ustconsumer_request_metadata(ctx, metadata->chan, 0, 0);
1caeb2eb 2348 pthread_mutex_lock(&metadata->lock);
94d49140
JD
2349 if (ret < 0) {
2350 goto end;
2351 }
2352
2353 ret = commit_one_metadata_packet(metadata);
2354 if (ret <= 0) {
2355 goto end;
2356 } else if (ret > 0) {
2357 retry = 1;
2358 }
2359
2360 ustctl_flush_buffer(metadata->ustream, 1);
2361 ret = ustctl_snapshot(metadata->ustream);
2362 if (ret < 0) {
2363 if (errno != EAGAIN) {
2364 ERR("Sync metadata, taking UST snapshot");
2365 goto end;
2366 }
2367 DBG("No new metadata when syncing them.");
2368 /* No new metadata, exit. */
2369 ret = ENODATA;
2370 goto end;
2371 }
2372
2373 /*
2374 * After this flush, we still need to extract metadata.
2375 */
2376 if (retry) {
2377 ret = EAGAIN;
2378 }
2379
2380end:
2381 return ret;
2382}
2383
02b3d176
DG
2384/*
2385 * Return 0 on success else a negative value.
2386 */
2387static int notify_if_more_data(struct lttng_consumer_stream *stream,
2388 struct lttng_consumer_local_data *ctx)
2389{
2390 int ret;
2391 struct ustctl_consumer_stream *ustream;
2392
2393 assert(stream);
2394 assert(ctx);
2395
2396 ustream = stream->ustream;
2397
2398 /*
2399 * First, we are going to check if there is a new subbuffer available
2400 * before reading the stream wait_fd.
2401 */
2402 /* Get the next subbuffer */
2403 ret = ustctl_get_next_subbuf(ustream);
2404 if (ret) {
2405 /* No more data found, flag the stream. */
2406 stream->has_data = 0;
2407 ret = 0;
2408 goto end;
2409 }
2410
5420e5db 2411 ret = ustctl_put_subbuf(ustream);
02b3d176
DG
2412 assert(!ret);
2413
2414 /* This stream still has data. Flag it and wake up the data thread. */
2415 stream->has_data = 1;
2416
2417 if (stream->monitor && !stream->hangup_flush_done && !ctx->has_wakeup) {
2418 ssize_t writelen;
2419
2420 writelen = lttng_pipe_write(ctx->consumer_wakeup_pipe, "!", 1);
2421 if (writelen < 0 && errno != EAGAIN && errno != EWOULDBLOCK) {
2422 ret = writelen;
2423 goto end;
2424 }
2425
2426 /* The wake up pipe has been notified. */
2427 ctx->has_wakeup = 1;
2428 }
2429 ret = 0;
2430
2431end:
2432 return ret;
2433}
2434
fb83fe64
JD
2435static
2436int update_stream_stats(struct lttng_consumer_stream *stream)
2437{
2438 int ret;
2439 uint64_t seq, discarded;
2440
2441 ret = ustctl_get_sequence_number(stream->ustream, &seq);
2442 if (ret < 0) {
2443 PERROR("ustctl_get_sequence_number");
2444 goto end;
2445 }
2446 /*
2447 * Start the sequence when we extract the first packet in case we don't
2448 * start at 0 (for example if a consumer is not connected to the
2449 * session immediately after the beginning).
2450 */
2451 if (stream->last_sequence_number == -1ULL) {
2452 stream->last_sequence_number = seq;
2453 } else if (seq > stream->last_sequence_number) {
2454 stream->chan->lost_packets += seq -
2455 stream->last_sequence_number - 1;
2456 } else {
2457 /* seq <= last_sequence_number */
2458 ERR("Sequence number inconsistent : prev = %" PRIu64
2459 ", current = %" PRIu64,
2460 stream->last_sequence_number, seq);
2461 ret = -1;
2462 goto end;
2463 }
2464 stream->last_sequence_number = seq;
2465
2466 ret = ustctl_get_events_discarded(stream->ustream, &discarded);
2467 if (ret < 0) {
2468 PERROR("kernctl_get_events_discarded");
2469 goto end;
2470 }
2471 if (discarded < stream->last_discarded_events) {
2472 /*
83f4233d
MJ
2473 * Overflow has occurred. We assume only one wrap-around
2474 * has occurred.
fb83fe64
JD
2475 */
2476 stream->chan->discarded_events +=
2477 (1ULL << (CAA_BITS_PER_LONG - 1)) -
2478 stream->last_discarded_events + discarded;
2479 } else {
2480 stream->chan->discarded_events += discarded -
2481 stream->last_discarded_events;
2482 }
2483 stream->last_discarded_events = discarded;
2484 ret = 0;
2485
2486end:
2487 return ret;
2488}
2489
94d49140
JD
2490/*
2491 * Read subbuffer from the given stream.
2492 *
2493 * Stream lock MUST be acquired.
2494 *
2495 * Return 0 on success else a negative value.
2496 */
d41f73b7
MD
2497int lttng_ustconsumer_read_subbuffer(struct lttng_consumer_stream *stream,
2498 struct lttng_consumer_local_data *ctx)
2499{
1d4dfdef 2500 unsigned long len, subbuf_size, padding;
1c20f0e2 2501 int err, write_index = 1;
d41f73b7 2502 long ret = 0;
ffe60014 2503 struct ustctl_consumer_stream *ustream;
50adc264 2504 struct ctf_packet_index index;
ffe60014
DG
2505
2506 assert(stream);
2507 assert(stream->ustream);
2508 assert(ctx);
d41f73b7 2509
3eb914c0 2510 DBG("In UST read_subbuffer (wait_fd: %d, name: %s)", stream->wait_fd,
ffe60014
DG
2511 stream->name);
2512
2513 /* Ease our life for what's next. */
2514 ustream = stream->ustream;
d41f73b7 2515
6cd525e8 2516 /*
02b3d176
DG
2517 * We can consume the 1 byte written into the wait_fd by UST. Don't trigger
2518 * error if we cannot read this one byte (read returns 0), or if the error
2519 * is EAGAIN or EWOULDBLOCK.
2520 *
2521 * This is only done when the stream is monitored by a thread, before the
2522 * flush is done after a hangup and if the stream is not flagged with data
2523 * since there might be nothing to consume in the wait fd but still have
2524 * data available flagged by the consumer wake up pipe.
6cd525e8 2525 */
02b3d176
DG
2526 if (stream->monitor && !stream->hangup_flush_done && !stream->has_data) {
2527 char dummy;
c617c0c6
MD
2528 ssize_t readlen;
2529
6cd525e8
MD
2530 readlen = lttng_read(stream->wait_fd, &dummy, 1);
2531 if (readlen < 0 && errno != EAGAIN && errno != EWOULDBLOCK) {
effcf122
MD
2532 ret = readlen;
2533 goto end;
2534 }
d41f73b7
MD
2535 }
2536
04ef1097 2537retry:
d41f73b7 2538 /* Get the next subbuffer */
ffe60014 2539 err = ustctl_get_next_subbuf(ustream);
d41f73b7 2540 if (err != 0) {
04ef1097
MD
2541 /*
2542 * Populate metadata info if the existing info has
2543 * already been read.
2544 */
2545 if (stream->metadata_flag) {
94d49140
JD
2546 ret = commit_one_metadata_packet(stream);
2547 if (ret <= 0) {
04ef1097
MD
2548 goto end;
2549 }
04ef1097
MD
2550 ustctl_flush_buffer(stream->ustream, 1);
2551 goto retry;
2552 }
2553
1d4dfdef 2554 ret = err; /* ustctl_get_next_subbuf returns negative, caller expect positive. */
d41f73b7
MD
2555 /*
2556 * This is a debug message even for single-threaded consumer,
2557 * because poll() have more relaxed criterions than get subbuf,
2558 * so get_subbuf may fail for short race windows where poll()
2559 * would issue wakeups.
2560 */
2561 DBG("Reserving sub buffer failed (everything is normal, "
ffe60014 2562 "it is due to concurrency) [ret: %d]", err);
d41f73b7
MD
2563 goto end;
2564 }
ffe60014 2565 assert(stream->chan->output == CONSUMER_CHANNEL_MMAP);
309167d2 2566
1c20f0e2 2567 if (!stream->metadata_flag) {
309167d2
JD
2568 index.offset = htobe64(stream->out_fd_offset);
2569 ret = get_index_values(&index, ustream);
2570 if (ret < 0) {
7b87473d
MD
2571 err = ustctl_put_subbuf(ustream);
2572 assert(err == 0);
309167d2
JD
2573 goto end;
2574 }
fb83fe64
JD
2575
2576 /* Update the stream's sequence and discarded events count. */
2577 ret = update_stream_stats(stream);
2578 if (ret < 0) {
2579 PERROR("kernctl_get_events_discarded");
7b87473d
MD
2580 err = ustctl_put_subbuf(ustream);
2581 assert(err == 0);
fb83fe64
JD
2582 goto end;
2583 }
1c20f0e2
JD
2584 } else {
2585 write_index = 0;
309167d2
JD
2586 }
2587
1d4dfdef 2588 /* Get the full padded subbuffer size */
ffe60014 2589 err = ustctl_get_padded_subbuf_size(ustream, &len);
effcf122 2590 assert(err == 0);
1d4dfdef
DG
2591
2592 /* Get subbuffer data size (without padding) */
ffe60014 2593 err = ustctl_get_subbuf_size(ustream, &subbuf_size);
1d4dfdef
DG
2594 assert(err == 0);
2595
2596 /* Make sure we don't get a subbuffer size bigger than the padded */
2597 assert(len >= subbuf_size);
2598
2599 padding = len - subbuf_size;
d41f73b7 2600 /* write the subbuffer to the tracefile */
309167d2 2601 ret = lttng_consumer_on_read_subbuffer_mmap(ctx, stream, subbuf_size, padding, &index);
91dfef6e
DG
2602 /*
2603 * The mmap operation should write subbuf_size amount of data when network
2604 * streaming or the full padding (len) size when we are _not_ streaming.
2605 */
d88aee68
DG
2606 if ((ret != subbuf_size && stream->net_seq_idx != (uint64_t) -1ULL) ||
2607 (ret != len && stream->net_seq_idx == (uint64_t) -1ULL)) {
d41f73b7 2608 /*
91dfef6e 2609 * Display the error but continue processing to try to release the
c5c45efa
DG
2610 * subbuffer. This is a DBG statement since any unexpected kill or
2611 * signal, the application gets unregistered, relayd gets closed or
2612 * anything that affects the buffer lifetime will trigger this error.
2613 * So, for the sake of the user, don't print this error since it can
2614 * happen and it is OK with the code flow.
d41f73b7 2615 */
c5c45efa 2616 DBG("Error writing to tracefile "
8fd623e0 2617 "(ret: %ld != len: %lu != subbuf_size: %lu)",
91dfef6e 2618 ret, len, subbuf_size);
309167d2 2619 write_index = 0;
d41f73b7 2620 }
ffe60014 2621 err = ustctl_put_next_subbuf(ustream);
effcf122 2622 assert(err == 0);
331744e3 2623
02b3d176
DG
2624 /*
2625 * This will consumer the byte on the wait_fd if and only if there is not
2626 * next subbuffer to be acquired.
2627 */
2628 if (!stream->metadata_flag) {
2629 ret = notify_if_more_data(stream, ctx);
2630 if (ret < 0) {
2631 goto end;
2632 }
2633 }
2634
309167d2 2635 /* Write index if needed. */
1c20f0e2
JD
2636 if (!write_index) {
2637 goto end;
2638 }
2639
94d49140
JD
2640 if (stream->chan->live_timer_interval && !stream->metadata_flag) {
2641 /*
2642 * In live, block until all the metadata is sent.
2643 */
c585821b
MD
2644 pthread_mutex_lock(&stream->metadata_timer_lock);
2645 assert(!stream->missed_metadata_flush);
2646 stream->waiting_on_metadata = true;
2647 pthread_mutex_unlock(&stream->metadata_timer_lock);
2648
94d49140 2649 err = consumer_stream_sync_metadata(ctx, stream->session_id);
c585821b
MD
2650
2651 pthread_mutex_lock(&stream->metadata_timer_lock);
2652 stream->waiting_on_metadata = false;
2653 if (stream->missed_metadata_flush) {
2654 stream->missed_metadata_flush = false;
2655 pthread_mutex_unlock(&stream->metadata_timer_lock);
2656 (void) consumer_flush_ust_index(stream);
2657 } else {
2658 pthread_mutex_unlock(&stream->metadata_timer_lock);
2659 }
2660
94d49140
JD
2661 if (err < 0) {
2662 goto end;
2663 }
2664 }
2665
1c20f0e2
JD
2666 assert(!stream->metadata_flag);
2667 err = consumer_stream_write_index(stream, &index);
2668 if (err < 0) {
2669 goto end;
309167d2
JD
2670 }
2671
d41f73b7
MD
2672end:
2673 return ret;
2674}
2675
ffe60014
DG
2676/*
2677 * Called when a stream is created.
fe4477ee
JD
2678 *
2679 * Return 0 on success or else a negative value.
ffe60014 2680 */
d41f73b7
MD
2681int lttng_ustconsumer_on_recv_stream(struct lttng_consumer_stream *stream)
2682{
fe4477ee
JD
2683 int ret;
2684
10a50311
JD
2685 assert(stream);
2686
fe4477ee 2687 /* Don't create anything if this is set for streaming. */
10a50311 2688 if (stream->net_seq_idx == (uint64_t) -1ULL && stream->chan->monitor) {
fe4477ee
JD
2689 ret = utils_create_stream_file(stream->chan->pathname, stream->name,
2690 stream->chan->tracefile_size, stream->tracefile_count_current,
309167d2 2691 stream->uid, stream->gid, NULL);
fe4477ee
JD
2692 if (ret < 0) {
2693 goto error;
2694 }
2695 stream->out_fd = ret;
2696 stream->tracefile_size_current = 0;
309167d2
JD
2697
2698 if (!stream->metadata_flag) {
f8f3885c
MD
2699 struct lttng_index_file *index_file;
2700
2701 index_file = lttng_index_file_create(stream->chan->pathname,
309167d2
JD
2702 stream->name, stream->uid, stream->gid,
2703 stream->chan->tracefile_size,
f8f3885c
MD
2704 stream->tracefile_count_current,
2705 CTF_INDEX_MAJOR, CTF_INDEX_MINOR);
2706 if (!index_file) {
309167d2
JD
2707 goto error;
2708 }
1b47ae58 2709 assert(!stream->index_file);
f8f3885c 2710 stream->index_file = index_file;
309167d2 2711 }
fe4477ee
JD
2712 }
2713 ret = 0;
2714
2715error:
2716 return ret;
d41f73b7 2717}
ca22feea
DG
2718
2719/*
2720 * Check if data is still being extracted from the buffers for a specific
4e9a4686
DG
2721 * stream. Consumer data lock MUST be acquired before calling this function
2722 * and the stream lock.
ca22feea 2723 *
6d805429 2724 * Return 1 if the traced data are still getting read else 0 meaning that the
ca22feea
DG
2725 * data is available for trace viewer reading.
2726 */
6d805429 2727int lttng_ustconsumer_data_pending(struct lttng_consumer_stream *stream)
ca22feea
DG
2728{
2729 int ret;
2730
2731 assert(stream);
ffe60014 2732 assert(stream->ustream);
ca22feea 2733
6d805429 2734 DBG("UST consumer checking data pending");
c8f59ee5 2735
ca6b395f
MD
2736 if (stream->endpoint_status != CONSUMER_ENDPOINT_ACTIVE) {
2737 ret = 0;
2738 goto end;
2739 }
2740
04ef1097 2741 if (stream->chan->type == CONSUMER_CHANNEL_TYPE_METADATA) {
e6ee4eab
DG
2742 uint64_t contiguous, pushed;
2743
2744 /* Ease our life a bit. */
c585821b 2745 contiguous = stream->chan->metadata_cache->max_offset;
e6ee4eab
DG
2746 pushed = stream->ust_metadata_pushed;
2747
04ef1097
MD
2748 /*
2749 * We can simply check whether all contiguously available data
2750 * has been pushed to the ring buffer, since the push operation
2751 * is performed within get_next_subbuf(), and because both
2752 * get_next_subbuf() and put_next_subbuf() are issued atomically
2753 * thanks to the stream lock within
2754 * lttng_ustconsumer_read_subbuffer(). This basically means that
2755 * whetnever ust_metadata_pushed is incremented, the associated
2756 * metadata has been consumed from the metadata stream.
2757 */
2758 DBG("UST consumer metadata pending check: contiguous %" PRIu64 " vs pushed %" PRIu64,
e6ee4eab 2759 contiguous, pushed);
aa01b94c 2760 assert(((int64_t) (contiguous - pushed)) >= 0);
e6ee4eab 2761 if ((contiguous != pushed) ||
6acdf328 2762 (((int64_t) contiguous - pushed) > 0 || contiguous == 0)) {
04ef1097
MD
2763 ret = 1; /* Data is pending */
2764 goto end;
2765 }
2766 } else {
2767 ret = ustctl_get_next_subbuf(stream->ustream);
2768 if (ret == 0) {
2769 /*
2770 * There is still data so let's put back this
2771 * subbuffer.
2772 */
2773 ret = ustctl_put_subbuf(stream->ustream);
2774 assert(ret == 0);
2775 ret = 1; /* Data is pending */
2776 goto end;
2777 }
ca22feea
DG
2778 }
2779
6d805429
DG
2780 /* Data is NOT pending so ready to be read. */
2781 ret = 0;
ca22feea 2782
6efae65e
DG
2783end:
2784 return ret;
ca22feea 2785}
d88aee68 2786
6d574024
DG
2787/*
2788 * Stop a given metadata channel timer if enabled and close the wait fd which
2789 * is the poll pipe of the metadata stream.
2790 *
2791 * This MUST be called with the metadata channel acquired.
2792 */
2793void lttng_ustconsumer_close_metadata(struct lttng_consumer_channel *metadata)
2794{
2795 int ret;
2796
2797 assert(metadata);
2798 assert(metadata->type == CONSUMER_CHANNEL_TYPE_METADATA);
2799
2800 DBG("Closing metadata channel key %" PRIu64, metadata->key);
2801
2802 if (metadata->switch_timer_enabled == 1) {
2803 consumer_timer_switch_stop(metadata);
2804 }
2805
2806 if (!metadata->metadata_stream) {
2807 goto end;
2808 }
2809
2810 /*
2811 * Closing write side so the thread monitoring the stream wakes up if any
2812 * and clean the metadata stream.
2813 */
2814 if (metadata->metadata_stream->ust_metadata_poll_pipe[1] >= 0) {
2815 ret = close(metadata->metadata_stream->ust_metadata_poll_pipe[1]);
2816 if (ret < 0) {
2817 PERROR("closing metadata pipe write side");
2818 }
2819 metadata->metadata_stream->ust_metadata_poll_pipe[1] = -1;
2820 }
2821
2822end:
2823 return;
2824}
2825
d88aee68
DG
2826/*
2827 * Close every metadata stream wait fd of the metadata hash table. This
2828 * function MUST be used very carefully so not to run into a race between the
2829 * metadata thread handling streams and this function closing their wait fd.
2830 *
2831 * For UST, this is used when the session daemon hangs up. Its the metadata
2832 * producer so calling this is safe because we are assured that no state change
2833 * can occur in the metadata thread for the streams in the hash table.
2834 */
6d574024 2835void lttng_ustconsumer_close_all_metadata(struct lttng_ht *metadata_ht)
d88aee68 2836{
d88aee68
DG
2837 struct lttng_ht_iter iter;
2838 struct lttng_consumer_stream *stream;
2839
2840 assert(metadata_ht);
2841 assert(metadata_ht->ht);
2842
2843 DBG("UST consumer closing all metadata streams");
2844
2845 rcu_read_lock();
2846 cds_lfht_for_each_entry(metadata_ht->ht, &iter.iter, stream,
2847 node.node) {
9ce5646a
MD
2848
2849 health_code_update();
2850
be2b50c7 2851 pthread_mutex_lock(&stream->chan->lock);
6d574024 2852 lttng_ustconsumer_close_metadata(stream->chan);
be2b50c7
DG
2853 pthread_mutex_unlock(&stream->chan->lock);
2854
d88aee68
DG
2855 }
2856 rcu_read_unlock();
2857}
d8ef542d
MD
2858
2859void lttng_ustconsumer_close_stream_wakeup(struct lttng_consumer_stream *stream)
2860{
2861 int ret;
2862
2863 ret = ustctl_stream_close_wakeup_fd(stream->ustream);
2864 if (ret < 0) {
2865 ERR("Unable to close wakeup fd");
2866 }
2867}
331744e3 2868
f666ae70
MD
2869/*
2870 * Please refer to consumer-timer.c before adding any lock within this
2871 * function or any of its callees. Timers have a very strict locking
2872 * semantic with respect to teardown. Failure to respect this semantic
2873 * introduces deadlocks.
c585821b
MD
2874 *
2875 * DON'T hold the metadata lock when calling this function, else this
2876 * can cause deadlock involving consumer awaiting for metadata to be
2877 * pushed out due to concurrent interaction with the session daemon.
f666ae70 2878 */
331744e3 2879int lttng_ustconsumer_request_metadata(struct lttng_consumer_local_data *ctx,
94d49140 2880 struct lttng_consumer_channel *channel, int timer, int wait)
331744e3
JD
2881{
2882 struct lttcomm_metadata_request_msg request;
2883 struct lttcomm_consumer_msg msg;
0c759fc9 2884 enum lttcomm_return_code ret_code = LTTCOMM_CONSUMERD_SUCCESS;
93ec662e 2885 uint64_t len, key, offset, version;
331744e3
JD
2886 int ret;
2887
2888 assert(channel);
2889 assert(channel->metadata_cache);
2890
53efb85a
MD
2891 memset(&request, 0, sizeof(request));
2892
331744e3
JD
2893 /* send the metadata request to sessiond */
2894 switch (consumer_data.type) {
2895 case LTTNG_CONSUMER64_UST:
2896 request.bits_per_long = 64;
2897 break;
2898 case LTTNG_CONSUMER32_UST:
2899 request.bits_per_long = 32;
2900 break;
2901 default:
2902 request.bits_per_long = 0;
2903 break;
2904 }
2905
2906 request.session_id = channel->session_id;
1950109e 2907 request.session_id_per_pid = channel->session_id_per_pid;
567eb353
DG
2908 /*
2909 * Request the application UID here so the metadata of that application can
2910 * be sent back. The channel UID corresponds to the user UID of the session
2911 * used for the rights on the stream file(s).
2912 */
2913 request.uid = channel->ust_app_uid;
331744e3 2914 request.key = channel->key;
567eb353 2915
1950109e 2916 DBG("Sending metadata request to sessiond, session id %" PRIu64
567eb353
DG
2917 ", per-pid %" PRIu64 ", app UID %u and channek key %" PRIu64,
2918 request.session_id, request.session_id_per_pid, request.uid,
2919 request.key);
331744e3 2920
75d83e50 2921 pthread_mutex_lock(&ctx->metadata_socket_lock);
9ce5646a
MD
2922
2923 health_code_update();
2924
331744e3
JD
2925 ret = lttcomm_send_unix_sock(ctx->consumer_metadata_socket, &request,
2926 sizeof(request));
2927 if (ret < 0) {
2928 ERR("Asking metadata to sessiond");
2929 goto end;
2930 }
2931
9ce5646a
MD
2932 health_code_update();
2933
331744e3
JD
2934 /* Receive the metadata from sessiond */
2935 ret = lttcomm_recv_unix_sock(ctx->consumer_metadata_socket, &msg,
2936 sizeof(msg));
2937 if (ret != sizeof(msg)) {
8fd623e0 2938 DBG("Consumer received unexpected message size %d (expects %zu)",
331744e3
JD
2939 ret, sizeof(msg));
2940 lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_ERROR_RECV_CMD);
2941 /*
2942 * The ret value might 0 meaning an orderly shutdown but this is ok
2943 * since the caller handles this.
2944 */
2945 goto end;
2946 }
2947
9ce5646a
MD
2948 health_code_update();
2949
331744e3
JD
2950 if (msg.cmd_type == LTTNG_ERR_UND) {
2951 /* No registry found */
2952 (void) consumer_send_status_msg(ctx->consumer_metadata_socket,
2953 ret_code);
2954 ret = 0;
2955 goto end;
2956 } else if (msg.cmd_type != LTTNG_CONSUMER_PUSH_METADATA) {
2957 ERR("Unexpected cmd_type received %d", msg.cmd_type);
2958 ret = -1;
2959 goto end;
2960 }
2961
2962 len = msg.u.push_metadata.len;
2963 key = msg.u.push_metadata.key;
2964 offset = msg.u.push_metadata.target_offset;
93ec662e 2965 version = msg.u.push_metadata.version;
331744e3
JD
2966
2967 assert(key == channel->key);
2968 if (len == 0) {
2969 DBG("No new metadata to receive for key %" PRIu64, key);
2970 }
2971
9ce5646a
MD
2972 health_code_update();
2973
331744e3
JD
2974 /* Tell session daemon we are ready to receive the metadata. */
2975 ret = consumer_send_status_msg(ctx->consumer_metadata_socket,
0c759fc9 2976 LTTCOMM_CONSUMERD_SUCCESS);
331744e3
JD
2977 if (ret < 0 || len == 0) {
2978 /*
2979 * Somehow, the session daemon is not responding anymore or there is
2980 * nothing to receive.
2981 */
2982 goto end;
2983 }
2984
9ce5646a
MD
2985 health_code_update();
2986
1eb682be 2987 ret = lttng_ustconsumer_recv_metadata(ctx->consumer_metadata_socket,
93ec662e 2988 key, offset, len, version, channel, timer, wait);
1eb682be 2989 if (ret >= 0) {
f2a444f1
DG
2990 /*
2991 * Only send the status msg if the sessiond is alive meaning a positive
2992 * ret code.
2993 */
1eb682be 2994 (void) consumer_send_status_msg(ctx->consumer_metadata_socket, ret);
f2a444f1 2995 }
331744e3
JD
2996 ret = 0;
2997
2998end:
9ce5646a
MD
2999 health_code_update();
3000
75d83e50 3001 pthread_mutex_unlock(&ctx->metadata_socket_lock);
331744e3
JD
3002 return ret;
3003}
70190e1c
DG
3004
3005/*
3006 * Return the ustctl call for the get stream id.
3007 */
3008int lttng_ustconsumer_get_stream_id(struct lttng_consumer_stream *stream,
3009 uint64_t *stream_id)
3010{
3011 assert(stream);
3012 assert(stream_id);
3013
3014 return ustctl_get_stream_id(stream->ustream, stream_id);
3015}
This page took 0.350802 seconds and 4 git commands to generate.