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