192217b4e0eea6557e02e25d4df648bc12c81987
[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 #include <assert.h>
21 #include <lttng/ust-ctl.h>
22 #include <poll.h>
23 #include <pthread.h>
24 #include <stdlib.h>
25 #include <string.h>
26 #include <sys/mman.h>
27 #include <sys/socket.h>
28 #include <sys/stat.h>
29 #include <sys/types.h>
30 #include <inttypes.h>
31 #include <unistd.h>
32 #include <urcu/list.h>
33 #include <signal.h>
34
35 #include <common/common.h>
36 #include <common/sessiond-comm/sessiond-comm.h>
37 #include <common/relayd/relayd.h>
38 #include <common/compat/fcntl.h>
39 #include <common/consumer-metadata-cache.h>
40 #include <common/consumer-stream.h>
41 #include <common/consumer-timer.h>
42 #include <common/utils.h>
43 #include <common/index/index.h>
44
45 #include "ust-consumer.h"
46
47 extern struct lttng_consumer_global_data consumer_data;
48 extern int consumer_poll_timeout;
49 extern volatile int consumer_quit;
50
51 /*
52 * Free channel object and all streams associated with it. This MUST be used
53 * only and only if the channel has _NEVER_ been added to the global channel
54 * hash table.
55 */
56 static void destroy_channel(struct lttng_consumer_channel *channel)
57 {
58 struct lttng_consumer_stream *stream, *stmp;
59
60 assert(channel);
61
62 DBG("UST consumer cleaning stream list");
63
64 cds_list_for_each_entry_safe(stream, stmp, &channel->streams.head,
65 send_node) {
66 cds_list_del(&stream->send_node);
67 ustctl_destroy_stream(stream->ustream);
68 free(stream);
69 }
70
71 /*
72 * If a channel is available meaning that was created before the streams
73 * were, delete it.
74 */
75 if (channel->uchan) {
76 lttng_ustconsumer_del_channel(channel);
77 }
78 free(channel);
79 }
80
81 /*
82 * Add channel to internal consumer state.
83 *
84 * Returns 0 on success or else a negative value.
85 */
86 static int add_channel(struct lttng_consumer_channel *channel,
87 struct lttng_consumer_local_data *ctx)
88 {
89 int ret = 0;
90
91 assert(channel);
92 assert(ctx);
93
94 if (ctx->on_recv_channel != NULL) {
95 ret = ctx->on_recv_channel(channel);
96 if (ret == 0) {
97 ret = consumer_add_channel(channel, ctx);
98 } else if (ret < 0) {
99 /* Most likely an ENOMEM. */
100 lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_OUTFD_ERROR);
101 goto error;
102 }
103 } else {
104 ret = consumer_add_channel(channel, ctx);
105 }
106
107 DBG("UST consumer channel added (key: %" PRIu64 ")", channel->key);
108
109 error:
110 return ret;
111 }
112
113 /*
114 * Allocate and return a consumer channel object.
115 */
116 static struct lttng_consumer_channel *allocate_channel(uint64_t session_id,
117 const char *pathname, const char *name, uid_t uid, gid_t gid,
118 uint64_t relayd_id, uint64_t key, enum lttng_event_output output,
119 uint64_t tracefile_size, uint64_t tracefile_count,
120 uint64_t session_id_per_pid, unsigned int monitor,
121 unsigned int live_timer_interval)
122 {
123 assert(pathname);
124 assert(name);
125
126 return consumer_allocate_channel(key, session_id, pathname, name, uid,
127 gid, relayd_id, output, tracefile_size,
128 tracefile_count, session_id_per_pid, monitor, live_timer_interval);
129 }
130
131 /*
132 * Allocate and return a consumer stream object. If _alloc_ret is not NULL, the
133 * error value if applicable is set in it else it is kept untouched.
134 *
135 * Return NULL on error else the newly allocated stream object.
136 */
137 static struct lttng_consumer_stream *allocate_stream(int cpu, int key,
138 struct lttng_consumer_channel *channel,
139 struct lttng_consumer_local_data *ctx, int *_alloc_ret)
140 {
141 int alloc_ret;
142 struct lttng_consumer_stream *stream = NULL;
143
144 assert(channel);
145 assert(ctx);
146
147 stream = consumer_allocate_stream(channel->key,
148 key,
149 LTTNG_CONSUMER_ACTIVE_STREAM,
150 channel->name,
151 channel->uid,
152 channel->gid,
153 channel->relayd_id,
154 channel->session_id,
155 cpu,
156 &alloc_ret,
157 channel->type,
158 channel->monitor);
159 if (stream == NULL) {
160 switch (alloc_ret) {
161 case -ENOENT:
162 /*
163 * We could not find the channel. Can happen if cpu hotplug
164 * happens while tearing down.
165 */
166 DBG3("Could not find channel");
167 break;
168 case -ENOMEM:
169 case -EINVAL:
170 default:
171 lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_OUTFD_ERROR);
172 break;
173 }
174 goto error;
175 }
176
177 stream->chan = channel;
178
179 error:
180 if (_alloc_ret) {
181 *_alloc_ret = alloc_ret;
182 }
183 return stream;
184 }
185
186 /*
187 * Send the given stream pointer to the corresponding thread.
188 *
189 * Returns 0 on success else a negative value.
190 */
191 static int send_stream_to_thread(struct lttng_consumer_stream *stream,
192 struct lttng_consumer_local_data *ctx)
193 {
194 int ret;
195 struct lttng_pipe *stream_pipe;
196
197 /* Get the right pipe where the stream will be sent. */
198 if (stream->metadata_flag) {
199 ret = consumer_add_metadata_stream(stream);
200 if (ret) {
201 ERR("Consumer add metadata stream %" PRIu64 " failed.",
202 stream->key);
203 goto error;
204 }
205 stream_pipe = ctx->consumer_metadata_pipe;
206 } else {
207 ret = consumer_add_data_stream(stream);
208 if (ret) {
209 ERR("Consumer add stream %" PRIu64 " failed.",
210 stream->key);
211 goto error;
212 }
213 stream_pipe = ctx->consumer_data_pipe;
214 }
215
216 /*
217 * From this point on, the stream's ownership has been moved away from
218 * the channel and becomes globally visible.
219 */
220 stream->globally_visible = 1;
221
222 ret = lttng_pipe_write(stream_pipe, &stream, sizeof(stream));
223 if (ret < 0) {
224 ERR("Consumer write %s stream to pipe %d",
225 stream->metadata_flag ? "metadata" : "data",
226 lttng_pipe_get_writefd(stream_pipe));
227 if (stream->metadata_flag) {
228 consumer_del_stream_for_metadata(stream);
229 } else {
230 consumer_del_stream_for_data(stream);
231 }
232 }
233 error:
234 return ret;
235 }
236
237 /*
238 * Create streams for the given channel using liblttng-ust-ctl.
239 *
240 * Return 0 on success else a negative value.
241 */
242 static int create_ust_streams(struct lttng_consumer_channel *channel,
243 struct lttng_consumer_local_data *ctx)
244 {
245 int ret, cpu = 0;
246 struct ustctl_consumer_stream *ustream;
247 struct lttng_consumer_stream *stream;
248
249 assert(channel);
250 assert(ctx);
251
252 /*
253 * While a stream is available from ustctl. When NULL is returned, we've
254 * reached the end of the possible stream for the channel.
255 */
256 while ((ustream = ustctl_create_stream(channel->uchan, cpu))) {
257 int wait_fd;
258 int ust_metadata_pipe[2];
259
260 if (channel->type == CONSUMER_CHANNEL_TYPE_METADATA && channel->monitor) {
261 ret = utils_create_pipe_cloexec_nonblock(ust_metadata_pipe);
262 if (ret < 0) {
263 ERR("Create ust metadata poll pipe");
264 goto error;
265 }
266 wait_fd = ust_metadata_pipe[0];
267 } else {
268 wait_fd = ustctl_stream_get_wait_fd(ustream);
269 }
270
271 /* Allocate consumer stream object. */
272 stream = allocate_stream(cpu, wait_fd, channel, ctx, &ret);
273 if (!stream) {
274 goto error_alloc;
275 }
276 stream->ustream = ustream;
277 /*
278 * Store it so we can save multiple function calls afterwards since
279 * this value is used heavily in the stream threads. This is UST
280 * specific so this is why it's done after allocation.
281 */
282 stream->wait_fd = wait_fd;
283
284 /*
285 * Increment channel refcount since the channel reference has now been
286 * assigned in the allocation process above.
287 */
288 if (stream->chan->monitor) {
289 uatomic_inc(&stream->chan->refcount);
290 }
291
292 /*
293 * Order is important this is why a list is used. On error, the caller
294 * should clean this list.
295 */
296 cds_list_add_tail(&stream->send_node, &channel->streams.head);
297
298 ret = ustctl_get_max_subbuf_size(stream->ustream,
299 &stream->max_sb_size);
300 if (ret < 0) {
301 ERR("ustctl_get_max_subbuf_size failed for stream %s",
302 stream->name);
303 goto error;
304 }
305
306 /* Do actions once stream has been received. */
307 if (ctx->on_recv_stream) {
308 ret = ctx->on_recv_stream(stream);
309 if (ret < 0) {
310 goto error;
311 }
312 }
313
314 DBG("UST consumer add stream %s (key: %" PRIu64 ") with relayd id %" PRIu64,
315 stream->name, stream->key, stream->relayd_stream_id);
316
317 /* Set next CPU stream. */
318 channel->streams.count = ++cpu;
319
320 /* Keep stream reference when creating metadata. */
321 if (channel->type == CONSUMER_CHANNEL_TYPE_METADATA) {
322 channel->metadata_stream = stream;
323 stream->ust_metadata_poll_pipe[0] = ust_metadata_pipe[0];
324 stream->ust_metadata_poll_pipe[1] = ust_metadata_pipe[1];
325 }
326 }
327
328 return 0;
329
330 error:
331 error_alloc:
332 return ret;
333 }
334
335 /*
336 * Create an UST channel with the given attributes and send it to the session
337 * daemon using the ust ctl API.
338 *
339 * Return 0 on success or else a negative value.
340 */
341 static int create_ust_channel(struct ustctl_consumer_channel_attr *attr,
342 struct ustctl_consumer_channel **chanp)
343 {
344 int ret;
345 struct ustctl_consumer_channel *channel;
346
347 assert(attr);
348 assert(chanp);
349
350 DBG3("Creating channel to ustctl with attr: [overwrite: %d, "
351 "subbuf_size: %" PRIu64 ", num_subbuf: %" PRIu64 ", "
352 "switch_timer_interval: %u, read_timer_interval: %u, "
353 "output: %d, type: %d", attr->overwrite, attr->subbuf_size,
354 attr->num_subbuf, attr->switch_timer_interval,
355 attr->read_timer_interval, attr->output, attr->type);
356
357 channel = ustctl_create_channel(attr);
358 if (!channel) {
359 ret = -1;
360 goto error_create;
361 }
362
363 *chanp = channel;
364
365 return 0;
366
367 error_create:
368 return ret;
369 }
370
371 /*
372 * Send a single given stream to the session daemon using the sock.
373 *
374 * Return 0 on success else a negative value.
375 */
376 static int send_sessiond_stream(int sock, struct lttng_consumer_stream *stream)
377 {
378 int ret;
379
380 assert(stream);
381 assert(sock >= 0);
382
383 DBG("UST consumer sending stream %" PRIu64 " to sessiond", stream->key);
384
385 /* Send stream to session daemon. */
386 ret = ustctl_send_stream_to_sessiond(sock, stream->ustream);
387 if (ret < 0) {
388 goto error;
389 }
390
391 error:
392 return ret;
393 }
394
395 /*
396 * Send channel to sessiond.
397 *
398 * Return 0 on success or else a negative value.
399 */
400 static int send_sessiond_channel(int sock,
401 struct lttng_consumer_channel *channel,
402 struct lttng_consumer_local_data *ctx, int *relayd_error)
403 {
404 int ret, ret_code = LTTNG_OK;
405 struct lttng_consumer_stream *stream;
406
407 assert(channel);
408 assert(ctx);
409 assert(sock >= 0);
410
411 DBG("UST consumer sending channel %s to sessiond", channel->name);
412
413 if (channel->relayd_id != (uint64_t) -1ULL) {
414 cds_list_for_each_entry(stream, &channel->streams.head, send_node) {
415 /* Try to send the stream to the relayd if one is available. */
416 ret = consumer_send_relayd_stream(stream, stream->chan->pathname);
417 if (ret < 0) {
418 /*
419 * Flag that the relayd was the problem here probably due to a
420 * communicaton error on the socket.
421 */
422 if (relayd_error) {
423 *relayd_error = 1;
424 }
425 ret_code = LTTNG_ERR_RELAYD_CONNECT_FAIL;
426 }
427 }
428 }
429
430 /* Inform sessiond that we are about to send channel and streams. */
431 ret = consumer_send_status_msg(sock, ret_code);
432 if (ret < 0 || ret_code != LTTNG_OK) {
433 /*
434 * Either the session daemon is not responding or the relayd died so we
435 * stop now.
436 */
437 goto error;
438 }
439
440 /* Send channel to sessiond. */
441 ret = ustctl_send_channel_to_sessiond(sock, channel->uchan);
442 if (ret < 0) {
443 goto error;
444 }
445
446 ret = ustctl_channel_close_wakeup_fd(channel->uchan);
447 if (ret < 0) {
448 goto error;
449 }
450
451 /* The channel was sent successfully to the sessiond at this point. */
452 cds_list_for_each_entry(stream, &channel->streams.head, send_node) {
453 /* Send stream to session daemon. */
454 ret = send_sessiond_stream(sock, stream);
455 if (ret < 0) {
456 goto error;
457 }
458 }
459
460 /* Tell sessiond there is no more stream. */
461 ret = ustctl_send_stream_to_sessiond(sock, NULL);
462 if (ret < 0) {
463 goto error;
464 }
465
466 DBG("UST consumer NULL stream sent to sessiond");
467
468 return 0;
469
470 error:
471 if (ret_code != LTTNG_OK) {
472 ret = -1;
473 }
474 return ret;
475 }
476
477 /*
478 * Creates a channel and streams and add the channel it to the channel internal
479 * state. The created stream must ONLY be sent once the GET_CHANNEL command is
480 * received.
481 *
482 * Return 0 on success or else, a negative value is returned and the channel
483 * MUST be destroyed by consumer_del_channel().
484 */
485 static int ask_channel(struct lttng_consumer_local_data *ctx, int sock,
486 struct lttng_consumer_channel *channel,
487 struct ustctl_consumer_channel_attr *attr)
488 {
489 int ret;
490
491 assert(ctx);
492 assert(channel);
493 assert(attr);
494
495 /*
496 * This value is still used by the kernel consumer since for the kernel,
497 * the stream ownership is not IN the consumer so we need to have the
498 * number of left stream that needs to be initialized so we can know when
499 * to delete the channel (see consumer.c).
500 *
501 * As for the user space tracer now, the consumer creates and sends the
502 * stream to the session daemon which only sends them to the application
503 * once every stream of a channel is received making this value useless
504 * because we they will be added to the poll thread before the application
505 * receives them. This ensures that a stream can not hang up during
506 * initilization of a channel.
507 */
508 channel->nb_init_stream_left = 0;
509
510 /* The reply msg status is handled in the following call. */
511 ret = create_ust_channel(attr, &channel->uchan);
512 if (ret < 0) {
513 goto end;
514 }
515
516 channel->wait_fd = ustctl_channel_get_wait_fd(channel->uchan);
517
518 /*
519 * For the snapshots (no monitor), we create the metadata streams
520 * on demand, not during the channel creation.
521 */
522 if (channel->type == CONSUMER_CHANNEL_TYPE_METADATA && !channel->monitor) {
523 ret = 0;
524 goto end;
525 }
526
527 /* Open all streams for this channel. */
528 ret = create_ust_streams(channel, ctx);
529 if (ret < 0) {
530 goto end;
531 }
532
533 end:
534 return ret;
535 }
536
537 /*
538 * Send all stream of a channel to the right thread handling it.
539 *
540 * On error, return a negative value else 0 on success.
541 */
542 static int send_streams_to_thread(struct lttng_consumer_channel *channel,
543 struct lttng_consumer_local_data *ctx)
544 {
545 int ret = 0;
546 struct lttng_consumer_stream *stream, *stmp;
547
548 assert(channel);
549 assert(ctx);
550
551 /* Send streams to the corresponding thread. */
552 cds_list_for_each_entry_safe(stream, stmp, &channel->streams.head,
553 send_node) {
554 /* Sending the stream to the thread. */
555 ret = send_stream_to_thread(stream, ctx);
556 if (ret < 0) {
557 /*
558 * If we are unable to send the stream to the thread, there is
559 * a big problem so just stop everything.
560 */
561 /* Remove node from the channel stream list. */
562 cds_list_del(&stream->send_node);
563 goto error;
564 }
565
566 /* Remove node from the channel stream list. */
567 cds_list_del(&stream->send_node);
568
569 }
570
571 error:
572 return ret;
573 }
574
575 /*
576 * Flush channel's streams using the given key to retrieve the channel.
577 *
578 * Return 0 on success else an LTTng error code.
579 */
580 static int flush_channel(uint64_t chan_key)
581 {
582 int ret = 0;
583 struct lttng_consumer_channel *channel;
584 struct lttng_consumer_stream *stream;
585 struct lttng_ht *ht;
586 struct lttng_ht_iter iter;
587
588 DBG("UST consumer flush channel key %" PRIu64, chan_key);
589
590 rcu_read_lock();
591 channel = consumer_find_channel(chan_key);
592 if (!channel) {
593 ERR("UST consumer flush channel %" PRIu64 " not found", chan_key);
594 ret = LTTNG_ERR_UST_CHAN_NOT_FOUND;
595 goto error;
596 }
597
598 ht = consumer_data.stream_per_chan_id_ht;
599
600 /* For each stream of the channel id, flush it. */
601 cds_lfht_for_each_entry_duplicate(ht->ht,
602 ht->hash_fct(&channel->key, lttng_ht_seed), ht->match_fct,
603 &channel->key, &iter.iter, stream, node_channel_id.node) {
604 ustctl_flush_buffer(stream->ustream, 1);
605 }
606 error:
607 rcu_read_unlock();
608 return ret;
609 }
610 /*
611 * Close metadata stream wakeup_fd using the given key to retrieve the channel.
612 * RCU read side lock MUST be acquired before calling this function.
613 *
614 * NOTE: This function does NOT take any channel nor stream lock.
615 *
616 * Return 0 on success else LTTng error code.
617 */
618 static int _close_metadata(struct lttng_consumer_channel *channel)
619 {
620 int ret = LTTNG_OK;
621
622 assert(channel);
623 assert(channel->type == CONSUMER_CHANNEL_TYPE_METADATA);
624
625 if (channel->switch_timer_enabled == 1) {
626 DBG("Deleting timer on metadata channel");
627 consumer_timer_switch_stop(channel);
628 }
629
630 if (channel->metadata_stream) {
631 ret = ustctl_stream_close_wakeup_fd(channel->metadata_stream->ustream);
632 if (ret < 0) {
633 ERR("UST consumer unable to close fd of metadata (ret: %d)", ret);
634 ret = LTTCOMM_CONSUMERD_ERROR_METADATA;
635 }
636
637 if (channel->monitor) {
638 /* Close the read-side in consumer_del_metadata_stream */
639 ret = close(channel->metadata_stream->ust_metadata_poll_pipe[1]);
640 if (ret < 0) {
641 PERROR("Close UST metadata write-side poll pipe");
642 ret = LTTCOMM_CONSUMERD_ERROR_METADATA;
643 }
644 }
645 }
646
647 return ret;
648 }
649
650 /*
651 * Close metadata stream wakeup_fd using the given key to retrieve the channel.
652 * RCU read side lock MUST be acquired before calling this function.
653 *
654 * Return 0 on success else an LTTng error code.
655 */
656 static int close_metadata(uint64_t chan_key)
657 {
658 int ret = 0;
659 struct lttng_consumer_channel *channel;
660
661 DBG("UST consumer close metadata key %" PRIu64, chan_key);
662
663 channel = consumer_find_channel(chan_key);
664 if (!channel) {
665 /*
666 * This is possible if the metadata thread has issue a delete because
667 * the endpoint point of the stream hung up. There is no way the
668 * session daemon can know about it thus use a DBG instead of an actual
669 * error.
670 */
671 DBG("UST consumer close metadata %" PRIu64 " not found", chan_key);
672 ret = LTTNG_ERR_UST_CHAN_NOT_FOUND;
673 goto error;
674 }
675
676 pthread_mutex_lock(&consumer_data.lock);
677 pthread_mutex_lock(&channel->lock);
678
679 if (cds_lfht_is_node_deleted(&channel->node.node)) {
680 goto error_unlock;
681 }
682
683 ret = _close_metadata(channel);
684
685 error_unlock:
686 pthread_mutex_unlock(&channel->lock);
687 pthread_mutex_unlock(&consumer_data.lock);
688 error:
689 return ret;
690 }
691
692 /*
693 * RCU read side lock MUST be acquired before calling this function.
694 *
695 * Return 0 on success else an LTTng error code.
696 */
697 static int setup_metadata(struct lttng_consumer_local_data *ctx, uint64_t key)
698 {
699 int ret;
700 struct lttng_consumer_channel *metadata;
701
702 DBG("UST consumer setup metadata key %" PRIu64, key);
703
704 metadata = consumer_find_channel(key);
705 if (!metadata) {
706 ERR("UST consumer push metadata %" PRIu64 " not found", key);
707 ret = LTTNG_ERR_UST_CHAN_NOT_FOUND;
708 goto end;
709 }
710
711 /*
712 * In no monitor mode, the metadata channel has no stream(s) so skip the
713 * ownership transfer to the metadata thread.
714 */
715 if (!metadata->monitor) {
716 DBG("Metadata channel in no monitor");
717 ret = 0;
718 goto end;
719 }
720
721 /*
722 * Send metadata stream to relayd if one available. Availability is
723 * known if the stream is still in the list of the channel.
724 */
725 if (cds_list_empty(&metadata->streams.head)) {
726 ERR("Metadata channel key %" PRIu64 ", no stream available.", key);
727 ret = LTTCOMM_CONSUMERD_ERROR_METADATA;
728 goto error_no_stream;
729 }
730
731 /* Send metadata stream to relayd if needed. */
732 if (metadata->metadata_stream->net_seq_idx != (uint64_t) -1ULL) {
733 ret = consumer_send_relayd_stream(metadata->metadata_stream,
734 metadata->pathname);
735 if (ret < 0) {
736 ret = LTTCOMM_CONSUMERD_ERROR_METADATA;
737 goto error;
738 }
739 }
740
741 ret = send_streams_to_thread(metadata, ctx);
742 if (ret < 0) {
743 /*
744 * If we are unable to send the stream to the thread, there is
745 * a big problem so just stop everything.
746 */
747 ret = LTTCOMM_CONSUMERD_FATAL;
748 goto error;
749 }
750 /* List MUST be empty after or else it could be reused. */
751 assert(cds_list_empty(&metadata->streams.head));
752
753 ret = 0;
754 goto end;
755
756 error:
757 /*
758 * Delete metadata channel on error. At this point, the metadata stream can
759 * NOT be monitored by the metadata thread thus having the guarantee that
760 * the stream is still in the local stream list of the channel. This call
761 * will make sure to clean that list.
762 */
763 cds_list_del(&metadata->metadata_stream->send_node);
764 consumer_stream_destroy(metadata->metadata_stream, NULL);
765 error_no_stream:
766 end:
767 return ret;
768 }
769
770 /*
771 * Snapshot the whole metadata.
772 *
773 * Returns 0 on success, < 0 on error
774 */
775 static int snapshot_metadata(uint64_t key, char *path, uint64_t relayd_id,
776 struct lttng_consumer_local_data *ctx)
777 {
778 int ret = 0;
779 struct lttng_consumer_channel *metadata_channel;
780 struct lttng_consumer_stream *metadata_stream;
781
782 assert(path);
783 assert(ctx);
784
785 DBG("UST consumer snapshot metadata with key %" PRIu64 " at path %s",
786 key, path);
787
788 rcu_read_lock();
789
790 metadata_channel = consumer_find_channel(key);
791 if (!metadata_channel) {
792 ERR("UST snapshot metadata channel not found for key %" PRIu64,
793 key);
794 ret = -1;
795 goto error;
796 }
797 assert(!metadata_channel->monitor);
798
799 /*
800 * Ask the sessiond if we have new metadata waiting and update the
801 * consumer metadata cache.
802 */
803 ret = lttng_ustconsumer_request_metadata(ctx, metadata_channel, 0);
804 if (ret < 0) {
805 goto error;
806 }
807
808 /*
809 * The metadata stream is NOT created in no monitor mode when the channel
810 * is created on a sessiond ask channel command.
811 */
812 ret = create_ust_streams(metadata_channel, ctx);
813 if (ret < 0) {
814 goto error;
815 }
816
817 metadata_stream = metadata_channel->metadata_stream;
818 assert(metadata_stream);
819
820 if (relayd_id != (uint64_t) -1ULL) {
821 metadata_stream->net_seq_idx = relayd_id;
822 ret = consumer_send_relayd_stream(metadata_stream, path);
823 if (ret < 0) {
824 goto error_stream;
825 }
826 } else {
827 ret = utils_create_stream_file(path, metadata_stream->name,
828 metadata_stream->chan->tracefile_size,
829 metadata_stream->tracefile_count_current,
830 metadata_stream->uid, metadata_stream->gid, NULL);
831 if (ret < 0) {
832 goto error_stream;
833 }
834 metadata_stream->out_fd = ret;
835 metadata_stream->tracefile_size_current = 0;
836 }
837
838 pthread_mutex_lock(&metadata_channel->metadata_cache->lock);
839
840 do {
841 ret = lttng_consumer_read_subbuffer(metadata_stream, ctx);
842 if (ret < 0) {
843 goto error_unlock;
844 }
845 } while (ret > 0);
846
847 error_unlock:
848 pthread_mutex_unlock(&metadata_channel->metadata_cache->lock);
849
850 error_stream:
851 /*
852 * Clean up the stream completly because the next snapshot will use a new
853 * metadata stream.
854 */
855 cds_list_del(&metadata_stream->send_node);
856 consumer_stream_destroy(metadata_stream, NULL);
857 metadata_channel->metadata_stream = NULL;
858
859 error:
860 rcu_read_unlock();
861 return ret;
862 }
863
864 /*
865 * Take a snapshot of all the stream of a channel.
866 *
867 * Returns 0 on success, < 0 on error
868 */
869 static int snapshot_channel(uint64_t key, char *path, uint64_t relayd_id,
870 uint64_t max_stream_size, struct lttng_consumer_local_data *ctx)
871 {
872 int ret;
873 unsigned use_relayd = 0;
874 unsigned long consumed_pos, produced_pos;
875 struct lttng_consumer_channel *channel;
876 struct lttng_consumer_stream *stream;
877
878 assert(path);
879 assert(ctx);
880
881 rcu_read_lock();
882
883 if (relayd_id != (uint64_t) -1ULL) {
884 use_relayd = 1;
885 }
886
887 channel = consumer_find_channel(key);
888 if (!channel) {
889 ERR("UST snapshot channel not found for key %" PRIu64, key);
890 ret = -1;
891 goto error;
892 }
893 assert(!channel->monitor);
894 DBG("UST consumer snapshot channel %" PRIu64, key);
895
896 cds_list_for_each_entry(stream, &channel->streams.head, send_node) {
897 /* Lock stream because we are about to change its state. */
898 pthread_mutex_lock(&stream->lock);
899 stream->net_seq_idx = relayd_id;
900
901 if (use_relayd) {
902 ret = consumer_send_relayd_stream(stream, path);
903 if (ret < 0) {
904 goto error_unlock;
905 }
906 } else {
907 ret = utils_create_stream_file(path, stream->name,
908 stream->chan->tracefile_size,
909 stream->tracefile_count_current,
910 stream->uid, stream->gid, NULL);
911 if (ret < 0) {
912 goto error_unlock;
913 }
914 stream->out_fd = ret;
915 stream->tracefile_size_current = 0;
916
917 DBG("UST consumer snapshot stream %s/%s (%" PRIu64 ")", path,
918 stream->name, stream->key);
919 }
920
921 ustctl_flush_buffer(stream->ustream, 1);
922
923 ret = lttng_ustconsumer_take_snapshot(stream);
924 if (ret < 0) {
925 ERR("Taking UST snapshot");
926 goto error_unlock;
927 }
928
929 ret = lttng_ustconsumer_get_produced_snapshot(stream, &produced_pos);
930 if (ret < 0) {
931 ERR("Produced UST snapshot position");
932 goto error_unlock;
933 }
934
935 ret = lttng_ustconsumer_get_consumed_snapshot(stream, &consumed_pos);
936 if (ret < 0) {
937 ERR("Consumerd UST snapshot position");
938 goto error_unlock;
939 }
940
941 /*
942 * The original value is sent back if max stream size is larger than
943 * the possible size of the snapshot. Also, we asume that the session
944 * daemon should never send a maximum stream size that is lower than
945 * subbuffer size.
946 */
947 consumed_pos = consumer_get_consumed_maxsize(consumed_pos,
948 produced_pos, max_stream_size);
949
950 while (consumed_pos < produced_pos) {
951 ssize_t read_len;
952 unsigned long len, padded_len;
953
954 DBG("UST consumer taking snapshot at pos %lu", consumed_pos);
955
956 ret = ustctl_get_subbuf(stream->ustream, &consumed_pos);
957 if (ret < 0) {
958 if (ret != -EAGAIN) {
959 PERROR("ustctl_get_subbuf snapshot");
960 goto error_close_stream;
961 }
962 DBG("UST consumer get subbuf failed. Skipping it.");
963 consumed_pos += stream->max_sb_size;
964 continue;
965 }
966
967 ret = ustctl_get_subbuf_size(stream->ustream, &len);
968 if (ret < 0) {
969 ERR("Snapshot ustctl_get_subbuf_size");
970 goto error_put_subbuf;
971 }
972
973 ret = ustctl_get_padded_subbuf_size(stream->ustream, &padded_len);
974 if (ret < 0) {
975 ERR("Snapshot ustctl_get_padded_subbuf_size");
976 goto error_put_subbuf;
977 }
978
979 read_len = lttng_consumer_on_read_subbuffer_mmap(ctx, stream, len,
980 padded_len - len, NULL);
981 if (use_relayd) {
982 if (read_len != len) {
983 ret = -EPERM;
984 goto error_put_subbuf;
985 }
986 } else {
987 if (read_len != padded_len) {
988 ret = -EPERM;
989 goto error_put_subbuf;
990 }
991 }
992
993 ret = ustctl_put_subbuf(stream->ustream);
994 if (ret < 0) {
995 ERR("Snapshot ustctl_put_subbuf");
996 goto error_close_stream;
997 }
998 consumed_pos += stream->max_sb_size;
999 }
1000
1001 /* Simply close the stream so we can use it on the next snapshot. */
1002 consumer_stream_close(stream);
1003 pthread_mutex_unlock(&stream->lock);
1004 }
1005
1006 rcu_read_unlock();
1007 return 0;
1008
1009 error_put_subbuf:
1010 if (ustctl_put_subbuf(stream->ustream) < 0) {
1011 ERR("Snapshot ustctl_put_subbuf");
1012 }
1013 error_close_stream:
1014 consumer_stream_close(stream);
1015 error_unlock:
1016 pthread_mutex_unlock(&stream->lock);
1017 error:
1018 rcu_read_unlock();
1019 return ret;
1020 }
1021
1022 /*
1023 * Receive the metadata updates from the sessiond.
1024 */
1025 int lttng_ustconsumer_recv_metadata(int sock, uint64_t key, uint64_t offset,
1026 uint64_t len, struct lttng_consumer_channel *channel,
1027 int timer)
1028 {
1029 int ret, ret_code = LTTNG_OK;
1030 char *metadata_str;
1031
1032 DBG("UST consumer push metadata key %" PRIu64 " of len %" PRIu64, key, len);
1033
1034 metadata_str = zmalloc(len * sizeof(char));
1035 if (!metadata_str) {
1036 PERROR("zmalloc metadata string");
1037 ret_code = LTTCOMM_CONSUMERD_ENOMEM;
1038 goto end;
1039 }
1040
1041 /* Receive metadata string. */
1042 ret = lttcomm_recv_unix_sock(sock, metadata_str, len);
1043 if (ret < 0) {
1044 /* Session daemon is dead so return gracefully. */
1045 ret_code = ret;
1046 goto end_free;
1047 }
1048
1049 pthread_mutex_lock(&channel->metadata_cache->lock);
1050 ret = consumer_metadata_cache_write(channel, offset, len, metadata_str);
1051 if (ret < 0) {
1052 /* Unable to handle metadata. Notify session daemon. */
1053 ret_code = LTTCOMM_CONSUMERD_ERROR_METADATA;
1054 /*
1055 * Skip metadata flush on write error since the offset and len might
1056 * not have been updated which could create an infinite loop below when
1057 * waiting for the metadata cache to be flushed.
1058 */
1059 pthread_mutex_unlock(&channel->metadata_cache->lock);
1060 goto end_free;
1061 }
1062 pthread_mutex_unlock(&channel->metadata_cache->lock);
1063
1064 while (consumer_metadata_cache_flushed(channel, offset + len, timer)) {
1065 DBG("Waiting for metadata to be flushed");
1066 usleep(DEFAULT_METADATA_AVAILABILITY_WAIT_TIME);
1067 }
1068
1069 end_free:
1070 free(metadata_str);
1071 end:
1072 return ret_code;
1073 }
1074
1075 /*
1076 * Receive command from session daemon and process it.
1077 *
1078 * Return 1 on success else a negative value or 0.
1079 */
1080 int lttng_ustconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
1081 int sock, struct pollfd *consumer_sockpoll)
1082 {
1083 ssize_t ret;
1084 enum lttng_error_code ret_code = LTTNG_OK;
1085 struct lttcomm_consumer_msg msg;
1086 struct lttng_consumer_channel *channel = NULL;
1087
1088 ret = lttcomm_recv_unix_sock(sock, &msg, sizeof(msg));
1089 if (ret != sizeof(msg)) {
1090 DBG("Consumer received unexpected message size %zd (expects %zu)",
1091 ret, sizeof(msg));
1092 /*
1093 * The ret value might 0 meaning an orderly shutdown but this is ok
1094 * since the caller handles this.
1095 */
1096 if (ret > 0) {
1097 lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_ERROR_RECV_CMD);
1098 ret = -1;
1099 }
1100 return ret;
1101 }
1102 if (msg.cmd_type == LTTNG_CONSUMER_STOP) {
1103 /*
1104 * Notify the session daemon that the command is completed.
1105 *
1106 * On transport layer error, the function call will print an error
1107 * message so handling the returned code is a bit useless since we
1108 * return an error code anyway.
1109 */
1110 (void) consumer_send_status_msg(sock, ret_code);
1111 return -ENOENT;
1112 }
1113
1114 /* relayd needs RCU read-side lock */
1115 rcu_read_lock();
1116
1117 switch (msg.cmd_type) {
1118 case LTTNG_CONSUMER_ADD_RELAYD_SOCKET:
1119 {
1120 /* Session daemon status message are handled in the following call. */
1121 ret = consumer_add_relayd_socket(msg.u.relayd_sock.net_index,
1122 msg.u.relayd_sock.type, ctx, sock, consumer_sockpoll,
1123 &msg.u.relayd_sock.sock, msg.u.relayd_sock.session_id,
1124 msg.u.relayd_sock.relayd_session_id);
1125 goto end_nosignal;
1126 }
1127 case LTTNG_CONSUMER_DESTROY_RELAYD:
1128 {
1129 uint64_t index = msg.u.destroy_relayd.net_seq_idx;
1130 struct consumer_relayd_sock_pair *relayd;
1131
1132 DBG("UST consumer destroying relayd %" PRIu64, index);
1133
1134 /* Get relayd reference if exists. */
1135 relayd = consumer_find_relayd(index);
1136 if (relayd == NULL) {
1137 DBG("Unable to find relayd %" PRIu64, index);
1138 ret_code = LTTNG_ERR_NO_CONSUMER;
1139 }
1140
1141 /*
1142 * Each relayd socket pair has a refcount of stream attached to it
1143 * which tells if the relayd is still active or not depending on the
1144 * refcount value.
1145 *
1146 * This will set the destroy flag of the relayd object and destroy it
1147 * if the refcount reaches zero when called.
1148 *
1149 * The destroy can happen either here or when a stream fd hangs up.
1150 */
1151 if (relayd) {
1152 consumer_flag_relayd_for_destroy(relayd);
1153 }
1154
1155 goto end_msg_sessiond;
1156 }
1157 case LTTNG_CONSUMER_UPDATE_STREAM:
1158 {
1159 rcu_read_unlock();
1160 return -ENOSYS;
1161 }
1162 case LTTNG_CONSUMER_DATA_PENDING:
1163 {
1164 int ret, is_data_pending;
1165 uint64_t id = msg.u.data_pending.session_id;
1166
1167 DBG("UST consumer data pending command for id %" PRIu64, id);
1168
1169 is_data_pending = consumer_data_pending(id);
1170
1171 /* Send back returned value to session daemon */
1172 ret = lttcomm_send_unix_sock(sock, &is_data_pending,
1173 sizeof(is_data_pending));
1174 if (ret < 0) {
1175 DBG("Error when sending the data pending ret code: %d", ret);
1176 goto error_fatal;
1177 }
1178
1179 /*
1180 * No need to send back a status message since the data pending
1181 * returned value is the response.
1182 */
1183 break;
1184 }
1185 case LTTNG_CONSUMER_ASK_CHANNEL_CREATION:
1186 {
1187 int ret;
1188 struct ustctl_consumer_channel_attr attr;
1189
1190 /* Create a plain object and reserve a channel key. */
1191 channel = allocate_channel(msg.u.ask_channel.session_id,
1192 msg.u.ask_channel.pathname, msg.u.ask_channel.name,
1193 msg.u.ask_channel.uid, msg.u.ask_channel.gid,
1194 msg.u.ask_channel.relayd_id, msg.u.ask_channel.key,
1195 (enum lttng_event_output) msg.u.ask_channel.output,
1196 msg.u.ask_channel.tracefile_size,
1197 msg.u.ask_channel.tracefile_count,
1198 msg.u.ask_channel.session_id_per_pid,
1199 msg.u.ask_channel.monitor,
1200 msg.u.ask_channel.live_timer_interval);
1201 if (!channel) {
1202 goto end_channel_error;
1203 }
1204
1205 /*
1206 * Assign UST application UID to the channel. This value is ignored for
1207 * per PID buffers. This is specific to UST thus setting this after the
1208 * allocation.
1209 */
1210 channel->ust_app_uid = msg.u.ask_channel.ust_app_uid;
1211
1212 /* Build channel attributes from received message. */
1213 attr.subbuf_size = msg.u.ask_channel.subbuf_size;
1214 attr.num_subbuf = msg.u.ask_channel.num_subbuf;
1215 attr.overwrite = msg.u.ask_channel.overwrite;
1216 attr.switch_timer_interval = msg.u.ask_channel.switch_timer_interval;
1217 attr.read_timer_interval = msg.u.ask_channel.read_timer_interval;
1218 attr.chan_id = msg.u.ask_channel.chan_id;
1219 attr.output = msg.u.ask_channel.output;
1220 memcpy(attr.uuid, msg.u.ask_channel.uuid, sizeof(attr.uuid));
1221
1222 /* Translate and save channel type. */
1223 switch (msg.u.ask_channel.type) {
1224 case LTTNG_UST_CHAN_PER_CPU:
1225 channel->type = CONSUMER_CHANNEL_TYPE_DATA;
1226 attr.type = LTTNG_UST_CHAN_PER_CPU;
1227 /*
1228 * Set refcount to 1 for owner. Below, we will
1229 * pass ownership to the
1230 * consumer_thread_channel_poll() thread.
1231 */
1232 channel->refcount = 1;
1233 break;
1234 case LTTNG_UST_CHAN_METADATA:
1235 channel->type = CONSUMER_CHANNEL_TYPE_METADATA;
1236 attr.type = LTTNG_UST_CHAN_METADATA;
1237 break;
1238 default:
1239 assert(0);
1240 goto error_fatal;
1241 };
1242
1243 ret = ask_channel(ctx, sock, channel, &attr);
1244 if (ret < 0) {
1245 goto end_channel_error;
1246 }
1247
1248 if (msg.u.ask_channel.type == LTTNG_UST_CHAN_METADATA) {
1249 ret = consumer_metadata_cache_allocate(channel);
1250 if (ret < 0) {
1251 ERR("Allocating metadata cache");
1252 goto end_channel_error;
1253 }
1254 consumer_timer_switch_start(channel, attr.switch_timer_interval);
1255 attr.switch_timer_interval = 0;
1256 }
1257
1258 consumer_timer_live_start(channel, msg.u.ask_channel.live_timer_interval);
1259
1260 /*
1261 * Add the channel to the internal state AFTER all streams were created
1262 * and successfully sent to session daemon. This way, all streams must
1263 * be ready before this channel is visible to the threads.
1264 * If add_channel succeeds, ownership of the channel is
1265 * passed to consumer_thread_channel_poll().
1266 */
1267 ret = add_channel(channel, ctx);
1268 if (ret < 0) {
1269 if (msg.u.ask_channel.type == LTTNG_UST_CHAN_METADATA) {
1270 if (channel->switch_timer_enabled == 1) {
1271 consumer_timer_switch_stop(channel);
1272 }
1273 consumer_metadata_cache_destroy(channel);
1274 }
1275 if (channel->live_timer_enabled == 1) {
1276 consumer_timer_live_stop(channel);
1277 }
1278 goto end_channel_error;
1279 }
1280
1281 /*
1282 * Channel and streams are now created. Inform the session daemon that
1283 * everything went well and should wait to receive the channel and
1284 * streams with ustctl API.
1285 */
1286 ret = consumer_send_status_channel(sock, channel);
1287 if (ret < 0) {
1288 /*
1289 * There is probably a problem on the socket.
1290 */
1291 goto error_fatal;
1292 }
1293
1294 break;
1295 }
1296 case LTTNG_CONSUMER_GET_CHANNEL:
1297 {
1298 int ret, relayd_err = 0;
1299 uint64_t key = msg.u.get_channel.key;
1300 struct lttng_consumer_channel *channel;
1301
1302 channel = consumer_find_channel(key);
1303 if (!channel) {
1304 ERR("UST consumer get channel key %" PRIu64 " not found", key);
1305 ret_code = LTTNG_ERR_UST_CHAN_NOT_FOUND;
1306 goto end_msg_sessiond;
1307 }
1308
1309 /* Send everything to sessiond. */
1310 ret = send_sessiond_channel(sock, channel, ctx, &relayd_err);
1311 if (ret < 0) {
1312 if (relayd_err) {
1313 /*
1314 * We were unable to send to the relayd the stream so avoid
1315 * sending back a fatal error to the thread since this is OK
1316 * and the consumer can continue its work. The above call
1317 * has sent the error status message to the sessiond.
1318 */
1319 goto end_nosignal;
1320 }
1321 /*
1322 * The communicaton was broken hence there is a bad state between
1323 * the consumer and sessiond so stop everything.
1324 */
1325 goto error_fatal;
1326 }
1327
1328 /*
1329 * In no monitor mode, the streams ownership is kept inside the channel
1330 * so don't send them to the data thread.
1331 */
1332 if (!channel->monitor) {
1333 goto end_msg_sessiond;
1334 }
1335
1336 ret = send_streams_to_thread(channel, ctx);
1337 if (ret < 0) {
1338 /*
1339 * If we are unable to send the stream to the thread, there is
1340 * a big problem so just stop everything.
1341 */
1342 goto error_fatal;
1343 }
1344 /* List MUST be empty after or else it could be reused. */
1345 assert(cds_list_empty(&channel->streams.head));
1346 goto end_msg_sessiond;
1347 }
1348 case LTTNG_CONSUMER_DESTROY_CHANNEL:
1349 {
1350 uint64_t key = msg.u.destroy_channel.key;
1351
1352 /*
1353 * Only called if streams have not been sent to stream
1354 * manager thread. However, channel has been sent to
1355 * channel manager thread.
1356 */
1357 notify_thread_del_channel(ctx, key);
1358 goto end_msg_sessiond;
1359 }
1360 case LTTNG_CONSUMER_CLOSE_METADATA:
1361 {
1362 int ret;
1363
1364 ret = close_metadata(msg.u.close_metadata.key);
1365 if (ret != 0) {
1366 ret_code = ret;
1367 }
1368
1369 goto end_msg_sessiond;
1370 }
1371 case LTTNG_CONSUMER_FLUSH_CHANNEL:
1372 {
1373 int ret;
1374
1375 ret = flush_channel(msg.u.flush_channel.key);
1376 if (ret != 0) {
1377 ret_code = ret;
1378 }
1379
1380 goto end_msg_sessiond;
1381 }
1382 case LTTNG_CONSUMER_PUSH_METADATA:
1383 {
1384 int ret;
1385 uint64_t len = msg.u.push_metadata.len;
1386 uint64_t key = msg.u.push_metadata.key;
1387 uint64_t offset = msg.u.push_metadata.target_offset;
1388 struct lttng_consumer_channel *channel;
1389
1390 DBG("UST consumer push metadata key %" PRIu64 " of len %" PRIu64, key,
1391 len);
1392
1393 channel = consumer_find_channel(key);
1394 if (!channel) {
1395 ERR("UST consumer push metadata %" PRIu64 " not found", key);
1396 ret_code = LTTNG_ERR_UST_CHAN_NOT_FOUND;
1397 goto end_msg_sessiond;
1398 }
1399
1400 /* Tell session daemon we are ready to receive the metadata. */
1401 ret = consumer_send_status_msg(sock, LTTNG_OK);
1402 if (ret < 0) {
1403 /* Somehow, the session daemon is not responding anymore. */
1404 goto error_fatal;
1405 }
1406
1407 /* Wait for more data. */
1408 if (lttng_consumer_poll_socket(consumer_sockpoll) < 0) {
1409 goto error_fatal;
1410 }
1411
1412 ret = lttng_ustconsumer_recv_metadata(sock, key, offset,
1413 len, channel, 0);
1414 if (ret < 0) {
1415 /* error receiving from sessiond */
1416 goto error_fatal;
1417 } else {
1418 ret_code = ret;
1419 goto end_msg_sessiond;
1420 }
1421 }
1422 case LTTNG_CONSUMER_SETUP_METADATA:
1423 {
1424 int ret;
1425
1426 ret = setup_metadata(ctx, msg.u.setup_metadata.key);
1427 if (ret) {
1428 ret_code = ret;
1429 }
1430 goto end_msg_sessiond;
1431 }
1432 case LTTNG_CONSUMER_SNAPSHOT_CHANNEL:
1433 {
1434 if (msg.u.snapshot_channel.metadata) {
1435 ret = snapshot_metadata(msg.u.snapshot_channel.key,
1436 msg.u.snapshot_channel.pathname,
1437 msg.u.snapshot_channel.relayd_id,
1438 ctx);
1439 if (ret < 0) {
1440 ERR("Snapshot metadata failed");
1441 ret_code = LTTNG_ERR_UST_META_FAIL;
1442 }
1443 } else {
1444 ret = snapshot_channel(msg.u.snapshot_channel.key,
1445 msg.u.snapshot_channel.pathname,
1446 msg.u.snapshot_channel.relayd_id,
1447 msg.u.snapshot_channel.max_stream_size,
1448 ctx);
1449 if (ret < 0) {
1450 ERR("Snapshot channel failed");
1451 ret_code = LTTNG_ERR_UST_CHAN_FAIL;
1452 }
1453 }
1454
1455 ret = consumer_send_status_msg(sock, ret_code);
1456 if (ret < 0) {
1457 /* Somehow, the session daemon is not responding anymore. */
1458 goto end_nosignal;
1459 }
1460 break;
1461 }
1462 default:
1463 break;
1464 }
1465
1466 end_nosignal:
1467 rcu_read_unlock();
1468
1469 /*
1470 * Return 1 to indicate success since the 0 value can be a socket
1471 * shutdown during the recv() or send() call.
1472 */
1473 return 1;
1474
1475 end_msg_sessiond:
1476 /*
1477 * The returned value here is not useful since either way we'll return 1 to
1478 * the caller because the session daemon socket management is done
1479 * elsewhere. Returning a negative code or 0 will shutdown the consumer.
1480 */
1481 ret = consumer_send_status_msg(sock, ret_code);
1482 if (ret < 0) {
1483 goto error_fatal;
1484 }
1485 rcu_read_unlock();
1486 return 1;
1487 end_channel_error:
1488 if (channel) {
1489 /*
1490 * Free channel here since no one has a reference to it. We don't
1491 * free after that because a stream can store this pointer.
1492 */
1493 destroy_channel(channel);
1494 }
1495 /* We have to send a status channel message indicating an error. */
1496 ret = consumer_send_status_channel(sock, NULL);
1497 if (ret < 0) {
1498 /* Stop everything if session daemon can not be notified. */
1499 goto error_fatal;
1500 }
1501 rcu_read_unlock();
1502 return 1;
1503 error_fatal:
1504 rcu_read_unlock();
1505 /* This will issue a consumer stop. */
1506 return -1;
1507 }
1508
1509 /*
1510 * Wrapper over the mmap() read offset from ust-ctl library. Since this can be
1511 * compiled out, we isolate it in this library.
1512 */
1513 int lttng_ustctl_get_mmap_read_offset(struct lttng_consumer_stream *stream,
1514 unsigned long *off)
1515 {
1516 assert(stream);
1517 assert(stream->ustream);
1518
1519 return ustctl_get_mmap_read_offset(stream->ustream, off);
1520 }
1521
1522 /*
1523 * Wrapper over the mmap() read offset from ust-ctl library. Since this can be
1524 * compiled out, we isolate it in this library.
1525 */
1526 void *lttng_ustctl_get_mmap_base(struct lttng_consumer_stream *stream)
1527 {
1528 assert(stream);
1529 assert(stream->ustream);
1530
1531 return ustctl_get_mmap_base(stream->ustream);
1532 }
1533
1534 /*
1535 * Take a snapshot for a specific fd
1536 *
1537 * Returns 0 on success, < 0 on error
1538 */
1539 int lttng_ustconsumer_take_snapshot(struct lttng_consumer_stream *stream)
1540 {
1541 assert(stream);
1542 assert(stream->ustream);
1543
1544 return ustctl_snapshot(stream->ustream);
1545 }
1546
1547 /*
1548 * Get the produced position
1549 *
1550 * Returns 0 on success, < 0 on error
1551 */
1552 int lttng_ustconsumer_get_produced_snapshot(
1553 struct lttng_consumer_stream *stream, unsigned long *pos)
1554 {
1555 assert(stream);
1556 assert(stream->ustream);
1557 assert(pos);
1558
1559 return ustctl_snapshot_get_produced(stream->ustream, pos);
1560 }
1561
1562 /*
1563 * Get the consumed position
1564 *
1565 * Returns 0 on success, < 0 on error
1566 */
1567 int lttng_ustconsumer_get_consumed_snapshot(
1568 struct lttng_consumer_stream *stream, unsigned long *pos)
1569 {
1570 assert(stream);
1571 assert(stream->ustream);
1572 assert(pos);
1573
1574 return ustctl_snapshot_get_consumed(stream->ustream, pos);
1575 }
1576
1577 /*
1578 * Called when the stream signal the consumer that it has hang up.
1579 */
1580 void lttng_ustconsumer_on_stream_hangup(struct lttng_consumer_stream *stream)
1581 {
1582 assert(stream);
1583 assert(stream->ustream);
1584
1585 ustctl_flush_buffer(stream->ustream, 0);
1586 stream->hangup_flush_done = 1;
1587 }
1588
1589 void lttng_ustconsumer_del_channel(struct lttng_consumer_channel *chan)
1590 {
1591 assert(chan);
1592 assert(chan->uchan);
1593
1594 if (chan->switch_timer_enabled == 1) {
1595 consumer_timer_switch_stop(chan);
1596 }
1597 consumer_metadata_cache_destroy(chan);
1598 ustctl_destroy_channel(chan->uchan);
1599 }
1600
1601 void lttng_ustconsumer_del_stream(struct lttng_consumer_stream *stream)
1602 {
1603 assert(stream);
1604 assert(stream->ustream);
1605
1606 if (stream->chan->switch_timer_enabled == 1) {
1607 consumer_timer_switch_stop(stream->chan);
1608 }
1609 ustctl_destroy_stream(stream->ustream);
1610 }
1611
1612 /*
1613 * Populate index values of a UST stream. Values are set in big endian order.
1614 *
1615 * Return 0 on success or else a negative value.
1616 */
1617 static int get_index_values(struct lttng_packet_index *index,
1618 struct ustctl_consumer_stream *ustream)
1619 {
1620 int ret;
1621
1622 ret = ustctl_get_timestamp_begin(ustream, &index->timestamp_begin);
1623 if (ret < 0) {
1624 PERROR("ustctl_get_timestamp_begin");
1625 goto error;
1626 }
1627 index->timestamp_begin = htobe64(index->timestamp_begin);
1628
1629 ret = ustctl_get_timestamp_end(ustream, &index->timestamp_end);
1630 if (ret < 0) {
1631 PERROR("ustctl_get_timestamp_end");
1632 goto error;
1633 }
1634 index->timestamp_end = htobe64(index->timestamp_end);
1635
1636 ret = ustctl_get_events_discarded(ustream, &index->events_discarded);
1637 if (ret < 0) {
1638 PERROR("ustctl_get_events_discarded");
1639 goto error;
1640 }
1641 index->events_discarded = htobe64(index->events_discarded);
1642
1643 ret = ustctl_get_content_size(ustream, &index->content_size);
1644 if (ret < 0) {
1645 PERROR("ustctl_get_content_size");
1646 goto error;
1647 }
1648 index->content_size = htobe64(index->content_size);
1649
1650 ret = ustctl_get_packet_size(ustream, &index->packet_size);
1651 if (ret < 0) {
1652 PERROR("ustctl_get_packet_size");
1653 goto error;
1654 }
1655 index->packet_size = htobe64(index->packet_size);
1656
1657 ret = ustctl_get_stream_id(ustream, &index->stream_id);
1658 if (ret < 0) {
1659 PERROR("ustctl_get_stream_id");
1660 goto error;
1661 }
1662 index->stream_id = htobe64(index->stream_id);
1663
1664 error:
1665 return ret;
1666 }
1667
1668
1669 int lttng_ustconsumer_read_subbuffer(struct lttng_consumer_stream *stream,
1670 struct lttng_consumer_local_data *ctx)
1671 {
1672 unsigned long len, subbuf_size, padding;
1673 int err, write_index = 1;
1674 long ret = 0;
1675 char dummy;
1676 struct ustctl_consumer_stream *ustream;
1677 struct lttng_packet_index index;
1678
1679 assert(stream);
1680 assert(stream->ustream);
1681 assert(ctx);
1682
1683 DBG("In UST read_subbuffer (wait_fd: %d, name: %s)", stream->wait_fd,
1684 stream->name);
1685
1686 /* Ease our life for what's next. */
1687 ustream = stream->ustream;
1688
1689 /* We can consume the 1 byte written into the wait_fd by UST */
1690 if (stream->monitor && !stream->hangup_flush_done) {
1691 ssize_t readlen;
1692
1693 do {
1694 readlen = read(stream->wait_fd, &dummy, 1);
1695 } while (readlen == -1 && errno == EINTR);
1696 if (readlen == -1 && errno != EAGAIN && errno != EWOULDBLOCK) {
1697 ret = readlen;
1698 goto end;
1699 }
1700 }
1701
1702 retry:
1703 /* Get the next subbuffer */
1704 err = ustctl_get_next_subbuf(ustream);
1705 if (err != 0) {
1706 /*
1707 * Populate metadata info if the existing info has
1708 * already been read.
1709 */
1710 if (stream->metadata_flag) {
1711 ssize_t write_len;
1712
1713 if (stream->chan->metadata_cache->contiguous
1714 == stream->ust_metadata_pushed) {
1715 ret = 0;
1716 goto end;
1717 }
1718
1719 write_len = ustctl_write_one_packet_to_channel(stream->chan->uchan,
1720 &stream->chan->metadata_cache->data[stream->ust_metadata_pushed],
1721 stream->chan->metadata_cache->contiguous
1722 - stream->ust_metadata_pushed);
1723 assert(write_len != 0);
1724 if (write_len < 0) {
1725 ERR("Writing one metadata packet");
1726 ret = -1;
1727 goto end;
1728 }
1729 stream->ust_metadata_pushed += write_len;
1730 ustctl_flush_buffer(stream->ustream, 1);
1731 goto retry;
1732 }
1733
1734 ret = err; /* ustctl_get_next_subbuf returns negative, caller expect positive. */
1735 /*
1736 * This is a debug message even for single-threaded consumer,
1737 * because poll() have more relaxed criterions than get subbuf,
1738 * so get_subbuf may fail for short race windows where poll()
1739 * would issue wakeups.
1740 */
1741 DBG("Reserving sub buffer failed (everything is normal, "
1742 "it is due to concurrency) [ret: %d]", err);
1743 goto end;
1744 }
1745 assert(stream->chan->output == CONSUMER_CHANNEL_MMAP);
1746
1747 if (!stream->metadata_flag) {
1748 index.offset = htobe64(stream->out_fd_offset);
1749 ret = get_index_values(&index, ustream);
1750 if (ret < 0) {
1751 goto end;
1752 }
1753 } else {
1754 write_index = 0;
1755 }
1756
1757 /* Get the full padded subbuffer size */
1758 err = ustctl_get_padded_subbuf_size(ustream, &len);
1759 assert(err == 0);
1760
1761 /* Get subbuffer data size (without padding) */
1762 err = ustctl_get_subbuf_size(ustream, &subbuf_size);
1763 assert(err == 0);
1764
1765 /* Make sure we don't get a subbuffer size bigger than the padded */
1766 assert(len >= subbuf_size);
1767
1768 padding = len - subbuf_size;
1769 /* write the subbuffer to the tracefile */
1770 ret = lttng_consumer_on_read_subbuffer_mmap(ctx, stream, subbuf_size, padding, &index);
1771 /*
1772 * The mmap operation should write subbuf_size amount of data when network
1773 * streaming or the full padding (len) size when we are _not_ streaming.
1774 */
1775 if ((ret != subbuf_size && stream->net_seq_idx != (uint64_t) -1ULL) ||
1776 (ret != len && stream->net_seq_idx == (uint64_t) -1ULL)) {
1777 /*
1778 * Display the error but continue processing to try to release the
1779 * subbuffer. This is a DBG statement since any unexpected kill or
1780 * signal, the application gets unregistered, relayd gets closed or
1781 * anything that affects the buffer lifetime will trigger this error.
1782 * So, for the sake of the user, don't print this error since it can
1783 * happen and it is OK with the code flow.
1784 */
1785 DBG("Error writing to tracefile "
1786 "(ret: %ld != len: %lu != subbuf_size: %lu)",
1787 ret, len, subbuf_size);
1788 write_index = 0;
1789 }
1790 err = ustctl_put_next_subbuf(ustream);
1791 assert(err == 0);
1792
1793 /* Write index if needed. */
1794 if (!write_index) {
1795 goto end;
1796 }
1797
1798 assert(!stream->metadata_flag);
1799 err = consumer_stream_write_index(stream, &index);
1800 if (err < 0) {
1801 goto end;
1802 }
1803
1804 end:
1805 return ret;
1806 }
1807
1808 /*
1809 * Called when a stream is created.
1810 *
1811 * Return 0 on success or else a negative value.
1812 */
1813 int lttng_ustconsumer_on_recv_stream(struct lttng_consumer_stream *stream)
1814 {
1815 int ret;
1816
1817 assert(stream);
1818
1819 /* Don't create anything if this is set for streaming. */
1820 if (stream->net_seq_idx == (uint64_t) -1ULL && stream->chan->monitor) {
1821 ret = utils_create_stream_file(stream->chan->pathname, stream->name,
1822 stream->chan->tracefile_size, stream->tracefile_count_current,
1823 stream->uid, stream->gid, NULL);
1824 if (ret < 0) {
1825 goto error;
1826 }
1827 stream->out_fd = ret;
1828 stream->tracefile_size_current = 0;
1829
1830 if (!stream->metadata_flag) {
1831 ret = index_create_file(stream->chan->pathname,
1832 stream->name, stream->uid, stream->gid,
1833 stream->chan->tracefile_size,
1834 stream->tracefile_count_current);
1835 if (ret < 0) {
1836 goto error;
1837 }
1838 stream->index_fd = ret;
1839 }
1840 }
1841 ret = 0;
1842
1843 error:
1844 return ret;
1845 }
1846
1847 /*
1848 * Check if data is still being extracted from the buffers for a specific
1849 * stream. Consumer data lock MUST be acquired before calling this function
1850 * and the stream lock.
1851 *
1852 * Return 1 if the traced data are still getting read else 0 meaning that the
1853 * data is available for trace viewer reading.
1854 */
1855 int lttng_ustconsumer_data_pending(struct lttng_consumer_stream *stream)
1856 {
1857 int ret;
1858
1859 assert(stream);
1860 assert(stream->ustream);
1861
1862 DBG("UST consumer checking data pending");
1863
1864 if (stream->endpoint_status != CONSUMER_ENDPOINT_ACTIVE) {
1865 ret = 0;
1866 goto end;
1867 }
1868
1869 if (stream->chan->type == CONSUMER_CHANNEL_TYPE_METADATA) {
1870 uint64_t contiguous, pushed;
1871
1872 /* Ease our life a bit. */
1873 contiguous = stream->chan->metadata_cache->contiguous;
1874 pushed = stream->ust_metadata_pushed;
1875
1876 /*
1877 * We can simply check whether all contiguously available data
1878 * has been pushed to the ring buffer, since the push operation
1879 * is performed within get_next_subbuf(), and because both
1880 * get_next_subbuf() and put_next_subbuf() are issued atomically
1881 * thanks to the stream lock within
1882 * lttng_ustconsumer_read_subbuffer(). This basically means that
1883 * whetnever ust_metadata_pushed is incremented, the associated
1884 * metadata has been consumed from the metadata stream.
1885 */
1886 DBG("UST consumer metadata pending check: contiguous %" PRIu64 " vs pushed %" PRIu64,
1887 contiguous, pushed);
1888 assert(((int64_t) contiguous - pushed) >= 0);
1889 if ((contiguous != pushed) ||
1890 (((int64_t) contiguous - pushed) > 0 || contiguous == 0)) {
1891 ret = 1; /* Data is pending */
1892 goto end;
1893 }
1894 } else {
1895 ret = ustctl_get_next_subbuf(stream->ustream);
1896 if (ret == 0) {
1897 /*
1898 * There is still data so let's put back this
1899 * subbuffer.
1900 */
1901 ret = ustctl_put_subbuf(stream->ustream);
1902 assert(ret == 0);
1903 ret = 1; /* Data is pending */
1904 goto end;
1905 }
1906 }
1907
1908 /* Data is NOT pending so ready to be read. */
1909 ret = 0;
1910
1911 end:
1912 return ret;
1913 }
1914
1915 /*
1916 * Close every metadata stream wait fd of the metadata hash table. This
1917 * function MUST be used very carefully so not to run into a race between the
1918 * metadata thread handling streams and this function closing their wait fd.
1919 *
1920 * For UST, this is used when the session daemon hangs up. Its the metadata
1921 * producer so calling this is safe because we are assured that no state change
1922 * can occur in the metadata thread for the streams in the hash table.
1923 */
1924 void lttng_ustconsumer_close_metadata(struct lttng_ht *metadata_ht)
1925 {
1926 struct lttng_ht_iter iter;
1927 struct lttng_consumer_stream *stream;
1928
1929 assert(metadata_ht);
1930 assert(metadata_ht->ht);
1931
1932 DBG("UST consumer closing all metadata streams");
1933
1934 rcu_read_lock();
1935 cds_lfht_for_each_entry(metadata_ht->ht, &iter.iter, stream,
1936 node.node) {
1937 pthread_mutex_lock(&stream->chan->lock);
1938 /*
1939 * Whatever returned value, we must continue to try to close everything
1940 * so ignore it.
1941 */
1942 (void) _close_metadata(stream->chan);
1943 DBG("Metadata wait fd %d and poll pipe fd %d closed", stream->wait_fd,
1944 stream->ust_metadata_poll_pipe[1]);
1945 pthread_mutex_unlock(&stream->chan->lock);
1946
1947 }
1948 rcu_read_unlock();
1949 }
1950
1951 void lttng_ustconsumer_close_stream_wakeup(struct lttng_consumer_stream *stream)
1952 {
1953 int ret;
1954
1955 ret = ustctl_stream_close_wakeup_fd(stream->ustream);
1956 if (ret < 0) {
1957 ERR("Unable to close wakeup fd");
1958 }
1959 }
1960
1961 /*
1962 * Please refer to consumer-timer.c before adding any lock within this
1963 * function or any of its callees. Timers have a very strict locking
1964 * semantic with respect to teardown. Failure to respect this semantic
1965 * introduces deadlocks.
1966 */
1967 int lttng_ustconsumer_request_metadata(struct lttng_consumer_local_data *ctx,
1968 struct lttng_consumer_channel *channel, int timer)
1969 {
1970 struct lttcomm_metadata_request_msg request;
1971 struct lttcomm_consumer_msg msg;
1972 enum lttng_error_code ret_code = LTTNG_OK;
1973 uint64_t len, key, offset;
1974 int ret;
1975
1976 assert(channel);
1977 assert(channel->metadata_cache);
1978
1979 /* send the metadata request to sessiond */
1980 switch (consumer_data.type) {
1981 case LTTNG_CONSUMER64_UST:
1982 request.bits_per_long = 64;
1983 break;
1984 case LTTNG_CONSUMER32_UST:
1985 request.bits_per_long = 32;
1986 break;
1987 default:
1988 request.bits_per_long = 0;
1989 break;
1990 }
1991
1992 request.session_id = channel->session_id;
1993 request.session_id_per_pid = channel->session_id_per_pid;
1994 /*
1995 * Request the application UID here so the metadata of that application can
1996 * be sent back. The channel UID corresponds to the user UID of the session
1997 * used for the rights on the stream file(s).
1998 */
1999 request.uid = channel->ust_app_uid;
2000 request.key = channel->key;
2001
2002 DBG("Sending metadata request to sessiond, session id %" PRIu64
2003 ", per-pid %" PRIu64 ", app UID %u and channek key %" PRIu64,
2004 request.session_id, request.session_id_per_pid, request.uid,
2005 request.key);
2006
2007 pthread_mutex_lock(&ctx->metadata_socket_lock);
2008 ret = lttcomm_send_unix_sock(ctx->consumer_metadata_socket, &request,
2009 sizeof(request));
2010 if (ret < 0) {
2011 ERR("Asking metadata to sessiond");
2012 goto end;
2013 }
2014
2015 /* Receive the metadata from sessiond */
2016 ret = lttcomm_recv_unix_sock(ctx->consumer_metadata_socket, &msg,
2017 sizeof(msg));
2018 if (ret != sizeof(msg)) {
2019 DBG("Consumer received unexpected message size %d (expects %zu)",
2020 ret, sizeof(msg));
2021 lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_ERROR_RECV_CMD);
2022 /*
2023 * The ret value might 0 meaning an orderly shutdown but this is ok
2024 * since the caller handles this.
2025 */
2026 goto end;
2027 }
2028
2029 if (msg.cmd_type == LTTNG_ERR_UND) {
2030 /* No registry found */
2031 (void) consumer_send_status_msg(ctx->consumer_metadata_socket,
2032 ret_code);
2033 ret = 0;
2034 goto end;
2035 } else if (msg.cmd_type != LTTNG_CONSUMER_PUSH_METADATA) {
2036 ERR("Unexpected cmd_type received %d", msg.cmd_type);
2037 ret = -1;
2038 goto end;
2039 }
2040
2041 len = msg.u.push_metadata.len;
2042 key = msg.u.push_metadata.key;
2043 offset = msg.u.push_metadata.target_offset;
2044
2045 assert(key == channel->key);
2046 if (len == 0) {
2047 DBG("No new metadata to receive for key %" PRIu64, key);
2048 }
2049
2050 /* Tell session daemon we are ready to receive the metadata. */
2051 ret = consumer_send_status_msg(ctx->consumer_metadata_socket,
2052 LTTNG_OK);
2053 if (ret < 0 || len == 0) {
2054 /*
2055 * Somehow, the session daemon is not responding anymore or there is
2056 * nothing to receive.
2057 */
2058 goto end;
2059 }
2060
2061 ret_code = lttng_ustconsumer_recv_metadata(ctx->consumer_metadata_socket,
2062 key, offset, len, channel, timer);
2063 if (ret_code >= 0) {
2064 /*
2065 * Only send the status msg if the sessiond is alive meaning a positive
2066 * ret code.
2067 */
2068 (void) consumer_send_status_msg(ctx->consumer_metadata_socket, ret_code);
2069 }
2070 ret = 0;
2071
2072 end:
2073 pthread_mutex_unlock(&ctx->metadata_socket_lock);
2074 return ret;
2075 }
This page took 0.120656 seconds and 3 git commands to generate.