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