Fix: intialization of ust_metadata_poll_pipe to garbage value
[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 *
5 * This program is free software; you can redistribute it and/or modify
6 * it under the terms of the GNU General Public License, version 2 only,
7 * as published by the Free Software Foundation.
8 *
9 * This program is distributed in the hope that it will be useful,
10 * but WITHOUT ANY WARRANTY; without even the implied warranty of
11 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
12 * GNU General Public License for more details.
13 *
14 * You should have received a copy of the GNU General Public License along
15 * with this program; if not, write to the Free Software Foundation, Inc.,
16 * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
17 */
18
19 #define _GNU_SOURCE
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
36 #include <bin/lttng-consumerd/health-consumerd.h>
37 #include <common/common.h>
38 #include <common/sessiond-comm/sessiond-comm.h>
39 #include <common/relayd/relayd.h>
40 #include <common/compat/fcntl.h>
41 #include <common/compat/endian.h>
42 #include <common/consumer-metadata-cache.h>
43 #include <common/consumer-stream.h>
44 #include <common/consumer-timer.h>
45 #include <common/utils.h>
46 #include <common/index/index.h>
47
48 #include "ust-consumer.h"
49
50 #define UINT_MAX_STR_LEN 11 /* includes \0 */
51
52 extern struct lttng_consumer_global_data consumer_data;
53 extern int consumer_poll_timeout;
54 extern volatile int consumer_quit;
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 }
86 free(channel);
87 }
88
89 /*
90 * Add channel to internal consumer state.
91 *
92 * Returns 0 on success or else a negative value.
93 */
94 static int add_channel(struct lttng_consumer_channel *channel,
95 struct lttng_consumer_local_data *ctx)
96 {
97 int ret = 0;
98
99 assert(channel);
100 assert(ctx);
101
102 if (ctx->on_recv_channel != NULL) {
103 ret = ctx->on_recv_channel(channel);
104 if (ret == 0) {
105 ret = consumer_add_channel(channel, ctx);
106 } else if (ret < 0) {
107 /* Most likely an ENOMEM. */
108 lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_OUTFD_ERROR);
109 goto error;
110 }
111 } else {
112 ret = consumer_add_channel(channel, ctx);
113 }
114
115 DBG("UST consumer channel added (key: %" PRIu64 ")", channel->key);
116
117 error:
118 return ret;
119 }
120
121 /*
122 * Allocate and return a consumer channel object.
123 */
124 static struct lttng_consumer_channel *allocate_channel(uint64_t session_id,
125 const char *pathname, const char *name, uid_t uid, gid_t gid,
126 uint64_t relayd_id, uint64_t key, enum lttng_event_output output,
127 uint64_t tracefile_size, uint64_t tracefile_count,
128 uint64_t session_id_per_pid, unsigned int monitor,
129 unsigned int live_timer_interval,
130 const char *root_shm_path, const char *shm_path)
131 {
132 assert(pathname);
133 assert(name);
134
135 return consumer_allocate_channel(key, session_id, pathname, name, uid,
136 gid, relayd_id, output, tracefile_size,
137 tracefile_count, session_id_per_pid, monitor,
138 live_timer_interval, root_shm_path, shm_path);
139 }
140
141 /*
142 * Allocate and return a consumer stream object. If _alloc_ret is not NULL, the
143 * error value if applicable is set in it else it is kept untouched.
144 *
145 * Return NULL on error else the newly allocated stream object.
146 */
147 static struct lttng_consumer_stream *allocate_stream(int cpu, int key,
148 struct lttng_consumer_channel *channel,
149 struct lttng_consumer_local_data *ctx, int *_alloc_ret)
150 {
151 int alloc_ret;
152 struct lttng_consumer_stream *stream = NULL;
153
154 assert(channel);
155 assert(ctx);
156
157 stream = consumer_allocate_stream(channel->key,
158 key,
159 LTTNG_CONSUMER_ACTIVE_STREAM,
160 channel->name,
161 channel->uid,
162 channel->gid,
163 channel->relayd_id,
164 channel->session_id,
165 cpu,
166 &alloc_ret,
167 channel->type,
168 channel->monitor);
169 if (stream == NULL) {
170 switch (alloc_ret) {
171 case -ENOENT:
172 /*
173 * We could not find the channel. Can happen if cpu hotplug
174 * happens while tearing down.
175 */
176 DBG3("Could not find channel");
177 break;
178 case -ENOMEM:
179 case -EINVAL:
180 default:
181 lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_OUTFD_ERROR);
182 break;
183 }
184 goto error;
185 }
186
187 stream->chan = channel;
188
189 error:
190 if (_alloc_ret) {
191 *_alloc_ret = alloc_ret;
192 }
193 return stream;
194 }
195
196 /*
197 * Send the given stream pointer to the corresponding thread.
198 *
199 * Returns 0 on success else a negative value.
200 */
201 static int send_stream_to_thread(struct lttng_consumer_stream *stream,
202 struct lttng_consumer_local_data *ctx)
203 {
204 int ret;
205 struct lttng_pipe *stream_pipe;
206
207 /* Get the right pipe where the stream will be sent. */
208 if (stream->metadata_flag) {
209 ret = consumer_add_metadata_stream(stream);
210 if (ret) {
211 ERR("Consumer add metadata stream %" PRIu64 " failed.",
212 stream->key);
213 goto error;
214 }
215 stream_pipe = ctx->consumer_metadata_pipe;
216 } else {
217 ret = consumer_add_data_stream(stream);
218 if (ret) {
219 ERR("Consumer add stream %" PRIu64 " failed.",
220 stream->key);
221 goto error;
222 }
223 stream_pipe = ctx->consumer_data_pipe;
224 }
225
226 /*
227 * From this point on, the stream's ownership has been moved away from
228 * the channel and becomes globally visible.
229 */
230 stream->globally_visible = 1;
231
232 ret = lttng_pipe_write(stream_pipe, &stream, sizeof(stream));
233 if (ret < 0) {
234 ERR("Consumer write %s stream to pipe %d",
235 stream->metadata_flag ? "metadata" : "data",
236 lttng_pipe_get_writefd(stream_pipe));
237 if (stream->metadata_flag) {
238 consumer_del_stream_for_metadata(stream);
239 } else {
240 consumer_del_stream_for_data(stream);
241 }
242 }
243 error:
244 return ret;
245 }
246
247 static
248 int get_stream_shm_path(char *stream_shm_path, const char *shm_path, int cpu)
249 {
250 char cpu_nr[UINT_MAX_STR_LEN]; /* unsigned int max len */
251 int ret;
252
253 strncpy(stream_shm_path, shm_path, PATH_MAX);
254 stream_shm_path[PATH_MAX - 1] = '\0';
255 ret = snprintf(cpu_nr, UINT_MAX_STR_LEN, "%u", cpu);
256 if (ret != 1) {
257 ret = -1;
258 goto end;
259 }
260 strncat(stream_shm_path, cpu_nr,
261 PATH_MAX - strlen(stream_shm_path) - 1);
262 ret = 0;
263 end:
264 return ret;
265 }
266
267 /*
268 * Create streams for the given channel using liblttng-ust-ctl.
269 *
270 * Return 0 on success else a negative value.
271 */
272 static int create_ust_streams(struct lttng_consumer_channel *channel,
273 struct lttng_consumer_local_data *ctx)
274 {
275 int ret, cpu = 0;
276 struct ustctl_consumer_stream *ustream;
277 struct lttng_consumer_stream *stream;
278
279 assert(channel);
280 assert(ctx);
281
282 /*
283 * While a stream is available from ustctl. When NULL is returned, we've
284 * reached the end of the possible stream for the channel.
285 */
286 while ((ustream = ustctl_create_stream(channel->uchan, cpu))) {
287 int wait_fd;
288 int ust_metadata_pipe[2];
289
290 health_code_update();
291
292 if (channel->type == CONSUMER_CHANNEL_TYPE_METADATA && channel->monitor) {
293 ret = utils_create_pipe_cloexec_nonblock(ust_metadata_pipe);
294 if (ret < 0) {
295 ERR("Create ust metadata poll pipe");
296 goto error;
297 }
298 wait_fd = ust_metadata_pipe[0];
299 } else {
300 wait_fd = ustctl_stream_get_wait_fd(ustream);
301 }
302
303 /* Allocate consumer stream object. */
304 stream = allocate_stream(cpu, wait_fd, channel, ctx, &ret);
305 if (!stream) {
306 goto error_alloc;
307 }
308 stream->ustream = ustream;
309 /*
310 * Store it so we can save multiple function calls afterwards since
311 * this value is used heavily in the stream threads. This is UST
312 * specific so this is why it's done after allocation.
313 */
314 stream->wait_fd = wait_fd;
315
316 /*
317 * Increment channel refcount since the channel reference has now been
318 * assigned in the allocation process above.
319 */
320 if (stream->chan->monitor) {
321 uatomic_inc(&stream->chan->refcount);
322 }
323
324 /*
325 * Order is important this is why a list is used. On error, the caller
326 * should clean this list.
327 */
328 cds_list_add_tail(&stream->send_node, &channel->streams.head);
329
330 ret = ustctl_get_max_subbuf_size(stream->ustream,
331 &stream->max_sb_size);
332 if (ret < 0) {
333 ERR("ustctl_get_max_subbuf_size failed for stream %s",
334 stream->name);
335 goto error;
336 }
337
338 /* Do actions once stream has been received. */
339 if (ctx->on_recv_stream) {
340 ret = ctx->on_recv_stream(stream);
341 if (ret < 0) {
342 goto error;
343 }
344 }
345
346 DBG("UST consumer add stream %s (key: %" PRIu64 ") with relayd id %" PRIu64,
347 stream->name, stream->key, stream->relayd_stream_id);
348
349 /* Set next CPU stream. */
350 channel->streams.count = ++cpu;
351
352 /* Keep stream reference when creating metadata. */
353 if (channel->type == CONSUMER_CHANNEL_TYPE_METADATA) {
354 channel->metadata_stream = stream;
355 if (channel->monitor) {
356 /* Set metadata poll pipe if we created one */
357 memcpy(stream->ust_metadata_poll_pipe,
358 ust_metadata_pipe,
359 sizeof(ust_metadata_pipe));
360 }
361 }
362 }
363
364 return 0;
365
366 error:
367 error_alloc:
368 return ret;
369 }
370
371 /*
372 * create_posix_shm is never called concurrently within a process.
373 */
374 static
375 int create_posix_shm(void)
376 {
377 char tmp_name[NAME_MAX];
378 int shmfd, ret;
379
380 ret = snprintf(tmp_name, NAME_MAX, "/ust-shm-consumer-%d", getpid());
381 if (ret < 0) {
382 PERROR("snprintf");
383 return -1;
384 }
385 /*
386 * Allocate shm, and immediately unlink its shm oject, keeping
387 * only the file descriptor as a reference to the object.
388 * We specifically do _not_ use the / at the beginning of the
389 * pathname so that some OS implementations can keep it local to
390 * the process (POSIX leaves this implementation-defined).
391 */
392 shmfd = shm_open(tmp_name, O_CREAT | O_EXCL | O_RDWR, 0700);
393 if (shmfd < 0) {
394 PERROR("shm_open");
395 goto error_shm_open;
396 }
397 ret = shm_unlink(tmp_name);
398 if (ret < 0 && errno != ENOENT) {
399 PERROR("shm_unlink");
400 goto error_shm_release;
401 }
402 return shmfd;
403
404 error_shm_release:
405 ret = close(shmfd);
406 if (ret) {
407 PERROR("close");
408 }
409 error_shm_open:
410 return -1;
411 }
412
413 static int open_ust_stream_fd(struct lttng_consumer_channel *channel,
414 struct ustctl_consumer_channel_attr *attr,
415 int cpu)
416 {
417 char shm_path[PATH_MAX];
418 int ret;
419
420 if (!channel->shm_path[0]) {
421 return create_posix_shm();
422 }
423 ret = get_stream_shm_path(shm_path, channel->shm_path, cpu);
424 if (ret) {
425 goto error_shm_path;
426 }
427 return run_as_open(shm_path,
428 O_RDWR | O_CREAT | O_EXCL, S_IRUSR | S_IWUSR,
429 channel->uid, channel->gid);
430
431 error_shm_path:
432 return -1;
433 }
434
435 /*
436 * Create an UST channel with the given attributes and send it to the session
437 * daemon using the ust ctl API.
438 *
439 * Return 0 on success or else a negative value.
440 */
441 static int create_ust_channel(struct lttng_consumer_channel *channel,
442 struct ustctl_consumer_channel_attr *attr,
443 struct ustctl_consumer_channel **ust_chanp)
444 {
445 int ret, nr_stream_fds, i, j;
446 int *stream_fds;
447 struct ustctl_consumer_channel *ust_channel;
448
449 assert(channel);
450 assert(attr);
451 assert(ust_chanp);
452
453 DBG3("Creating channel to ustctl with attr: [overwrite: %d, "
454 "subbuf_size: %" PRIu64 ", num_subbuf: %" PRIu64 ", "
455 "switch_timer_interval: %u, read_timer_interval: %u, "
456 "output: %d, type: %d", attr->overwrite, attr->subbuf_size,
457 attr->num_subbuf, attr->switch_timer_interval,
458 attr->read_timer_interval, attr->output, attr->type);
459
460 if (channel->type == CONSUMER_CHANNEL_TYPE_METADATA)
461 nr_stream_fds = 1;
462 else
463 nr_stream_fds = ustctl_get_nr_stream_per_channel();
464 stream_fds = zmalloc(nr_stream_fds * sizeof(*stream_fds));
465 if (!stream_fds) {
466 ret = -1;
467 goto error_alloc;
468 }
469 for (i = 0; i < nr_stream_fds; i++) {
470 stream_fds[i] = open_ust_stream_fd(channel, attr, i);
471 if (stream_fds[i] < 0) {
472 ret = -1;
473 goto error_open;
474 }
475 }
476 ust_channel = ustctl_create_channel(attr, stream_fds, nr_stream_fds);
477 if (!ust_channel) {
478 ret = -1;
479 goto error_create;
480 }
481 channel->nr_stream_fds = nr_stream_fds;
482 channel->stream_fds = stream_fds;
483 *ust_chanp = ust_channel;
484
485 return 0;
486
487 error_create:
488 error_open:
489 for (j = i - 1; j >= 0; j--) {
490 int closeret;
491
492 closeret = close(stream_fds[j]);
493 if (closeret) {
494 PERROR("close");
495 }
496 if (channel->shm_path[0]) {
497 char shm_path[PATH_MAX];
498
499 closeret = get_stream_shm_path(shm_path,
500 channel->shm_path, j);
501 if (closeret) {
502 ERR("Cannot get stream shm path");
503 }
504 closeret = run_as_unlink(shm_path,
505 channel->uid, channel->gid);
506 if (closeret) {
507 errno = -closeret;
508 PERROR("unlink %s", shm_path);
509 }
510 }
511 }
512 /* Try to rmdir all directories under shm_path root. */
513 if (channel->root_shm_path[0]) {
514 (void) run_as_recursive_rmdir(channel->root_shm_path,
515 channel->uid, channel->gid);
516 }
517 free(stream_fds);
518 error_alloc:
519 return ret;
520 }
521
522 /*
523 * Send a single given stream to the session daemon using the sock.
524 *
525 * Return 0 on success else a negative value.
526 */
527 static int send_sessiond_stream(int sock, struct lttng_consumer_stream *stream)
528 {
529 int ret;
530
531 assert(stream);
532 assert(sock >= 0);
533
534 DBG("UST consumer sending stream %" PRIu64 " to sessiond", stream->key);
535
536 /* Send stream to session daemon. */
537 ret = ustctl_send_stream_to_sessiond(sock, stream->ustream);
538 if (ret < 0) {
539 goto error;
540 }
541
542 error:
543 return ret;
544 }
545
546 /*
547 * Send channel to sessiond.
548 *
549 * Return 0 on success or else a negative value.
550 */
551 static int send_sessiond_channel(int sock,
552 struct lttng_consumer_channel *channel,
553 struct lttng_consumer_local_data *ctx, int *relayd_error)
554 {
555 int ret, ret_code = LTTCOMM_CONSUMERD_SUCCESS;
556 struct lttng_consumer_stream *stream;
557 uint64_t net_seq_idx = -1ULL;
558
559 assert(channel);
560 assert(ctx);
561 assert(sock >= 0);
562
563 DBG("UST consumer sending channel %s to sessiond", channel->name);
564
565 if (channel->relayd_id != (uint64_t) -1ULL) {
566 cds_list_for_each_entry(stream, &channel->streams.head, send_node) {
567
568 health_code_update();
569
570 /* Try to send the stream to the relayd if one is available. */
571 ret = consumer_send_relayd_stream(stream, stream->chan->pathname);
572 if (ret < 0) {
573 /*
574 * Flag that the relayd was the problem here probably due to a
575 * communicaton error on the socket.
576 */
577 if (relayd_error) {
578 *relayd_error = 1;
579 }
580 ret_code = LTTCOMM_CONSUMERD_RELAYD_FAIL;
581 }
582 if (net_seq_idx == -1ULL) {
583 net_seq_idx = stream->net_seq_idx;
584 }
585 }
586 }
587
588 /* Inform sessiond that we are about to send channel and streams. */
589 ret = consumer_send_status_msg(sock, ret_code);
590 if (ret < 0 || ret_code != LTTCOMM_CONSUMERD_SUCCESS) {
591 /*
592 * Either the session daemon is not responding or the relayd died so we
593 * stop now.
594 */
595 goto error;
596 }
597
598 /* Send channel to sessiond. */
599 ret = ustctl_send_channel_to_sessiond(sock, channel->uchan);
600 if (ret < 0) {
601 goto error;
602 }
603
604 ret = ustctl_channel_close_wakeup_fd(channel->uchan);
605 if (ret < 0) {
606 goto error;
607 }
608
609 /* The channel was sent successfully to the sessiond at this point. */
610 cds_list_for_each_entry(stream, &channel->streams.head, send_node) {
611
612 health_code_update();
613
614 /* Send stream to session daemon. */
615 ret = send_sessiond_stream(sock, stream);
616 if (ret < 0) {
617 goto error;
618 }
619 }
620
621 /* Tell sessiond there is no more stream. */
622 ret = ustctl_send_stream_to_sessiond(sock, NULL);
623 if (ret < 0) {
624 goto error;
625 }
626
627 DBG("UST consumer NULL stream sent to sessiond");
628
629 return 0;
630
631 error:
632 if (ret_code != LTTCOMM_CONSUMERD_SUCCESS) {
633 ret = -1;
634 }
635 return ret;
636 }
637
638 /*
639 * Creates a channel and streams and add the channel it to the channel internal
640 * state. The created stream must ONLY be sent once the GET_CHANNEL command is
641 * received.
642 *
643 * Return 0 on success or else, a negative value is returned and the channel
644 * MUST be destroyed by consumer_del_channel().
645 */
646 static int ask_channel(struct lttng_consumer_local_data *ctx, int sock,
647 struct lttng_consumer_channel *channel,
648 struct ustctl_consumer_channel_attr *attr)
649 {
650 int ret;
651
652 assert(ctx);
653 assert(channel);
654 assert(attr);
655
656 /*
657 * This value is still used by the kernel consumer since for the kernel,
658 * the stream ownership is not IN the consumer so we need to have the
659 * number of left stream that needs to be initialized so we can know when
660 * to delete the channel (see consumer.c).
661 *
662 * As for the user space tracer now, the consumer creates and sends the
663 * stream to the session daemon which only sends them to the application
664 * once every stream of a channel is received making this value useless
665 * because we they will be added to the poll thread before the application
666 * receives them. This ensures that a stream can not hang up during
667 * initilization of a channel.
668 */
669 channel->nb_init_stream_left = 0;
670
671 /* The reply msg status is handled in the following call. */
672 ret = create_ust_channel(channel, attr, &channel->uchan);
673 if (ret < 0) {
674 goto end;
675 }
676
677 channel->wait_fd = ustctl_channel_get_wait_fd(channel->uchan);
678
679 /*
680 * For the snapshots (no monitor), we create the metadata streams
681 * on demand, not during the channel creation.
682 */
683 if (channel->type == CONSUMER_CHANNEL_TYPE_METADATA && !channel->monitor) {
684 ret = 0;
685 goto end;
686 }
687
688 /* Open all streams for this channel. */
689 ret = create_ust_streams(channel, ctx);
690 if (ret < 0) {
691 goto end;
692 }
693
694 end:
695 return ret;
696 }
697
698 /*
699 * Send all stream of a channel to the right thread handling it.
700 *
701 * On error, return a negative value else 0 on success.
702 */
703 static int send_streams_to_thread(struct lttng_consumer_channel *channel,
704 struct lttng_consumer_local_data *ctx)
705 {
706 int ret = 0;
707 struct lttng_consumer_stream *stream, *stmp;
708
709 assert(channel);
710 assert(ctx);
711
712 /* Send streams to the corresponding thread. */
713 cds_list_for_each_entry_safe(stream, stmp, &channel->streams.head,
714 send_node) {
715
716 health_code_update();
717
718 /* Sending the stream to the thread. */
719 ret = send_stream_to_thread(stream, ctx);
720 if (ret < 0) {
721 /*
722 * If we are unable to send the stream to the thread, there is
723 * a big problem so just stop everything.
724 */
725 /* Remove node from the channel stream list. */
726 cds_list_del(&stream->send_node);
727 goto error;
728 }
729
730 /* Remove node from the channel stream list. */
731 cds_list_del(&stream->send_node);
732
733 }
734
735 error:
736 return ret;
737 }
738
739 /*
740 * Flush channel's streams using the given key to retrieve the channel.
741 *
742 * Return 0 on success else an LTTng error code.
743 */
744 static int flush_channel(uint64_t chan_key)
745 {
746 int ret = 0;
747 struct lttng_consumer_channel *channel;
748 struct lttng_consumer_stream *stream;
749 struct lttng_ht *ht;
750 struct lttng_ht_iter iter;
751
752 DBG("UST consumer flush channel key %" PRIu64, chan_key);
753
754 rcu_read_lock();
755 channel = consumer_find_channel(chan_key);
756 if (!channel) {
757 ERR("UST consumer flush channel %" PRIu64 " not found", chan_key);
758 ret = LTTNG_ERR_UST_CHAN_NOT_FOUND;
759 goto error;
760 }
761
762 ht = consumer_data.stream_per_chan_id_ht;
763
764 /* For each stream of the channel id, flush it. */
765 cds_lfht_for_each_entry_duplicate(ht->ht,
766 ht->hash_fct(&channel->key, lttng_ht_seed), ht->match_fct,
767 &channel->key, &iter.iter, stream, node_channel_id.node) {
768
769 health_code_update();
770
771 ustctl_flush_buffer(stream->ustream, 1);
772 }
773 error:
774 rcu_read_unlock();
775 return ret;
776 }
777
778 /*
779 * Close metadata stream wakeup_fd using the given key to retrieve the channel.
780 * RCU read side lock MUST be acquired before calling this function.
781 *
782 * Return 0 on success else an LTTng error code.
783 */
784 static int close_metadata(uint64_t chan_key)
785 {
786 int ret = 0;
787 struct lttng_consumer_channel *channel;
788
789 DBG("UST consumer close metadata key %" PRIu64, chan_key);
790
791 channel = consumer_find_channel(chan_key);
792 if (!channel) {
793 /*
794 * This is possible if the metadata thread has issue a delete because
795 * the endpoint point of the stream hung up. There is no way the
796 * session daemon can know about it thus use a DBG instead of an actual
797 * error.
798 */
799 DBG("UST consumer close metadata %" PRIu64 " not found", chan_key);
800 ret = LTTNG_ERR_UST_CHAN_NOT_FOUND;
801 goto error;
802 }
803
804 pthread_mutex_lock(&consumer_data.lock);
805 pthread_mutex_lock(&channel->lock);
806
807 if (cds_lfht_is_node_deleted(&channel->node.node)) {
808 goto error_unlock;
809 }
810
811 lttng_ustconsumer_close_metadata(channel);
812
813 error_unlock:
814 pthread_mutex_unlock(&channel->lock);
815 pthread_mutex_unlock(&consumer_data.lock);
816 error:
817 return ret;
818 }
819
820 /*
821 * RCU read side lock MUST be acquired before calling this function.
822 *
823 * Return 0 on success else an LTTng error code.
824 */
825 static int setup_metadata(struct lttng_consumer_local_data *ctx, uint64_t key)
826 {
827 int ret;
828 struct lttng_consumer_channel *metadata;
829
830 DBG("UST consumer setup metadata key %" PRIu64, key);
831
832 metadata = consumer_find_channel(key);
833 if (!metadata) {
834 ERR("UST consumer push metadata %" PRIu64 " not found", key);
835 ret = LTTNG_ERR_UST_CHAN_NOT_FOUND;
836 goto end;
837 }
838
839 /*
840 * In no monitor mode, the metadata channel has no stream(s) so skip the
841 * ownership transfer to the metadata thread.
842 */
843 if (!metadata->monitor) {
844 DBG("Metadata channel in no monitor");
845 ret = 0;
846 goto end;
847 }
848
849 /*
850 * Send metadata stream to relayd if one available. Availability is
851 * known if the stream is still in the list of the channel.
852 */
853 if (cds_list_empty(&metadata->streams.head)) {
854 ERR("Metadata channel key %" PRIu64 ", no stream available.", key);
855 ret = LTTCOMM_CONSUMERD_ERROR_METADATA;
856 goto error_no_stream;
857 }
858
859 /* Send metadata stream to relayd if needed. */
860 if (metadata->metadata_stream->net_seq_idx != (uint64_t) -1ULL) {
861 ret = consumer_send_relayd_stream(metadata->metadata_stream,
862 metadata->pathname);
863 if (ret < 0) {
864 ret = LTTCOMM_CONSUMERD_ERROR_METADATA;
865 goto error;
866 }
867 ret = consumer_send_relayd_streams_sent(
868 metadata->metadata_stream->net_seq_idx);
869 if (ret < 0) {
870 ret = LTTCOMM_CONSUMERD_RELAYD_FAIL;
871 goto error;
872 }
873 }
874
875 ret = send_streams_to_thread(metadata, ctx);
876 if (ret < 0) {
877 /*
878 * If we are unable to send the stream to the thread, there is
879 * a big problem so just stop everything.
880 */
881 ret = LTTCOMM_CONSUMERD_FATAL;
882 goto error;
883 }
884 /* List MUST be empty after or else it could be reused. */
885 assert(cds_list_empty(&metadata->streams.head));
886
887 ret = 0;
888 goto end;
889
890 error:
891 /*
892 * Delete metadata channel on error. At this point, the metadata stream can
893 * NOT be monitored by the metadata thread thus having the guarantee that
894 * the stream is still in the local stream list of the channel. This call
895 * will make sure to clean that list.
896 */
897 consumer_stream_destroy(metadata->metadata_stream, NULL);
898 cds_list_del(&metadata->metadata_stream->send_node);
899 metadata->metadata_stream = NULL;
900 error_no_stream:
901 end:
902 return ret;
903 }
904
905 /*
906 * Snapshot the whole metadata.
907 *
908 * Returns 0 on success, < 0 on error
909 */
910 static int snapshot_metadata(uint64_t key, char *path, uint64_t relayd_id,
911 struct lttng_consumer_local_data *ctx)
912 {
913 int ret = 0;
914 struct lttng_consumer_channel *metadata_channel;
915 struct lttng_consumer_stream *metadata_stream;
916
917 assert(path);
918 assert(ctx);
919
920 DBG("UST consumer snapshot metadata with key %" PRIu64 " at path %s",
921 key, path);
922
923 rcu_read_lock();
924
925 metadata_channel = consumer_find_channel(key);
926 if (!metadata_channel) {
927 ERR("UST snapshot metadata channel not found for key %" PRIu64,
928 key);
929 ret = -1;
930 goto error;
931 }
932 assert(!metadata_channel->monitor);
933
934 health_code_update();
935
936 /*
937 * Ask the sessiond if we have new metadata waiting and update the
938 * consumer metadata cache.
939 */
940 ret = lttng_ustconsumer_request_metadata(ctx, metadata_channel, 0, 1);
941 if (ret < 0) {
942 goto error;
943 }
944
945 health_code_update();
946
947 /*
948 * The metadata stream is NOT created in no monitor mode when the channel
949 * is created on a sessiond ask channel command.
950 */
951 ret = create_ust_streams(metadata_channel, ctx);
952 if (ret < 0) {
953 goto error;
954 }
955
956 metadata_stream = metadata_channel->metadata_stream;
957 assert(metadata_stream);
958
959 if (relayd_id != (uint64_t) -1ULL) {
960 metadata_stream->net_seq_idx = relayd_id;
961 ret = consumer_send_relayd_stream(metadata_stream, path);
962 if (ret < 0) {
963 goto error_stream;
964 }
965 } else {
966 ret = utils_create_stream_file(path, metadata_stream->name,
967 metadata_stream->chan->tracefile_size,
968 metadata_stream->tracefile_count_current,
969 metadata_stream->uid, metadata_stream->gid, NULL);
970 if (ret < 0) {
971 goto error_stream;
972 }
973 metadata_stream->out_fd = ret;
974 metadata_stream->tracefile_size_current = 0;
975 }
976
977 do {
978 health_code_update();
979
980 ret = lttng_consumer_read_subbuffer(metadata_stream, ctx);
981 if (ret < 0) {
982 goto error_stream;
983 }
984 } while (ret > 0);
985
986 error_stream:
987 /*
988 * Clean up the stream completly because the next snapshot will use a new
989 * metadata stream.
990 */
991 consumer_stream_destroy(metadata_stream, NULL);
992 cds_list_del(&metadata_stream->send_node);
993 metadata_channel->metadata_stream = NULL;
994
995 error:
996 rcu_read_unlock();
997 return ret;
998 }
999
1000 /*
1001 * Take a snapshot of all the stream of a channel.
1002 *
1003 * Returns 0 on success, < 0 on error
1004 */
1005 static int snapshot_channel(uint64_t key, char *path, uint64_t relayd_id,
1006 uint64_t nb_packets_per_stream, struct lttng_consumer_local_data *ctx)
1007 {
1008 int ret;
1009 unsigned use_relayd = 0;
1010 unsigned long consumed_pos, produced_pos;
1011 struct lttng_consumer_channel *channel;
1012 struct lttng_consumer_stream *stream;
1013
1014 assert(path);
1015 assert(ctx);
1016
1017 rcu_read_lock();
1018
1019 if (relayd_id != (uint64_t) -1ULL) {
1020 use_relayd = 1;
1021 }
1022
1023 channel = consumer_find_channel(key);
1024 if (!channel) {
1025 ERR("UST snapshot channel not found for key %" PRIu64, key);
1026 ret = -1;
1027 goto error;
1028 }
1029 assert(!channel->monitor);
1030 DBG("UST consumer snapshot channel %" PRIu64, key);
1031
1032 cds_list_for_each_entry(stream, &channel->streams.head, send_node) {
1033
1034 health_code_update();
1035
1036 /* Lock stream because we are about to change its state. */
1037 pthread_mutex_lock(&stream->lock);
1038 stream->net_seq_idx = relayd_id;
1039
1040 if (use_relayd) {
1041 ret = consumer_send_relayd_stream(stream, path);
1042 if (ret < 0) {
1043 goto error_unlock;
1044 }
1045 } else {
1046 ret = utils_create_stream_file(path, stream->name,
1047 stream->chan->tracefile_size,
1048 stream->tracefile_count_current,
1049 stream->uid, stream->gid, NULL);
1050 if (ret < 0) {
1051 goto error_unlock;
1052 }
1053 stream->out_fd = ret;
1054 stream->tracefile_size_current = 0;
1055
1056 DBG("UST consumer snapshot stream %s/%s (%" PRIu64 ")", path,
1057 stream->name, stream->key);
1058 }
1059 if (relayd_id != -1ULL) {
1060 ret = consumer_send_relayd_streams_sent(relayd_id);
1061 if (ret < 0) {
1062 goto error_unlock;
1063 }
1064 }
1065
1066 ustctl_flush_buffer(stream->ustream, 1);
1067
1068 ret = lttng_ustconsumer_take_snapshot(stream);
1069 if (ret < 0) {
1070 ERR("Taking UST snapshot");
1071 goto error_unlock;
1072 }
1073
1074 ret = lttng_ustconsumer_get_produced_snapshot(stream, &produced_pos);
1075 if (ret < 0) {
1076 ERR("Produced UST snapshot position");
1077 goto error_unlock;
1078 }
1079
1080 ret = lttng_ustconsumer_get_consumed_snapshot(stream, &consumed_pos);
1081 if (ret < 0) {
1082 ERR("Consumerd UST snapshot position");
1083 goto error_unlock;
1084 }
1085
1086 /*
1087 * The original value is sent back if max stream size is larger than
1088 * the possible size of the snapshot. Also, we assume that the session
1089 * daemon should never send a maximum stream size that is lower than
1090 * subbuffer size.
1091 */
1092 consumed_pos = consumer_get_consume_start_pos(consumed_pos,
1093 produced_pos, nb_packets_per_stream,
1094 stream->max_sb_size);
1095
1096 while (consumed_pos < produced_pos) {
1097 ssize_t read_len;
1098 unsigned long len, padded_len;
1099
1100 health_code_update();
1101
1102 DBG("UST consumer taking snapshot at pos %lu", consumed_pos);
1103
1104 ret = ustctl_get_subbuf(stream->ustream, &consumed_pos);
1105 if (ret < 0) {
1106 if (ret != -EAGAIN) {
1107 PERROR("ustctl_get_subbuf snapshot");
1108 goto error_close_stream;
1109 }
1110 DBG("UST consumer get subbuf failed. Skipping it.");
1111 consumed_pos += stream->max_sb_size;
1112 continue;
1113 }
1114
1115 ret = ustctl_get_subbuf_size(stream->ustream, &len);
1116 if (ret < 0) {
1117 ERR("Snapshot ustctl_get_subbuf_size");
1118 goto error_put_subbuf;
1119 }
1120
1121 ret = ustctl_get_padded_subbuf_size(stream->ustream, &padded_len);
1122 if (ret < 0) {
1123 ERR("Snapshot ustctl_get_padded_subbuf_size");
1124 goto error_put_subbuf;
1125 }
1126
1127 read_len = lttng_consumer_on_read_subbuffer_mmap(ctx, stream, len,
1128 padded_len - len, NULL);
1129 if (use_relayd) {
1130 if (read_len != len) {
1131 ret = -EPERM;
1132 goto error_put_subbuf;
1133 }
1134 } else {
1135 if (read_len != padded_len) {
1136 ret = -EPERM;
1137 goto error_put_subbuf;
1138 }
1139 }
1140
1141 ret = ustctl_put_subbuf(stream->ustream);
1142 if (ret < 0) {
1143 ERR("Snapshot ustctl_put_subbuf");
1144 goto error_close_stream;
1145 }
1146 consumed_pos += stream->max_sb_size;
1147 }
1148
1149 /* Simply close the stream so we can use it on the next snapshot. */
1150 consumer_stream_close(stream);
1151 pthread_mutex_unlock(&stream->lock);
1152 }
1153
1154 rcu_read_unlock();
1155 return 0;
1156
1157 error_put_subbuf:
1158 if (ustctl_put_subbuf(stream->ustream) < 0) {
1159 ERR("Snapshot ustctl_put_subbuf");
1160 }
1161 error_close_stream:
1162 consumer_stream_close(stream);
1163 error_unlock:
1164 pthread_mutex_unlock(&stream->lock);
1165 error:
1166 rcu_read_unlock();
1167 return ret;
1168 }
1169
1170 /*
1171 * Receive the metadata updates from the sessiond.
1172 */
1173 int lttng_ustconsumer_recv_metadata(int sock, uint64_t key, uint64_t offset,
1174 uint64_t len, struct lttng_consumer_channel *channel,
1175 int timer, int wait)
1176 {
1177 int ret, ret_code = LTTCOMM_CONSUMERD_SUCCESS;
1178 char *metadata_str;
1179
1180 DBG("UST consumer push metadata key %" PRIu64 " of len %" PRIu64, key, len);
1181
1182 metadata_str = zmalloc(len * sizeof(char));
1183 if (!metadata_str) {
1184 PERROR("zmalloc metadata string");
1185 ret_code = LTTCOMM_CONSUMERD_ENOMEM;
1186 goto end;
1187 }
1188
1189 health_code_update();
1190
1191 /* Receive metadata string. */
1192 ret = lttcomm_recv_unix_sock(sock, metadata_str, len);
1193 if (ret < 0) {
1194 /* Session daemon is dead so return gracefully. */
1195 ret_code = ret;
1196 goto end_free;
1197 }
1198
1199 health_code_update();
1200
1201 pthread_mutex_lock(&channel->metadata_cache->lock);
1202 ret = consumer_metadata_cache_write(channel, offset, len, metadata_str);
1203 if (ret < 0) {
1204 /* Unable to handle metadata. Notify session daemon. */
1205 ret_code = LTTCOMM_CONSUMERD_ERROR_METADATA;
1206 /*
1207 * Skip metadata flush on write error since the offset and len might
1208 * not have been updated which could create an infinite loop below when
1209 * waiting for the metadata cache to be flushed.
1210 */
1211 pthread_mutex_unlock(&channel->metadata_cache->lock);
1212 goto end_free;
1213 }
1214 pthread_mutex_unlock(&channel->metadata_cache->lock);
1215
1216 if (!wait) {
1217 goto end_free;
1218 }
1219 while (consumer_metadata_cache_flushed(channel, offset + len, timer)) {
1220 DBG("Waiting for metadata to be flushed");
1221
1222 health_code_update();
1223
1224 usleep(DEFAULT_METADATA_AVAILABILITY_WAIT_TIME);
1225 }
1226
1227 end_free:
1228 free(metadata_str);
1229 end:
1230 return ret_code;
1231 }
1232
1233 /*
1234 * Receive command from session daemon and process it.
1235 *
1236 * Return 1 on success else a negative value or 0.
1237 */
1238 int lttng_ustconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
1239 int sock, struct pollfd *consumer_sockpoll)
1240 {
1241 ssize_t ret;
1242 enum lttcomm_return_code ret_code = LTTCOMM_CONSUMERD_SUCCESS;
1243 struct lttcomm_consumer_msg msg;
1244 struct lttng_consumer_channel *channel = NULL;
1245
1246 health_code_update();
1247
1248 ret = lttcomm_recv_unix_sock(sock, &msg, sizeof(msg));
1249 if (ret != sizeof(msg)) {
1250 DBG("Consumer received unexpected message size %zd (expects %zu)",
1251 ret, sizeof(msg));
1252 /*
1253 * The ret value might 0 meaning an orderly shutdown but this is ok
1254 * since the caller handles this.
1255 */
1256 if (ret > 0) {
1257 lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_ERROR_RECV_CMD);
1258 ret = -1;
1259 }
1260 return ret;
1261 }
1262
1263 health_code_update();
1264
1265 /* deprecated */
1266 assert(msg.cmd_type != LTTNG_CONSUMER_STOP);
1267
1268 health_code_update();
1269
1270 /* relayd needs RCU read-side lock */
1271 rcu_read_lock();
1272
1273 switch (msg.cmd_type) {
1274 case LTTNG_CONSUMER_ADD_RELAYD_SOCKET:
1275 {
1276 /* Session daemon status message are handled in the following call. */
1277 ret = consumer_add_relayd_socket(msg.u.relayd_sock.net_index,
1278 msg.u.relayd_sock.type, ctx, sock, consumer_sockpoll,
1279 &msg.u.relayd_sock.sock, msg.u.relayd_sock.session_id,
1280 msg.u.relayd_sock.relayd_session_id);
1281 goto end_nosignal;
1282 }
1283 case LTTNG_CONSUMER_DESTROY_RELAYD:
1284 {
1285 uint64_t index = msg.u.destroy_relayd.net_seq_idx;
1286 struct consumer_relayd_sock_pair *relayd;
1287
1288 DBG("UST consumer destroying relayd %" PRIu64, index);
1289
1290 /* Get relayd reference if exists. */
1291 relayd = consumer_find_relayd(index);
1292 if (relayd == NULL) {
1293 DBG("Unable to find relayd %" PRIu64, index);
1294 ret_code = LTTCOMM_CONSUMERD_RELAYD_FAIL;
1295 }
1296
1297 /*
1298 * Each relayd socket pair has a refcount of stream attached to it
1299 * which tells if the relayd is still active or not depending on the
1300 * refcount value.
1301 *
1302 * This will set the destroy flag of the relayd object and destroy it
1303 * if the refcount reaches zero when called.
1304 *
1305 * The destroy can happen either here or when a stream fd hangs up.
1306 */
1307 if (relayd) {
1308 consumer_flag_relayd_for_destroy(relayd);
1309 }
1310
1311 goto end_msg_sessiond;
1312 }
1313 case LTTNG_CONSUMER_UPDATE_STREAM:
1314 {
1315 rcu_read_unlock();
1316 return -ENOSYS;
1317 }
1318 case LTTNG_CONSUMER_DATA_PENDING:
1319 {
1320 int ret, is_data_pending;
1321 uint64_t id = msg.u.data_pending.session_id;
1322
1323 DBG("UST consumer data pending command for id %" PRIu64, id);
1324
1325 is_data_pending = consumer_data_pending(id);
1326
1327 /* Send back returned value to session daemon */
1328 ret = lttcomm_send_unix_sock(sock, &is_data_pending,
1329 sizeof(is_data_pending));
1330 if (ret < 0) {
1331 DBG("Error when sending the data pending ret code: %d", ret);
1332 goto error_fatal;
1333 }
1334
1335 /*
1336 * No need to send back a status message since the data pending
1337 * returned value is the response.
1338 */
1339 break;
1340 }
1341 case LTTNG_CONSUMER_ASK_CHANNEL_CREATION:
1342 {
1343 int ret;
1344 struct ustctl_consumer_channel_attr attr;
1345
1346 /* Create a plain object and reserve a channel key. */
1347 channel = allocate_channel(msg.u.ask_channel.session_id,
1348 msg.u.ask_channel.pathname, msg.u.ask_channel.name,
1349 msg.u.ask_channel.uid, msg.u.ask_channel.gid,
1350 msg.u.ask_channel.relayd_id, msg.u.ask_channel.key,
1351 (enum lttng_event_output) msg.u.ask_channel.output,
1352 msg.u.ask_channel.tracefile_size,
1353 msg.u.ask_channel.tracefile_count,
1354 msg.u.ask_channel.session_id_per_pid,
1355 msg.u.ask_channel.monitor,
1356 msg.u.ask_channel.live_timer_interval,
1357 msg.u.ask_channel.root_shm_path,
1358 msg.u.ask_channel.shm_path);
1359 if (!channel) {
1360 goto end_channel_error;
1361 }
1362
1363 /*
1364 * Assign UST application UID to the channel. This value is ignored for
1365 * per PID buffers. This is specific to UST thus setting this after the
1366 * allocation.
1367 */
1368 channel->ust_app_uid = msg.u.ask_channel.ust_app_uid;
1369
1370 /* Build channel attributes from received message. */
1371 attr.subbuf_size = msg.u.ask_channel.subbuf_size;
1372 attr.num_subbuf = msg.u.ask_channel.num_subbuf;
1373 attr.overwrite = msg.u.ask_channel.overwrite;
1374 attr.switch_timer_interval = msg.u.ask_channel.switch_timer_interval;
1375 attr.read_timer_interval = msg.u.ask_channel.read_timer_interval;
1376 attr.chan_id = msg.u.ask_channel.chan_id;
1377 memcpy(attr.uuid, msg.u.ask_channel.uuid, sizeof(attr.uuid));
1378
1379 /* Match channel buffer type to the UST abi. */
1380 switch (msg.u.ask_channel.output) {
1381 case LTTNG_EVENT_MMAP:
1382 default:
1383 attr.output = LTTNG_UST_MMAP;
1384 break;
1385 }
1386
1387 /* Translate and save channel type. */
1388 switch (msg.u.ask_channel.type) {
1389 case LTTNG_UST_CHAN_PER_CPU:
1390 channel->type = CONSUMER_CHANNEL_TYPE_DATA;
1391 attr.type = LTTNG_UST_CHAN_PER_CPU;
1392 /*
1393 * Set refcount to 1 for owner. Below, we will
1394 * pass ownership to the
1395 * consumer_thread_channel_poll() thread.
1396 */
1397 channel->refcount = 1;
1398 break;
1399 case LTTNG_UST_CHAN_METADATA:
1400 channel->type = CONSUMER_CHANNEL_TYPE_METADATA;
1401 attr.type = LTTNG_UST_CHAN_METADATA;
1402 break;
1403 default:
1404 assert(0);
1405 goto error_fatal;
1406 };
1407
1408 health_code_update();
1409
1410 ret = ask_channel(ctx, sock, channel, &attr);
1411 if (ret < 0) {
1412 goto end_channel_error;
1413 }
1414
1415 if (msg.u.ask_channel.type == LTTNG_UST_CHAN_METADATA) {
1416 ret = consumer_metadata_cache_allocate(channel);
1417 if (ret < 0) {
1418 ERR("Allocating metadata cache");
1419 goto end_channel_error;
1420 }
1421 consumer_timer_switch_start(channel, attr.switch_timer_interval);
1422 attr.switch_timer_interval = 0;
1423 } else {
1424 consumer_timer_live_start(channel,
1425 msg.u.ask_channel.live_timer_interval);
1426 }
1427
1428 health_code_update();
1429
1430 /*
1431 * Add the channel to the internal state AFTER all streams were created
1432 * and successfully sent to session daemon. This way, all streams must
1433 * be ready before this channel is visible to the threads.
1434 * If add_channel succeeds, ownership of the channel is
1435 * passed to consumer_thread_channel_poll().
1436 */
1437 ret = add_channel(channel, ctx);
1438 if (ret < 0) {
1439 if (msg.u.ask_channel.type == LTTNG_UST_CHAN_METADATA) {
1440 if (channel->switch_timer_enabled == 1) {
1441 consumer_timer_switch_stop(channel);
1442 }
1443 consumer_metadata_cache_destroy(channel);
1444 }
1445 if (channel->live_timer_enabled == 1) {
1446 consumer_timer_live_stop(channel);
1447 }
1448 goto end_channel_error;
1449 }
1450
1451 health_code_update();
1452
1453 /*
1454 * Channel and streams are now created. Inform the session daemon that
1455 * everything went well and should wait to receive the channel and
1456 * streams with ustctl API.
1457 */
1458 ret = consumer_send_status_channel(sock, channel);
1459 if (ret < 0) {
1460 /*
1461 * There is probably a problem on the socket.
1462 */
1463 goto error_fatal;
1464 }
1465
1466 break;
1467 }
1468 case LTTNG_CONSUMER_GET_CHANNEL:
1469 {
1470 int ret, relayd_err = 0;
1471 uint64_t key = msg.u.get_channel.key;
1472 struct lttng_consumer_channel *channel;
1473
1474 channel = consumer_find_channel(key);
1475 if (!channel) {
1476 ERR("UST consumer get channel key %" PRIu64 " not found", key);
1477 ret_code = LTTCOMM_CONSUMERD_CHAN_NOT_FOUND;
1478 goto end_msg_sessiond;
1479 }
1480
1481 health_code_update();
1482
1483 /* Send everything to sessiond. */
1484 ret = send_sessiond_channel(sock, channel, ctx, &relayd_err);
1485 if (ret < 0) {
1486 if (relayd_err) {
1487 /*
1488 * We were unable to send to the relayd the stream so avoid
1489 * sending back a fatal error to the thread since this is OK
1490 * and the consumer can continue its work. The above call
1491 * has sent the error status message to the sessiond.
1492 */
1493 goto end_nosignal;
1494 }
1495 /*
1496 * The communicaton was broken hence there is a bad state between
1497 * the consumer and sessiond so stop everything.
1498 */
1499 goto error_fatal;
1500 }
1501
1502 health_code_update();
1503
1504 /*
1505 * In no monitor mode, the streams ownership is kept inside the channel
1506 * so don't send them to the data thread.
1507 */
1508 if (!channel->monitor) {
1509 goto end_msg_sessiond;
1510 }
1511
1512 ret = send_streams_to_thread(channel, ctx);
1513 if (ret < 0) {
1514 /*
1515 * If we are unable to send the stream to the thread, there is
1516 * a big problem so just stop everything.
1517 */
1518 goto error_fatal;
1519 }
1520 /* List MUST be empty after or else it could be reused. */
1521 assert(cds_list_empty(&channel->streams.head));
1522 goto end_msg_sessiond;
1523 }
1524 case LTTNG_CONSUMER_DESTROY_CHANNEL:
1525 {
1526 uint64_t key = msg.u.destroy_channel.key;
1527
1528 /*
1529 * Only called if streams have not been sent to stream
1530 * manager thread. However, channel has been sent to
1531 * channel manager thread.
1532 */
1533 notify_thread_del_channel(ctx, key);
1534 goto end_msg_sessiond;
1535 }
1536 case LTTNG_CONSUMER_CLOSE_METADATA:
1537 {
1538 int ret;
1539
1540 ret = close_metadata(msg.u.close_metadata.key);
1541 if (ret != 0) {
1542 ret_code = ret;
1543 }
1544
1545 goto end_msg_sessiond;
1546 }
1547 case LTTNG_CONSUMER_FLUSH_CHANNEL:
1548 {
1549 int ret;
1550
1551 ret = flush_channel(msg.u.flush_channel.key);
1552 if (ret != 0) {
1553 ret_code = ret;
1554 }
1555
1556 goto end_msg_sessiond;
1557 }
1558 case LTTNG_CONSUMER_PUSH_METADATA:
1559 {
1560 int ret;
1561 uint64_t len = msg.u.push_metadata.len;
1562 uint64_t key = msg.u.push_metadata.key;
1563 uint64_t offset = msg.u.push_metadata.target_offset;
1564 struct lttng_consumer_channel *channel;
1565
1566 DBG("UST consumer push metadata key %" PRIu64 " of len %" PRIu64, key,
1567 len);
1568
1569 channel = consumer_find_channel(key);
1570 if (!channel) {
1571 /*
1572 * This is possible if the metadata creation on the consumer side
1573 * is in flight vis-a-vis a concurrent push metadata from the
1574 * session daemon. Simply return that the channel failed and the
1575 * session daemon will handle that message correctly considering
1576 * that this race is acceptable thus the DBG() statement here.
1577 */
1578 DBG("UST consumer push metadata %" PRIu64 " not found", key);
1579 ret_code = LTTCOMM_CONSUMERD_CHANNEL_FAIL;
1580 goto end_msg_sessiond;
1581 }
1582
1583 health_code_update();
1584
1585 /* Tell session daemon we are ready to receive the metadata. */
1586 ret = consumer_send_status_msg(sock, LTTCOMM_CONSUMERD_SUCCESS);
1587 if (ret < 0) {
1588 /* Somehow, the session daemon is not responding anymore. */
1589 goto error_fatal;
1590 }
1591
1592 health_code_update();
1593
1594 /* Wait for more data. */
1595 health_poll_entry();
1596 ret = lttng_consumer_poll_socket(consumer_sockpoll);
1597 health_poll_exit();
1598 if (ret) {
1599 goto error_fatal;
1600 }
1601
1602 health_code_update();
1603
1604 ret = lttng_ustconsumer_recv_metadata(sock, key, offset,
1605 len, channel, 0, 1);
1606 if (ret < 0) {
1607 /* error receiving from sessiond */
1608 goto error_fatal;
1609 } else {
1610 ret_code = ret;
1611 goto end_msg_sessiond;
1612 }
1613 }
1614 case LTTNG_CONSUMER_SETUP_METADATA:
1615 {
1616 int ret;
1617
1618 ret = setup_metadata(ctx, msg.u.setup_metadata.key);
1619 if (ret) {
1620 ret_code = ret;
1621 }
1622 goto end_msg_sessiond;
1623 }
1624 case LTTNG_CONSUMER_SNAPSHOT_CHANNEL:
1625 {
1626 if (msg.u.snapshot_channel.metadata) {
1627 ret = snapshot_metadata(msg.u.snapshot_channel.key,
1628 msg.u.snapshot_channel.pathname,
1629 msg.u.snapshot_channel.relayd_id,
1630 ctx);
1631 if (ret < 0) {
1632 ERR("Snapshot metadata failed");
1633 ret_code = LTTCOMM_CONSUMERD_ERROR_METADATA;
1634 }
1635 } else {
1636 ret = snapshot_channel(msg.u.snapshot_channel.key,
1637 msg.u.snapshot_channel.pathname,
1638 msg.u.snapshot_channel.relayd_id,
1639 msg.u.snapshot_channel.nb_packets_per_stream,
1640 ctx);
1641 if (ret < 0) {
1642 ERR("Snapshot channel failed");
1643 ret_code = LTTCOMM_CONSUMERD_CHANNEL_FAIL;
1644 }
1645 }
1646
1647 health_code_update();
1648 ret = consumer_send_status_msg(sock, ret_code);
1649 if (ret < 0) {
1650 /* Somehow, the session daemon is not responding anymore. */
1651 goto end_nosignal;
1652 }
1653 health_code_update();
1654 break;
1655 }
1656 default:
1657 break;
1658 }
1659
1660 end_nosignal:
1661 rcu_read_unlock();
1662
1663 health_code_update();
1664
1665 /*
1666 * Return 1 to indicate success since the 0 value can be a socket
1667 * shutdown during the recv() or send() call.
1668 */
1669 return 1;
1670
1671 end_msg_sessiond:
1672 /*
1673 * The returned value here is not useful since either way we'll return 1 to
1674 * the caller because the session daemon socket management is done
1675 * elsewhere. Returning a negative code or 0 will shutdown the consumer.
1676 */
1677 ret = consumer_send_status_msg(sock, ret_code);
1678 if (ret < 0) {
1679 goto error_fatal;
1680 }
1681 rcu_read_unlock();
1682
1683 health_code_update();
1684
1685 return 1;
1686 end_channel_error:
1687 if (channel) {
1688 /*
1689 * Free channel here since no one has a reference to it. We don't
1690 * free after that because a stream can store this pointer.
1691 */
1692 destroy_channel(channel);
1693 }
1694 /* We have to send a status channel message indicating an error. */
1695 ret = consumer_send_status_channel(sock, NULL);
1696 if (ret < 0) {
1697 /* Stop everything if session daemon can not be notified. */
1698 goto error_fatal;
1699 }
1700 rcu_read_unlock();
1701
1702 health_code_update();
1703
1704 return 1;
1705 error_fatal:
1706 rcu_read_unlock();
1707 /* This will issue a consumer stop. */
1708 return -1;
1709 }
1710
1711 /*
1712 * Wrapper over the mmap() read offset from ust-ctl library. Since this can be
1713 * compiled out, we isolate it in this library.
1714 */
1715 int lttng_ustctl_get_mmap_read_offset(struct lttng_consumer_stream *stream,
1716 unsigned long *off)
1717 {
1718 assert(stream);
1719 assert(stream->ustream);
1720
1721 return ustctl_get_mmap_read_offset(stream->ustream, off);
1722 }
1723
1724 /*
1725 * Wrapper over the mmap() read offset from ust-ctl library. Since this can be
1726 * compiled out, we isolate it in this library.
1727 */
1728 void *lttng_ustctl_get_mmap_base(struct lttng_consumer_stream *stream)
1729 {
1730 assert(stream);
1731 assert(stream->ustream);
1732
1733 return ustctl_get_mmap_base(stream->ustream);
1734 }
1735
1736 /*
1737 * Take a snapshot for a specific fd
1738 *
1739 * Returns 0 on success, < 0 on error
1740 */
1741 int lttng_ustconsumer_take_snapshot(struct lttng_consumer_stream *stream)
1742 {
1743 assert(stream);
1744 assert(stream->ustream);
1745
1746 return ustctl_snapshot(stream->ustream);
1747 }
1748
1749 /*
1750 * Get the produced position
1751 *
1752 * Returns 0 on success, < 0 on error
1753 */
1754 int lttng_ustconsumer_get_produced_snapshot(
1755 struct lttng_consumer_stream *stream, unsigned long *pos)
1756 {
1757 assert(stream);
1758 assert(stream->ustream);
1759 assert(pos);
1760
1761 return ustctl_snapshot_get_produced(stream->ustream, pos);
1762 }
1763
1764 /*
1765 * Get the consumed position
1766 *
1767 * Returns 0 on success, < 0 on error
1768 */
1769 int lttng_ustconsumer_get_consumed_snapshot(
1770 struct lttng_consumer_stream *stream, unsigned long *pos)
1771 {
1772 assert(stream);
1773 assert(stream->ustream);
1774 assert(pos);
1775
1776 return ustctl_snapshot_get_consumed(stream->ustream, pos);
1777 }
1778
1779 void lttng_ustconsumer_flush_buffer(struct lttng_consumer_stream *stream,
1780 int producer)
1781 {
1782 assert(stream);
1783 assert(stream->ustream);
1784
1785 ustctl_flush_buffer(stream->ustream, producer);
1786 }
1787
1788 int lttng_ustconsumer_get_current_timestamp(
1789 struct lttng_consumer_stream *stream, uint64_t *ts)
1790 {
1791 assert(stream);
1792 assert(stream->ustream);
1793 assert(ts);
1794
1795 return ustctl_get_current_timestamp(stream->ustream, ts);
1796 }
1797
1798 /*
1799 * Called when the stream signal the consumer that it has hang up.
1800 */
1801 void lttng_ustconsumer_on_stream_hangup(struct lttng_consumer_stream *stream)
1802 {
1803 assert(stream);
1804 assert(stream->ustream);
1805
1806 ustctl_flush_buffer(stream->ustream, 0);
1807 stream->hangup_flush_done = 1;
1808 }
1809
1810 void lttng_ustconsumer_del_channel(struct lttng_consumer_channel *chan)
1811 {
1812 int i;
1813
1814 assert(chan);
1815 assert(chan->uchan);
1816
1817 if (chan->switch_timer_enabled == 1) {
1818 consumer_timer_switch_stop(chan);
1819 }
1820 consumer_metadata_cache_destroy(chan);
1821 ustctl_destroy_channel(chan->uchan);
1822 for (i = 0; i < chan->nr_stream_fds; i++) {
1823 int ret;
1824
1825 ret = close(chan->stream_fds[i]);
1826 if (ret) {
1827 PERROR("close");
1828 }
1829 if (chan->shm_path[0]) {
1830 char shm_path[PATH_MAX];
1831
1832 ret = get_stream_shm_path(shm_path, chan->shm_path, i);
1833 if (ret) {
1834 ERR("Cannot get stream shm path");
1835 }
1836 ret = run_as_unlink(shm_path, chan->uid, chan->gid);
1837 if (ret) {
1838 errno = -ret;
1839 PERROR("unlink %s", shm_path);
1840 }
1841 }
1842 }
1843 free(chan->stream_fds);
1844 /* Try to rmdir all directories under shm_path root. */
1845 if (chan->root_shm_path[0]) {
1846 (void) run_as_recursive_rmdir(chan->root_shm_path,
1847 chan->uid, chan->gid);
1848 }
1849 }
1850
1851 void lttng_ustconsumer_del_stream(struct lttng_consumer_stream *stream)
1852 {
1853 assert(stream);
1854 assert(stream->ustream);
1855
1856 if (stream->chan->switch_timer_enabled == 1) {
1857 consumer_timer_switch_stop(stream->chan);
1858 }
1859 ustctl_destroy_stream(stream->ustream);
1860 }
1861
1862 int lttng_ustconsumer_get_wakeup_fd(struct lttng_consumer_stream *stream)
1863 {
1864 assert(stream);
1865 assert(stream->ustream);
1866
1867 return ustctl_stream_get_wakeup_fd(stream->ustream);
1868 }
1869
1870 int lttng_ustconsumer_close_wakeup_fd(struct lttng_consumer_stream *stream)
1871 {
1872 assert(stream);
1873 assert(stream->ustream);
1874
1875 return ustctl_stream_close_wakeup_fd(stream->ustream);
1876 }
1877
1878 /*
1879 * Populate index values of a UST stream. Values are set in big endian order.
1880 *
1881 * Return 0 on success or else a negative value.
1882 */
1883 static int get_index_values(struct ctf_packet_index *index,
1884 struct ustctl_consumer_stream *ustream)
1885 {
1886 int ret;
1887
1888 ret = ustctl_get_timestamp_begin(ustream, &index->timestamp_begin);
1889 if (ret < 0) {
1890 PERROR("ustctl_get_timestamp_begin");
1891 goto error;
1892 }
1893 index->timestamp_begin = htobe64(index->timestamp_begin);
1894
1895 ret = ustctl_get_timestamp_end(ustream, &index->timestamp_end);
1896 if (ret < 0) {
1897 PERROR("ustctl_get_timestamp_end");
1898 goto error;
1899 }
1900 index->timestamp_end = htobe64(index->timestamp_end);
1901
1902 ret = ustctl_get_events_discarded(ustream, &index->events_discarded);
1903 if (ret < 0) {
1904 PERROR("ustctl_get_events_discarded");
1905 goto error;
1906 }
1907 index->events_discarded = htobe64(index->events_discarded);
1908
1909 ret = ustctl_get_content_size(ustream, &index->content_size);
1910 if (ret < 0) {
1911 PERROR("ustctl_get_content_size");
1912 goto error;
1913 }
1914 index->content_size = htobe64(index->content_size);
1915
1916 ret = ustctl_get_packet_size(ustream, &index->packet_size);
1917 if (ret < 0) {
1918 PERROR("ustctl_get_packet_size");
1919 goto error;
1920 }
1921 index->packet_size = htobe64(index->packet_size);
1922
1923 ret = ustctl_get_stream_id(ustream, &index->stream_id);
1924 if (ret < 0) {
1925 PERROR("ustctl_get_stream_id");
1926 goto error;
1927 }
1928 index->stream_id = htobe64(index->stream_id);
1929
1930 error:
1931 return ret;
1932 }
1933
1934 /*
1935 * Write up to one packet from the metadata cache to the channel.
1936 *
1937 * Returns the number of bytes pushed in the cache, or a negative value
1938 * on error.
1939 */
1940 static
1941 int commit_one_metadata_packet(struct lttng_consumer_stream *stream)
1942 {
1943 ssize_t write_len;
1944 int ret;
1945
1946 pthread_mutex_lock(&stream->chan->metadata_cache->lock);
1947 if (stream->chan->metadata_cache->contiguous
1948 == stream->ust_metadata_pushed) {
1949 ret = 0;
1950 goto end;
1951 }
1952
1953 write_len = ustctl_write_one_packet_to_channel(stream->chan->uchan,
1954 &stream->chan->metadata_cache->data[stream->ust_metadata_pushed],
1955 stream->chan->metadata_cache->contiguous
1956 - stream->ust_metadata_pushed);
1957 assert(write_len != 0);
1958 if (write_len < 0) {
1959 ERR("Writing one metadata packet");
1960 ret = -1;
1961 goto end;
1962 }
1963 stream->ust_metadata_pushed += write_len;
1964
1965 assert(stream->chan->metadata_cache->contiguous >=
1966 stream->ust_metadata_pushed);
1967 ret = write_len;
1968
1969 end:
1970 pthread_mutex_unlock(&stream->chan->metadata_cache->lock);
1971 return ret;
1972 }
1973
1974
1975 /*
1976 * Sync metadata meaning request them to the session daemon and snapshot to the
1977 * metadata thread can consumer them.
1978 *
1979 * Metadata stream lock MUST be acquired.
1980 *
1981 * Return 0 if new metadatda is available, EAGAIN if the metadata stream
1982 * is empty or a negative value on error.
1983 */
1984 int lttng_ustconsumer_sync_metadata(struct lttng_consumer_local_data *ctx,
1985 struct lttng_consumer_stream *metadata)
1986 {
1987 int ret;
1988 int retry = 0;
1989
1990 assert(ctx);
1991 assert(metadata);
1992
1993 /*
1994 * Request metadata from the sessiond, but don't wait for the flush
1995 * because we locked the metadata thread.
1996 */
1997 ret = lttng_ustconsumer_request_metadata(ctx, metadata->chan, 0, 0);
1998 if (ret < 0) {
1999 goto end;
2000 }
2001
2002 ret = commit_one_metadata_packet(metadata);
2003 if (ret <= 0) {
2004 goto end;
2005 } else if (ret > 0) {
2006 retry = 1;
2007 }
2008
2009 ustctl_flush_buffer(metadata->ustream, 1);
2010 ret = ustctl_snapshot(metadata->ustream);
2011 if (ret < 0) {
2012 if (errno != EAGAIN) {
2013 ERR("Sync metadata, taking UST snapshot");
2014 goto end;
2015 }
2016 DBG("No new metadata when syncing them.");
2017 /* No new metadata, exit. */
2018 ret = ENODATA;
2019 goto end;
2020 }
2021
2022 /*
2023 * After this flush, we still need to extract metadata.
2024 */
2025 if (retry) {
2026 ret = EAGAIN;
2027 }
2028
2029 end:
2030 return ret;
2031 }
2032
2033 /*
2034 * Return 0 on success else a negative value.
2035 */
2036 static int notify_if_more_data(struct lttng_consumer_stream *stream,
2037 struct lttng_consumer_local_data *ctx)
2038 {
2039 int ret;
2040 struct ustctl_consumer_stream *ustream;
2041
2042 assert(stream);
2043 assert(ctx);
2044
2045 ustream = stream->ustream;
2046
2047 /*
2048 * First, we are going to check if there is a new subbuffer available
2049 * before reading the stream wait_fd.
2050 */
2051 /* Get the next subbuffer */
2052 ret = ustctl_get_next_subbuf(ustream);
2053 if (ret) {
2054 /* No more data found, flag the stream. */
2055 stream->has_data = 0;
2056 ret = 0;
2057 goto end;
2058 }
2059
2060 ret = ustctl_put_subbuf(ustream);
2061 assert(!ret);
2062
2063 /* This stream still has data. Flag it and wake up the data thread. */
2064 stream->has_data = 1;
2065
2066 if (stream->monitor && !stream->hangup_flush_done && !ctx->has_wakeup) {
2067 ssize_t writelen;
2068
2069 writelen = lttng_pipe_write(ctx->consumer_wakeup_pipe, "!", 1);
2070 if (writelen < 0 && errno != EAGAIN && errno != EWOULDBLOCK) {
2071 ret = writelen;
2072 goto end;
2073 }
2074
2075 /* The wake up pipe has been notified. */
2076 ctx->has_wakeup = 1;
2077 }
2078 ret = 0;
2079
2080 end:
2081 return ret;
2082 }
2083
2084 /*
2085 * Read subbuffer from the given stream.
2086 *
2087 * Stream lock MUST be acquired.
2088 *
2089 * Return 0 on success else a negative value.
2090 */
2091 int lttng_ustconsumer_read_subbuffer(struct lttng_consumer_stream *stream,
2092 struct lttng_consumer_local_data *ctx)
2093 {
2094 unsigned long len, subbuf_size, padding;
2095 int err, write_index = 1;
2096 long ret = 0;
2097 struct ustctl_consumer_stream *ustream;
2098 struct ctf_packet_index index;
2099
2100 assert(stream);
2101 assert(stream->ustream);
2102 assert(ctx);
2103
2104 DBG("In UST read_subbuffer (wait_fd: %d, name: %s)", stream->wait_fd,
2105 stream->name);
2106
2107 /* Ease our life for what's next. */
2108 ustream = stream->ustream;
2109
2110 /*
2111 * We can consume the 1 byte written into the wait_fd by UST. Don't trigger
2112 * error if we cannot read this one byte (read returns 0), or if the error
2113 * is EAGAIN or EWOULDBLOCK.
2114 *
2115 * This is only done when the stream is monitored by a thread, before the
2116 * flush is done after a hangup and if the stream is not flagged with data
2117 * since there might be nothing to consume in the wait fd but still have
2118 * data available flagged by the consumer wake up pipe.
2119 */
2120 if (stream->monitor && !stream->hangup_flush_done && !stream->has_data) {
2121 char dummy;
2122 ssize_t readlen;
2123
2124 readlen = lttng_read(stream->wait_fd, &dummy, 1);
2125 if (readlen < 0 && errno != EAGAIN && errno != EWOULDBLOCK) {
2126 ret = readlen;
2127 goto end;
2128 }
2129 }
2130
2131 retry:
2132 /* Get the next subbuffer */
2133 err = ustctl_get_next_subbuf(ustream);
2134 if (err != 0) {
2135 /*
2136 * Populate metadata info if the existing info has
2137 * already been read.
2138 */
2139 if (stream->metadata_flag) {
2140 ret = commit_one_metadata_packet(stream);
2141 if (ret <= 0) {
2142 goto end;
2143 }
2144 ustctl_flush_buffer(stream->ustream, 1);
2145 goto retry;
2146 }
2147
2148 ret = err; /* ustctl_get_next_subbuf returns negative, caller expect positive. */
2149 /*
2150 * This is a debug message even for single-threaded consumer,
2151 * because poll() have more relaxed criterions than get subbuf,
2152 * so get_subbuf may fail for short race windows where poll()
2153 * would issue wakeups.
2154 */
2155 DBG("Reserving sub buffer failed (everything is normal, "
2156 "it is due to concurrency) [ret: %d]", err);
2157 goto end;
2158 }
2159 assert(stream->chan->output == CONSUMER_CHANNEL_MMAP);
2160
2161 if (!stream->metadata_flag) {
2162 index.offset = htobe64(stream->out_fd_offset);
2163 ret = get_index_values(&index, ustream);
2164 if (ret < 0) {
2165 goto end;
2166 }
2167 } else {
2168 write_index = 0;
2169 }
2170
2171 /* Get the full padded subbuffer size */
2172 err = ustctl_get_padded_subbuf_size(ustream, &len);
2173 assert(err == 0);
2174
2175 /* Get subbuffer data size (without padding) */
2176 err = ustctl_get_subbuf_size(ustream, &subbuf_size);
2177 assert(err == 0);
2178
2179 /* Make sure we don't get a subbuffer size bigger than the padded */
2180 assert(len >= subbuf_size);
2181
2182 padding = len - subbuf_size;
2183 /* write the subbuffer to the tracefile */
2184 ret = lttng_consumer_on_read_subbuffer_mmap(ctx, stream, subbuf_size, padding, &index);
2185 /*
2186 * The mmap operation should write subbuf_size amount of data when network
2187 * streaming or the full padding (len) size when we are _not_ streaming.
2188 */
2189 if ((ret != subbuf_size && stream->net_seq_idx != (uint64_t) -1ULL) ||
2190 (ret != len && stream->net_seq_idx == (uint64_t) -1ULL)) {
2191 /*
2192 * Display the error but continue processing to try to release the
2193 * subbuffer. This is a DBG statement since any unexpected kill or
2194 * signal, the application gets unregistered, relayd gets closed or
2195 * anything that affects the buffer lifetime will trigger this error.
2196 * So, for the sake of the user, don't print this error since it can
2197 * happen and it is OK with the code flow.
2198 */
2199 DBG("Error writing to tracefile "
2200 "(ret: %ld != len: %lu != subbuf_size: %lu)",
2201 ret, len, subbuf_size);
2202 write_index = 0;
2203 }
2204 err = ustctl_put_next_subbuf(ustream);
2205 assert(err == 0);
2206
2207 /*
2208 * This will consumer the byte on the wait_fd if and only if there is not
2209 * next subbuffer to be acquired.
2210 */
2211 if (!stream->metadata_flag) {
2212 ret = notify_if_more_data(stream, ctx);
2213 if (ret < 0) {
2214 goto end;
2215 }
2216 }
2217
2218 /* Write index if needed. */
2219 if (!write_index) {
2220 goto end;
2221 }
2222
2223 if (stream->chan->live_timer_interval && !stream->metadata_flag) {
2224 /*
2225 * In live, block until all the metadata is sent.
2226 */
2227 err = consumer_stream_sync_metadata(ctx, stream->session_id);
2228 if (err < 0) {
2229 goto end;
2230 }
2231 }
2232
2233 assert(!stream->metadata_flag);
2234 err = consumer_stream_write_index(stream, &index);
2235 if (err < 0) {
2236 goto end;
2237 }
2238
2239 end:
2240 return ret;
2241 }
2242
2243 /*
2244 * Called when a stream is created.
2245 *
2246 * Return 0 on success or else a negative value.
2247 */
2248 int lttng_ustconsumer_on_recv_stream(struct lttng_consumer_stream *stream)
2249 {
2250 int ret;
2251
2252 assert(stream);
2253
2254 /* Don't create anything if this is set for streaming. */
2255 if (stream->net_seq_idx == (uint64_t) -1ULL && stream->chan->monitor) {
2256 ret = utils_create_stream_file(stream->chan->pathname, stream->name,
2257 stream->chan->tracefile_size, stream->tracefile_count_current,
2258 stream->uid, stream->gid, NULL);
2259 if (ret < 0) {
2260 goto error;
2261 }
2262 stream->out_fd = ret;
2263 stream->tracefile_size_current = 0;
2264
2265 if (!stream->metadata_flag) {
2266 ret = index_create_file(stream->chan->pathname,
2267 stream->name, stream->uid, stream->gid,
2268 stream->chan->tracefile_size,
2269 stream->tracefile_count_current);
2270 if (ret < 0) {
2271 goto error;
2272 }
2273 stream->index_fd = ret;
2274 }
2275 }
2276 ret = 0;
2277
2278 error:
2279 return ret;
2280 }
2281
2282 /*
2283 * Check if data is still being extracted from the buffers for a specific
2284 * stream. Consumer data lock MUST be acquired before calling this function
2285 * and the stream lock.
2286 *
2287 * Return 1 if the traced data are still getting read else 0 meaning that the
2288 * data is available for trace viewer reading.
2289 */
2290 int lttng_ustconsumer_data_pending(struct lttng_consumer_stream *stream)
2291 {
2292 int ret;
2293
2294 assert(stream);
2295 assert(stream->ustream);
2296
2297 DBG("UST consumer checking data pending");
2298
2299 if (stream->endpoint_status != CONSUMER_ENDPOINT_ACTIVE) {
2300 ret = 0;
2301 goto end;
2302 }
2303
2304 if (stream->chan->type == CONSUMER_CHANNEL_TYPE_METADATA) {
2305 uint64_t contiguous, pushed;
2306
2307 /* Ease our life a bit. */
2308 contiguous = stream->chan->metadata_cache->contiguous;
2309 pushed = stream->ust_metadata_pushed;
2310
2311 /*
2312 * We can simply check whether all contiguously available data
2313 * has been pushed to the ring buffer, since the push operation
2314 * is performed within get_next_subbuf(), and because both
2315 * get_next_subbuf() and put_next_subbuf() are issued atomically
2316 * thanks to the stream lock within
2317 * lttng_ustconsumer_read_subbuffer(). This basically means that
2318 * whetnever ust_metadata_pushed is incremented, the associated
2319 * metadata has been consumed from the metadata stream.
2320 */
2321 DBG("UST consumer metadata pending check: contiguous %" PRIu64 " vs pushed %" PRIu64,
2322 contiguous, pushed);
2323 assert(((int64_t) (contiguous - pushed)) >= 0);
2324 if ((contiguous != pushed) ||
2325 (((int64_t) contiguous - pushed) > 0 || contiguous == 0)) {
2326 ret = 1; /* Data is pending */
2327 goto end;
2328 }
2329 } else {
2330 ret = ustctl_get_next_subbuf(stream->ustream);
2331 if (ret == 0) {
2332 /*
2333 * There is still data so let's put back this
2334 * subbuffer.
2335 */
2336 ret = ustctl_put_subbuf(stream->ustream);
2337 assert(ret == 0);
2338 ret = 1; /* Data is pending */
2339 goto end;
2340 }
2341 }
2342
2343 /* Data is NOT pending so ready to be read. */
2344 ret = 0;
2345
2346 end:
2347 return ret;
2348 }
2349
2350 /*
2351 * Stop a given metadata channel timer if enabled and close the wait fd which
2352 * is the poll pipe of the metadata stream.
2353 *
2354 * This MUST be called with the metadata channel acquired.
2355 */
2356 void lttng_ustconsumer_close_metadata(struct lttng_consumer_channel *metadata)
2357 {
2358 int ret;
2359
2360 assert(metadata);
2361 assert(metadata->type == CONSUMER_CHANNEL_TYPE_METADATA);
2362
2363 DBG("Closing metadata channel key %" PRIu64, metadata->key);
2364
2365 if (metadata->switch_timer_enabled == 1) {
2366 consumer_timer_switch_stop(metadata);
2367 }
2368
2369 if (!metadata->metadata_stream) {
2370 goto end;
2371 }
2372
2373 /*
2374 * Closing write side so the thread monitoring the stream wakes up if any
2375 * and clean the metadata stream.
2376 */
2377 if (metadata->metadata_stream->ust_metadata_poll_pipe[1] >= 0) {
2378 ret = close(metadata->metadata_stream->ust_metadata_poll_pipe[1]);
2379 if (ret < 0) {
2380 PERROR("closing metadata pipe write side");
2381 }
2382 metadata->metadata_stream->ust_metadata_poll_pipe[1] = -1;
2383 }
2384
2385 end:
2386 return;
2387 }
2388
2389 /*
2390 * Close every metadata stream wait fd of the metadata hash table. This
2391 * function MUST be used very carefully so not to run into a race between the
2392 * metadata thread handling streams and this function closing their wait fd.
2393 *
2394 * For UST, this is used when the session daemon hangs up. Its the metadata
2395 * producer so calling this is safe because we are assured that no state change
2396 * can occur in the metadata thread for the streams in the hash table.
2397 */
2398 void lttng_ustconsumer_close_all_metadata(struct lttng_ht *metadata_ht)
2399 {
2400 struct lttng_ht_iter iter;
2401 struct lttng_consumer_stream *stream;
2402
2403 assert(metadata_ht);
2404 assert(metadata_ht->ht);
2405
2406 DBG("UST consumer closing all metadata streams");
2407
2408 rcu_read_lock();
2409 cds_lfht_for_each_entry(metadata_ht->ht, &iter.iter, stream,
2410 node.node) {
2411
2412 health_code_update();
2413
2414 pthread_mutex_lock(&stream->chan->lock);
2415 lttng_ustconsumer_close_metadata(stream->chan);
2416 pthread_mutex_unlock(&stream->chan->lock);
2417
2418 }
2419 rcu_read_unlock();
2420 }
2421
2422 void lttng_ustconsumer_close_stream_wakeup(struct lttng_consumer_stream *stream)
2423 {
2424 int ret;
2425
2426 ret = ustctl_stream_close_wakeup_fd(stream->ustream);
2427 if (ret < 0) {
2428 ERR("Unable to close wakeup fd");
2429 }
2430 }
2431
2432 /*
2433 * Please refer to consumer-timer.c before adding any lock within this
2434 * function or any of its callees. Timers have a very strict locking
2435 * semantic with respect to teardown. Failure to respect this semantic
2436 * introduces deadlocks.
2437 */
2438 int lttng_ustconsumer_request_metadata(struct lttng_consumer_local_data *ctx,
2439 struct lttng_consumer_channel *channel, int timer, int wait)
2440 {
2441 struct lttcomm_metadata_request_msg request;
2442 struct lttcomm_consumer_msg msg;
2443 enum lttcomm_return_code ret_code = LTTCOMM_CONSUMERD_SUCCESS;
2444 uint64_t len, key, offset;
2445 int ret;
2446
2447 assert(channel);
2448 assert(channel->metadata_cache);
2449
2450 memset(&request, 0, sizeof(request));
2451
2452 /* send the metadata request to sessiond */
2453 switch (consumer_data.type) {
2454 case LTTNG_CONSUMER64_UST:
2455 request.bits_per_long = 64;
2456 break;
2457 case LTTNG_CONSUMER32_UST:
2458 request.bits_per_long = 32;
2459 break;
2460 default:
2461 request.bits_per_long = 0;
2462 break;
2463 }
2464
2465 request.session_id = channel->session_id;
2466 request.session_id_per_pid = channel->session_id_per_pid;
2467 /*
2468 * Request the application UID here so the metadata of that application can
2469 * be sent back. The channel UID corresponds to the user UID of the session
2470 * used for the rights on the stream file(s).
2471 */
2472 request.uid = channel->ust_app_uid;
2473 request.key = channel->key;
2474
2475 DBG("Sending metadata request to sessiond, session id %" PRIu64
2476 ", per-pid %" PRIu64 ", app UID %u and channek key %" PRIu64,
2477 request.session_id, request.session_id_per_pid, request.uid,
2478 request.key);
2479
2480 pthread_mutex_lock(&ctx->metadata_socket_lock);
2481
2482 health_code_update();
2483
2484 ret = lttcomm_send_unix_sock(ctx->consumer_metadata_socket, &request,
2485 sizeof(request));
2486 if (ret < 0) {
2487 ERR("Asking metadata to sessiond");
2488 goto end;
2489 }
2490
2491 health_code_update();
2492
2493 /* Receive the metadata from sessiond */
2494 ret = lttcomm_recv_unix_sock(ctx->consumer_metadata_socket, &msg,
2495 sizeof(msg));
2496 if (ret != sizeof(msg)) {
2497 DBG("Consumer received unexpected message size %d (expects %zu)",
2498 ret, sizeof(msg));
2499 lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_ERROR_RECV_CMD);
2500 /*
2501 * The ret value might 0 meaning an orderly shutdown but this is ok
2502 * since the caller handles this.
2503 */
2504 goto end;
2505 }
2506
2507 health_code_update();
2508
2509 if (msg.cmd_type == LTTNG_ERR_UND) {
2510 /* No registry found */
2511 (void) consumer_send_status_msg(ctx->consumer_metadata_socket,
2512 ret_code);
2513 ret = 0;
2514 goto end;
2515 } else if (msg.cmd_type != LTTNG_CONSUMER_PUSH_METADATA) {
2516 ERR("Unexpected cmd_type received %d", msg.cmd_type);
2517 ret = -1;
2518 goto end;
2519 }
2520
2521 len = msg.u.push_metadata.len;
2522 key = msg.u.push_metadata.key;
2523 offset = msg.u.push_metadata.target_offset;
2524
2525 assert(key == channel->key);
2526 if (len == 0) {
2527 DBG("No new metadata to receive for key %" PRIu64, key);
2528 }
2529
2530 health_code_update();
2531
2532 /* Tell session daemon we are ready to receive the metadata. */
2533 ret = consumer_send_status_msg(ctx->consumer_metadata_socket,
2534 LTTCOMM_CONSUMERD_SUCCESS);
2535 if (ret < 0 || len == 0) {
2536 /*
2537 * Somehow, the session daemon is not responding anymore or there is
2538 * nothing to receive.
2539 */
2540 goto end;
2541 }
2542
2543 health_code_update();
2544
2545 ret = lttng_ustconsumer_recv_metadata(ctx->consumer_metadata_socket,
2546 key, offset, len, channel, timer, wait);
2547 if (ret >= 0) {
2548 /*
2549 * Only send the status msg if the sessiond is alive meaning a positive
2550 * ret code.
2551 */
2552 (void) consumer_send_status_msg(ctx->consumer_metadata_socket, ret);
2553 }
2554 ret = 0;
2555
2556 end:
2557 health_code_update();
2558
2559 pthread_mutex_unlock(&ctx->metadata_socket_lock);
2560 return ret;
2561 }
2562
2563 /*
2564 * Return the ustctl call for the get stream id.
2565 */
2566 int lttng_ustconsumer_get_stream_id(struct lttng_consumer_stream *stream,
2567 uint64_t *stream_id)
2568 {
2569 assert(stream);
2570 assert(stream_id);
2571
2572 return ustctl_get_stream_id(stream->ustream, stream_id);
2573 }
This page took 0.146869 seconds and 4 git commands to generate.