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