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