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