Cleanup: Move `create_posix_shm()` to common/shm.c
[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>
b7fc068d 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]) {
b7fc068d 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,
ff588497
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,
ff588497
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,
ff588497
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
b1316da1
JG
1245static
1246void metadata_stream_reset_cache_consumed_position(
1247 struct lttng_consumer_stream *stream)
1248{
1249 ASSERT_LOCKED(stream->lock);
1250
1251 DBG("Reset metadata cache of session %" PRIu64,
1252 stream->chan->session_id);
1253 stream->ust_metadata_pushed = 0;
1254}
1255
331744e3 1256/*
c585821b
MD
1257 * Receive the metadata updates from the sessiond. Supports receiving
1258 * overlapping metadata, but is needs to always belong to a contiguous
1259 * range starting from 0.
1260 * Be careful about the locks held when calling this function: it needs
1261 * the metadata cache flush to concurrently progress in order to
1262 * complete.
331744e3
JD
1263 */
1264int lttng_ustconsumer_recv_metadata(int sock, uint64_t key, uint64_t offset,
93ec662e
JD
1265 uint64_t len, uint64_t version,
1266 struct lttng_consumer_channel *channel, int timer, int wait)
331744e3 1267{
0c759fc9 1268 int ret, ret_code = LTTCOMM_CONSUMERD_SUCCESS;
331744e3 1269 char *metadata_str;
b1316da1 1270 enum consumer_metadata_cache_write_status cache_write_status;
331744e3 1271
8fd623e0 1272 DBG("UST consumer push metadata key %" PRIu64 " of len %" PRIu64, key, len);
331744e3
JD
1273
1274 metadata_str = zmalloc(len * sizeof(char));
1275 if (!metadata_str) {
1276 PERROR("zmalloc metadata string");
1277 ret_code = LTTCOMM_CONSUMERD_ENOMEM;
1278 goto end;
1279 }
1280
9ce5646a
MD
1281 health_code_update();
1282
331744e3
JD
1283 /* Receive metadata string. */
1284 ret = lttcomm_recv_unix_sock(sock, metadata_str, len);
1285 if (ret < 0) {
1286 /* Session daemon is dead so return gracefully. */
1287 ret_code = ret;
1288 goto end_free;
1289 }
1290
9ce5646a
MD
1291 health_code_update();
1292
331744e3 1293 pthread_mutex_lock(&channel->metadata_cache->lock);
b1316da1 1294 cache_write_status = consumer_metadata_cache_write(
a25d34bc
JG
1295 channel->metadata_cache, offset, len, version,
1296 metadata_str);
3bdc49f3 1297 pthread_mutex_unlock(&channel->metadata_cache->lock);
b1316da1
JG
1298 switch (cache_write_status) {
1299 case CONSUMER_METADATA_CACHE_WRITE_STATUS_NO_CHANGE:
1300 /*
1301 * The write entirely overlapped with existing contents of the
1302 * same metadata version (same content); there is nothing to do.
1303 */
1304 break;
1305 case CONSUMER_METADATA_CACHE_WRITE_STATUS_INVALIDATED:
1306 /*
1307 * The metadata cache was invalidated (previously pushed
1308 * content has been overwritten). Reset the stream's consumed
1309 * metadata position to ensure the metadata poll thread consumes
1310 * the whole cache.
1311 */
1312 pthread_mutex_lock(&channel->metadata_stream->lock);
1313 metadata_stream_reset_cache_consumed_position(
1314 channel->metadata_stream);
1315 pthread_mutex_unlock(&channel->metadata_stream->lock);
1316 /* Fall-through. */
1317 case CONSUMER_METADATA_CACHE_WRITE_STATUS_APPENDED_CONTENT:
1318 /*
1319 * In both cases, the metadata poll thread has new data to
1320 * consume.
1321 */
1322 ret = consumer_metadata_wakeup_pipe(channel);
1323 if (ret) {
1324 ret_code = LTTCOMM_CONSUMERD_ERROR_METADATA;
1325 goto end_free;
1326 }
1327 break;
1328 case CONSUMER_METADATA_CACHE_WRITE_STATUS_ERROR:
331744e3
JD
1329 /* Unable to handle metadata. Notify session daemon. */
1330 ret_code = LTTCOMM_CONSUMERD_ERROR_METADATA;
a32bd775
DG
1331 /*
1332 * Skip metadata flush on write error since the offset and len might
1333 * not have been updated which could create an infinite loop below when
1334 * waiting for the metadata cache to be flushed.
1335 */
a32bd775 1336 goto end_free;
b1316da1
JG
1337 default:
1338 abort();
331744e3 1339 }
331744e3 1340
94d49140
JD
1341 if (!wait) {
1342 goto end_free;
1343 }
5e41ebe1 1344 while (consumer_metadata_cache_flushed(channel, offset + len, timer)) {
331744e3 1345 DBG("Waiting for metadata to be flushed");
9ce5646a
MD
1346
1347 health_code_update();
1348
331744e3
JD
1349 usleep(DEFAULT_METADATA_AVAILABILITY_WAIT_TIME);
1350 }
1351
1352end_free:
1353 free(metadata_str);
1354end:
1355 return ret_code;
1356}
1357
4cbc1a04
DG
1358/*
1359 * Receive command from session daemon and process it.
1360 *
1361 * Return 1 on success else a negative value or 0.
1362 */
3bd1e081
MD
1363int lttng_ustconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
1364 int sock, struct pollfd *consumer_sockpoll)
1365{
1366 ssize_t ret;
0c759fc9 1367 enum lttcomm_return_code ret_code = LTTCOMM_CONSUMERD_SUCCESS;
3bd1e081 1368 struct lttcomm_consumer_msg msg;
ffe60014 1369 struct lttng_consumer_channel *channel = NULL;
3bd1e081 1370
9ce5646a
MD
1371 health_code_update();
1372
3bd1e081
MD
1373 ret = lttcomm_recv_unix_sock(sock, &msg, sizeof(msg));
1374 if (ret != sizeof(msg)) {
173af62f
DG
1375 DBG("Consumer received unexpected message size %zd (expects %zu)",
1376 ret, sizeof(msg));
3be74084
DG
1377 /*
1378 * The ret value might 0 meaning an orderly shutdown but this is ok
1379 * since the caller handles this.
1380 */
489f70e9 1381 if (ret > 0) {
c6857fcf 1382 lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_ERROR_RECV_CMD);
489f70e9
MD
1383 ret = -1;
1384 }
3bd1e081
MD
1385 return ret;
1386 }
9ce5646a
MD
1387
1388 health_code_update();
1389
84382d49
MD
1390 /* deprecated */
1391 assert(msg.cmd_type != LTTNG_CONSUMER_STOP);
3bd1e081 1392
9ce5646a
MD
1393 health_code_update();
1394
3f8e211f 1395 /* relayd needs RCU read-side lock */
b0b335c8
MD
1396 rcu_read_lock();
1397
3bd1e081 1398 switch (msg.cmd_type) {
00e2e675
DG
1399 case LTTNG_CONSUMER_ADD_RELAYD_SOCKET:
1400 {
f50f23d9 1401 /* Session daemon status message are handled in the following call. */
2527bf85 1402 consumer_add_relayd_socket(msg.u.relayd_sock.net_index,
7735ef9e 1403 msg.u.relayd_sock.type, ctx, sock, consumer_sockpoll,
d3e2ba59
JD
1404 &msg.u.relayd_sock.sock, msg.u.relayd_sock.session_id,
1405 msg.u.relayd_sock.relayd_session_id);
00e2e675
DG
1406 goto end_nosignal;
1407 }
173af62f
DG
1408 case LTTNG_CONSUMER_DESTROY_RELAYD:
1409 {
a6ba4fe1 1410 uint64_t index = msg.u.destroy_relayd.net_seq_idx;
173af62f
DG
1411 struct consumer_relayd_sock_pair *relayd;
1412
a6ba4fe1 1413 DBG("UST consumer destroying relayd %" PRIu64, index);
173af62f
DG
1414
1415 /* Get relayd reference if exists. */
a6ba4fe1 1416 relayd = consumer_find_relayd(index);
173af62f 1417 if (relayd == NULL) {
3448e266 1418 DBG("Unable to find relayd %" PRIu64, index);
e462382a 1419 ret_code = LTTCOMM_CONSUMERD_RELAYD_FAIL;
173af62f
DG
1420 }
1421
a6ba4fe1
DG
1422 /*
1423 * Each relayd socket pair has a refcount of stream attached to it
1424 * which tells if the relayd is still active or not depending on the
1425 * refcount value.
1426 *
1427 * This will set the destroy flag of the relayd object and destroy it
1428 * if the refcount reaches zero when called.
1429 *
1430 * The destroy can happen either here or when a stream fd hangs up.
1431 */
f50f23d9
DG
1432 if (relayd) {
1433 consumer_flag_relayd_for_destroy(relayd);
1434 }
1435
d88aee68 1436 goto end_msg_sessiond;
173af62f 1437 }
3bd1e081
MD
1438 case LTTNG_CONSUMER_UPDATE_STREAM:
1439 {
3f8e211f 1440 rcu_read_unlock();
7ad0a0cb 1441 return -ENOSYS;
3bd1e081 1442 }
6d805429 1443 case LTTNG_CONSUMER_DATA_PENDING:
53632229 1444 {
3be74084 1445 int ret, is_data_pending;
6d805429 1446 uint64_t id = msg.u.data_pending.session_id;
ca22feea 1447
6d805429 1448 DBG("UST consumer data pending command for id %" PRIu64, id);
ca22feea 1449
3be74084 1450 is_data_pending = consumer_data_pending(id);
ca22feea
DG
1451
1452 /* Send back returned value to session daemon */
3be74084
DG
1453 ret = lttcomm_send_unix_sock(sock, &is_data_pending,
1454 sizeof(is_data_pending));
ca22feea 1455 if (ret < 0) {
3be74084 1456 DBG("Error when sending the data pending ret code: %d", ret);
489f70e9 1457 goto error_fatal;
ca22feea 1458 }
f50f23d9
DG
1459
1460 /*
1461 * No need to send back a status message since the data pending
1462 * returned value is the response.
1463 */
ca22feea 1464 break;
53632229 1465 }
ffe60014
DG
1466 case LTTNG_CONSUMER_ASK_CHANNEL_CREATION:
1467 {
1468 int ret;
1469 struct ustctl_consumer_channel_attr attr;
d2956687
JG
1470 const uint64_t chunk_id = msg.u.ask_channel.chunk_id.value;
1471 const struct lttng_credentials buffer_credentials = {
ff588497
JR
1472 .uid = LTTNG_OPTIONAL_INIT_VALUE(msg.u.ask_channel.buffer_credentials.uid),
1473 .gid = LTTNG_OPTIONAL_INIT_VALUE(msg.u.ask_channel.buffer_credentials.gid),
d2956687 1474 };
ffe60014
DG
1475
1476 /* Create a plain object and reserve a channel key. */
a2814ea7
JG
1477 channel = consumer_allocate_channel(
1478 msg.u.ask_channel.key,
1479 msg.u.ask_channel.session_id,
d2956687
JG
1480 msg.u.ask_channel.chunk_id.is_set ?
1481 &chunk_id : NULL,
1482 msg.u.ask_channel.pathname,
1483 msg.u.ask_channel.name,
1484 msg.u.ask_channel.relayd_id,
1624d5b7
JD
1485 (enum lttng_event_output) msg.u.ask_channel.output,
1486 msg.u.ask_channel.tracefile_size,
2bba9e53 1487 msg.u.ask_channel.tracefile_count,
1950109e 1488 msg.u.ask_channel.session_id_per_pid,
ecc48a90 1489 msg.u.ask_channel.monitor,
d7ba1388 1490 msg.u.ask_channel.live_timer_interval,
a2814ea7 1491 msg.u.ask_channel.is_live,
3d071855 1492 msg.u.ask_channel.root_shm_path,
d7ba1388 1493 msg.u.ask_channel.shm_path);
ffe60014
DG
1494 if (!channel) {
1495 goto end_channel_error;
1496 }
1497
d2956687
JG
1498 LTTNG_OPTIONAL_SET(&channel->buffer_credentials,
1499 buffer_credentials);
1500
567eb353
DG
1501 /*
1502 * Assign UST application UID to the channel. This value is ignored for
1503 * per PID buffers. This is specific to UST thus setting this after the
1504 * allocation.
1505 */
1506 channel->ust_app_uid = msg.u.ask_channel.ust_app_uid;
1507
ffe60014
DG
1508 /* Build channel attributes from received message. */
1509 attr.subbuf_size = msg.u.ask_channel.subbuf_size;
1510 attr.num_subbuf = msg.u.ask_channel.num_subbuf;
1511 attr.overwrite = msg.u.ask_channel.overwrite;
1512 attr.switch_timer_interval = msg.u.ask_channel.switch_timer_interval;
1513 attr.read_timer_interval = msg.u.ask_channel.read_timer_interval;
7972aab2 1514 attr.chan_id = msg.u.ask_channel.chan_id;
ffe60014 1515 memcpy(attr.uuid, msg.u.ask_channel.uuid, sizeof(attr.uuid));
491d1539 1516 attr.blocking_timeout= msg.u.ask_channel.blocking_timeout;
ffe60014 1517
0c759fc9
DG
1518 /* Match channel buffer type to the UST abi. */
1519 switch (msg.u.ask_channel.output) {
1520 case LTTNG_EVENT_MMAP:
1521 default:
fc4b93fa 1522 attr.output = LTTNG_UST_ABI_MMAP;
0c759fc9
DG
1523 break;
1524 }
1525
ffe60014
DG
1526 /* Translate and save channel type. */
1527 switch (msg.u.ask_channel.type) {
fc4b93fa 1528 case LTTNG_UST_ABI_CHAN_PER_CPU:
ffe60014 1529 channel->type = CONSUMER_CHANNEL_TYPE_DATA;
fc4b93fa 1530 attr.type = LTTNG_UST_ABI_CHAN_PER_CPU;
8633d6e3
MD
1531 /*
1532 * Set refcount to 1 for owner. Below, we will
1533 * pass ownership to the
1534 * consumer_thread_channel_poll() thread.
1535 */
1536 channel->refcount = 1;
ffe60014 1537 break;
fc4b93fa 1538 case LTTNG_UST_ABI_CHAN_METADATA:
ffe60014 1539 channel->type = CONSUMER_CHANNEL_TYPE_METADATA;
fc4b93fa 1540 attr.type = LTTNG_UST_ABI_CHAN_METADATA;
ffe60014
DG
1541 break;
1542 default:
1543 assert(0);
1544 goto error_fatal;
1545 };
1546
9ce5646a
MD
1547 health_code_update();
1548
d2956687 1549 ret = ask_channel(ctx, channel, &attr);
ffe60014
DG
1550 if (ret < 0) {
1551 goto end_channel_error;
1552 }
1553
fc4b93fa 1554 if (msg.u.ask_channel.type == LTTNG_UST_ABI_CHAN_METADATA) {
fc643247
MD
1555 ret = consumer_metadata_cache_allocate(channel);
1556 if (ret < 0) {
1557 ERR("Allocating metadata cache");
1558 goto end_channel_error;
1559 }
1560 consumer_timer_switch_start(channel, attr.switch_timer_interval);
1561 attr.switch_timer_interval = 0;
94d49140 1562 } else {
e9404c27
JG
1563 int monitor_start_ret;
1564
94d49140
JD
1565 consumer_timer_live_start(channel,
1566 msg.u.ask_channel.live_timer_interval);
e9404c27
JG
1567 monitor_start_ret = consumer_timer_monitor_start(
1568 channel,
1569 msg.u.ask_channel.monitor_timer_interval);
1570 if (monitor_start_ret < 0) {
1571 ERR("Starting channel monitoring timer failed");
1572 goto end_channel_error;
1573 }
fc643247
MD
1574 }
1575
9ce5646a
MD
1576 health_code_update();
1577
ffe60014
DG
1578 /*
1579 * Add the channel to the internal state AFTER all streams were created
1580 * and successfully sent to session daemon. This way, all streams must
1581 * be ready before this channel is visible to the threads.
fc643247
MD
1582 * If add_channel succeeds, ownership of the channel is
1583 * passed to consumer_thread_channel_poll().
ffe60014
DG
1584 */
1585 ret = add_channel(channel, ctx);
1586 if (ret < 0) {
fc4b93fa 1587 if (msg.u.ask_channel.type == LTTNG_UST_ABI_CHAN_METADATA) {
ea88ca2a
MD
1588 if (channel->switch_timer_enabled == 1) {
1589 consumer_timer_switch_stop(channel);
1590 }
1591 consumer_metadata_cache_destroy(channel);
1592 }
d3e2ba59
JD
1593 if (channel->live_timer_enabled == 1) {
1594 consumer_timer_live_stop(channel);
1595 }
e9404c27
JG
1596 if (channel->monitor_timer_enabled == 1) {
1597 consumer_timer_monitor_stop(channel);
1598 }
ffe60014
DG
1599 goto end_channel_error;
1600 }
1601
9ce5646a
MD
1602 health_code_update();
1603
ffe60014
DG
1604 /*
1605 * Channel and streams are now created. Inform the session daemon that
1606 * everything went well and should wait to receive the channel and
1607 * streams with ustctl API.
1608 */
1609 ret = consumer_send_status_channel(sock, channel);
1610 if (ret < 0) {
1611 /*
489f70e9 1612 * There is probably a problem on the socket.
ffe60014 1613 */
489f70e9 1614 goto error_fatal;
ffe60014
DG
1615 }
1616
1617 break;
1618 }
1619 case LTTNG_CONSUMER_GET_CHANNEL:
1620 {
1621 int ret, relayd_err = 0;
d88aee68 1622 uint64_t key = msg.u.get_channel.key;
ffe60014 1623 struct lttng_consumer_channel *channel;
ffe60014
DG
1624
1625 channel = consumer_find_channel(key);
1626 if (!channel) {
8fd623e0 1627 ERR("UST consumer get channel key %" PRIu64 " not found", key);
e462382a 1628 ret_code = LTTCOMM_CONSUMERD_CHAN_NOT_FOUND;
f3a1fc7b 1629 goto end_get_channel;
ffe60014
DG
1630 }
1631
9ce5646a
MD
1632 health_code_update();
1633
a3a86f35
JG
1634 /* Send the channel to sessiond (and relayd, if applicable). */
1635 ret = send_channel_to_sessiond_and_relayd(sock, channel, ctx,
1636 &relayd_err);
ffe60014
DG
1637 if (ret < 0) {
1638 if (relayd_err) {
1639 /*
1640 * We were unable to send to the relayd the stream so avoid
1641 * sending back a fatal error to the thread since this is OK
f2a444f1
DG
1642 * and the consumer can continue its work. The above call
1643 * has sent the error status message to the sessiond.
ffe60014 1644 */
f3a1fc7b 1645 goto end_get_channel_nosignal;
ffe60014
DG
1646 }
1647 /*
1648 * The communicaton was broken hence there is a bad state between
1649 * the consumer and sessiond so stop everything.
1650 */
f3a1fc7b 1651 goto error_get_channel_fatal;
ffe60014
DG
1652 }
1653
9ce5646a
MD
1654 health_code_update();
1655
10a50311
JD
1656 /*
1657 * In no monitor mode, the streams ownership is kept inside the channel
1658 * so don't send them to the data thread.
1659 */
1660 if (!channel->monitor) {
f3a1fc7b 1661 goto end_get_channel;
10a50311
JD
1662 }
1663
d88aee68
DG
1664 ret = send_streams_to_thread(channel, ctx);
1665 if (ret < 0) {
1666 /*
1667 * If we are unable to send the stream to the thread, there is
1668 * a big problem so just stop everything.
1669 */
f3a1fc7b 1670 goto error_get_channel_fatal;
ffe60014 1671 }
ffe60014
DG
1672 /* List MUST be empty after or else it could be reused. */
1673 assert(cds_list_empty(&channel->streams.head));
f3a1fc7b 1674end_get_channel:
d88aee68 1675 goto end_msg_sessiond;
f3a1fc7b
JG
1676error_get_channel_fatal:
1677 goto error_fatal;
1678end_get_channel_nosignal:
1679 goto end_nosignal;
d88aee68
DG
1680 }
1681 case LTTNG_CONSUMER_DESTROY_CHANNEL:
1682 {
1683 uint64_t key = msg.u.destroy_channel.key;
d88aee68 1684
a0cbdd2e
MD
1685 /*
1686 * Only called if streams have not been sent to stream
1687 * manager thread. However, channel has been sent to
1688 * channel manager thread.
1689 */
1690 notify_thread_del_channel(ctx, key);
d88aee68 1691 goto end_msg_sessiond;
ffe60014 1692 }
d88aee68
DG
1693 case LTTNG_CONSUMER_CLOSE_METADATA:
1694 {
1695 int ret;
1696
1697 ret = close_metadata(msg.u.close_metadata.key);
1698 if (ret != 0) {
1699 ret_code = ret;
1700 }
1701
1702 goto end_msg_sessiond;
1703 }
7972aab2
DG
1704 case LTTNG_CONSUMER_FLUSH_CHANNEL:
1705 {
1706 int ret;
1707
1708 ret = flush_channel(msg.u.flush_channel.key);
1709 if (ret != 0) {
1710 ret_code = ret;
1711 }
1712
1713 goto end_msg_sessiond;
1714 }
0dd01979
MD
1715 case LTTNG_CONSUMER_CLEAR_QUIESCENT_CHANNEL:
1716 {
1717 int ret;
1718
1719 ret = clear_quiescent_channel(
1720 msg.u.clear_quiescent_channel.key);
1721 if (ret != 0) {
1722 ret_code = ret;
1723 }
1724
1725 goto end_msg_sessiond;
1726 }
d88aee68 1727 case LTTNG_CONSUMER_PUSH_METADATA:
ffe60014
DG
1728 {
1729 int ret;
d88aee68 1730 uint64_t len = msg.u.push_metadata.len;
d88aee68 1731 uint64_t key = msg.u.push_metadata.key;
331744e3 1732 uint64_t offset = msg.u.push_metadata.target_offset;
93ec662e 1733 uint64_t version = msg.u.push_metadata.version;
ffe60014
DG
1734 struct lttng_consumer_channel *channel;
1735
8fd623e0
DG
1736 DBG("UST consumer push metadata key %" PRIu64 " of len %" PRIu64, key,
1737 len);
ffe60014
DG
1738
1739 channel = consumer_find_channel(key);
1740 if (!channel) {
000baf6a
DG
1741 /*
1742 * This is possible if the metadata creation on the consumer side
1743 * is in flight vis-a-vis a concurrent push metadata from the
1744 * session daemon. Simply return that the channel failed and the
1745 * session daemon will handle that message correctly considering
1746 * that this race is acceptable thus the DBG() statement here.
1747 */
1748 DBG("UST consumer push metadata %" PRIu64 " not found", key);
1749 ret_code = LTTCOMM_CONSUMERD_CHANNEL_FAIL;
a8ffe244 1750 goto end_push_metadata_msg_sessiond;
d88aee68
DG
1751 }
1752
9ce5646a
MD
1753 health_code_update();
1754
c585821b
MD
1755 if (!len) {
1756 /*
1757 * There is nothing to receive. We have simply
1758 * checked whether the channel can be found.
1759 */
1760 ret_code = LTTCOMM_CONSUMERD_SUCCESS;
a8ffe244 1761 goto end_push_metadata_msg_sessiond;
c585821b
MD
1762 }
1763
d88aee68 1764 /* Tell session daemon we are ready to receive the metadata. */
0c759fc9 1765 ret = consumer_send_status_msg(sock, LTTCOMM_CONSUMERD_SUCCESS);
ffe60014
DG
1766 if (ret < 0) {
1767 /* Somehow, the session daemon is not responding anymore. */
a8ffe244 1768 goto error_push_metadata_fatal;
d88aee68
DG
1769 }
1770
9ce5646a
MD
1771 health_code_update();
1772
d88aee68 1773 /* Wait for more data. */
9ce5646a
MD
1774 health_poll_entry();
1775 ret = lttng_consumer_poll_socket(consumer_sockpoll);
1776 health_poll_exit();
84382d49 1777 if (ret) {
a8ffe244 1778 goto error_push_metadata_fatal;
d88aee68
DG
1779 }
1780
9ce5646a
MD
1781 health_code_update();
1782
331744e3 1783 ret = lttng_ustconsumer_recv_metadata(sock, key, offset,
93ec662e 1784 len, version, channel, 0, 1);
d88aee68 1785 if (ret < 0) {
331744e3 1786 /* error receiving from sessiond */
a8ffe244 1787 goto error_push_metadata_fatal;
331744e3
JD
1788 } else {
1789 ret_code = ret;
a8ffe244 1790 goto end_push_metadata_msg_sessiond;
d88aee68 1791 }
a8ffe244
JG
1792end_push_metadata_msg_sessiond:
1793 goto end_msg_sessiond;
1794error_push_metadata_fatal:
1795 goto error_fatal;
d88aee68
DG
1796 }
1797 case LTTNG_CONSUMER_SETUP_METADATA:
1798 {
1799 int ret;
1800
1801 ret = setup_metadata(ctx, msg.u.setup_metadata.key);
1802 if (ret) {
1803 ret_code = ret;
1804 }
1805 goto end_msg_sessiond;
ffe60014 1806 }
6dc3064a
DG
1807 case LTTNG_CONSUMER_SNAPSHOT_CHANNEL:
1808 {
3eb928aa
MD
1809 struct lttng_consumer_channel *channel;
1810 uint64_t key = msg.u.snapshot_channel.key;
1811
1812 channel = consumer_find_channel(key);
1813 if (!channel) {
1814 DBG("UST snapshot channel not found for key %" PRIu64, key);
1815 ret_code = LTTCOMM_CONSUMERD_CHAN_NOT_FOUND;
10a50311 1816 } else {
3eb928aa
MD
1817 if (msg.u.snapshot_channel.metadata) {
1818 ret = snapshot_metadata(channel, key,
1819 msg.u.snapshot_channel.pathname,
1820 msg.u.snapshot_channel.relayd_id,
d2956687 1821 ctx);
3eb928aa
MD
1822 if (ret < 0) {
1823 ERR("Snapshot metadata failed");
1824 ret_code = LTTCOMM_CONSUMERD_SNAPSHOT_FAILED;
1825 }
1826 } else {
1827 ret = snapshot_channel(channel, key,
1828 msg.u.snapshot_channel.pathname,
1829 msg.u.snapshot_channel.relayd_id,
1830 msg.u.snapshot_channel.nb_packets_per_stream,
1831 ctx);
1832 if (ret < 0) {
1833 ERR("Snapshot channel failed");
1834 ret_code = LTTCOMM_CONSUMERD_SNAPSHOT_FAILED;
1835 }
10a50311
JD
1836 }
1837 }
9ce5646a 1838 health_code_update();
6dc3064a
DG
1839 ret = consumer_send_status_msg(sock, ret_code);
1840 if (ret < 0) {
1841 /* Somehow, the session daemon is not responding anymore. */
1842 goto end_nosignal;
1843 }
9ce5646a 1844 health_code_update();
6dc3064a
DG
1845 break;
1846 }
fb83fe64
JD
1847 case LTTNG_CONSUMER_DISCARDED_EVENTS:
1848 {
beb59458
MJ
1849 int ret = 0;
1850 uint64_t discarded_events;
fb83fe64
JD
1851 struct lttng_ht_iter iter;
1852 struct lttng_ht *ht;
1853 struct lttng_consumer_stream *stream;
1854 uint64_t id = msg.u.discarded_events.session_id;
1855 uint64_t key = msg.u.discarded_events.channel_key;
1856
1857 DBG("UST consumer discarded events command for session id %"
1858 PRIu64, id);
1859 rcu_read_lock();
1860 pthread_mutex_lock(&consumer_data.lock);
1861
1862 ht = consumer_data.stream_list_ht;
1863
1864 /*
1865 * We only need a reference to the channel, but they are not
1866 * directly indexed, so we just use the first matching stream
1867 * to extract the information we need, we default to 0 if not
1868 * found (no events are dropped if the channel is not yet in
1869 * use).
1870 */
beb59458 1871 discarded_events = 0;
fb83fe64
JD
1872 cds_lfht_for_each_entry_duplicate(ht->ht,
1873 ht->hash_fct(&id, lttng_ht_seed),
1874 ht->match_fct, &id,
1875 &iter.iter, stream, node_session_id.node) {
1876 if (stream->chan->key == key) {
beb59458 1877 discarded_events = stream->chan->discarded_events;
fb83fe64
JD
1878 break;
1879 }
1880 }
1881 pthread_mutex_unlock(&consumer_data.lock);
1882 rcu_read_unlock();
1883
1884 DBG("UST consumer discarded events command for session id %"
1885 PRIu64 ", channel key %" PRIu64, id, key);
1886
1887 health_code_update();
1888
1889 /* Send back returned value to session daemon */
beb59458 1890 ret = lttcomm_send_unix_sock(sock, &discarded_events, sizeof(discarded_events));
fb83fe64
JD
1891 if (ret < 0) {
1892 PERROR("send discarded events");
1893 goto error_fatal;
1894 }
1895
1896 break;
1897 }
1898 case LTTNG_CONSUMER_LOST_PACKETS:
1899 {
9a06e8d4
JG
1900 int ret;
1901 uint64_t lost_packets;
fb83fe64
JD
1902 struct lttng_ht_iter iter;
1903 struct lttng_ht *ht;
1904 struct lttng_consumer_stream *stream;
1905 uint64_t id = msg.u.lost_packets.session_id;
1906 uint64_t key = msg.u.lost_packets.channel_key;
1907
1908 DBG("UST consumer lost packets command for session id %"
1909 PRIu64, id);
1910 rcu_read_lock();
1911 pthread_mutex_lock(&consumer_data.lock);
1912
1913 ht = consumer_data.stream_list_ht;
1914
1915 /*
1916 * We only need a reference to the channel, but they are not
1917 * directly indexed, so we just use the first matching stream
1918 * to extract the information we need, we default to 0 if not
1919 * found (no packets lost if the channel is not yet in use).
1920 */
9a06e8d4 1921 lost_packets = 0;
fb83fe64
JD
1922 cds_lfht_for_each_entry_duplicate(ht->ht,
1923 ht->hash_fct(&id, lttng_ht_seed),
1924 ht->match_fct, &id,
1925 &iter.iter, stream, node_session_id.node) {
1926 if (stream->chan->key == key) {
9a06e8d4 1927 lost_packets = stream->chan->lost_packets;
fb83fe64
JD
1928 break;
1929 }
1930 }
1931 pthread_mutex_unlock(&consumer_data.lock);
1932 rcu_read_unlock();
1933
1934 DBG("UST consumer lost packets command for session id %"
1935 PRIu64 ", channel key %" PRIu64, id, key);
1936
1937 health_code_update();
1938
1939 /* Send back returned value to session daemon */
9a06e8d4
JG
1940 ret = lttcomm_send_unix_sock(sock, &lost_packets,
1941 sizeof(lost_packets));
fb83fe64
JD
1942 if (ret < 0) {
1943 PERROR("send lost packets");
1944 goto error_fatal;
1945 }
1946
1947 break;
1948 }
b3530820
JG
1949 case LTTNG_CONSUMER_SET_CHANNEL_MONITOR_PIPE:
1950 {
1951 int channel_monitor_pipe;
1952
1953 ret_code = LTTCOMM_CONSUMERD_SUCCESS;
1954 /* Successfully received the command's type. */
1955 ret = consumer_send_status_msg(sock, ret_code);
1956 if (ret < 0) {
1957 goto error_fatal;
1958 }
1959
1960 ret = lttcomm_recv_fds_unix_sock(sock, &channel_monitor_pipe,
1961 1);
1962 if (ret != sizeof(channel_monitor_pipe)) {
1963 ERR("Failed to receive channel monitor pipe");
1964 goto error_fatal;
1965 }
1966
1967 DBG("Received channel monitor pipe (%d)", channel_monitor_pipe);
1968 ret = consumer_timer_thread_set_channel_monitor_pipe(
1969 channel_monitor_pipe);
1970 if (!ret) {
1971 int flags;
1972
1973 ret_code = LTTCOMM_CONSUMERD_SUCCESS;
1974 /* Set the pipe as non-blocking. */
1975 ret = fcntl(channel_monitor_pipe, F_GETFL, 0);
1976 if (ret == -1) {
1977 PERROR("fcntl get flags of the channel monitoring pipe");
1978 goto error_fatal;
1979 }
1980 flags = ret;
1981
1982 ret = fcntl(channel_monitor_pipe, F_SETFL,
1983 flags | O_NONBLOCK);
1984 if (ret == -1) {
1985 PERROR("fcntl set O_NONBLOCK flag of the channel monitoring pipe");
1986 goto error_fatal;
1987 }
1988 DBG("Channel monitor pipe set as non-blocking");
1989 } else {
1990 ret_code = LTTCOMM_CONSUMERD_ALREADY_SET;
1991 }
1992 goto end_msg_sessiond;
1993 }
b99a8d42
JD
1994 case LTTNG_CONSUMER_ROTATE_CHANNEL:
1995 {
92b7a7f8
MD
1996 struct lttng_consumer_channel *channel;
1997 uint64_t key = msg.u.rotate_channel.key;
b99a8d42 1998
92b7a7f8
MD
1999 channel = consumer_find_channel(key);
2000 if (!channel) {
2001 DBG("Channel %" PRIu64 " not found", key);
2002 ret_code = LTTCOMM_CONSUMERD_CHAN_NOT_FOUND;
2003 } else {
2004 /*
2005 * Sample the rotate position of all the streams in
2006 * this channel.
2007 */
2008 ret = lttng_consumer_rotate_channel(channel, key,
92b7a7f8
MD
2009 msg.u.rotate_channel.relayd_id,
2010 msg.u.rotate_channel.metadata,
92b7a7f8
MD
2011 ctx);
2012 if (ret < 0) {
2013 ERR("Rotate channel failed");
2014 ret_code = LTTCOMM_CONSUMERD_ROTATION_FAIL;
2015 }
b99a8d42 2016
92b7a7f8
MD
2017 health_code_update();
2018 }
b99a8d42
JD
2019 ret = consumer_send_status_msg(sock, ret_code);
2020 if (ret < 0) {
2021 /* Somehow, the session daemon is not responding anymore. */
41c7a76d 2022 goto end_rotate_channel_nosignal;
b99a8d42
JD
2023 }
2024
2025 /*
2026 * Rotate the streams that are ready right now.
2027 * FIXME: this is a second consecutive iteration over the
2028 * streams in a channel, there is probably a better way to
2029 * handle this, but it needs to be after the
2030 * consumer_send_status_msg() call.
2031 */
92b7a7f8
MD
2032 if (channel) {
2033 ret = lttng_consumer_rotate_ready_streams(
2034 channel, key, ctx);
2035 if (ret < 0) {
2036 ERR("Rotate channel failed");
2037 }
b99a8d42
JD
2038 }
2039 break;
41c7a76d
JG
2040end_rotate_channel_nosignal:
2041 goto end_nosignal;
b99a8d42 2042 }
5f3aff8b
MD
2043 case LTTNG_CONSUMER_CLEAR_CHANNEL:
2044 {
2045 struct lttng_consumer_channel *channel;
2046 uint64_t key = msg.u.clear_channel.key;
2047
2048 channel = consumer_find_channel(key);
2049 if (!channel) {
2050 DBG("Channel %" PRIu64 " not found", key);
2051 ret_code = LTTCOMM_CONSUMERD_CHAN_NOT_FOUND;
2052 } else {
2053 ret = lttng_consumer_clear_channel(channel);
2054 if (ret) {
2055 ERR("Clear channel failed key %" PRIu64, key);
2056 ret_code = ret;
2057 }
2058
2059 health_code_update();
2060 }
2061 ret = consumer_send_status_msg(sock, ret_code);
2062 if (ret < 0) {
2063 /* Somehow, the session daemon is not responding anymore. */
2064 goto end_nosignal;
2065 }
2066 break;
2067 }
d2956687 2068 case LTTNG_CONSUMER_INIT:
00fb02ac 2069 {
d2956687
JG
2070 ret_code = lttng_consumer_init_command(ctx,
2071 msg.u.init.sessiond_uuid);
d88744a4 2072 health_code_update();
d88744a4
JD
2073 ret = consumer_send_status_msg(sock, ret_code);
2074 if (ret < 0) {
2075 /* Somehow, the session daemon is not responding anymore. */
2076 goto end_nosignal;
2077 }
2078 break;
2079 }
d2956687 2080 case LTTNG_CONSUMER_CREATE_TRACE_CHUNK:
d88744a4 2081 {
d2956687 2082 const struct lttng_credentials credentials = {
ff588497
JR
2083 .uid = LTTNG_OPTIONAL_INIT_VALUE(msg.u.create_trace_chunk.credentials.value.uid),
2084 .gid = LTTNG_OPTIONAL_INIT_VALUE(msg.u.create_trace_chunk.credentials.value.gid),
d2956687
JG
2085 };
2086 const bool is_local_trace =
2087 !msg.u.create_trace_chunk.relayd_id.is_set;
2088 const uint64_t relayd_id =
2089 msg.u.create_trace_chunk.relayd_id.value;
2090 const char *chunk_override_name =
2091 *msg.u.create_trace_chunk.override_name ?
2092 msg.u.create_trace_chunk.override_name :
2093 NULL;
cbf53d23 2094 struct lttng_directory_handle *chunk_directory_handle = NULL;
d88744a4 2095
d2956687
JG
2096 /*
2097 * The session daemon will only provide a chunk directory file
2098 * descriptor for local traces.
2099 */
2100 if (is_local_trace) {
2101 int chunk_dirfd;
19990ed5 2102
d2956687
JG
2103 /* Acnowledge the reception of the command. */
2104 ret = consumer_send_status_msg(sock,
2105 LTTCOMM_CONSUMERD_SUCCESS);
2106 if (ret < 0) {
2107 /* Somehow, the session daemon is not responding anymore. */
2108 goto end_nosignal;
2109 }
92816cc3 2110
5da88b0f
MD
2111 /*
2112 * Receive trace chunk domain dirfd.
2113 */
d2956687
JG
2114 ret = lttcomm_recv_fds_unix_sock(sock, &chunk_dirfd, 1);
2115 if (ret != sizeof(chunk_dirfd)) {
5da88b0f 2116 ERR("Failed to receive trace chunk domain directory file descriptor");
d2956687
JG
2117 goto error_fatal;
2118 }
92816cc3 2119
5da88b0f 2120 DBG("Received trace chunk domain directory fd (%d)",
d2956687 2121 chunk_dirfd);
cbf53d23 2122 chunk_directory_handle = lttng_directory_handle_create_from_dirfd(
d2956687 2123 chunk_dirfd);
cbf53d23 2124 if (!chunk_directory_handle) {
5da88b0f 2125 ERR("Failed to initialize chunk domain directory handle from directory file descriptor");
d2956687
JG
2126 if (close(chunk_dirfd)) {
2127 PERROR("Failed to close chunk directory file descriptor");
2128 }
2129 goto error_fatal;
2130 }
92816cc3
JG
2131 }
2132
d2956687
JG
2133 ret_code = lttng_consumer_create_trace_chunk(
2134 !is_local_trace ? &relayd_id : NULL,
2135 msg.u.create_trace_chunk.session_id,
2136 msg.u.create_trace_chunk.chunk_id,
e5add6d0
JG
2137 (time_t) msg.u.create_trace_chunk
2138 .creation_timestamp,
d2956687 2139 chunk_override_name,
e5add6d0
JG
2140 msg.u.create_trace_chunk.credentials.is_set ?
2141 &credentials :
2142 NULL,
cbf53d23
JG
2143 chunk_directory_handle);
2144 lttng_directory_handle_put(chunk_directory_handle);
d2956687 2145 goto end_msg_sessiond;
00fb02ac 2146 }
d2956687 2147 case LTTNG_CONSUMER_CLOSE_TRACE_CHUNK:
a1ae2ea5 2148 {
bbc4768c
JG
2149 enum lttng_trace_chunk_command_type close_command =
2150 msg.u.close_trace_chunk.close_command.value;
d2956687
JG
2151 const uint64_t relayd_id =
2152 msg.u.close_trace_chunk.relayd_id.value;
ecd1a12f
MD
2153 struct lttcomm_consumer_close_trace_chunk_reply reply;
2154 char closed_trace_chunk_path[LTTNG_PATH_MAX];
2155 int ret;
d2956687
JG
2156
2157 ret_code = lttng_consumer_close_trace_chunk(
2158 msg.u.close_trace_chunk.relayd_id.is_set ?
bbc4768c
JG
2159 &relayd_id :
2160 NULL,
d2956687
JG
2161 msg.u.close_trace_chunk.session_id,
2162 msg.u.close_trace_chunk.chunk_id,
bbc4768c
JG
2163 (time_t) msg.u.close_trace_chunk.close_timestamp,
2164 msg.u.close_trace_chunk.close_command.is_set ?
2165 &close_command :
ecd1a12f
MD
2166 NULL, closed_trace_chunk_path);
2167 reply.ret_code = ret_code;
2168 reply.path_length = strlen(closed_trace_chunk_path) + 1;
2169 ret = lttcomm_send_unix_sock(sock, &reply, sizeof(reply));
2170 if (ret != sizeof(reply)) {
2171 goto error_fatal;
2172 }
2173 ret = lttcomm_send_unix_sock(sock, closed_trace_chunk_path,
2174 reply.path_length);
2175 if (ret != reply.path_length) {
2176 goto error_fatal;
2177 }
2178 goto end_nosignal;
a1ae2ea5 2179 }
d2956687 2180 case LTTNG_CONSUMER_TRACE_CHUNK_EXISTS:
3654ed19 2181 {
d2956687
JG
2182 const uint64_t relayd_id =
2183 msg.u.trace_chunk_exists.relayd_id.value;
2184
2185 ret_code = lttng_consumer_trace_chunk_exists(
2186 msg.u.trace_chunk_exists.relayd_id.is_set ?
2187 &relayd_id : NULL,
2188 msg.u.trace_chunk_exists.session_id,
2189 msg.u.trace_chunk_exists.chunk_id);
2190 goto end_msg_sessiond;
04ed9e10
JG
2191 }
2192 case LTTNG_CONSUMER_OPEN_CHANNEL_PACKETS:
2193 {
2194 const uint64_t key = msg.u.open_channel_packets.key;
2195 struct lttng_consumer_channel *channel =
2196 consumer_find_channel(key);
2197
2198 if (channel) {
2199 pthread_mutex_lock(&channel->lock);
2200 ret_code = lttng_consumer_open_channel_packets(channel);
2201 pthread_mutex_unlock(&channel->lock);
2202 } else {
2203 /*
2204 * The channel could have disappeared in per-pid
2205 * buffering mode.
2206 */
2207 DBG("Channel %" PRIu64 " not found", key);
2208 ret_code = LTTCOMM_CONSUMERD_CHAN_NOT_FOUND;
2209 }
2210
2211 health_code_update();
2212 goto end_msg_sessiond;
3654ed19 2213 }
3bd1e081
MD
2214 default:
2215 break;
2216 }
3f8e211f 2217
3bd1e081 2218end_nosignal:
4cbc1a04
DG
2219 /*
2220 * Return 1 to indicate success since the 0 value can be a socket
2221 * shutdown during the recv() or send() call.
2222 */
f3a1fc7b
JG
2223 ret = 1;
2224 goto end;
ffe60014
DG
2225
2226end_msg_sessiond:
2227 /*
2228 * The returned value here is not useful since either way we'll return 1 to
2229 * the caller because the session daemon socket management is done
2230 * elsewhere. Returning a negative code or 0 will shutdown the consumer.
2231 */
489f70e9
MD
2232 ret = consumer_send_status_msg(sock, ret_code);
2233 if (ret < 0) {
2234 goto error_fatal;
2235 }
f3a1fc7b
JG
2236 ret = 1;
2237 goto end;
9ce5646a 2238
ffe60014
DG
2239end_channel_error:
2240 if (channel) {
2241 /*
2242 * Free channel here since no one has a reference to it. We don't
2243 * free after that because a stream can store this pointer.
2244 */
2245 destroy_channel(channel);
2246 }
2247 /* We have to send a status channel message indicating an error. */
2248 ret = consumer_send_status_channel(sock, NULL);
2249 if (ret < 0) {
2250 /* Stop everything if session daemon can not be notified. */
2251 goto error_fatal;
2252 }
f3a1fc7b
JG
2253 ret = 1;
2254 goto end;
9ce5646a 2255
ffe60014 2256error_fatal:
ffe60014 2257 /* This will issue a consumer stop. */
f3a1fc7b
JG
2258 ret = -1;
2259 goto end;
2260
2261end:
2262 rcu_read_unlock();
2263 health_code_update();
2264 return ret;
3bd1e081
MD
2265}
2266
fc6d7a51
JD
2267void lttng_ustctl_flush_buffer(struct lttng_consumer_stream *stream,
2268 int producer_active)
2269{
2270 assert(stream);
2271 assert(stream->ustream);
2272
2273 ustctl_flush_buffer(stream->ustream, producer_active);
2274}
2275
ffe60014 2276/*
e9404c27 2277 * Take a snapshot for a specific stream.
ffe60014
DG
2278 *
2279 * Returns 0 on success, < 0 on error
2280 */
2281int lttng_ustconsumer_take_snapshot(struct lttng_consumer_stream *stream)
3bd1e081 2282{
ffe60014
DG
2283 assert(stream);
2284 assert(stream->ustream);
2285
2286 return ustctl_snapshot(stream->ustream);
3bd1e081
MD
2287}
2288
e9404c27
JG
2289/*
2290 * Sample consumed and produced positions for a specific stream.
2291 *
2292 * Returns 0 on success, < 0 on error.
2293 */
2294int lttng_ustconsumer_sample_snapshot_positions(
2295 struct lttng_consumer_stream *stream)
2296{
2297 assert(stream);
2298 assert(stream->ustream);
2299
2300 return ustctl_snapshot_sample_positions(stream->ustream);
2301}
2302
ffe60014
DG
2303/*
2304 * Get the produced position
2305 *
2306 * Returns 0 on success, < 0 on error
2307 */
2308int lttng_ustconsumer_get_produced_snapshot(
2309 struct lttng_consumer_stream *stream, unsigned long *pos)
3bd1e081 2310{
ffe60014
DG
2311 assert(stream);
2312 assert(stream->ustream);
2313 assert(pos);
7a57cf92 2314
ffe60014
DG
2315 return ustctl_snapshot_get_produced(stream->ustream, pos);
2316}
7a57cf92 2317
10a50311
JD
2318/*
2319 * Get the consumed position
2320 *
2321 * Returns 0 on success, < 0 on error
2322 */
2323int lttng_ustconsumer_get_consumed_snapshot(
2324 struct lttng_consumer_stream *stream, unsigned long *pos)
2325{
2326 assert(stream);
2327 assert(stream->ustream);
2328 assert(pos);
2329
2330 return ustctl_snapshot_get_consumed(stream->ustream, pos);
2331}
2332
84a182ce
DG
2333void lttng_ustconsumer_flush_buffer(struct lttng_consumer_stream *stream,
2334 int producer)
2335{
2336 assert(stream);
2337 assert(stream->ustream);
2338
2339 ustctl_flush_buffer(stream->ustream, producer);
2340}
2341
214f70e0
JR
2342void lttng_ustconsumer_clear_buffer(struct lttng_consumer_stream *stream)
2343{
2344 assert(stream);
2345 assert(stream->ustream);
2346
2347 ustctl_clear_buffer(stream->ustream);
2348}
2349
84a182ce
DG
2350int lttng_ustconsumer_get_current_timestamp(
2351 struct lttng_consumer_stream *stream, uint64_t *ts)
2352{
2353 assert(stream);
2354 assert(stream->ustream);
2355 assert(ts);
2356
2357 return ustctl_get_current_timestamp(stream->ustream, ts);
2358}
2359
fb83fe64
JD
2360int lttng_ustconsumer_get_sequence_number(
2361 struct lttng_consumer_stream *stream, uint64_t *seq)
2362{
2363 assert(stream);
2364 assert(stream->ustream);
2365 assert(seq);
2366
2367 return ustctl_get_sequence_number(stream->ustream, seq);
2368}
2369
ffe60014 2370/*
0dd01979 2371 * Called when the stream signals the consumer that it has hung up.
ffe60014
DG
2372 */
2373void lttng_ustconsumer_on_stream_hangup(struct lttng_consumer_stream *stream)
2374{
2375 assert(stream);
2376 assert(stream->ustream);
2c1dd183 2377
0dd01979
MD
2378 pthread_mutex_lock(&stream->lock);
2379 if (!stream->quiescent) {
2380 ustctl_flush_buffer(stream->ustream, 0);
2381 stream->quiescent = true;
2382 }
2383 pthread_mutex_unlock(&stream->lock);
ffe60014
DG
2384 stream->hangup_flush_done = 1;
2385}
ee77a7b0 2386
ffe60014
DG
2387void lttng_ustconsumer_del_channel(struct lttng_consumer_channel *chan)
2388{
4628484a
MD
2389 int i;
2390
ffe60014
DG
2391 assert(chan);
2392 assert(chan->uchan);
d2956687 2393 assert(chan->buffer_credentials.is_set);
e316aad5 2394
ea88ca2a
MD
2395 if (chan->switch_timer_enabled == 1) {
2396 consumer_timer_switch_stop(chan);
2397 }
4628484a
MD
2398 for (i = 0; i < chan->nr_stream_fds; i++) {
2399 int ret;
2400
2401 ret = close(chan->stream_fds[i]);
2402 if (ret) {
2403 PERROR("close");
2404 }
2405 if (chan->shm_path[0]) {
2406 char shm_path[PATH_MAX];
2407
2408 ret = get_stream_shm_path(shm_path, chan->shm_path, i);
2409 if (ret) {
2410 ERR("Cannot get stream shm path");
2411 }
d2956687 2412 ret = run_as_unlink(shm_path,
ff588497
JR
2413 lttng_credentials_get_uid(LTTNG_OPTIONAL_GET_PTR(
2414 chan->buffer_credentials)),
2415 lttng_credentials_get_gid(LTTNG_OPTIONAL_GET_PTR(
2416 chan->buffer_credentials)));
4628484a 2417 if (ret) {
4628484a
MD
2418 PERROR("unlink %s", shm_path);
2419 }
2420 }
2421 }
3bd1e081
MD
2422}
2423
b83e03c4
MD
2424void lttng_ustconsumer_free_channel(struct lttng_consumer_channel *chan)
2425{
2426 assert(chan);
2427 assert(chan->uchan);
d2956687 2428 assert(chan->buffer_credentials.is_set);
b83e03c4
MD
2429
2430 consumer_metadata_cache_destroy(chan);
2431 ustctl_destroy_channel(chan->uchan);
ea853771
JR
2432 /* Try to rmdir all directories under shm_path root. */
2433 if (chan->root_shm_path[0]) {
602766ec 2434 (void) run_as_rmdir_recursive(chan->root_shm_path,
ff588497
JR
2435 lttng_credentials_get_uid(LTTNG_OPTIONAL_GET_PTR(
2436 chan->buffer_credentials)),
2437 lttng_credentials_get_gid(LTTNG_OPTIONAL_GET_PTR(
2438 chan->buffer_credentials)),
f75c5439 2439 LTTNG_DIRECTORY_HANDLE_SKIP_NON_EMPTY_FLAG);
ea853771 2440 }
b83e03c4
MD
2441 free(chan->stream_fds);
2442}
2443
3bd1e081
MD
2444void lttng_ustconsumer_del_stream(struct lttng_consumer_stream *stream)
2445{
ffe60014
DG
2446 assert(stream);
2447 assert(stream->ustream);
d41f73b7 2448
ea88ca2a
MD
2449 if (stream->chan->switch_timer_enabled == 1) {
2450 consumer_timer_switch_stop(stream->chan);
2451 }
ffe60014
DG
2452 ustctl_destroy_stream(stream->ustream);
2453}
d41f73b7 2454
6d574024
DG
2455int lttng_ustconsumer_get_wakeup_fd(struct lttng_consumer_stream *stream)
2456{
2457 assert(stream);
2458 assert(stream->ustream);
2459
2460 return ustctl_stream_get_wakeup_fd(stream->ustream);
2461}
2462
2463int lttng_ustconsumer_close_wakeup_fd(struct lttng_consumer_stream *stream)
2464{
2465 assert(stream);
2466 assert(stream->ustream);
2467
2468 return ustctl_stream_close_wakeup_fd(stream->ustream);
2469}
2470
94d49140
JD
2471/*
2472 * Write up to one packet from the metadata cache to the channel.
2473 *
577eea73
JG
2474 * Returns the number of bytes pushed from the cache into the ring buffer, or a
2475 * negative value on error.
94d49140
JD
2476 */
2477static
2478int commit_one_metadata_packet(struct lttng_consumer_stream *stream)
2479{
2480 ssize_t write_len;
2481 int ret;
2482
2483 pthread_mutex_lock(&stream->chan->metadata_cache->lock);
9eac9828
JG
2484 if (stream->chan->metadata_cache->contents.size ==
2485 stream->ust_metadata_pushed) {
55954e07
JG
2486 /*
2487 * In the context of a user space metadata channel, a
2488 * change in version can be detected in two ways:
2489 * 1) During the pre-consume of the `read_subbuffer` loop,
2490 * 2) When populating the metadata ring buffer (i.e. here).
2491 *
2492 * This function is invoked when there is no metadata
2493 * available in the ring-buffer. If all data was consumed
2494 * up to the size of the metadata cache, there is no metadata
2495 * to insert in the ring-buffer.
2496 *
2497 * However, the metadata version could still have changed (a
2498 * regeneration without any new data will yield the same cache
2499 * size).
2500 *
2501 * The cache's version is checked for a version change and the
2502 * consumed position is reset if one occurred.
2503 *
2504 * This check is only necessary for the user space domain as
2505 * it has to manage the cache explicitly. If this reset was not
2506 * performed, no metadata would be consumed (and no reset would
2507 * occur as part of the pre-consume) until the metadata size
2508 * exceeded the cache size.
2509 */
2510 if (stream->metadata_version !=
2511 stream->chan->metadata_cache->version) {
2512 metadata_stream_reset_cache_consumed_position(stream);
2513 consumer_stream_metadata_set_version(stream,
2514 stream->chan->metadata_cache->version);
2515 } else {
2516 ret = 0;
2517 goto end;
2518 }
94d49140
JD
2519 }
2520
2521 write_len = ustctl_write_one_packet_to_channel(stream->chan->uchan,
9eac9828
JG
2522 &stream->chan->metadata_cache->contents.data[stream->ust_metadata_pushed],
2523 stream->chan->metadata_cache->contents.size -
2524 stream->ust_metadata_pushed);
94d49140
JD
2525 assert(write_len != 0);
2526 if (write_len < 0) {
2527 ERR("Writing one metadata packet");
f5ba75b4 2528 ret = write_len;
94d49140
JD
2529 goto end;
2530 }
2531 stream->ust_metadata_pushed += write_len;
2532
9eac9828 2533 assert(stream->chan->metadata_cache->contents.size >=
94d49140
JD
2534 stream->ust_metadata_pushed);
2535 ret = write_len;
2536
0d88e046
JG
2537 /*
2538 * Switch packet (but don't open the next one) on every commit of
2539 * a metadata packet. Since the subbuffer is fully filled (with padding,
2540 * if needed), the stream is "quiescent" after this commit.
2541 */
2542 ustctl_flush_buffer(stream->ustream, 1);
2543 stream->quiescent = true;
94d49140
JD
2544end:
2545 pthread_mutex_unlock(&stream->chan->metadata_cache->lock);
2546 return ret;
2547}
2548
309167d2 2549
94d49140
JD
2550/*
2551 * Sync metadata meaning request them to the session daemon and snapshot to the
2552 * metadata thread can consumer them.
2553 *
c585821b
MD
2554 * Metadata stream lock is held here, but we need to release it when
2555 * interacting with sessiond, else we cause a deadlock with live
2556 * awaiting on metadata to be pushed out.
94d49140 2557 *
cdb72e4e 2558 * The RCU read side lock must be held by the caller.
94d49140 2559 */
577eea73
JG
2560enum sync_metadata_status lttng_ustconsumer_sync_metadata(
2561 struct lttng_consumer_local_data *ctx,
cdb72e4e 2562 struct lttng_consumer_stream *metadata_stream)
94d49140
JD
2563{
2564 int ret;
577eea73 2565 enum sync_metadata_status status;
cdb72e4e 2566 struct lttng_consumer_channel *metadata_channel;
94d49140
JD
2567
2568 assert(ctx);
cdb72e4e 2569 assert(metadata_stream);
94d49140 2570
cdb72e4e
JG
2571 metadata_channel = metadata_stream->chan;
2572 pthread_mutex_unlock(&metadata_stream->lock);
94d49140
JD
2573 /*
2574 * Request metadata from the sessiond, but don't wait for the flush
2575 * because we locked the metadata thread.
2576 */
cdb72e4e
JG
2577 ret = lttng_ustconsumer_request_metadata(ctx, metadata_channel, 0, 0);
2578 pthread_mutex_lock(&metadata_stream->lock);
94d49140 2579 if (ret < 0) {
577eea73 2580 status = SYNC_METADATA_STATUS_ERROR;
94d49140
JD
2581 goto end;
2582 }
2583
cdb72e4e
JG
2584 /*
2585 * The metadata stream and channel can be deleted while the
2586 * metadata stream lock was released. The streamed is checked
2587 * for deletion before we use it further.
2588 *
2589 * Note that it is safe to access a logically-deleted stream since its
2590 * existence is still guaranteed by the RCU read side lock. However,
2591 * it should no longer be used. The close/deletion of the metadata
2592 * channel and stream already guarantees that all metadata has been
2593 * consumed. Therefore, there is nothing left to do in this function.
2594 */
2595 if (consumer_stream_is_deleted(metadata_stream)) {
2596 DBG("Metadata stream %" PRIu64 " was deleted during the metadata synchronization",
2597 metadata_stream->key);
577eea73 2598 status = SYNC_METADATA_STATUS_NO_DATA;
cdb72e4e
JG
2599 goto end;
2600 }
2601
2602 ret = commit_one_metadata_packet(metadata_stream);
577eea73
JG
2603 if (ret < 0) {
2604 status = SYNC_METADATA_STATUS_ERROR;
94d49140
JD
2605 goto end;
2606 } else if (ret > 0) {
577eea73
JG
2607 status = SYNC_METADATA_STATUS_NEW_DATA;
2608 } else /* ret == 0 */ {
2609 status = SYNC_METADATA_STATUS_NO_DATA;
2610 goto end;
94d49140
JD
2611 }
2612
cdb72e4e 2613 ret = ustctl_snapshot(metadata_stream->ustream);
94d49140 2614 if (ret < 0) {
577eea73
JG
2615 ERR("Failed to take a snapshot of the metadata ring-buffer positions, ret = %d", ret);
2616 status = SYNC_METADATA_STATUS_ERROR;
94d49140
JD
2617 goto end;
2618 }
2619
94d49140 2620end:
577eea73 2621 return status;
94d49140
JD
2622}
2623
02b3d176
DG
2624/*
2625 * Return 0 on success else a negative value.
2626 */
2627static int notify_if_more_data(struct lttng_consumer_stream *stream,
2628 struct lttng_consumer_local_data *ctx)
2629{
2630 int ret;
2631 struct ustctl_consumer_stream *ustream;
2632
2633 assert(stream);
2634 assert(ctx);
2635
2636 ustream = stream->ustream;
2637
2638 /*
2639 * First, we are going to check if there is a new subbuffer available
2640 * before reading the stream wait_fd.
2641 */
2642 /* Get the next subbuffer */
2643 ret = ustctl_get_next_subbuf(ustream);
2644 if (ret) {
2645 /* No more data found, flag the stream. */
2646 stream->has_data = 0;
2647 ret = 0;
2648 goto end;
2649 }
2650
5420e5db 2651 ret = ustctl_put_subbuf(ustream);
02b3d176
DG
2652 assert(!ret);
2653
2654 /* This stream still has data. Flag it and wake up the data thread. */
2655 stream->has_data = 1;
2656
2657 if (stream->monitor && !stream->hangup_flush_done && !ctx->has_wakeup) {
2658 ssize_t writelen;
2659
2660 writelen = lttng_pipe_write(ctx->consumer_wakeup_pipe, "!", 1);
2661 if (writelen < 0 && errno != EAGAIN && errno != EWOULDBLOCK) {
2662 ret = writelen;
2663 goto end;
2664 }
2665
2666 /* The wake up pipe has been notified. */
2667 ctx->has_wakeup = 1;
2668 }
2669 ret = 0;
2670
2671end:
2672 return ret;
2673}
2674
6f9449c2 2675static int consumer_stream_ust_on_wake_up(struct lttng_consumer_stream *stream)
fb83fe64 2676{
6f9449c2 2677 int ret = 0;
fb83fe64 2678
fb83fe64 2679 /*
6f9449c2
JG
2680 * We can consume the 1 byte written into the wait_fd by
2681 * UST. Don't trigger error if we cannot read this one byte
2682 * (read returns 0), or if the error is EAGAIN or EWOULDBLOCK.
2683 *
2684 * This is only done when the stream is monitored by a thread,
2685 * before the flush is done after a hangup and if the stream
2686 * is not flagged with data since there might be nothing to
2687 * consume in the wait fd but still have data available
2688 * flagged by the consumer wake up pipe.
fb83fe64 2689 */
6f9449c2
JG
2690 if (stream->monitor && !stream->hangup_flush_done && !stream->has_data) {
2691 char dummy;
2692 ssize_t readlen;
2693
2694 readlen = lttng_read(stream->wait_fd, &dummy, 1);
2695 if (readlen < 0 && errno != EAGAIN && errno != EWOULDBLOCK) {
2696 ret = readlen;
2697 }
fb83fe64 2698 }
fb83fe64 2699
6f9449c2
JG
2700 return ret;
2701}
2702
2703static int extract_common_subbuffer_info(struct lttng_consumer_stream *stream,
2704 struct stream_subbuffer *subbuf)
2705{
2706 int ret;
2707
2708 ret = ustctl_get_subbuf_size(
2709 stream->ustream, &subbuf->info.data.subbuf_size);
2710 if (ret) {
fb83fe64
JD
2711 goto end;
2712 }
6f9449c2
JG
2713
2714 ret = ustctl_get_padded_subbuf_size(
2715 stream->ustream, &subbuf->info.data.padded_subbuf_size);
2716 if (ret) {
2717 goto end;
fb83fe64 2718 }
fb83fe64
JD
2719
2720end:
2721 return ret;
2722}
2723
6f9449c2
JG
2724static int extract_metadata_subbuffer_info(struct lttng_consumer_stream *stream,
2725 struct stream_subbuffer *subbuf)
d41f73b7 2726{
6f9449c2 2727 int ret;
ffe60014 2728
6f9449c2
JG
2729 ret = extract_common_subbuffer_info(stream, subbuf);
2730 if (ret) {
2731 goto end;
2732 }
d41f73b7 2733
55954e07 2734 subbuf->info.metadata.version = stream->metadata_version;
ffe60014 2735
6f9449c2
JG
2736end:
2737 return ret;
2738}
d41f73b7 2739
6f9449c2
JG
2740static int extract_data_subbuffer_info(struct lttng_consumer_stream *stream,
2741 struct stream_subbuffer *subbuf)
2742{
2743 int ret;
c617c0c6 2744
6f9449c2
JG
2745 ret = extract_common_subbuffer_info(stream, subbuf);
2746 if (ret) {
2747 goto end;
02d02e31
JD
2748 }
2749
6f9449c2
JG
2750 ret = ustctl_get_packet_size(
2751 stream->ustream, &subbuf->info.data.packet_size);
2752 if (ret < 0) {
2753 PERROR("Failed to get sub-buffer packet size");
2754 goto end;
2755 }
04ef1097 2756
6f9449c2
JG
2757 ret = ustctl_get_content_size(
2758 stream->ustream, &subbuf->info.data.content_size);
2759 if (ret < 0) {
2760 PERROR("Failed to get sub-buffer content size");
2761 goto end;
d41f73b7 2762 }
309167d2 2763
6f9449c2
JG
2764 ret = ustctl_get_timestamp_begin(
2765 stream->ustream, &subbuf->info.data.timestamp_begin);
2766 if (ret < 0) {
2767 PERROR("Failed to get sub-buffer begin timestamp");
2768 goto end;
2769 }
fb83fe64 2770
6f9449c2
JG
2771 ret = ustctl_get_timestamp_end(
2772 stream->ustream, &subbuf->info.data.timestamp_end);
2773 if (ret < 0) {
2774 PERROR("Failed to get sub-buffer end timestamp");
2775 goto end;
2776 }
2777
2778 ret = ustctl_get_events_discarded(
2779 stream->ustream, &subbuf->info.data.events_discarded);
2780 if (ret) {
2781 PERROR("Failed to get sub-buffer events discarded count");
2782 goto end;
2783 }
2784
2785 ret = ustctl_get_sequence_number(stream->ustream,
2786 &subbuf->info.data.sequence_number.value);
2787 if (ret) {
2788 /* May not be supported by older LTTng-modules. */
2789 if (ret != -ENOTTY) {
2790 PERROR("Failed to get sub-buffer sequence number");
2791 goto end;
fb83fe64 2792 }
1c20f0e2 2793 } else {
6f9449c2 2794 subbuf->info.data.sequence_number.is_set = true;
309167d2
JD
2795 }
2796
6f9449c2
JG
2797 ret = ustctl_get_stream_id(
2798 stream->ustream, &subbuf->info.data.stream_id);
2799 if (ret < 0) {
2800 PERROR("Failed to get stream id");
2801 goto end;
2802 }
1d4dfdef 2803
6f9449c2
JG
2804 ret = ustctl_get_instance_id(stream->ustream,
2805 &subbuf->info.data.stream_instance_id.value);
2806 if (ret) {
2807 /* May not be supported by older LTTng-modules. */
2808 if (ret != -ENOTTY) {
2809 PERROR("Failed to get stream instance id");
2810 goto end;
2811 }
2812 } else {
2813 subbuf->info.data.stream_instance_id.is_set = true;
2814 }
2815end:
2816 return ret;
2817}
1d4dfdef 2818
6f9449c2
JG
2819static int get_next_subbuffer_common(struct lttng_consumer_stream *stream,
2820 struct stream_subbuffer *subbuffer)
2821{
2822 int ret;
2823 const char *addr;
1d4dfdef 2824
6f9449c2
JG
2825 ret = stream->read_subbuffer_ops.extract_subbuffer_info(
2826 stream, subbuffer);
2827 if (ret) {
2828 goto end;
2829 }
02d02e31 2830
6f9449c2 2831 ret = get_current_subbuf_addr(stream, &addr);
128708c3 2832 if (ret) {
6f9449c2 2833 goto end;
128708c3
JG
2834 }
2835
6f9449c2
JG
2836 subbuffer->buffer.buffer = lttng_buffer_view_init(
2837 addr, 0, subbuffer->info.data.padded_subbuf_size);
2838 assert(subbuffer->buffer.buffer.data != NULL);
2839end:
2840 return ret;
2841}
fd424d99 2842
6f9449c2
JG
2843static int get_next_subbuffer(struct lttng_consumer_stream *stream,
2844 struct stream_subbuffer *subbuffer)
2845{
2846 int ret;
331744e3 2847
6f9449c2
JG
2848 ret = ustctl_get_next_subbuf(stream->ustream);
2849 if (ret) {
2850 goto end;
02b3d176
DG
2851 }
2852
6f9449c2
JG
2853 ret = get_next_subbuffer_common(stream, subbuffer);
2854 if (ret) {
23d56598 2855 goto end;
1c20f0e2 2856 }
6f9449c2
JG
2857end:
2858 return ret;
2859}
1c20f0e2 2860
6f9449c2
JG
2861static int get_next_subbuffer_metadata(struct lttng_consumer_stream *stream,
2862 struct stream_subbuffer *subbuffer)
2863{
2864 int ret;
f5ba75b4
JG
2865 bool cache_empty;
2866 bool got_subbuffer;
2867 bool coherent;
2868 bool buffer_empty;
2869 unsigned long consumed_pos, produced_pos;
6f9449c2 2870
f5ba75b4
JG
2871 do {
2872 ret = ustctl_get_next_subbuf(stream->ustream);
2873 if (ret == 0) {
2874 got_subbuffer = true;
2875 } else {
2876 got_subbuffer = false;
2877 if (ret != -EAGAIN) {
2878 /* Fatal error. */
2879 goto end;
2880 }
c585821b
MD
2881 }
2882
f5ba75b4
JG
2883 /*
2884 * Determine if the cache is empty and ensure that a sub-buffer
2885 * is made available if the cache is not empty.
2886 */
2887 if (!got_subbuffer) {
2888 ret = commit_one_metadata_packet(stream);
2889 if (ret < 0 && ret != -ENOBUFS) {
2890 goto end;
2891 } else if (ret == 0) {
2892 /* Not an error, the cache is empty. */
2893 cache_empty = true;
2894 ret = -ENODATA;
2895 goto end;
2896 } else {
2897 cache_empty = false;
2898 }
2899 } else {
2900 pthread_mutex_lock(&stream->chan->metadata_cache->lock);
9eac9828
JG
2901 cache_empty = stream->chan->metadata_cache->contents.size ==
2902 stream->ust_metadata_pushed;
f5ba75b4 2903 pthread_mutex_unlock(&stream->chan->metadata_cache->lock);
94d49140 2904 }
f5ba75b4 2905 } while (!got_subbuffer);
94d49140 2906
f5ba75b4 2907 /* Populate sub-buffer infos and view. */
6f9449c2
JG
2908 ret = get_next_subbuffer_common(stream, subbuffer);
2909 if (ret) {
2910 goto end;
309167d2 2911 }
f5ba75b4
JG
2912
2913 ret = lttng_ustconsumer_sample_snapshot_positions(stream);
2914 if (ret < 0) {
2915 /*
2916 * -EAGAIN is not expected since we got a sub-buffer and haven't
2917 * pushed the consumption position yet (on put_next).
2918 */
2919 PERROR("Failed to take a snapshot of metadata buffer positions");
2920 goto end;
2921 }
2922
2923 ret = lttng_ustconsumer_get_consumed_snapshot(stream, &consumed_pos);
2924 if (ret) {
2925 PERROR("Failed to get metadata consumed position");
2926 goto end;
2927 }
2928
2929 ret = lttng_ustconsumer_get_produced_snapshot(stream, &produced_pos);
2930 if (ret) {
2931 PERROR("Failed to get metadata produced position");
2932 goto end;
2933 }
2934
2935 /* Last sub-buffer of the ring buffer ? */
2936 buffer_empty = (consumed_pos + stream->max_sb_size) == produced_pos;
2937
2938 /*
2939 * The sessiond registry lock ensures that coherent units of metadata
2940 * are pushed to the consumer daemon at once. Hence, if a sub-buffer is
2941 * acquired, the cache is empty, and it is the only available sub-buffer
2942 * available, it is safe to assume that it is "coherent".
2943 */
2944 coherent = got_subbuffer && cache_empty && buffer_empty;
2945
2946 LTTNG_OPTIONAL_SET(&subbuffer->info.metadata.coherent, coherent);
23d56598 2947end:
d41f73b7
MD
2948 return ret;
2949}
2950
6f9449c2
JG
2951static int put_next_subbuffer(struct lttng_consumer_stream *stream,
2952 struct stream_subbuffer *subbuffer)
2953{
2954 const int ret = ustctl_put_next_subbuf(stream->ustream);
2955
2956 assert(ret == 0);
2957 return ret;
2958}
2959
2960static int signal_metadata(struct lttng_consumer_stream *stream,
2961 struct lttng_consumer_local_data *ctx)
2962{
8db3acaf 2963 ASSERT_LOCKED(stream->metadata_rdv_lock);
6f9449c2
JG
2964 return pthread_cond_broadcast(&stream->metadata_rdv) ? -errno : 0;
2965}
2966
f5ba75b4 2967static int lttng_ustconsumer_set_stream_ops(
6f9449c2
JG
2968 struct lttng_consumer_stream *stream)
2969{
f5ba75b4
JG
2970 int ret = 0;
2971
6f9449c2
JG
2972 stream->read_subbuffer_ops.on_wake_up = consumer_stream_ust_on_wake_up;
2973 if (stream->metadata_flag) {
2974 stream->read_subbuffer_ops.get_next_subbuffer =
2975 get_next_subbuffer_metadata;
2976 stream->read_subbuffer_ops.extract_subbuffer_info =
2977 extract_metadata_subbuffer_info;
2978 stream->read_subbuffer_ops.reset_metadata =
55954e07 2979 metadata_stream_reset_cache_consumed_position;
f5ba75b4
JG
2980 if (stream->chan->is_live) {
2981 stream->read_subbuffer_ops.on_sleep = signal_metadata;
2982 ret = consumer_stream_enable_metadata_bucketization(
2983 stream);
2984 if (ret) {
2985 goto end;
2986 }
2987 }
6f9449c2
JG
2988 } else {
2989 stream->read_subbuffer_ops.get_next_subbuffer =
2990 get_next_subbuffer;
2991 stream->read_subbuffer_ops.extract_subbuffer_info =
2992 extract_data_subbuffer_info;
2993 stream->read_subbuffer_ops.on_sleep = notify_if_more_data;
2994 if (stream->chan->is_live) {
2995 stream->read_subbuffer_ops.send_live_beacon =
2996 consumer_flush_ust_index;
2997 }
2998 }
2999
3000 stream->read_subbuffer_ops.put_next_subbuffer = put_next_subbuffer;
f5ba75b4
JG
3001end:
3002 return ret;
6f9449c2
JG
3003}
3004
ffe60014
DG
3005/*
3006 * Called when a stream is created.
fe4477ee
JD
3007 *
3008 * Return 0 on success or else a negative value.
ffe60014 3009 */
d41f73b7
MD
3010int lttng_ustconsumer_on_recv_stream(struct lttng_consumer_stream *stream)
3011{
fe4477ee
JD
3012 int ret;
3013
10a50311
JD
3014 assert(stream);
3015
d2956687
JG
3016 /*
3017 * Don't create anything if this is set for streaming or if there is
3018 * no current trace chunk on the parent channel.
3019 */
3020 if (stream->net_seq_idx == (uint64_t) -1ULL && stream->chan->monitor &&
3021 stream->chan->trace_chunk) {
3022 ret = consumer_stream_create_output_files(stream, true);
3023 if (ret) {
fe4477ee
JD
3024 goto error;
3025 }
fe4477ee 3026 }
6f9449c2
JG
3027
3028 lttng_ustconsumer_set_stream_ops(stream);
fe4477ee
JD
3029 ret = 0;
3030
3031error:
3032 return ret;
d41f73b7 3033}
ca22feea
DG
3034
3035/*
3036 * Check if data is still being extracted from the buffers for a specific
4e9a4686
DG
3037 * stream. Consumer data lock MUST be acquired before calling this function
3038 * and the stream lock.
ca22feea 3039 *
6d805429 3040 * Return 1 if the traced data are still getting read else 0 meaning that the
ca22feea
DG
3041 * data is available for trace viewer reading.
3042 */
6d805429 3043int lttng_ustconsumer_data_pending(struct lttng_consumer_stream *stream)
ca22feea
DG
3044{
3045 int ret;
3046
3047 assert(stream);
ffe60014 3048 assert(stream->ustream);
b1316da1 3049 ASSERT_LOCKED(stream->lock);
ca22feea 3050
6d805429 3051 DBG("UST consumer checking data pending");
c8f59ee5 3052
ca6b395f
MD
3053 if (stream->endpoint_status != CONSUMER_ENDPOINT_ACTIVE) {
3054 ret = 0;
3055 goto end;
3056 }
3057
04ef1097 3058 if (stream->chan->type == CONSUMER_CHANNEL_TYPE_METADATA) {
e6ee4eab
DG
3059 uint64_t contiguous, pushed;
3060
3061 /* Ease our life a bit. */
934ba8fd 3062 pthread_mutex_lock(&stream->chan->metadata_cache->lock);
9eac9828 3063 contiguous = stream->chan->metadata_cache->contents.size;
934ba8fd 3064 pthread_mutex_unlock(&stream->chan->metadata_cache->lock);
e6ee4eab
DG
3065 pushed = stream->ust_metadata_pushed;
3066
04ef1097
MD
3067 /*
3068 * We can simply check whether all contiguously available data
3069 * has been pushed to the ring buffer, since the push operation
3070 * is performed within get_next_subbuf(), and because both
3071 * get_next_subbuf() and put_next_subbuf() are issued atomically
3072 * thanks to the stream lock within
3073 * lttng_ustconsumer_read_subbuffer(). This basically means that
3074 * whetnever ust_metadata_pushed is incremented, the associated
3075 * metadata has been consumed from the metadata stream.
3076 */
3077 DBG("UST consumer metadata pending check: contiguous %" PRIu64 " vs pushed %" PRIu64,
e6ee4eab 3078 contiguous, pushed);
aa01b94c 3079 assert(((int64_t) (contiguous - pushed)) >= 0);
e6ee4eab 3080 if ((contiguous != pushed) ||
6acdf328 3081 (((int64_t) contiguous - pushed) > 0 || contiguous == 0)) {
04ef1097
MD
3082 ret = 1; /* Data is pending */
3083 goto end;
3084 }
3085 } else {
3086 ret = ustctl_get_next_subbuf(stream->ustream);
3087 if (ret == 0) {
3088 /*
3089 * There is still data so let's put back this
3090 * subbuffer.
3091 */
3092 ret = ustctl_put_subbuf(stream->ustream);
3093 assert(ret == 0);
3094 ret = 1; /* Data is pending */
3095 goto end;
3096 }
ca22feea
DG
3097 }
3098
6d805429
DG
3099 /* Data is NOT pending so ready to be read. */
3100 ret = 0;
ca22feea 3101
6efae65e
DG
3102end:
3103 return ret;
ca22feea 3104}
d88aee68 3105
6d574024
DG
3106/*
3107 * Stop a given metadata channel timer if enabled and close the wait fd which
3108 * is the poll pipe of the metadata stream.
3109 *
d2c82a5a 3110 * This MUST be called with the metadata channel lock acquired.
6d574024
DG
3111 */
3112void lttng_ustconsumer_close_metadata(struct lttng_consumer_channel *metadata)
3113{
3114 int ret;
3115
3116 assert(metadata);
3117 assert(metadata->type == CONSUMER_CHANNEL_TYPE_METADATA);
3118
3119 DBG("Closing metadata channel key %" PRIu64, metadata->key);
3120
3121 if (metadata->switch_timer_enabled == 1) {
3122 consumer_timer_switch_stop(metadata);
3123 }
3124
3125 if (!metadata->metadata_stream) {
3126 goto end;
3127 }
3128
3129 /*
3130 * Closing write side so the thread monitoring the stream wakes up if any
3131 * and clean the metadata stream.
3132 */
3133 if (metadata->metadata_stream->ust_metadata_poll_pipe[1] >= 0) {
3134 ret = close(metadata->metadata_stream->ust_metadata_poll_pipe[1]);
3135 if (ret < 0) {
3136 PERROR("closing metadata pipe write side");
3137 }
3138 metadata->metadata_stream->ust_metadata_poll_pipe[1] = -1;
3139 }
3140
3141end:
3142 return;
3143}
3144
d88aee68
DG
3145/*
3146 * Close every metadata stream wait fd of the metadata hash table. This
3147 * function MUST be used very carefully so not to run into a race between the
3148 * metadata thread handling streams and this function closing their wait fd.
3149 *
3150 * For UST, this is used when the session daemon hangs up. Its the metadata
3151 * producer so calling this is safe because we are assured that no state change
3152 * can occur in the metadata thread for the streams in the hash table.
3153 */
6d574024 3154void lttng_ustconsumer_close_all_metadata(struct lttng_ht *metadata_ht)
d88aee68 3155{
d88aee68
DG
3156 struct lttng_ht_iter iter;
3157 struct lttng_consumer_stream *stream;
3158
3159 assert(metadata_ht);
3160 assert(metadata_ht->ht);
3161
3162 DBG("UST consumer closing all metadata streams");
3163
3164 rcu_read_lock();
3165 cds_lfht_for_each_entry(metadata_ht->ht, &iter.iter, stream,
3166 node.node) {
9ce5646a
MD
3167
3168 health_code_update();
3169
be2b50c7 3170 pthread_mutex_lock(&stream->chan->lock);
6d574024 3171 lttng_ustconsumer_close_metadata(stream->chan);
be2b50c7
DG
3172 pthread_mutex_unlock(&stream->chan->lock);
3173
d88aee68
DG
3174 }
3175 rcu_read_unlock();
3176}
d8ef542d
MD
3177
3178void lttng_ustconsumer_close_stream_wakeup(struct lttng_consumer_stream *stream)
3179{
3180 int ret;
3181
3182 ret = ustctl_stream_close_wakeup_fd(stream->ustream);
3183 if (ret < 0) {
3184 ERR("Unable to close wakeup fd");
3185 }
3186}
331744e3 3187
f666ae70
MD
3188/*
3189 * Please refer to consumer-timer.c before adding any lock within this
3190 * function or any of its callees. Timers have a very strict locking
3191 * semantic with respect to teardown. Failure to respect this semantic
3192 * introduces deadlocks.
c585821b
MD
3193 *
3194 * DON'T hold the metadata lock when calling this function, else this
3195 * can cause deadlock involving consumer awaiting for metadata to be
3196 * pushed out due to concurrent interaction with the session daemon.
f666ae70 3197 */
331744e3 3198int lttng_ustconsumer_request_metadata(struct lttng_consumer_local_data *ctx,
94d49140 3199 struct lttng_consumer_channel *channel, int timer, int wait)
331744e3
JD
3200{
3201 struct lttcomm_metadata_request_msg request;
3202 struct lttcomm_consumer_msg msg;
0c759fc9 3203 enum lttcomm_return_code ret_code = LTTCOMM_CONSUMERD_SUCCESS;
93ec662e 3204 uint64_t len, key, offset, version;
331744e3
JD
3205 int ret;
3206
3207 assert(channel);
3208 assert(channel->metadata_cache);
3209
53efb85a
MD
3210 memset(&request, 0, sizeof(request));
3211
331744e3
JD
3212 /* send the metadata request to sessiond */
3213 switch (consumer_data.type) {
3214 case LTTNG_CONSUMER64_UST:
3215 request.bits_per_long = 64;