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