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