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