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