aede53658736e2037d8f90bb53b050c4bb2c2997
[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 ssize_t ret;
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 ret = lttcomm_recv_unix_sock(sock, &msg, sizeof(msg));
1374 if (ret != sizeof(msg)) {
1375 DBG("Consumer received unexpected message size %zd (expects %zu)",
1376 ret, sizeof(msg));
1377 /*
1378 * The ret value might 0 meaning an orderly shutdown but this is ok
1379 * since the caller handles this.
1380 */
1381 if (ret > 0) {
1382 lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_ERROR_RECV_CMD);
1383 ret = -1;
1384 }
1385 return ret;
1386 }
1387
1388 health_code_update();
1389
1390 /* deprecated */
1391 assert(msg.cmd_type != LTTNG_CONSUMER_STOP);
1392
1393 health_code_update();
1394
1395 /* relayd needs RCU read-side lock */
1396 rcu_read_lock();
1397
1398 switch (msg.cmd_type) {
1399 case LTTNG_CONSUMER_ADD_RELAYD_SOCKET:
1400 {
1401 /* Session daemon status message are handled in the following call. */
1402 consumer_add_relayd_socket(msg.u.relayd_sock.net_index,
1403 msg.u.relayd_sock.type, ctx, sock, consumer_sockpoll,
1404 &msg.u.relayd_sock.sock, msg.u.relayd_sock.session_id,
1405 msg.u.relayd_sock.relayd_session_id);
1406 goto end_nosignal;
1407 }
1408 case LTTNG_CONSUMER_DESTROY_RELAYD:
1409 {
1410 uint64_t index = msg.u.destroy_relayd.net_seq_idx;
1411 struct consumer_relayd_sock_pair *relayd;
1412
1413 DBG("UST consumer destroying relayd %" PRIu64, index);
1414
1415 /* Get relayd reference if exists. */
1416 relayd = consumer_find_relayd(index);
1417 if (relayd == NULL) {
1418 DBG("Unable to find relayd %" PRIu64, index);
1419 ret_code = LTTCOMM_CONSUMERD_RELAYD_FAIL;
1420 }
1421
1422 /*
1423 * Each relayd socket pair has a refcount of stream attached to it
1424 * which tells if the relayd is still active or not depending on the
1425 * refcount value.
1426 *
1427 * This will set the destroy flag of the relayd object and destroy it
1428 * if the refcount reaches zero when called.
1429 *
1430 * The destroy can happen either here or when a stream fd hangs up.
1431 */
1432 if (relayd) {
1433 consumer_flag_relayd_for_destroy(relayd);
1434 }
1435
1436 goto end_msg_sessiond;
1437 }
1438 case LTTNG_CONSUMER_UPDATE_STREAM:
1439 {
1440 rcu_read_unlock();
1441 return -ENOSYS;
1442 }
1443 case LTTNG_CONSUMER_DATA_PENDING:
1444 {
1445 int ret, is_data_pending;
1446 uint64_t id = msg.u.data_pending.session_id;
1447
1448 DBG("UST consumer data pending command for id %" PRIu64, id);
1449
1450 is_data_pending = consumer_data_pending(id);
1451
1452 /* Send back returned value to session daemon */
1453 ret = lttcomm_send_unix_sock(sock, &is_data_pending,
1454 sizeof(is_data_pending));
1455 if (ret < 0) {
1456 DBG("Error when sending the data pending ret code: %d", ret);
1457 goto error_fatal;
1458 }
1459
1460 /*
1461 * No need to send back a status message since the data pending
1462 * returned value is the response.
1463 */
1464 break;
1465 }
1466 case LTTNG_CONSUMER_ASK_CHANNEL_CREATION:
1467 {
1468 int ret;
1469 struct ustctl_consumer_channel_attr attr;
1470 const uint64_t chunk_id = msg.u.ask_channel.chunk_id.value;
1471 const struct lttng_credentials buffer_credentials = {
1472 .uid = LTTNG_OPTIONAL_INIT_VALUE(msg.u.ask_channel.buffer_credentials.uid),
1473 .gid = LTTNG_OPTIONAL_INIT_VALUE(msg.u.ask_channel.buffer_credentials.gid),
1474 };
1475
1476 /* Create a plain object and reserve a channel key. */
1477 channel = consumer_allocate_channel(
1478 msg.u.ask_channel.key,
1479 msg.u.ask_channel.session_id,
1480 msg.u.ask_channel.chunk_id.is_set ?
1481 &chunk_id : NULL,
1482 msg.u.ask_channel.pathname,
1483 msg.u.ask_channel.name,
1484 msg.u.ask_channel.relayd_id,
1485 (enum lttng_event_output) msg.u.ask_channel.output,
1486 msg.u.ask_channel.tracefile_size,
1487 msg.u.ask_channel.tracefile_count,
1488 msg.u.ask_channel.session_id_per_pid,
1489 msg.u.ask_channel.monitor,
1490 msg.u.ask_channel.live_timer_interval,
1491 msg.u.ask_channel.is_live,
1492 msg.u.ask_channel.root_shm_path,
1493 msg.u.ask_channel.shm_path);
1494 if (!channel) {
1495 goto end_channel_error;
1496 }
1497
1498 LTTNG_OPTIONAL_SET(&channel->buffer_credentials,
1499 buffer_credentials);
1500
1501 /*
1502 * Assign UST application UID to the channel. This value is ignored for
1503 * per PID buffers. This is specific to UST thus setting this after the
1504 * allocation.
1505 */
1506 channel->ust_app_uid = msg.u.ask_channel.ust_app_uid;
1507
1508 /* Build channel attributes from received message. */
1509 attr.subbuf_size = msg.u.ask_channel.subbuf_size;
1510 attr.num_subbuf = msg.u.ask_channel.num_subbuf;
1511 attr.overwrite = msg.u.ask_channel.overwrite;
1512 attr.switch_timer_interval = msg.u.ask_channel.switch_timer_interval;
1513 attr.read_timer_interval = msg.u.ask_channel.read_timer_interval;
1514 attr.chan_id = msg.u.ask_channel.chan_id;
1515 memcpy(attr.uuid, msg.u.ask_channel.uuid, sizeof(attr.uuid));
1516 attr.blocking_timeout= msg.u.ask_channel.blocking_timeout;
1517
1518 /* Match channel buffer type to the UST abi. */
1519 switch (msg.u.ask_channel.output) {
1520 case LTTNG_EVENT_MMAP:
1521 default:
1522 attr.output = LTTNG_UST_ABI_MMAP;
1523 break;
1524 }
1525
1526 /* Translate and save channel type. */
1527 switch (msg.u.ask_channel.type) {
1528 case LTTNG_UST_ABI_CHAN_PER_CPU:
1529 channel->type = CONSUMER_CHANNEL_TYPE_DATA;
1530 attr.type = LTTNG_UST_ABI_CHAN_PER_CPU;
1531 /*
1532 * Set refcount to 1 for owner. Below, we will
1533 * pass ownership to the
1534 * consumer_thread_channel_poll() thread.
1535 */
1536 channel->refcount = 1;
1537 break;
1538 case LTTNG_UST_ABI_CHAN_METADATA:
1539 channel->type = CONSUMER_CHANNEL_TYPE_METADATA;
1540 attr.type = LTTNG_UST_ABI_CHAN_METADATA;
1541 break;
1542 default:
1543 assert(0);
1544 goto error_fatal;
1545 };
1546
1547 health_code_update();
1548
1549 ret = ask_channel(ctx, channel, &attr);
1550 if (ret < 0) {
1551 goto end_channel_error;
1552 }
1553
1554 if (msg.u.ask_channel.type == LTTNG_UST_ABI_CHAN_METADATA) {
1555 ret = consumer_metadata_cache_allocate(channel);
1556 if (ret < 0) {
1557 ERR("Allocating metadata cache");
1558 goto end_channel_error;
1559 }
1560 consumer_timer_switch_start(channel, attr.switch_timer_interval);
1561 attr.switch_timer_interval = 0;
1562 } else {
1563 int monitor_start_ret;
1564
1565 consumer_timer_live_start(channel,
1566 msg.u.ask_channel.live_timer_interval);
1567 monitor_start_ret = consumer_timer_monitor_start(
1568 channel,
1569 msg.u.ask_channel.monitor_timer_interval);
1570 if (monitor_start_ret < 0) {
1571 ERR("Starting channel monitoring timer failed");
1572 goto end_channel_error;
1573 }
1574 }
1575
1576 health_code_update();
1577
1578 /*
1579 * Add the channel to the internal state AFTER all streams were created
1580 * and successfully sent to session daemon. This way, all streams must
1581 * be ready before this channel is visible to the threads.
1582 * If add_channel succeeds, ownership of the channel is
1583 * passed to consumer_thread_channel_poll().
1584 */
1585 ret = add_channel(channel, ctx);
1586 if (ret < 0) {
1587 if (msg.u.ask_channel.type == LTTNG_UST_ABI_CHAN_METADATA) {
1588 if (channel->switch_timer_enabled == 1) {
1589 consumer_timer_switch_stop(channel);
1590 }
1591 consumer_metadata_cache_destroy(channel);
1592 }
1593 if (channel->live_timer_enabled == 1) {
1594 consumer_timer_live_stop(channel);
1595 }
1596 if (channel->monitor_timer_enabled == 1) {
1597 consumer_timer_monitor_stop(channel);
1598 }
1599 goto end_channel_error;
1600 }
1601
1602 health_code_update();
1603
1604 /*
1605 * Channel and streams are now created. Inform the session daemon that
1606 * everything went well and should wait to receive the channel and
1607 * streams with ustctl API.
1608 */
1609 ret = consumer_send_status_channel(sock, channel);
1610 if (ret < 0) {
1611 /*
1612 * There is probably a problem on the socket.
1613 */
1614 goto error_fatal;
1615 }
1616
1617 break;
1618 }
1619 case LTTNG_CONSUMER_GET_CHANNEL:
1620 {
1621 int ret, relayd_err = 0;
1622 uint64_t key = msg.u.get_channel.key;
1623 struct lttng_consumer_channel *channel;
1624
1625 channel = consumer_find_channel(key);
1626 if (!channel) {
1627 ERR("UST consumer get channel key %" PRIu64 " not found", key);
1628 ret_code = LTTCOMM_CONSUMERD_CHAN_NOT_FOUND;
1629 goto end_get_channel;
1630 }
1631
1632 health_code_update();
1633
1634 /* Send the channel to sessiond (and relayd, if applicable). */
1635 ret = send_channel_to_sessiond_and_relayd(sock, channel, ctx,
1636 &relayd_err);
1637 if (ret < 0) {
1638 if (relayd_err) {
1639 /*
1640 * We were unable to send to the relayd the stream so avoid
1641 * sending back a fatal error to the thread since this is OK
1642 * and the consumer can continue its work. The above call
1643 * has sent the error status message to the sessiond.
1644 */
1645 goto end_get_channel_nosignal;
1646 }
1647 /*
1648 * The communicaton was broken hence there is a bad state between
1649 * the consumer and sessiond so stop everything.
1650 */
1651 goto error_get_channel_fatal;
1652 }
1653
1654 health_code_update();
1655
1656 /*
1657 * In no monitor mode, the streams ownership is kept inside the channel
1658 * so don't send them to the data thread.
1659 */
1660 if (!channel->monitor) {
1661 goto end_get_channel;
1662 }
1663
1664 ret = send_streams_to_thread(channel, ctx);
1665 if (ret < 0) {
1666 /*
1667 * If we are unable to send the stream to the thread, there is
1668 * a big problem so just stop everything.
1669 */
1670 goto error_get_channel_fatal;
1671 }
1672 /* List MUST be empty after or else it could be reused. */
1673 assert(cds_list_empty(&channel->streams.head));
1674 end_get_channel:
1675 goto end_msg_sessiond;
1676 error_get_channel_fatal:
1677 goto error_fatal;
1678 end_get_channel_nosignal:
1679 goto end_nosignal;
1680 }
1681 case LTTNG_CONSUMER_DESTROY_CHANNEL:
1682 {
1683 uint64_t key = msg.u.destroy_channel.key;
1684
1685 /*
1686 * Only called if streams have not been sent to stream
1687 * manager thread. However, channel has been sent to
1688 * channel manager thread.
1689 */
1690 notify_thread_del_channel(ctx, key);
1691 goto end_msg_sessiond;
1692 }
1693 case LTTNG_CONSUMER_CLOSE_METADATA:
1694 {
1695 int ret;
1696
1697 ret = close_metadata(msg.u.close_metadata.key);
1698 if (ret != 0) {
1699 ret_code = ret;
1700 }
1701
1702 goto end_msg_sessiond;
1703 }
1704 case LTTNG_CONSUMER_FLUSH_CHANNEL:
1705 {
1706 int ret;
1707
1708 ret = flush_channel(msg.u.flush_channel.key);
1709 if (ret != 0) {
1710 ret_code = ret;
1711 }
1712
1713 goto end_msg_sessiond;
1714 }
1715 case LTTNG_CONSUMER_CLEAR_QUIESCENT_CHANNEL:
1716 {
1717 int ret;
1718
1719 ret = clear_quiescent_channel(
1720 msg.u.clear_quiescent_channel.key);
1721 if (ret != 0) {
1722 ret_code = ret;
1723 }
1724
1725 goto end_msg_sessiond;
1726 }
1727 case LTTNG_CONSUMER_PUSH_METADATA:
1728 {
1729 int ret;
1730 uint64_t len = msg.u.push_metadata.len;
1731 uint64_t key = msg.u.push_metadata.key;
1732 uint64_t offset = msg.u.push_metadata.target_offset;
1733 uint64_t version = msg.u.push_metadata.version;
1734 struct lttng_consumer_channel *channel;
1735
1736 DBG("UST consumer push metadata key %" PRIu64 " of len %" PRIu64, key,
1737 len);
1738
1739 channel = consumer_find_channel(key);
1740 if (!channel) {
1741 /*
1742 * This is possible if the metadata creation on the consumer side
1743 * is in flight vis-a-vis a concurrent push metadata from the
1744 * session daemon. Simply return that the channel failed and the
1745 * session daemon will handle that message correctly considering
1746 * that this race is acceptable thus the DBG() statement here.
1747 */
1748 DBG("UST consumer push metadata %" PRIu64 " not found", key);
1749 ret_code = LTTCOMM_CONSUMERD_CHANNEL_FAIL;
1750 goto end_push_metadata_msg_sessiond;
1751 }
1752
1753 health_code_update();
1754
1755 if (!len) {
1756 /*
1757 * There is nothing to receive. We have simply
1758 * checked whether the channel can be found.
1759 */
1760 ret_code = LTTCOMM_CONSUMERD_SUCCESS;
1761 goto end_push_metadata_msg_sessiond;
1762 }
1763
1764 /* Tell session daemon we are ready to receive the metadata. */
1765 ret = consumer_send_status_msg(sock, LTTCOMM_CONSUMERD_SUCCESS);
1766 if (ret < 0) {
1767 /* Somehow, the session daemon is not responding anymore. */
1768 goto error_push_metadata_fatal;
1769 }
1770
1771 health_code_update();
1772
1773 /* Wait for more data. */
1774 health_poll_entry();
1775 ret = lttng_consumer_poll_socket(consumer_sockpoll);
1776 health_poll_exit();
1777 if (ret) {
1778 goto error_push_metadata_fatal;
1779 }
1780
1781 health_code_update();
1782
1783 ret = lttng_ustconsumer_recv_metadata(sock, key, offset,
1784 len, version, channel, 0, 1);
1785 if (ret < 0) {
1786 /* error receiving from sessiond */
1787 goto error_push_metadata_fatal;
1788 } else {
1789 ret_code = ret;
1790 goto end_push_metadata_msg_sessiond;
1791 }
1792 end_push_metadata_msg_sessiond:
1793 goto end_msg_sessiond;
1794 error_push_metadata_fatal:
1795 goto error_fatal;
1796 }
1797 case LTTNG_CONSUMER_SETUP_METADATA:
1798 {
1799 int ret;
1800
1801 ret = setup_metadata(ctx, msg.u.setup_metadata.key);
1802 if (ret) {
1803 ret_code = ret;
1804 }
1805 goto end_msg_sessiond;
1806 }
1807 case LTTNG_CONSUMER_SNAPSHOT_CHANNEL:
1808 {
1809 struct lttng_consumer_channel *channel;
1810 uint64_t key = msg.u.snapshot_channel.key;
1811
1812 channel = consumer_find_channel(key);
1813 if (!channel) {
1814 DBG("UST snapshot channel not found for key %" PRIu64, key);
1815 ret_code = LTTCOMM_CONSUMERD_CHAN_NOT_FOUND;
1816 } else {
1817 if (msg.u.snapshot_channel.metadata) {
1818 ret = snapshot_metadata(channel, key,
1819 msg.u.snapshot_channel.pathname,
1820 msg.u.snapshot_channel.relayd_id,
1821 ctx);
1822 if (ret < 0) {
1823 ERR("Snapshot metadata failed");
1824 ret_code = LTTCOMM_CONSUMERD_SNAPSHOT_FAILED;
1825 }
1826 } else {
1827 ret = snapshot_channel(channel, key,
1828 msg.u.snapshot_channel.pathname,
1829 msg.u.snapshot_channel.relayd_id,
1830 msg.u.snapshot_channel.nb_packets_per_stream,
1831 ctx);
1832 if (ret < 0) {
1833 ERR("Snapshot channel failed");
1834 ret_code = LTTCOMM_CONSUMERD_SNAPSHOT_FAILED;
1835 }
1836 }
1837 }
1838 health_code_update();
1839 ret = consumer_send_status_msg(sock, ret_code);
1840 if (ret < 0) {
1841 /* Somehow, the session daemon is not responding anymore. */
1842 goto end_nosignal;
1843 }
1844 health_code_update();
1845 break;
1846 }
1847 case LTTNG_CONSUMER_DISCARDED_EVENTS:
1848 {
1849 int ret = 0;
1850 uint64_t discarded_events;
1851 struct lttng_ht_iter iter;
1852 struct lttng_ht *ht;
1853 struct lttng_consumer_stream *stream;
1854 uint64_t id = msg.u.discarded_events.session_id;
1855 uint64_t key = msg.u.discarded_events.channel_key;
1856
1857 DBG("UST consumer discarded events command for session id %"
1858 PRIu64, id);
1859 rcu_read_lock();
1860 pthread_mutex_lock(&consumer_data.lock);
1861
1862 ht = consumer_data.stream_list_ht;
1863
1864 /*
1865 * We only need a reference to the channel, but they are not
1866 * directly indexed, so we just use the first matching stream
1867 * to extract the information we need, we default to 0 if not
1868 * found (no events are dropped if the channel is not yet in
1869 * use).
1870 */
1871 discarded_events = 0;
1872 cds_lfht_for_each_entry_duplicate(ht->ht,
1873 ht->hash_fct(&id, lttng_ht_seed),
1874 ht->match_fct, &id,
1875 &iter.iter, stream, node_session_id.node) {
1876 if (stream->chan->key == key) {
1877 discarded_events = stream->chan->discarded_events;
1878 break;
1879 }
1880 }
1881 pthread_mutex_unlock(&consumer_data.lock);
1882 rcu_read_unlock();
1883
1884 DBG("UST consumer discarded events command for session id %"
1885 PRIu64 ", channel key %" PRIu64, id, key);
1886
1887 health_code_update();
1888
1889 /* Send back returned value to session daemon */
1890 ret = lttcomm_send_unix_sock(sock, &discarded_events, sizeof(discarded_events));
1891 if (ret < 0) {
1892 PERROR("send discarded events");
1893 goto error_fatal;
1894 }
1895
1896 break;
1897 }
1898 case LTTNG_CONSUMER_LOST_PACKETS:
1899 {
1900 int ret;
1901 uint64_t lost_packets;
1902 struct lttng_ht_iter iter;
1903 struct lttng_ht *ht;
1904 struct lttng_consumer_stream *stream;
1905 uint64_t id = msg.u.lost_packets.session_id;
1906 uint64_t key = msg.u.lost_packets.channel_key;
1907
1908 DBG("UST consumer lost packets command for session id %"
1909 PRIu64, id);
1910 rcu_read_lock();
1911 pthread_mutex_lock(&consumer_data.lock);
1912
1913 ht = consumer_data.stream_list_ht;
1914
1915 /*
1916 * We only need a reference to the channel, but they are not
1917 * directly indexed, so we just use the first matching stream
1918 * to extract the information we need, we default to 0 if not
1919 * found (no packets lost if the channel is not yet in use).
1920 */
1921 lost_packets = 0;
1922 cds_lfht_for_each_entry_duplicate(ht->ht,
1923 ht->hash_fct(&id, lttng_ht_seed),
1924 ht->match_fct, &id,
1925 &iter.iter, stream, node_session_id.node) {
1926 if (stream->chan->key == key) {
1927 lost_packets = stream->chan->lost_packets;
1928 break;
1929 }
1930 }
1931 pthread_mutex_unlock(&consumer_data.lock);
1932 rcu_read_unlock();
1933
1934 DBG("UST consumer lost packets command for session id %"
1935 PRIu64 ", channel key %" PRIu64, id, key);
1936
1937 health_code_update();
1938
1939 /* Send back returned value to session daemon */
1940 ret = lttcomm_send_unix_sock(sock, &lost_packets,
1941 sizeof(lost_packets));
1942 if (ret < 0) {
1943 PERROR("send lost packets");
1944 goto error_fatal;
1945 }
1946
1947 break;
1948 }
1949 case LTTNG_CONSUMER_SET_CHANNEL_MONITOR_PIPE:
1950 {
1951 int channel_monitor_pipe;
1952
1953 ret_code = LTTCOMM_CONSUMERD_SUCCESS;
1954 /* Successfully received the command's type. */
1955 ret = consumer_send_status_msg(sock, ret_code);
1956 if (ret < 0) {
1957 goto error_fatal;
1958 }
1959
1960 ret = lttcomm_recv_fds_unix_sock(sock, &channel_monitor_pipe,
1961 1);
1962 if (ret != sizeof(channel_monitor_pipe)) {
1963 ERR("Failed to receive channel monitor pipe");
1964 goto error_fatal;
1965 }
1966
1967 DBG("Received channel monitor pipe (%d)", channel_monitor_pipe);
1968 ret = consumer_timer_thread_set_channel_monitor_pipe(
1969 channel_monitor_pipe);
1970 if (!ret) {
1971 int flags;
1972
1973 ret_code = LTTCOMM_CONSUMERD_SUCCESS;
1974 /* Set the pipe as non-blocking. */
1975 ret = fcntl(channel_monitor_pipe, F_GETFL, 0);
1976 if (ret == -1) {
1977 PERROR("fcntl get flags of the channel monitoring pipe");
1978 goto error_fatal;
1979 }
1980 flags = ret;
1981
1982 ret = fcntl(channel_monitor_pipe, F_SETFL,
1983 flags | O_NONBLOCK);
1984 if (ret == -1) {
1985 PERROR("fcntl set O_NONBLOCK flag of the channel monitoring pipe");
1986 goto error_fatal;
1987 }
1988 DBG("Channel monitor pipe set as non-blocking");
1989 } else {
1990 ret_code = LTTCOMM_CONSUMERD_ALREADY_SET;
1991 }
1992 goto end_msg_sessiond;
1993 }
1994 case LTTNG_CONSUMER_ROTATE_CHANNEL:
1995 {
1996 struct lttng_consumer_channel *channel;
1997 uint64_t key = msg.u.rotate_channel.key;
1998
1999 channel = consumer_find_channel(key);
2000 if (!channel) {
2001 DBG("Channel %" PRIu64 " not found", key);
2002 ret_code = LTTCOMM_CONSUMERD_CHAN_NOT_FOUND;
2003 } else {
2004 /*
2005 * Sample the rotate position of all the streams in
2006 * this channel.
2007 */
2008 ret = lttng_consumer_rotate_channel(channel, key,
2009 msg.u.rotate_channel.relayd_id,
2010 msg.u.rotate_channel.metadata,
2011 ctx);
2012 if (ret < 0) {
2013 ERR("Rotate channel failed");
2014 ret_code = LTTCOMM_CONSUMERD_ROTATION_FAIL;
2015 }
2016
2017 health_code_update();
2018 }
2019 ret = consumer_send_status_msg(sock, ret_code);
2020 if (ret < 0) {
2021 /* Somehow, the session daemon is not responding anymore. */
2022 goto end_rotate_channel_nosignal;
2023 }
2024
2025 /*
2026 * Rotate the streams that are ready right now.
2027 * FIXME: this is a second consecutive iteration over the
2028 * streams in a channel, there is probably a better way to
2029 * handle this, but it needs to be after the
2030 * consumer_send_status_msg() call.
2031 */
2032 if (channel) {
2033 ret = lttng_consumer_rotate_ready_streams(
2034 channel, key, ctx);
2035 if (ret < 0) {
2036 ERR("Rotate channel failed");
2037 }
2038 }
2039 break;
2040 end_rotate_channel_nosignal:
2041 goto end_nosignal;
2042 }
2043 case LTTNG_CONSUMER_CLEAR_CHANNEL:
2044 {
2045 struct lttng_consumer_channel *channel;
2046 uint64_t key = msg.u.clear_channel.key;
2047
2048 channel = consumer_find_channel(key);
2049 if (!channel) {
2050 DBG("Channel %" PRIu64 " not found", key);
2051 ret_code = LTTCOMM_CONSUMERD_CHAN_NOT_FOUND;
2052 } else {
2053 ret = lttng_consumer_clear_channel(channel);
2054 if (ret) {
2055 ERR("Clear channel failed key %" PRIu64, key);
2056 ret_code = ret;
2057 }
2058
2059 health_code_update();
2060 }
2061 ret = consumer_send_status_msg(sock, ret_code);
2062 if (ret < 0) {
2063 /* Somehow, the session daemon is not responding anymore. */
2064 goto end_nosignal;
2065 }
2066 break;
2067 }
2068 case LTTNG_CONSUMER_INIT:
2069 {
2070 ret_code = lttng_consumer_init_command(ctx,
2071 msg.u.init.sessiond_uuid);
2072 health_code_update();
2073 ret = consumer_send_status_msg(sock, ret_code);
2074 if (ret < 0) {
2075 /* Somehow, the session daemon is not responding anymore. */
2076 goto end_nosignal;
2077 }
2078 break;
2079 }
2080 case LTTNG_CONSUMER_CREATE_TRACE_CHUNK:
2081 {
2082 const struct lttng_credentials credentials = {
2083 .uid = LTTNG_OPTIONAL_INIT_VALUE(msg.u.create_trace_chunk.credentials.value.uid),
2084 .gid = LTTNG_OPTIONAL_INIT_VALUE(msg.u.create_trace_chunk.credentials.value.gid),
2085 };
2086 const bool is_local_trace =
2087 !msg.u.create_trace_chunk.relayd_id.is_set;
2088 const uint64_t relayd_id =
2089 msg.u.create_trace_chunk.relayd_id.value;
2090 const char *chunk_override_name =
2091 *msg.u.create_trace_chunk.override_name ?
2092 msg.u.create_trace_chunk.override_name :
2093 NULL;
2094 struct lttng_directory_handle *chunk_directory_handle = NULL;
2095
2096 /*
2097 * The session daemon will only provide a chunk directory file
2098 * descriptor for local traces.
2099 */
2100 if (is_local_trace) {
2101 int chunk_dirfd;
2102
2103 /* Acnowledge the reception of the command. */
2104 ret = consumer_send_status_msg(sock,
2105 LTTCOMM_CONSUMERD_SUCCESS);
2106 if (ret < 0) {
2107 /* Somehow, the session daemon is not responding anymore. */
2108 goto end_nosignal;
2109 }
2110
2111 /*
2112 * Receive trace chunk domain dirfd.
2113 */
2114 ret = lttcomm_recv_fds_unix_sock(sock, &chunk_dirfd, 1);
2115 if (ret != sizeof(chunk_dirfd)) {
2116 ERR("Failed to receive trace chunk domain directory file descriptor");
2117 goto error_fatal;
2118 }
2119
2120 DBG("Received trace chunk domain directory fd (%d)",
2121 chunk_dirfd);
2122 chunk_directory_handle = lttng_directory_handle_create_from_dirfd(
2123 chunk_dirfd);
2124 if (!chunk_directory_handle) {
2125 ERR("Failed to initialize chunk domain directory handle from directory file descriptor");
2126 if (close(chunk_dirfd)) {
2127 PERROR("Failed to close chunk directory file descriptor");
2128 }
2129 goto error_fatal;
2130 }
2131 }
2132
2133 ret_code = lttng_consumer_create_trace_chunk(
2134 !is_local_trace ? &relayd_id : NULL,
2135 msg.u.create_trace_chunk.session_id,
2136 msg.u.create_trace_chunk.chunk_id,
2137 (time_t) msg.u.create_trace_chunk
2138 .creation_timestamp,
2139 chunk_override_name,
2140 msg.u.create_trace_chunk.credentials.is_set ?
2141 &credentials :
2142 NULL,
2143 chunk_directory_handle);
2144 lttng_directory_handle_put(chunk_directory_handle);
2145 goto end_msg_sessiond;
2146 }
2147 case LTTNG_CONSUMER_CLOSE_TRACE_CHUNK:
2148 {
2149 enum lttng_trace_chunk_command_type close_command =
2150 msg.u.close_trace_chunk.close_command.value;
2151 const uint64_t relayd_id =
2152 msg.u.close_trace_chunk.relayd_id.value;
2153 struct lttcomm_consumer_close_trace_chunk_reply reply;
2154 char closed_trace_chunk_path[LTTNG_PATH_MAX];
2155 int ret;
2156
2157 ret_code = lttng_consumer_close_trace_chunk(
2158 msg.u.close_trace_chunk.relayd_id.is_set ?
2159 &relayd_id :
2160 NULL,
2161 msg.u.close_trace_chunk.session_id,
2162 msg.u.close_trace_chunk.chunk_id,
2163 (time_t) msg.u.close_trace_chunk.close_timestamp,
2164 msg.u.close_trace_chunk.close_command.is_set ?
2165 &close_command :
2166 NULL, closed_trace_chunk_path);
2167 reply.ret_code = ret_code;
2168 reply.path_length = strlen(closed_trace_chunk_path) + 1;
2169 ret = lttcomm_send_unix_sock(sock, &reply, sizeof(reply));
2170 if (ret != sizeof(reply)) {
2171 goto error_fatal;
2172 }
2173 ret = lttcomm_send_unix_sock(sock, closed_trace_chunk_path,
2174 reply.path_length);
2175 if (ret != reply.path_length) {
2176 goto error_fatal;
2177 }
2178 goto end_nosignal;
2179 }
2180 case LTTNG_CONSUMER_TRACE_CHUNK_EXISTS:
2181 {
2182 const uint64_t relayd_id =
2183 msg.u.trace_chunk_exists.relayd_id.value;
2184
2185 ret_code = lttng_consumer_trace_chunk_exists(
2186 msg.u.trace_chunk_exists.relayd_id.is_set ?
2187 &relayd_id : NULL,
2188 msg.u.trace_chunk_exists.session_id,
2189 msg.u.trace_chunk_exists.chunk_id);
2190 goto end_msg_sessiond;
2191 }
2192 case LTTNG_CONSUMER_OPEN_CHANNEL_PACKETS:
2193 {
2194 const uint64_t key = msg.u.open_channel_packets.key;
2195 struct lttng_consumer_channel *channel =
2196 consumer_find_channel(key);
2197
2198 if (channel) {
2199 pthread_mutex_lock(&channel->lock);
2200 ret_code = lttng_consumer_open_channel_packets(channel);
2201 pthread_mutex_unlock(&channel->lock);
2202 } else {
2203 /*
2204 * The channel could have disappeared in per-pid
2205 * buffering mode.
2206 */
2207 DBG("Channel %" PRIu64 " not found", key);
2208 ret_code = LTTCOMM_CONSUMERD_CHAN_NOT_FOUND;
2209 }
2210
2211 health_code_update();
2212 goto end_msg_sessiond;
2213 }
2214 default:
2215 break;
2216 }
2217
2218 end_nosignal:
2219 /*
2220 * Return 1 to indicate success since the 0 value can be a socket
2221 * shutdown during the recv() or send() call.
2222 */
2223 ret = 1;
2224 goto end;
2225
2226 end_msg_sessiond:
2227 /*
2228 * The returned value here is not useful since either way we'll return 1 to
2229 * the caller because the session daemon socket management is done
2230 * elsewhere. Returning a negative code or 0 will shutdown the consumer.
2231 */
2232 ret = consumer_send_status_msg(sock, ret_code);
2233 if (ret < 0) {
2234 goto error_fatal;
2235 }
2236 ret = 1;
2237 goto end;
2238
2239 end_channel_error:
2240 if (channel) {
2241 /*
2242 * Free channel here since no one has a reference to it. We don't
2243 * free after that because a stream can store this pointer.
2244 */
2245 destroy_channel(channel);
2246 }
2247 /* We have to send a status channel message indicating an error. */
2248 ret = consumer_send_status_channel(sock, NULL);
2249 if (ret < 0) {
2250 /* Stop everything if session daemon can not be notified. */
2251 goto error_fatal;
2252 }
2253 ret = 1;
2254 goto end;
2255
2256 error_fatal:
2257 /* This will issue a consumer stop. */
2258 ret = -1;
2259 goto end;
2260
2261 end:
2262 rcu_read_unlock();
2263 health_code_update();
2264 return ret;
2265 }
2266
2267 void lttng_ustctl_flush_buffer(struct lttng_consumer_stream *stream,
2268 int producer_active)
2269 {
2270 assert(stream);
2271 assert(stream->ustream);
2272
2273 ustctl_flush_buffer(stream->ustream, producer_active);
2274 }
2275
2276 /*
2277 * Take a snapshot for a specific stream.
2278 *
2279 * Returns 0 on success, < 0 on error
2280 */
2281 int lttng_ustconsumer_take_snapshot(struct lttng_consumer_stream *stream)
2282 {
2283 assert(stream);
2284 assert(stream->ustream);
2285
2286 return ustctl_snapshot(stream->ustream);
2287 }
2288
2289 /*
2290 * Sample consumed and produced positions for a specific stream.
2291 *
2292 * Returns 0 on success, < 0 on error.
2293 */
2294 int lttng_ustconsumer_sample_snapshot_positions(
2295 struct lttng_consumer_stream *stream)
2296 {
2297 assert(stream);
2298 assert(stream->ustream);
2299
2300 return ustctl_snapshot_sample_positions(stream->ustream);
2301 }
2302
2303 /*
2304 * Get the produced position
2305 *
2306 * Returns 0 on success, < 0 on error
2307 */
2308 int lttng_ustconsumer_get_produced_snapshot(
2309 struct lttng_consumer_stream *stream, unsigned long *pos)
2310 {
2311 assert(stream);
2312 assert(stream->ustream);
2313 assert(pos);
2314
2315 return ustctl_snapshot_get_produced(stream->ustream, pos);
2316 }
2317
2318 /*
2319 * Get the consumed position
2320 *
2321 * Returns 0 on success, < 0 on error
2322 */
2323 int lttng_ustconsumer_get_consumed_snapshot(
2324 struct lttng_consumer_stream *stream, unsigned long *pos)
2325 {
2326 assert(stream);
2327 assert(stream->ustream);
2328 assert(pos);
2329
2330 return ustctl_snapshot_get_consumed(stream->ustream, pos);
2331 }
2332
2333 void lttng_ustconsumer_flush_buffer(struct lttng_consumer_stream *stream,
2334 int producer)
2335 {
2336 assert(stream);
2337 assert(stream->ustream);
2338
2339 ustctl_flush_buffer(stream->ustream, producer);
2340 }
2341
2342 void lttng_ustconsumer_clear_buffer(struct lttng_consumer_stream *stream)
2343 {
2344 assert(stream);
2345 assert(stream->ustream);
2346
2347 ustctl_clear_buffer(stream->ustream);
2348 }
2349
2350 int lttng_ustconsumer_get_current_timestamp(
2351 struct lttng_consumer_stream *stream, uint64_t *ts)
2352 {
2353 assert(stream);
2354 assert(stream->ustream);
2355 assert(ts);
2356
2357 return ustctl_get_current_timestamp(stream->ustream, ts);
2358 }
2359
2360 int lttng_ustconsumer_get_sequence_number(
2361 struct lttng_consumer_stream *stream, uint64_t *seq)
2362 {
2363 assert(stream);
2364 assert(stream->ustream);
2365 assert(seq);
2366
2367 return ustctl_get_sequence_number(stream->ustream, seq);
2368 }
2369
2370 /*
2371 * Called when the stream signals the consumer that it has hung up.
2372 */
2373 void lttng_ustconsumer_on_stream_hangup(struct lttng_consumer_stream *stream)
2374 {
2375 assert(stream);
2376 assert(stream->ustream);
2377
2378 pthread_mutex_lock(&stream->lock);
2379 if (!stream->quiescent) {
2380 ustctl_flush_buffer(stream->ustream, 0);
2381 stream->quiescent = true;
2382 }
2383 pthread_mutex_unlock(&stream->lock);
2384 stream->hangup_flush_done = 1;
2385 }
2386
2387 void lttng_ustconsumer_del_channel(struct lttng_consumer_channel *chan)
2388 {
2389 int i;
2390
2391 assert(chan);
2392 assert(chan->uchan);
2393 assert(chan->buffer_credentials.is_set);
2394
2395 if (chan->switch_timer_enabled == 1) {
2396 consumer_timer_switch_stop(chan);
2397 }
2398 for (i = 0; i < chan->nr_stream_fds; i++) {
2399 int ret;
2400
2401 ret = close(chan->stream_fds[i]);
2402 if (ret) {
2403 PERROR("close");
2404 }
2405 if (chan->shm_path[0]) {
2406 char shm_path[PATH_MAX];
2407
2408 ret = get_stream_shm_path(shm_path, chan->shm_path, i);
2409 if (ret) {
2410 ERR("Cannot get stream shm path");
2411 }
2412 ret = run_as_unlink(shm_path,
2413 lttng_credentials_get_uid(LTTNG_OPTIONAL_GET_PTR(
2414 chan->buffer_credentials)),
2415 lttng_credentials_get_gid(LTTNG_OPTIONAL_GET_PTR(
2416 chan->buffer_credentials)));
2417 if (ret) {
2418 PERROR("unlink %s", shm_path);
2419 }
2420 }
2421 }
2422 }
2423
2424 void lttng_ustconsumer_free_channel(struct lttng_consumer_channel *chan)
2425 {
2426 assert(chan);
2427 assert(chan->uchan);
2428 assert(chan->buffer_credentials.is_set);
2429
2430 consumer_metadata_cache_destroy(chan);
2431 ustctl_destroy_channel(chan->uchan);
2432 /* Try to rmdir all directories under shm_path root. */
2433 if (chan->root_shm_path[0]) {
2434 (void) run_as_rmdir_recursive(chan->root_shm_path,
2435 lttng_credentials_get_uid(LTTNG_OPTIONAL_GET_PTR(
2436 chan->buffer_credentials)),
2437 lttng_credentials_get_gid(LTTNG_OPTIONAL_GET_PTR(
2438 chan->buffer_credentials)),
2439 LTTNG_DIRECTORY_HANDLE_SKIP_NON_EMPTY_FLAG);
2440 }
2441 free(chan->stream_fds);
2442 }
2443
2444 void lttng_ustconsumer_del_stream(struct lttng_consumer_stream *stream)
2445 {
2446 assert(stream);
2447 assert(stream->ustream);
2448
2449 if (stream->chan->switch_timer_enabled == 1) {
2450 consumer_timer_switch_stop(stream->chan);
2451 }
2452 ustctl_destroy_stream(stream->ustream);
2453 }
2454
2455 int lttng_ustconsumer_get_wakeup_fd(struct lttng_consumer_stream *stream)
2456 {
2457 assert(stream);
2458 assert(stream->ustream);
2459
2460 return ustctl_stream_get_wakeup_fd(stream->ustream);
2461 }
2462
2463 int lttng_ustconsumer_close_wakeup_fd(struct lttng_consumer_stream *stream)
2464 {
2465 assert(stream);
2466 assert(stream->ustream);
2467
2468 return ustctl_stream_close_wakeup_fd(stream->ustream);
2469 }
2470
2471 /*
2472 * Write up to one packet from the metadata cache to the channel.
2473 *
2474 * Returns the number of bytes pushed from the cache into the ring buffer, or a
2475 * negative value on error.
2476 */
2477 static
2478 int commit_one_metadata_packet(struct lttng_consumer_stream *stream)
2479 {
2480 ssize_t write_len;
2481 int ret;
2482
2483 pthread_mutex_lock(&stream->chan->metadata_cache->lock);
2484 if (stream->chan->metadata_cache->contents.size ==
2485 stream->ust_metadata_pushed) {
2486 /*
2487 * In the context of a user space metadata channel, a
2488 * change in version can be detected in two ways:
2489 * 1) During the pre-consume of the `read_subbuffer` loop,
2490 * 2) When populating the metadata ring buffer (i.e. here).
2491 *
2492 * This function is invoked when there is no metadata
2493 * available in the ring-buffer. If all data was consumed
2494 * up to the size of the metadata cache, there is no metadata
2495 * to insert in the ring-buffer.
2496 *
2497 * However, the metadata version could still have changed (a
2498 * regeneration without any new data will yield the same cache
2499 * size).
2500 *
2501 * The cache's version is checked for a version change and the
2502 * consumed position is reset if one occurred.
2503 *
2504 * This check is only necessary for the user space domain as
2505 * it has to manage the cache explicitly. If this reset was not
2506 * performed, no metadata would be consumed (and no reset would
2507 * occur as part of the pre-consume) until the metadata size
2508 * exceeded the cache size.
2509 */
2510 if (stream->metadata_version !=
2511 stream->chan->metadata_cache->version) {
2512 metadata_stream_reset_cache_consumed_position(stream);
2513 consumer_stream_metadata_set_version(stream,
2514 stream->chan->metadata_cache->version);
2515 } else {
2516 ret = 0;
2517 goto end;
2518 }
2519 }
2520
2521 write_len = ustctl_write_one_packet_to_channel(stream->chan->uchan,
2522 &stream->chan->metadata_cache->contents.data[stream->ust_metadata_pushed],
2523 stream->chan->metadata_cache->contents.size -
2524 stream->ust_metadata_pushed);
2525 assert(write_len != 0);
2526 if (write_len < 0) {
2527 ERR("Writing one metadata packet");
2528 ret = write_len;
2529 goto end;
2530 }
2531 stream->ust_metadata_pushed += write_len;
2532
2533 assert(stream->chan->metadata_cache->contents.size >=
2534 stream->ust_metadata_pushed);
2535 ret = write_len;
2536
2537 /*
2538 * Switch packet (but don't open the next one) on every commit of
2539 * a metadata packet. Since the subbuffer is fully filled (with padding,
2540 * if needed), the stream is "quiescent" after this commit.
2541 */
2542 ustctl_flush_buffer(stream->ustream, 1);
2543 stream->quiescent = true;
2544 end:
2545 pthread_mutex_unlock(&stream->chan->metadata_cache->lock);
2546 return ret;
2547 }
2548
2549
2550 /*
2551 * Sync metadata meaning request them to the session daemon and snapshot to the
2552 * metadata thread can consumer them.
2553 *
2554 * Metadata stream lock is held here, but we need to release it when
2555 * interacting with sessiond, else we cause a deadlock with live
2556 * awaiting on metadata to be pushed out.
2557 *
2558 * The RCU read side lock must be held by the caller.
2559 */
2560 enum sync_metadata_status lttng_ustconsumer_sync_metadata(
2561 struct lttng_consumer_local_data *ctx,
2562 struct lttng_consumer_stream *metadata_stream)
2563 {
2564 int ret;
2565 enum sync_metadata_status status;
2566 struct lttng_consumer_channel *metadata_channel;
2567
2568 assert(ctx);
2569 assert(metadata_stream);
2570
2571 metadata_channel = metadata_stream->chan;
2572 pthread_mutex_unlock(&metadata_stream->lock);
2573 /*
2574 * Request metadata from the sessiond, but don't wait for the flush
2575 * because we locked the metadata thread.
2576 */
2577 ret = lttng_ustconsumer_request_metadata(ctx, metadata_channel, 0, 0);
2578 pthread_mutex_lock(&metadata_stream->lock);
2579 if (ret < 0) {
2580 status = SYNC_METADATA_STATUS_ERROR;
2581 goto end;
2582 }
2583
2584 /*
2585 * The metadata stream and channel can be deleted while the
2586 * metadata stream lock was released. The streamed is checked
2587 * for deletion before we use it further.
2588 *
2589 * Note that it is safe to access a logically-deleted stream since its
2590 * existence is still guaranteed by the RCU read side lock. However,
2591 * it should no longer be used. The close/deletion of the metadata
2592 * channel and stream already guarantees that all metadata has been
2593 * consumed. Therefore, there is nothing left to do in this function.
2594 */
2595 if (consumer_stream_is_deleted(metadata_stream)) {
2596 DBG("Metadata stream %" PRIu64 " was deleted during the metadata synchronization",
2597 metadata_stream->key);
2598 status = SYNC_METADATA_STATUS_NO_DATA;
2599 goto end;
2600 }
2601
2602 ret = commit_one_metadata_packet(metadata_stream);
2603 if (ret < 0) {
2604 status = SYNC_METADATA_STATUS_ERROR;
2605 goto end;
2606 } else if (ret > 0) {
2607 status = SYNC_METADATA_STATUS_NEW_DATA;
2608 } else /* ret == 0 */ {
2609 status = SYNC_METADATA_STATUS_NO_DATA;
2610 goto end;
2611 }
2612
2613 ret = ustctl_snapshot(metadata_stream->ustream);
2614 if (ret < 0) {
2615 ERR("Failed to take a snapshot of the metadata ring-buffer positions, ret = %d", ret);
2616 status = SYNC_METADATA_STATUS_ERROR;
2617 goto end;
2618 }
2619
2620 end:
2621 return status;
2622 }
2623
2624 /*
2625 * Return 0 on success else a negative value.
2626 */
2627 static int notify_if_more_data(struct lttng_consumer_stream *stream,
2628 struct lttng_consumer_local_data *ctx)
2629 {
2630 int ret;
2631 struct ustctl_consumer_stream *ustream;
2632
2633 assert(stream);
2634 assert(ctx);
2635
2636 ustream = stream->ustream;
2637
2638 /*
2639 * First, we are going to check if there is a new subbuffer available
2640 * before reading the stream wait_fd.
2641 */
2642 /* Get the next subbuffer */
2643 ret = ustctl_get_next_subbuf(ustream);
2644 if (ret) {
2645 /* No more data found, flag the stream. */
2646 stream->has_data = 0;
2647 ret = 0;
2648 goto end;
2649 }
2650
2651 ret = ustctl_put_subbuf(ustream);
2652 assert(!ret);
2653
2654 /* This stream still has data. Flag it and wake up the data thread. */
2655 stream->has_data = 1;
2656
2657 if (stream->monitor && !stream->hangup_flush_done && !ctx->has_wakeup) {
2658 ssize_t writelen;
2659
2660 writelen = lttng_pipe_write(ctx->consumer_wakeup_pipe, "!", 1);
2661 if (writelen < 0 && errno != EAGAIN && errno != EWOULDBLOCK) {
2662 ret = writelen;
2663 goto end;
2664 }
2665
2666 /* The wake up pipe has been notified. */
2667 ctx->has_wakeup = 1;
2668 }
2669 ret = 0;
2670
2671 end:
2672 return ret;
2673 }
2674
2675 static int consumer_stream_ust_on_wake_up(struct lttng_consumer_stream *stream)
2676 {
2677 int ret = 0;
2678
2679 /*
2680 * We can consume the 1 byte written into the wait_fd by
2681 * UST. Don't trigger error if we cannot read this one byte
2682 * (read returns 0), or if the error is EAGAIN or EWOULDBLOCK.
2683 *
2684 * This is only done when the stream is monitored by a thread,
2685 * before the flush is done after a hangup and if the stream
2686 * is not flagged with data since there might be nothing to
2687 * consume in the wait fd but still have data available
2688 * flagged by the consumer wake up pipe.
2689 */
2690 if (stream->monitor && !stream->hangup_flush_done && !stream->has_data) {
2691 char dummy;
2692 ssize_t readlen;
2693
2694 readlen = lttng_read(stream->wait_fd, &dummy, 1);
2695 if (readlen < 0 && errno != EAGAIN && errno != EWOULDBLOCK) {
2696 ret = readlen;
2697 }
2698 }
2699
2700 return ret;
2701 }
2702
2703 static int extract_common_subbuffer_info(struct lttng_consumer_stream *stream,
2704 struct stream_subbuffer *subbuf)
2705 {
2706 int ret;
2707
2708 ret = ustctl_get_subbuf_size(
2709 stream->ustream, &subbuf->info.data.subbuf_size);
2710 if (ret) {
2711 goto end;
2712 }
2713
2714 ret = ustctl_get_padded_subbuf_size(
2715 stream->ustream, &subbuf->info.data.padded_subbuf_size);
2716 if (ret) {
2717 goto end;
2718 }
2719
2720 end:
2721 return ret;
2722 }
2723
2724 static int extract_metadata_subbuffer_info(struct lttng_consumer_stream *stream,
2725 struct stream_subbuffer *subbuf)
2726 {
2727 int ret;
2728
2729 ret = extract_common_subbuffer_info(stream, subbuf);
2730 if (ret) {
2731 goto end;
2732 }
2733
2734 subbuf->info.metadata.version = stream->metadata_version;
2735
2736 end:
2737 return ret;
2738 }
2739
2740 static int extract_data_subbuffer_info(struct lttng_consumer_stream *stream,
2741 struct stream_subbuffer *subbuf)
2742 {
2743 int ret;
2744
2745 ret = extract_common_subbuffer_info(stream, subbuf);
2746 if (ret) {
2747 goto end;
2748 }
2749
2750 ret = ustctl_get_packet_size(
2751 stream->ustream, &subbuf->info.data.packet_size);
2752 if (ret < 0) {
2753 PERROR("Failed to get sub-buffer packet size");
2754 goto end;
2755 }
2756
2757 ret = ustctl_get_content_size(
2758 stream->ustream, &subbuf->info.data.content_size);
2759 if (ret < 0) {
2760 PERROR("Failed to get sub-buffer content size");
2761 goto end;
2762 }
2763
2764 ret = ustctl_get_timestamp_begin(
2765 stream->ustream, &subbuf->info.data.timestamp_begin);
2766 if (ret < 0) {
2767 PERROR("Failed to get sub-buffer begin timestamp");
2768 goto end;
2769 }
2770
2771 ret = ustctl_get_timestamp_end(
2772 stream->ustream, &subbuf->info.data.timestamp_end);
2773 if (ret < 0) {
2774 PERROR("Failed to get sub-buffer end timestamp");
2775 goto end;
2776 }
2777
2778 ret = ustctl_get_events_discarded(
2779 stream->ustream, &subbuf->info.data.events_discarded);
2780 if (ret) {
2781 PERROR("Failed to get sub-buffer events discarded count");
2782 goto end;
2783 }
2784
2785 ret = ustctl_get_sequence_number(stream->ustream,
2786 &subbuf->info.data.sequence_number.value);
2787 if (ret) {
2788 /* May not be supported by older LTTng-modules. */
2789 if (ret != -ENOTTY) {
2790 PERROR("Failed to get sub-buffer sequence number");
2791 goto end;
2792 }
2793 } else {
2794 subbuf->info.data.sequence_number.is_set = true;
2795 }
2796
2797 ret = ustctl_get_stream_id(
2798 stream->ustream, &subbuf->info.data.stream_id);
2799 if (ret < 0) {
2800 PERROR("Failed to get stream id");
2801 goto end;
2802 }
2803
2804 ret = ustctl_get_instance_id(stream->ustream,
2805 &subbuf->info.data.stream_instance_id.value);
2806 if (ret) {
2807 /* May not be supported by older LTTng-modules. */
2808 if (ret != -ENOTTY) {
2809 PERROR("Failed to get stream instance id");
2810 goto end;
2811 }
2812 } else {
2813 subbuf->info.data.stream_instance_id.is_set = true;
2814 }
2815 end:
2816 return ret;
2817 }
2818
2819 static int get_next_subbuffer_common(struct lttng_consumer_stream *stream,
2820 struct stream_subbuffer *subbuffer)
2821 {
2822 int ret;
2823 const char *addr;
2824
2825 ret = stream->read_subbuffer_ops.extract_subbuffer_info(
2826 stream, subbuffer);
2827 if (ret) {
2828 goto end;
2829 }
2830
2831 ret = get_current_subbuf_addr(stream, &addr);
2832 if (ret) {
2833 goto end;
2834 }
2835
2836 subbuffer->buffer.buffer = lttng_buffer_view_init(
2837 addr, 0, subbuffer->info.data.padded_subbuf_size);
2838 assert(subbuffer->buffer.buffer.data != NULL);
2839 end:
2840 return ret;
2841 }
2842
2843 static int get_next_subbuffer(struct lttng_consumer_stream *stream,
2844 struct stream_subbuffer *subbuffer)
2845 {
2846 int ret;
2847
2848 ret = ustctl_get_next_subbuf(stream->ustream);
2849 if (ret) {
2850 goto end;
2851 }
2852
2853 ret = get_next_subbuffer_common(stream, subbuffer);
2854 if (ret) {
2855 goto end;
2856 }
2857 end:
2858 return ret;
2859 }
2860
2861 static int get_next_subbuffer_metadata(struct lttng_consumer_stream *stream,
2862 struct stream_subbuffer *subbuffer)
2863 {
2864 int ret;
2865 bool cache_empty;
2866 bool got_subbuffer;
2867 bool coherent;
2868 bool buffer_empty;
2869 unsigned long consumed_pos, produced_pos;
2870
2871 do {
2872 ret = ustctl_get_next_subbuf(stream->ustream);
2873 if (ret == 0) {
2874 got_subbuffer = true;
2875 } else {
2876 got_subbuffer = false;
2877 if (ret != -EAGAIN) {
2878 /* Fatal error. */
2879 goto end;
2880 }
2881 }
2882
2883 /*
2884 * Determine if the cache is empty and ensure that a sub-buffer
2885 * is made available if the cache is not empty.
2886 */
2887 if (!got_subbuffer) {
2888 ret = commit_one_metadata_packet(stream);
2889 if (ret < 0 && ret != -ENOBUFS) {
2890 goto end;
2891 } else if (ret == 0) {
2892 /* Not an error, the cache is empty. */
2893 cache_empty = true;
2894 ret = -ENODATA;
2895 goto end;
2896 } else {
2897 cache_empty = false;
2898 }
2899 } else {
2900 pthread_mutex_lock(&stream->chan->metadata_cache->lock);
2901 cache_empty = stream->chan->metadata_cache->contents.size ==
2902 stream->ust_metadata_pushed;
2903 pthread_mutex_unlock(&stream->chan->metadata_cache->lock);
2904 }
2905 } while (!got_subbuffer);
2906
2907 /* Populate sub-buffer infos and view. */
2908 ret = get_next_subbuffer_common(stream, subbuffer);
2909 if (ret) {
2910 goto end;
2911 }
2912
2913 ret = lttng_ustconsumer_sample_snapshot_positions(stream);
2914 if (ret < 0) {
2915 /*
2916 * -EAGAIN is not expected since we got a sub-buffer and haven't
2917 * pushed the consumption position yet (on put_next).
2918 */
2919 PERROR("Failed to take a snapshot of metadata buffer positions");
2920 goto end;
2921 }
2922
2923 ret = lttng_ustconsumer_get_consumed_snapshot(stream, &consumed_pos);
2924 if (ret) {
2925 PERROR("Failed to get metadata consumed position");
2926 goto end;
2927 }
2928
2929 ret = lttng_ustconsumer_get_produced_snapshot(stream, &produced_pos);
2930 if (ret) {
2931 PERROR("Failed to get metadata produced position");
2932 goto end;
2933 }
2934
2935 /* Last sub-buffer of the ring buffer ? */
2936 buffer_empty = (consumed_pos + stream->max_sb_size) == produced_pos;
2937
2938 /*
2939 * The sessiond registry lock ensures that coherent units of metadata
2940 * are pushed to the consumer daemon at once. Hence, if a sub-buffer is
2941 * acquired, the cache is empty, and it is the only available sub-buffer
2942 * available, it is safe to assume that it is "coherent".
2943 */
2944 coherent = got_subbuffer && cache_empty && buffer_empty;
2945
2946 LTTNG_OPTIONAL_SET(&subbuffer->info.metadata.coherent, coherent);
2947 end:
2948 return ret;
2949 }
2950
2951 static int put_next_subbuffer(struct lttng_consumer_stream *stream,
2952 struct stream_subbuffer *subbuffer)
2953 {
2954 const int ret = ustctl_put_next_subbuf(stream->ustream);
2955
2956 assert(ret == 0);
2957 return ret;
2958 }
2959
2960 static int signal_metadata(struct lttng_consumer_stream *stream,
2961 struct lttng_consumer_local_data *ctx)
2962 {
2963 ASSERT_LOCKED(stream->metadata_rdv_lock);
2964 return pthread_cond_broadcast(&stream->metadata_rdv) ? -errno : 0;
2965 }
2966
2967 static int lttng_ustconsumer_set_stream_ops(
2968 struct lttng_consumer_stream *stream)
2969 {
2970 int ret = 0;
2971
2972 stream->read_subbuffer_ops.on_wake_up = consumer_stream_ust_on_wake_up;
2973 if (stream->metadata_flag) {
2974 stream->read_subbuffer_ops.get_next_subbuffer =
2975 get_next_subbuffer_metadata;
2976 stream->read_subbuffer_ops.extract_subbuffer_info =
2977 extract_metadata_subbuffer_info;
2978 stream->read_subbuffer_ops.reset_metadata =
2979 metadata_stream_reset_cache_consumed_position;
2980 if (stream->chan->is_live) {
2981 stream->read_subbuffer_ops.on_sleep = signal_metadata;
2982 ret = consumer_stream_enable_metadata_bucketization(
2983 stream);
2984 if (ret) {
2985 goto end;
2986 }
2987 }
2988 } else {
2989 stream->read_subbuffer_ops.get_next_subbuffer =
2990 get_next_subbuffer;
2991 stream->read_subbuffer_ops.extract_subbuffer_info =
2992 extract_data_subbuffer_info;
2993 stream->read_subbuffer_ops.on_sleep = notify_if_more_data;
2994 if (stream->chan->is_live) {
2995 stream->read_subbuffer_ops.send_live_beacon =
2996 consumer_flush_ust_index;
2997 }
2998 }
2999
3000 stream->read_subbuffer_ops.put_next_subbuffer = put_next_subbuffer;
3001 end:
3002 return ret;
3003 }
3004
3005 /*
3006 * Called when a stream is created.
3007 *
3008 * Return 0 on success or else a negative value.
3009 */
3010 int lttng_ustconsumer_on_recv_stream(struct lttng_consumer_stream *stream)
3011 {
3012 int ret;
3013
3014 assert(stream);
3015
3016 /*
3017 * Don't create anything if this is set for streaming or if there is
3018 * no current trace chunk on the parent channel.
3019 */
3020 if (stream->net_seq_idx == (uint64_t) -1ULL && stream->chan->monitor &&
3021 stream->chan->trace_chunk) {
3022 ret = consumer_stream_create_output_files(stream, true);
3023 if (ret) {
3024 goto error;
3025 }
3026 }
3027
3028 lttng_ustconsumer_set_stream_ops(stream);
3029 ret = 0;
3030
3031 error:
3032 return ret;
3033 }
3034
3035 /*
3036 * Check if data is still being extracted from the buffers for a specific
3037 * stream. Consumer data lock MUST be acquired before calling this function
3038 * and the stream lock.
3039 *
3040 * Return 1 if the traced data are still getting read else 0 meaning that the
3041 * data is available for trace viewer reading.
3042 */
3043 int lttng_ustconsumer_data_pending(struct lttng_consumer_stream *stream)
3044 {
3045 int ret;
3046
3047 assert(stream);
3048 assert(stream->ustream);
3049 ASSERT_LOCKED(stream->lock);
3050
3051 DBG("UST consumer checking data pending");
3052
3053 if (stream->endpoint_status != CONSUMER_ENDPOINT_ACTIVE) {
3054 ret = 0;
3055 goto end;
3056 }
3057
3058 if (stream->chan->type == CONSUMER_CHANNEL_TYPE_METADATA) {
3059 uint64_t contiguous, pushed;
3060
3061 /* Ease our life a bit. */
3062 pthread_mutex_lock(&stream->chan->metadata_cache->lock);
3063 contiguous = stream->chan->metadata_cache->contents.size;
3064 pthread_mutex_unlock(&stream->chan->metadata_cache->lock);
3065 pushed = stream->ust_metadata_pushed;
3066
3067 /*
3068 * We can simply check whether all contiguously available data
3069 * has been pushed to the ring buffer, since the push operation
3070 * is performed within get_next_subbuf(), and because both
3071 * get_next_subbuf() and put_next_subbuf() are issued atomically
3072 * thanks to the stream lock within
3073 * lttng_ustconsumer_read_subbuffer(). This basically means that
3074 * whetnever ust_metadata_pushed is incremented, the associated
3075 * metadata has been consumed from the metadata stream.
3076 */
3077 DBG("UST consumer metadata pending check: contiguous %" PRIu64 " vs pushed %" PRIu64,
3078 contiguous, pushed);
3079 assert(((int64_t) (contiguous - pushed)) >= 0);
3080 if ((contiguous != pushed) ||
3081 (((int64_t) contiguous - pushed) > 0 || contiguous == 0)) {
3082 ret = 1; /* Data is pending */
3083 goto end;
3084 }
3085 } else {
3086 ret = ustctl_get_next_subbuf(stream->ustream);
3087 if (ret == 0) {
3088 /*
3089 * There is still data so let's put back this
3090 * subbuffer.
3091 */
3092 ret = ustctl_put_subbuf(stream->ustream);
3093 assert(ret == 0);
3094 ret = 1; /* Data is pending */
3095 goto end;
3096 }
3097 }
3098
3099 /* Data is NOT pending so ready to be read. */
3100 ret = 0;
3101
3102 end:
3103 return ret;
3104 }
3105
3106 /*
3107 * Stop a given metadata channel timer if enabled and close the wait fd which
3108 * is the poll pipe of the metadata stream.
3109 *
3110 * This MUST be called with the metadata channel lock acquired.
3111 */
3112 void lttng_ustconsumer_close_metadata(struct lttng_consumer_channel *metadata)
3113 {
3114 int ret;
3115
3116 assert(metadata);
3117 assert(metadata->type == CONSUMER_CHANNEL_TYPE_METADATA);
3118
3119 DBG("Closing metadata channel key %" PRIu64, metadata->key);
3120
3121 if (metadata->switch_timer_enabled == 1) {
3122 consumer_timer_switch_stop(metadata);
3123 }
3124
3125 if (!metadata->metadata_stream) {
3126 goto end;
3127 }
3128
3129 /*
3130 * Closing write side so the thread monitoring the stream wakes up if any
3131 * and clean the metadata stream.
3132 */
3133 if (metadata->metadata_stream->ust_metadata_poll_pipe[1] >= 0) {
3134 ret = close(metadata->metadata_stream->ust_metadata_poll_pipe[1]);
3135 if (ret < 0) {
3136 PERROR("closing metadata pipe write side");
3137 }
3138 metadata->metadata_stream->ust_metadata_poll_pipe[1] = -1;
3139 }
3140
3141 end:
3142 return;
3143 }
3144
3145 /*
3146 * Close every metadata stream wait fd of the metadata hash table. This
3147 * function MUST be used very carefully so not to run into a race between the
3148 * metadata thread handling streams and this function closing their wait fd.
3149 *
3150 * For UST, this is used when the session daemon hangs up. Its the metadata
3151 * producer so calling this is safe because we are assured that no state change
3152 * can occur in the metadata thread for the streams in the hash table.
3153 */
3154 void lttng_ustconsumer_close_all_metadata(struct lttng_ht *metadata_ht)
3155 {
3156 struct lttng_ht_iter iter;
3157 struct lttng_consumer_stream *stream;
3158
3159 assert(metadata_ht);
3160 assert(metadata_ht->ht);
3161
3162 DBG("UST consumer closing all metadata streams");
3163
3164 rcu_read_lock();
3165 cds_lfht_for_each_entry(metadata_ht->ht, &iter.iter, stream,
3166 node.node) {
3167
3168 health_code_update();
3169
3170 pthread_mutex_lock(&stream->chan->lock);
3171 lttng_ustconsumer_close_metadata(stream->chan);
3172 pthread_mutex_unlock(&stream->chan->lock);
3173
3174 }
3175 rcu_read_unlock();
3176 }
3177
3178 void lttng_ustconsumer_close_stream_wakeup(struct lttng_consumer_stream *stream)
3179 {
3180 int ret;
3181
3182 ret = ustctl_stream_close_wakeup_fd(stream->ustream);
3183 if (ret < 0) {
3184 ERR("Unable to close wakeup fd");
3185 }
3186 }
3187
3188 /*
3189 * Please refer to consumer-timer.c before adding any lock within this
3190 * function or any of its callees. Timers have a very strict locking
3191 * semantic with respect to teardown. Failure to respect this semantic
3192 * introduces deadlocks.
3193 *
3194 * DON'T hold the metadata lock when calling this function, else this
3195 * can cause deadlock involving consumer awaiting for metadata to be
3196 * pushed out due to concurrent interaction with the session daemon.
3197 */
3198 int lttng_ustconsumer_request_metadata(struct lttng_consumer_local_data *ctx,
3199 struct lttng_consumer_channel *channel, int timer, int wait)
3200 {
3201 struct lttcomm_metadata_request_msg request;
3202 struct lttcomm_consumer_msg msg;
3203 enum lttcomm_return_code ret_code = LTTCOMM_CONSUMERD_SUCCESS;
3204 uint64_t len, key, offset, version;
3205 int ret;
3206
3207 assert(channel);
3208 assert(channel->metadata_cache);
3209
3210 memset(&request, 0, sizeof(request));
3211
3212 /* send the metadata request to sessiond */
3213 switch (consumer_data.type) {
3214 case LTTNG_CONSUMER64_UST:
3215 request.bits_per_long = 64;
3216 break;
3217 case LTTNG_CONSUMER32_UST:
3218 request.bits_per_long = 32;
3219 break;
3220 default:
3221 request.bits_per_long = 0;
3222 break;
3223 }
3224
3225 request.session_id = channel->session_id;
3226 request.session_id_per_pid = channel->session_id_per_pid;
3227 /*
3228 * Request the application UID here so the metadata of that application can
3229 * be sent back. The channel UID corresponds to the user UID of the session
3230 * used for the rights on the stream file(s).
3231 */
3232 request.uid = channel->ust_app_uid;
3233 request.key = channel->key;
3234
3235 DBG("Sending metadata request to sessiond, session id %" PRIu64
3236 ", per-pid %" PRIu64 ", app UID %u and channel key %" PRIu64,
3237 request.session_id, request.session_id_per_pid, request.uid,
3238 request.key);
3239
3240 pthread_mutex_lock(&ctx->metadata_socket_lock);
3241
3242 health_code_update();
3243
3244 ret = lttcomm_send_unix_sock(ctx->consumer_metadata_socket, &request,
3245 sizeof(request));
3246 if (ret < 0) {
3247 ERR("Asking metadata to sessiond");
3248 goto end;
3249 }
3250
3251 health_code_update();
3252
3253 /* Receive the metadata from sessiond */
3254 ret = lttcomm_recv_unix_sock(ctx->consumer_metadata_socket, &msg,
3255 sizeof(msg));
3256 if (ret != sizeof(msg)) {
3257 DBG("Consumer received unexpected message size %d (expects %zu)",
3258 ret, sizeof(msg));
3259 lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_ERROR_RECV_CMD);
3260 /*
3261 * The ret value might 0 meaning an orderly shutdown but this is ok
3262 * since the caller handles this.
3263 */
3264 goto end;
3265 }
3266
3267 health_code_update();
3268
3269 if (msg.cmd_type == LTTNG_ERR_UND) {
3270 /* No registry found */
3271 (void) consumer_send_status_msg(ctx->consumer_metadata_socket,
3272 ret_code);
3273 ret = 0;
3274 goto end;
3275 } else if (msg.cmd_type != LTTNG_CONSUMER_PUSH_METADATA) {
3276 ERR("Unexpected cmd_type received %d", msg.cmd_type);
3277 ret = -1;
3278 goto end;
3279 }
3280
3281 len = msg.u.push_metadata.len;
3282 key = msg.u.push_metadata.key;
3283 offset = msg.u.push_metadata.target_offset;
3284 version = msg.u.push_metadata.version;
3285
3286 assert(key == channel->key);
3287 if (len == 0) {
3288 DBG("No new metadata to receive for key %" PRIu64, key);
3289 }
3290
3291 health_code_update();
3292
3293 /* Tell session daemon we are ready to receive the metadata. */
3294 ret = consumer_send_status_msg(ctx->consumer_metadata_socket,
3295 LTTCOMM_CONSUMERD_SUCCESS);
3296 if (ret < 0 || len == 0) {
3297 /*
3298 * Somehow, the session daemon is not responding anymore or there is
3299 * nothing to receive.
3300 */
3301 goto end;
3302 }
3303
3304 health_code_update();
3305
3306 ret = lttng_ustconsumer_recv_metadata(ctx->consumer_metadata_socket,
3307 key, offset, len, version, channel, timer, wait);
3308 if (ret >= 0) {
3309 /*
3310 * Only send the status msg if the sessiond is alive meaning a positive
3311 * ret code.
3312 */
3313 (void) consumer_send_status_msg(ctx->consumer_metadata_socket, ret);
3314 }
3315 ret = 0;
3316
3317 end:
3318 health_code_update();
3319
3320 pthread_mutex_unlock(&ctx->metadata_socket_lock);
3321 return ret;
3322 }
3323
3324 /*
3325 * Return the ustctl call for the get stream id.
3326 */
3327 int lttng_ustconsumer_get_stream_id(struct lttng_consumer_stream *stream,
3328 uint64_t *stream_id)
3329 {
3330 assert(stream);
3331 assert(stream_id);
3332
3333 return ustctl_get_stream_id(stream->ustream, stream_id);
3334 }
This page took 0.167954 seconds and 3 git commands to generate.