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