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