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