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