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