Fix: Missing rcu read side lock in consumer
[lttng-tools.git] / src / common / consumer.c
1 /*
2 * Copyright (C) 2011 - Julien Desfossez <julien.desfossez@polymtl.ca>
3 * Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
4 * 2012 - David Goulet <dgoulet@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, but WITHOUT
11 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
12 * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
13 * 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 _GNU_SOURCE
21 #include <assert.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/types.h>
29 #include <unistd.h>
30 #include <inttypes.h>
31
32 #include <common/common.h>
33 #include <common/utils.h>
34 #include <common/compat/poll.h>
35 #include <common/kernel-ctl/kernel-ctl.h>
36 #include <common/sessiond-comm/relayd.h>
37 #include <common/sessiond-comm/sessiond-comm.h>
38 #include <common/kernel-consumer/kernel-consumer.h>
39 #include <common/relayd/relayd.h>
40 #include <common/ust-consumer/ust-consumer.h>
41
42 #include "consumer.h"
43
44 struct lttng_consumer_global_data consumer_data = {
45 .stream_count = 0,
46 .need_update = 1,
47 .type = LTTNG_CONSUMER_UNKNOWN,
48 };
49
50 /* timeout parameter, to control the polling thread grace period. */
51 int consumer_poll_timeout = -1;
52
53 /*
54 * Flag to inform the polling thread to quit when all fd hung up. Updated by
55 * the consumer_thread_receive_fds when it notices that all fds has hung up.
56 * Also updated by the signal handler (consumer_should_exit()). Read by the
57 * polling threads.
58 */
59 volatile int consumer_quit = 0;
60
61 /*
62 * Find a stream. The consumer_data.lock must be locked during this
63 * call.
64 */
65 static struct lttng_consumer_stream *consumer_find_stream(int key)
66 {
67 struct lttng_ht_iter iter;
68 struct lttng_ht_node_ulong *node;
69 struct lttng_consumer_stream *stream = NULL;
70
71 /* Negative keys are lookup failures */
72 if (key < 0)
73 return NULL;
74
75 rcu_read_lock();
76
77 lttng_ht_lookup(consumer_data.stream_ht, (void *)((unsigned long) key),
78 &iter);
79 node = lttng_ht_iter_get_node_ulong(&iter);
80 if (node != NULL) {
81 stream = caa_container_of(node, struct lttng_consumer_stream, node);
82 }
83
84 rcu_read_unlock();
85
86 return stream;
87 }
88
89 static void consumer_steal_stream_key(int key)
90 {
91 struct lttng_consumer_stream *stream;
92
93 rcu_read_lock();
94 stream = consumer_find_stream(key);
95 if (stream) {
96 stream->key = -1;
97 /*
98 * We don't want the lookup to match, but we still need
99 * to iterate on this stream when iterating over the hash table. Just
100 * change the node key.
101 */
102 stream->node.key = -1;
103 }
104 rcu_read_unlock();
105 }
106
107 static struct lttng_consumer_channel *consumer_find_channel(int key)
108 {
109 struct lttng_ht_iter iter;
110 struct lttng_ht_node_ulong *node;
111 struct lttng_consumer_channel *channel = NULL;
112
113 /* Negative keys are lookup failures */
114 if (key < 0)
115 return NULL;
116
117 rcu_read_lock();
118
119 lttng_ht_lookup(consumer_data.channel_ht, (void *)((unsigned long) key),
120 &iter);
121 node = lttng_ht_iter_get_node_ulong(&iter);
122 if (node != NULL) {
123 channel = caa_container_of(node, struct lttng_consumer_channel, node);
124 }
125
126 rcu_read_unlock();
127
128 return channel;
129 }
130
131 static void consumer_steal_channel_key(int key)
132 {
133 struct lttng_consumer_channel *channel;
134
135 rcu_read_lock();
136 channel = consumer_find_channel(key);
137 if (channel) {
138 channel->key = -1;
139 /*
140 * We don't want the lookup to match, but we still need
141 * to iterate on this channel when iterating over the hash table. Just
142 * change the node key.
143 */
144 channel->node.key = -1;
145 }
146 rcu_read_unlock();
147 }
148
149 static
150 void consumer_free_stream(struct rcu_head *head)
151 {
152 struct lttng_ht_node_ulong *node =
153 caa_container_of(head, struct lttng_ht_node_ulong, head);
154 struct lttng_consumer_stream *stream =
155 caa_container_of(node, struct lttng_consumer_stream, node);
156
157 free(stream);
158 }
159
160 /*
161 * RCU protected relayd socket pair free.
162 */
163 static void consumer_rcu_free_relayd(struct rcu_head *head)
164 {
165 struct lttng_ht_node_ulong *node =
166 caa_container_of(head, struct lttng_ht_node_ulong, head);
167 struct consumer_relayd_sock_pair *relayd =
168 caa_container_of(node, struct consumer_relayd_sock_pair, node);
169
170 free(relayd);
171 }
172
173 /*
174 * Destroy and free relayd socket pair object.
175 *
176 * This function MUST be called with the consumer_data lock acquired.
177 */
178 static void destroy_relayd(struct consumer_relayd_sock_pair *relayd)
179 {
180 int ret;
181 struct lttng_ht_iter iter;
182
183 if (relayd == NULL) {
184 return;
185 }
186
187 DBG("Consumer destroy and close relayd socket pair");
188
189 iter.iter.node = &relayd->node.node;
190 ret = lttng_ht_del(consumer_data.relayd_ht, &iter);
191 if (ret != 0) {
192 /* We assume the relayd was already destroyed */
193 return;
194 }
195
196 /* Close all sockets */
197 pthread_mutex_lock(&relayd->ctrl_sock_mutex);
198 (void) relayd_close(&relayd->control_sock);
199 pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
200 (void) relayd_close(&relayd->data_sock);
201
202 /* RCU free() call */
203 call_rcu(&relayd->node.head, consumer_rcu_free_relayd);
204 }
205
206 /*
207 * Flag a relayd socket pair for destruction. Destroy it if the refcount
208 * reaches zero.
209 *
210 * RCU read side lock MUST be aquired before calling this function.
211 */
212 void consumer_flag_relayd_for_destroy(struct consumer_relayd_sock_pair *relayd)
213 {
214 assert(relayd);
215
216 /* Set destroy flag for this object */
217 uatomic_set(&relayd->destroy_flag, 1);
218
219 /* Destroy the relayd if refcount is 0 */
220 if (uatomic_read(&relayd->refcount) == 0) {
221 destroy_relayd(relayd);
222 }
223 }
224
225 /*
226 * Remove a stream from the global list protected by a mutex. This
227 * function is also responsible for freeing its data structures.
228 */
229 void consumer_del_stream(struct lttng_consumer_stream *stream)
230 {
231 int ret;
232 struct lttng_ht_iter iter;
233 struct lttng_consumer_channel *free_chan = NULL;
234 struct consumer_relayd_sock_pair *relayd;
235
236 assert(stream);
237
238 pthread_mutex_lock(&consumer_data.lock);
239
240 switch (consumer_data.type) {
241 case LTTNG_CONSUMER_KERNEL:
242 if (stream->mmap_base != NULL) {
243 ret = munmap(stream->mmap_base, stream->mmap_len);
244 if (ret != 0) {
245 perror("munmap");
246 }
247 }
248 break;
249 case LTTNG_CONSUMER32_UST:
250 case LTTNG_CONSUMER64_UST:
251 lttng_ustconsumer_del_stream(stream);
252 break;
253 default:
254 ERR("Unknown consumer_data type");
255 assert(0);
256 goto end;
257 }
258
259 rcu_read_lock();
260 iter.iter.node = &stream->node.node;
261 ret = lttng_ht_del(consumer_data.stream_ht, &iter);
262 assert(!ret);
263
264 rcu_read_unlock();
265
266 if (consumer_data.stream_count <= 0) {
267 goto end;
268 }
269 consumer_data.stream_count--;
270 if (!stream) {
271 goto end;
272 }
273 if (stream->out_fd >= 0) {
274 ret = close(stream->out_fd);
275 if (ret) {
276 PERROR("close");
277 }
278 }
279 if (stream->wait_fd >= 0 && !stream->wait_fd_is_copy) {
280 ret = close(stream->wait_fd);
281 if (ret) {
282 PERROR("close");
283 }
284 }
285 if (stream->shm_fd >= 0 && stream->wait_fd != stream->shm_fd) {
286 ret = close(stream->shm_fd);
287 if (ret) {
288 PERROR("close");
289 }
290 }
291
292 /* Check and cleanup relayd */
293 rcu_read_lock();
294 relayd = consumer_find_relayd(stream->net_seq_idx);
295 if (relayd != NULL) {
296 uatomic_dec(&relayd->refcount);
297 assert(uatomic_read(&relayd->refcount) >= 0);
298
299 /* Closing streams requires to lock the control socket. */
300 pthread_mutex_lock(&relayd->ctrl_sock_mutex);
301 ret = relayd_send_close_stream(&relayd->control_sock,
302 stream->relayd_stream_id,
303 stream->next_net_seq_num - 1);
304 pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
305 if (ret < 0) {
306 DBG("Unable to close stream on the relayd. Continuing");
307 /*
308 * Continue here. There is nothing we can do for the relayd.
309 * Chances are that the relayd has closed the socket so we just
310 * continue cleaning up.
311 */
312 }
313
314 /* Both conditions are met, we destroy the relayd. */
315 if (uatomic_read(&relayd->refcount) == 0 &&
316 uatomic_read(&relayd->destroy_flag)) {
317 destroy_relayd(relayd);
318 }
319 }
320 rcu_read_unlock();
321
322 if (!--stream->chan->refcount) {
323 free_chan = stream->chan;
324 }
325
326
327 call_rcu(&stream->node.head, consumer_free_stream);
328 end:
329 consumer_data.need_update = 1;
330 pthread_mutex_unlock(&consumer_data.lock);
331
332 if (free_chan)
333 consumer_del_channel(free_chan);
334 }
335
336 struct lttng_consumer_stream *consumer_allocate_stream(
337 int channel_key, int stream_key,
338 int shm_fd, int wait_fd,
339 enum lttng_consumer_stream_state state,
340 uint64_t mmap_len,
341 enum lttng_event_output output,
342 const char *path_name,
343 uid_t uid,
344 gid_t gid,
345 int net_index,
346 int metadata_flag)
347 {
348 struct lttng_consumer_stream *stream;
349 int ret;
350
351 stream = zmalloc(sizeof(*stream));
352 if (stream == NULL) {
353 perror("malloc struct lttng_consumer_stream");
354 goto end;
355 }
356 stream->chan = consumer_find_channel(channel_key);
357 if (!stream->chan) {
358 perror("Unable to find channel key");
359 goto end;
360 }
361 stream->chan->refcount++;
362 stream->key = stream_key;
363 stream->shm_fd = shm_fd;
364 stream->wait_fd = wait_fd;
365 stream->out_fd = -1;
366 stream->out_fd_offset = 0;
367 stream->state = state;
368 stream->mmap_len = mmap_len;
369 stream->mmap_base = NULL;
370 stream->output = output;
371 stream->uid = uid;
372 stream->gid = gid;
373 stream->net_seq_idx = net_index;
374 stream->metadata_flag = metadata_flag;
375 strncpy(stream->path_name, path_name, sizeof(stream->path_name));
376 stream->path_name[sizeof(stream->path_name) - 1] = '\0';
377 lttng_ht_node_init_ulong(&stream->node, stream->key);
378 lttng_ht_node_init_ulong(&stream->waitfd_node, stream->wait_fd);
379
380 switch (consumer_data.type) {
381 case LTTNG_CONSUMER_KERNEL:
382 break;
383 case LTTNG_CONSUMER32_UST:
384 case LTTNG_CONSUMER64_UST:
385 stream->cpu = stream->chan->cpucount++;
386 ret = lttng_ustconsumer_allocate_stream(stream);
387 if (ret) {
388 free(stream);
389 return NULL;
390 }
391 break;
392 default:
393 ERR("Unknown consumer_data type");
394 assert(0);
395 goto end;
396 }
397 DBG("Allocated stream %s (key %d, shm_fd %d, wait_fd %d, mmap_len %llu, out_fd %d, net_seq_idx %d)",
398 stream->path_name, stream->key,
399 stream->shm_fd,
400 stream->wait_fd,
401 (unsigned long long) stream->mmap_len,
402 stream->out_fd,
403 stream->net_seq_idx);
404 end:
405 return stream;
406 }
407
408 /*
409 * Add a stream to the global list protected by a mutex.
410 */
411 int consumer_add_stream(struct lttng_consumer_stream *stream)
412 {
413 int ret = 0;
414 struct lttng_ht_node_ulong *node;
415 struct lttng_ht_iter iter;
416 struct consumer_relayd_sock_pair *relayd;
417
418 pthread_mutex_lock(&consumer_data.lock);
419 /* Steal stream identifier, for UST */
420 consumer_steal_stream_key(stream->key);
421
422 rcu_read_lock();
423 lttng_ht_lookup(consumer_data.stream_ht,
424 (void *)((unsigned long) stream->key), &iter);
425 node = lttng_ht_iter_get_node_ulong(&iter);
426 if (node != NULL) {
427 rcu_read_unlock();
428 /* Stream already exist. Ignore the insertion */
429 goto end;
430 }
431
432 lttng_ht_add_unique_ulong(consumer_data.stream_ht, &stream->node);
433
434 /* Check and cleanup relayd */
435 relayd = consumer_find_relayd(stream->net_seq_idx);
436 if (relayd != NULL) {
437 uatomic_inc(&relayd->refcount);
438 }
439 rcu_read_unlock();
440
441 /* Update consumer data */
442 consumer_data.stream_count++;
443 consumer_data.need_update = 1;
444
445 end:
446 pthread_mutex_unlock(&consumer_data.lock);
447
448 return ret;
449 }
450
451 /*
452 * Add relayd socket to global consumer data hashtable. RCU read side lock MUST
453 * be acquired before calling this.
454 */
455 static int add_relayd(struct consumer_relayd_sock_pair *relayd)
456 {
457 int ret = 0;
458 struct lttng_ht_node_ulong *node;
459 struct lttng_ht_iter iter;
460
461 if (relayd == NULL) {
462 ret = -1;
463 goto end;
464 }
465
466 lttng_ht_lookup(consumer_data.relayd_ht,
467 (void *)((unsigned long) relayd->net_seq_idx), &iter);
468 node = lttng_ht_iter_get_node_ulong(&iter);
469 if (node != NULL) {
470 /* Relayd already exist. Ignore the insertion */
471 goto end;
472 }
473 lttng_ht_add_unique_ulong(consumer_data.relayd_ht, &relayd->node);
474
475 end:
476 return ret;
477 }
478
479 /*
480 * Allocate and return a consumer relayd socket.
481 */
482 struct consumer_relayd_sock_pair *consumer_allocate_relayd_sock_pair(
483 int net_seq_idx)
484 {
485 struct consumer_relayd_sock_pair *obj = NULL;
486
487 /* Negative net sequence index is a failure */
488 if (net_seq_idx < 0) {
489 goto error;
490 }
491
492 obj = zmalloc(sizeof(struct consumer_relayd_sock_pair));
493 if (obj == NULL) {
494 PERROR("zmalloc relayd sock");
495 goto error;
496 }
497
498 obj->net_seq_idx = net_seq_idx;
499 obj->refcount = 0;
500 obj->destroy_flag = 0;
501 lttng_ht_node_init_ulong(&obj->node, obj->net_seq_idx);
502 pthread_mutex_init(&obj->ctrl_sock_mutex, NULL);
503
504 error:
505 return obj;
506 }
507
508 /*
509 * Find a relayd socket pair in the global consumer data.
510 *
511 * Return the object if found else NULL.
512 * RCU read-side lock must be held across this call and while using the
513 * returned object.
514 */
515 struct consumer_relayd_sock_pair *consumer_find_relayd(int key)
516 {
517 struct lttng_ht_iter iter;
518 struct lttng_ht_node_ulong *node;
519 struct consumer_relayd_sock_pair *relayd = NULL;
520
521 /* Negative keys are lookup failures */
522 if (key < 0) {
523 goto error;
524 }
525
526 lttng_ht_lookup(consumer_data.relayd_ht, (void *)((unsigned long) key),
527 &iter);
528 node = lttng_ht_iter_get_node_ulong(&iter);
529 if (node != NULL) {
530 relayd = caa_container_of(node, struct consumer_relayd_sock_pair, node);
531 }
532
533 error:
534 return relayd;
535 }
536
537 /*
538 * Handle stream for relayd transmission if the stream applies for network
539 * streaming where the net sequence index is set.
540 *
541 * Return destination file descriptor or negative value on error.
542 */
543 static int write_relayd_stream_header(struct lttng_consumer_stream *stream,
544 size_t data_size, unsigned long padding,
545 struct consumer_relayd_sock_pair *relayd)
546 {
547 int outfd = -1, ret;
548 struct lttcomm_relayd_data_hdr data_hdr;
549
550 /* Safety net */
551 assert(stream);
552 assert(relayd);
553
554 /* Reset data header */
555 memset(&data_hdr, 0, sizeof(data_hdr));
556
557 if (stream->metadata_flag) {
558 /* Caller MUST acquire the relayd control socket lock */
559 ret = relayd_send_metadata(&relayd->control_sock, data_size);
560 if (ret < 0) {
561 goto error;
562 }
563
564 /* Metadata are always sent on the control socket. */
565 outfd = relayd->control_sock.fd;
566 } else {
567 /* Set header with stream information */
568 data_hdr.stream_id = htobe64(stream->relayd_stream_id);
569 data_hdr.data_size = htobe32(data_size);
570 data_hdr.padding_size = htobe32(padding);
571 data_hdr.net_seq_num = htobe64(stream->next_net_seq_num++);
572 /* Other fields are zeroed previously */
573
574 ret = relayd_send_data_hdr(&relayd->data_sock, &data_hdr,
575 sizeof(data_hdr));
576 if (ret < 0) {
577 goto error;
578 }
579
580 /* Set to go on data socket */
581 outfd = relayd->data_sock.fd;
582 }
583
584 error:
585 return outfd;
586 }
587
588 /*
589 * Update a stream according to what we just received.
590 */
591 void consumer_change_stream_state(int stream_key,
592 enum lttng_consumer_stream_state state)
593 {
594 struct lttng_consumer_stream *stream;
595
596 pthread_mutex_lock(&consumer_data.lock);
597 stream = consumer_find_stream(stream_key);
598 if (stream) {
599 stream->state = state;
600 }
601 consumer_data.need_update = 1;
602 pthread_mutex_unlock(&consumer_data.lock);
603 }
604
605 static
606 void consumer_free_channel(struct rcu_head *head)
607 {
608 struct lttng_ht_node_ulong *node =
609 caa_container_of(head, struct lttng_ht_node_ulong, head);
610 struct lttng_consumer_channel *channel =
611 caa_container_of(node, struct lttng_consumer_channel, node);
612
613 free(channel);
614 }
615
616 /*
617 * Remove a channel from the global list protected by a mutex. This
618 * function is also responsible for freeing its data structures.
619 */
620 void consumer_del_channel(struct lttng_consumer_channel *channel)
621 {
622 int ret;
623 struct lttng_ht_iter iter;
624
625 pthread_mutex_lock(&consumer_data.lock);
626
627 switch (consumer_data.type) {
628 case LTTNG_CONSUMER_KERNEL:
629 break;
630 case LTTNG_CONSUMER32_UST:
631 case LTTNG_CONSUMER64_UST:
632 lttng_ustconsumer_del_channel(channel);
633 break;
634 default:
635 ERR("Unknown consumer_data type");
636 assert(0);
637 goto end;
638 }
639
640 rcu_read_lock();
641 iter.iter.node = &channel->node.node;
642 ret = lttng_ht_del(consumer_data.channel_ht, &iter);
643 assert(!ret);
644 rcu_read_unlock();
645
646 if (channel->mmap_base != NULL) {
647 ret = munmap(channel->mmap_base, channel->mmap_len);
648 if (ret != 0) {
649 perror("munmap");
650 }
651 }
652 if (channel->wait_fd >= 0 && !channel->wait_fd_is_copy) {
653 ret = close(channel->wait_fd);
654 if (ret) {
655 PERROR("close");
656 }
657 }
658 if (channel->shm_fd >= 0 && channel->wait_fd != channel->shm_fd) {
659 ret = close(channel->shm_fd);
660 if (ret) {
661 PERROR("close");
662 }
663 }
664
665 call_rcu(&channel->node.head, consumer_free_channel);
666 end:
667 pthread_mutex_unlock(&consumer_data.lock);
668 }
669
670 struct lttng_consumer_channel *consumer_allocate_channel(
671 int channel_key,
672 int shm_fd, int wait_fd,
673 uint64_t mmap_len,
674 uint64_t max_sb_size)
675 {
676 struct lttng_consumer_channel *channel;
677 int ret;
678
679 channel = zmalloc(sizeof(*channel));
680 if (channel == NULL) {
681 perror("malloc struct lttng_consumer_channel");
682 goto end;
683 }
684 channel->key = channel_key;
685 channel->shm_fd = shm_fd;
686 channel->wait_fd = wait_fd;
687 channel->mmap_len = mmap_len;
688 channel->max_sb_size = max_sb_size;
689 channel->refcount = 0;
690 lttng_ht_node_init_ulong(&channel->node, channel->key);
691
692 switch (consumer_data.type) {
693 case LTTNG_CONSUMER_KERNEL:
694 channel->mmap_base = NULL;
695 channel->mmap_len = 0;
696 break;
697 case LTTNG_CONSUMER32_UST:
698 case LTTNG_CONSUMER64_UST:
699 ret = lttng_ustconsumer_allocate_channel(channel);
700 if (ret) {
701 free(channel);
702 return NULL;
703 }
704 break;
705 default:
706 ERR("Unknown consumer_data type");
707 assert(0);
708 goto end;
709 }
710 DBG("Allocated channel (key %d, shm_fd %d, wait_fd %d, mmap_len %llu, max_sb_size %llu)",
711 channel->key, channel->shm_fd, channel->wait_fd,
712 (unsigned long long) channel->mmap_len,
713 (unsigned long long) channel->max_sb_size);
714 end:
715 return channel;
716 }
717
718 /*
719 * Add a channel to the global list protected by a mutex.
720 */
721 int consumer_add_channel(struct lttng_consumer_channel *channel)
722 {
723 struct lttng_ht_node_ulong *node;
724 struct lttng_ht_iter iter;
725
726 pthread_mutex_lock(&consumer_data.lock);
727 /* Steal channel identifier, for UST */
728 consumer_steal_channel_key(channel->key);
729 rcu_read_lock();
730
731 lttng_ht_lookup(consumer_data.channel_ht,
732 (void *)((unsigned long) channel->key), &iter);
733 node = lttng_ht_iter_get_node_ulong(&iter);
734 if (node != NULL) {
735 /* Channel already exist. Ignore the insertion */
736 goto end;
737 }
738
739 lttng_ht_add_unique_ulong(consumer_data.channel_ht, &channel->node);
740
741 end:
742 rcu_read_unlock();
743 pthread_mutex_unlock(&consumer_data.lock);
744
745 return 0;
746 }
747
748 /*
749 * Allocate the pollfd structure and the local view of the out fds to avoid
750 * doing a lookup in the linked list and concurrency issues when writing is
751 * needed. Called with consumer_data.lock held.
752 *
753 * Returns the number of fds in the structures.
754 */
755 int consumer_update_poll_array(
756 struct lttng_consumer_local_data *ctx, struct pollfd **pollfd,
757 struct lttng_consumer_stream **local_stream)
758 {
759 int i = 0;
760 struct lttng_ht_iter iter;
761 struct lttng_consumer_stream *stream;
762
763 DBG("Updating poll fd array");
764 rcu_read_lock();
765 cds_lfht_for_each_entry(consumer_data.stream_ht->ht, &iter.iter, stream,
766 node.node) {
767 if (stream->state != LTTNG_CONSUMER_ACTIVE_STREAM) {
768 continue;
769 }
770 DBG("Active FD %d", stream->wait_fd);
771 (*pollfd)[i].fd = stream->wait_fd;
772 (*pollfd)[i].events = POLLIN | POLLPRI;
773 local_stream[i] = stream;
774 i++;
775 }
776 rcu_read_unlock();
777
778 /*
779 * Insert the consumer_poll_pipe at the end of the array and don't
780 * increment i so nb_fd is the number of real FD.
781 */
782 (*pollfd)[i].fd = ctx->consumer_poll_pipe[0];
783 (*pollfd)[i].events = POLLIN | POLLPRI;
784 return i;
785 }
786
787 /*
788 * Poll on the should_quit pipe and the command socket return -1 on error and
789 * should exit, 0 if data is available on the command socket
790 */
791 int lttng_consumer_poll_socket(struct pollfd *consumer_sockpoll)
792 {
793 int num_rdy;
794
795 restart:
796 num_rdy = poll(consumer_sockpoll, 2, -1);
797 if (num_rdy == -1) {
798 /*
799 * Restart interrupted system call.
800 */
801 if (errno == EINTR) {
802 goto restart;
803 }
804 perror("Poll error");
805 goto exit;
806 }
807 if (consumer_sockpoll[0].revents & (POLLIN | POLLPRI)) {
808 DBG("consumer_should_quit wake up");
809 goto exit;
810 }
811 return 0;
812
813 exit:
814 return -1;
815 }
816
817 /*
818 * Set the error socket.
819 */
820 void lttng_consumer_set_error_sock(
821 struct lttng_consumer_local_data *ctx, int sock)
822 {
823 ctx->consumer_error_socket = sock;
824 }
825
826 /*
827 * Set the command socket path.
828 */
829 void lttng_consumer_set_command_sock_path(
830 struct lttng_consumer_local_data *ctx, char *sock)
831 {
832 ctx->consumer_command_sock_path = sock;
833 }
834
835 /*
836 * Send return code to the session daemon.
837 * If the socket is not defined, we return 0, it is not a fatal error
838 */
839 int lttng_consumer_send_error(
840 struct lttng_consumer_local_data *ctx, int cmd)
841 {
842 if (ctx->consumer_error_socket > 0) {
843 return lttcomm_send_unix_sock(ctx->consumer_error_socket, &cmd,
844 sizeof(enum lttcomm_sessiond_command));
845 }
846
847 return 0;
848 }
849
850 /*
851 * Close all the tracefiles and stream fds, should be called when all instances
852 * are destroyed.
853 */
854 void lttng_consumer_cleanup(void)
855 {
856 struct lttng_ht_iter iter;
857 struct lttng_ht_node_ulong *node;
858
859 rcu_read_lock();
860
861 /*
862 * close all outfd. Called when there are no more threads running (after
863 * joining on the threads), no need to protect list iteration with mutex.
864 */
865 cds_lfht_for_each_entry(consumer_data.stream_ht->ht, &iter.iter, node,
866 node) {
867 struct lttng_consumer_stream *stream =
868 caa_container_of(node, struct lttng_consumer_stream, node);
869 consumer_del_stream(stream);
870 }
871
872 cds_lfht_for_each_entry(consumer_data.channel_ht->ht, &iter.iter, node,
873 node) {
874 struct lttng_consumer_channel *channel =
875 caa_container_of(node, struct lttng_consumer_channel, node);
876 consumer_del_channel(channel);
877 }
878
879 rcu_read_unlock();
880
881 lttng_ht_destroy(consumer_data.stream_ht);
882 lttng_ht_destroy(consumer_data.channel_ht);
883 }
884
885 /*
886 * Called from signal handler.
887 */
888 void lttng_consumer_should_exit(struct lttng_consumer_local_data *ctx)
889 {
890 int ret;
891 consumer_quit = 1;
892 do {
893 ret = write(ctx->consumer_should_quit[1], "4", 1);
894 } while (ret < 0 && errno == EINTR);
895 if (ret < 0) {
896 perror("write consumer quit");
897 }
898 }
899
900 void lttng_consumer_sync_trace_file(struct lttng_consumer_stream *stream,
901 off_t orig_offset)
902 {
903 int outfd = stream->out_fd;
904
905 /*
906 * This does a blocking write-and-wait on any page that belongs to the
907 * subbuffer prior to the one we just wrote.
908 * Don't care about error values, as these are just hints and ways to
909 * limit the amount of page cache used.
910 */
911 if (orig_offset < stream->chan->max_sb_size) {
912 return;
913 }
914 lttng_sync_file_range(outfd, orig_offset - stream->chan->max_sb_size,
915 stream->chan->max_sb_size,
916 SYNC_FILE_RANGE_WAIT_BEFORE
917 | SYNC_FILE_RANGE_WRITE
918 | SYNC_FILE_RANGE_WAIT_AFTER);
919 /*
920 * Give hints to the kernel about how we access the file:
921 * POSIX_FADV_DONTNEED : we won't re-access data in a near future after
922 * we write it.
923 *
924 * We need to call fadvise again after the file grows because the
925 * kernel does not seem to apply fadvise to non-existing parts of the
926 * file.
927 *
928 * Call fadvise _after_ having waited for the page writeback to
929 * complete because the dirty page writeback semantic is not well
930 * defined. So it can be expected to lead to lower throughput in
931 * streaming.
932 */
933 posix_fadvise(outfd, orig_offset - stream->chan->max_sb_size,
934 stream->chan->max_sb_size, POSIX_FADV_DONTNEED);
935 }
936
937 /*
938 * Initialise the necessary environnement :
939 * - create a new context
940 * - create the poll_pipe
941 * - create the should_quit pipe (for signal handler)
942 * - create the thread pipe (for splice)
943 *
944 * Takes a function pointer as argument, this function is called when data is
945 * available on a buffer. This function is responsible to do the
946 * kernctl_get_next_subbuf, read the data with mmap or splice depending on the
947 * buffer configuration and then kernctl_put_next_subbuf at the end.
948 *
949 * Returns a pointer to the new context or NULL on error.
950 */
951 struct lttng_consumer_local_data *lttng_consumer_create(
952 enum lttng_consumer_type type,
953 ssize_t (*buffer_ready)(struct lttng_consumer_stream *stream,
954 struct lttng_consumer_local_data *ctx),
955 int (*recv_channel)(struct lttng_consumer_channel *channel),
956 int (*recv_stream)(struct lttng_consumer_stream *stream),
957 int (*update_stream)(int stream_key, uint32_t state))
958 {
959 int ret, i;
960 struct lttng_consumer_local_data *ctx;
961
962 assert(consumer_data.type == LTTNG_CONSUMER_UNKNOWN ||
963 consumer_data.type == type);
964 consumer_data.type = type;
965
966 ctx = zmalloc(sizeof(struct lttng_consumer_local_data));
967 if (ctx == NULL) {
968 perror("allocating context");
969 goto error;
970 }
971
972 ctx->consumer_error_socket = -1;
973 /* assign the callbacks */
974 ctx->on_buffer_ready = buffer_ready;
975 ctx->on_recv_channel = recv_channel;
976 ctx->on_recv_stream = recv_stream;
977 ctx->on_update_stream = update_stream;
978
979 ret = pipe(ctx->consumer_poll_pipe);
980 if (ret < 0) {
981 perror("Error creating poll pipe");
982 goto error_poll_pipe;
983 }
984
985 /* set read end of the pipe to non-blocking */
986 ret = fcntl(ctx->consumer_poll_pipe[0], F_SETFL, O_NONBLOCK);
987 if (ret < 0) {
988 perror("fcntl O_NONBLOCK");
989 goto error_poll_fcntl;
990 }
991
992 /* set write end of the pipe to non-blocking */
993 ret = fcntl(ctx->consumer_poll_pipe[1], F_SETFL, O_NONBLOCK);
994 if (ret < 0) {
995 perror("fcntl O_NONBLOCK");
996 goto error_poll_fcntl;
997 }
998
999 ret = pipe(ctx->consumer_should_quit);
1000 if (ret < 0) {
1001 perror("Error creating recv pipe");
1002 goto error_quit_pipe;
1003 }
1004
1005 ret = pipe(ctx->consumer_thread_pipe);
1006 if (ret < 0) {
1007 perror("Error creating thread pipe");
1008 goto error_thread_pipe;
1009 }
1010
1011 ret = utils_create_pipe(ctx->consumer_metadata_pipe);
1012 if (ret < 0) {
1013 goto error_metadata_pipe;
1014 }
1015
1016 ret = utils_create_pipe(ctx->consumer_splice_metadata_pipe);
1017 if (ret < 0) {
1018 goto error_splice_pipe;
1019 }
1020
1021 return ctx;
1022
1023 error_splice_pipe:
1024 utils_close_pipe(ctx->consumer_metadata_pipe);
1025 error_metadata_pipe:
1026 utils_close_pipe(ctx->consumer_thread_pipe);
1027 error_thread_pipe:
1028 for (i = 0; i < 2; i++) {
1029 int err;
1030
1031 err = close(ctx->consumer_should_quit[i]);
1032 if (err) {
1033 PERROR("close");
1034 }
1035 }
1036 error_poll_fcntl:
1037 error_quit_pipe:
1038 for (i = 0; i < 2; i++) {
1039 int err;
1040
1041 err = close(ctx->consumer_poll_pipe[i]);
1042 if (err) {
1043 PERROR("close");
1044 }
1045 }
1046 error_poll_pipe:
1047 free(ctx);
1048 error:
1049 return NULL;
1050 }
1051
1052 /*
1053 * Close all fds associated with the instance and free the context.
1054 */
1055 void lttng_consumer_destroy(struct lttng_consumer_local_data *ctx)
1056 {
1057 int ret;
1058
1059 ret = close(ctx->consumer_error_socket);
1060 if (ret) {
1061 PERROR("close");
1062 }
1063 ret = close(ctx->consumer_thread_pipe[0]);
1064 if (ret) {
1065 PERROR("close");
1066 }
1067 ret = close(ctx->consumer_thread_pipe[1]);
1068 if (ret) {
1069 PERROR("close");
1070 }
1071 ret = close(ctx->consumer_poll_pipe[0]);
1072 if (ret) {
1073 PERROR("close");
1074 }
1075 ret = close(ctx->consumer_poll_pipe[1]);
1076 if (ret) {
1077 PERROR("close");
1078 }
1079 ret = close(ctx->consumer_should_quit[0]);
1080 if (ret) {
1081 PERROR("close");
1082 }
1083 ret = close(ctx->consumer_should_quit[1]);
1084 if (ret) {
1085 PERROR("close");
1086 }
1087 utils_close_pipe(ctx->consumer_splice_metadata_pipe);
1088
1089 unlink(ctx->consumer_command_sock_path);
1090 free(ctx);
1091 }
1092
1093 /*
1094 * Write the metadata stream id on the specified file descriptor.
1095 */
1096 static int write_relayd_metadata_id(int fd,
1097 struct lttng_consumer_stream *stream,
1098 struct consumer_relayd_sock_pair *relayd,
1099 unsigned long padding)
1100 {
1101 int ret;
1102 struct lttcomm_relayd_metadata_payload hdr;
1103
1104 hdr.stream_id = htobe64(stream->relayd_stream_id);
1105 hdr.padding_size = htobe32(padding);
1106 do {
1107 ret = write(fd, (void *) &hdr, sizeof(hdr));
1108 } while (ret < 0 && errno == EINTR);
1109 if (ret < 0) {
1110 PERROR("write metadata stream id");
1111 goto end;
1112 }
1113 DBG("Metadata stream id %" PRIu64 " with padding %lu written before data",
1114 stream->relayd_stream_id, padding);
1115
1116 end:
1117 return ret;
1118 }
1119
1120 /*
1121 * Mmap the ring buffer, read it and write the data to the tracefile. This is a
1122 * core function for writing trace buffers to either the local filesystem or
1123 * the network.
1124 *
1125 * Careful review MUST be put if any changes occur!
1126 *
1127 * Returns the number of bytes written
1128 */
1129 ssize_t lttng_consumer_on_read_subbuffer_mmap(
1130 struct lttng_consumer_local_data *ctx,
1131 struct lttng_consumer_stream *stream, unsigned long len,
1132 unsigned long padding)
1133 {
1134 unsigned long mmap_offset;
1135 ssize_t ret = 0, written = 0;
1136 off_t orig_offset = stream->out_fd_offset;
1137 /* Default is on the disk */
1138 int outfd = stream->out_fd;
1139 struct consumer_relayd_sock_pair *relayd = NULL;
1140
1141 /* RCU lock for the relayd pointer */
1142 rcu_read_lock();
1143
1144 /* Flag that the current stream if set for network streaming. */
1145 if (stream->net_seq_idx != -1) {
1146 relayd = consumer_find_relayd(stream->net_seq_idx);
1147 if (relayd == NULL) {
1148 goto end;
1149 }
1150 }
1151
1152 /* get the offset inside the fd to mmap */
1153 switch (consumer_data.type) {
1154 case LTTNG_CONSUMER_KERNEL:
1155 ret = kernctl_get_mmap_read_offset(stream->wait_fd, &mmap_offset);
1156 break;
1157 case LTTNG_CONSUMER32_UST:
1158 case LTTNG_CONSUMER64_UST:
1159 ret = lttng_ustctl_get_mmap_read_offset(stream->chan->handle,
1160 stream->buf, &mmap_offset);
1161 break;
1162 default:
1163 ERR("Unknown consumer_data type");
1164 assert(0);
1165 }
1166 if (ret != 0) {
1167 errno = -ret;
1168 PERROR("tracer ctl get_mmap_read_offset");
1169 written = ret;
1170 goto end;
1171 }
1172
1173 /* Handle stream on the relayd if the output is on the network */
1174 if (relayd) {
1175 unsigned long netlen = len;
1176
1177 /*
1178 * Lock the control socket for the complete duration of the function
1179 * since from this point on we will use the socket.
1180 */
1181 if (stream->metadata_flag) {
1182 /* Metadata requires the control socket. */
1183 pthread_mutex_lock(&relayd->ctrl_sock_mutex);
1184 netlen += sizeof(struct lttcomm_relayd_metadata_payload);
1185 }
1186
1187 ret = write_relayd_stream_header(stream, netlen, padding, relayd);
1188 if (ret >= 0) {
1189 /* Use the returned socket. */
1190 outfd = ret;
1191
1192 /* Write metadata stream id before payload */
1193 if (stream->metadata_flag) {
1194 ret = write_relayd_metadata_id(outfd, stream, relayd, padding);
1195 if (ret < 0) {
1196 written = ret;
1197 goto end;
1198 }
1199 }
1200 }
1201 /* Else, use the default set before which is the filesystem. */
1202 } else {
1203 /* No streaming, we have to set the len with the full padding */
1204 len += padding;
1205 }
1206
1207 while (len > 0) {
1208 do {
1209 ret = write(outfd, stream->mmap_base + mmap_offset, len);
1210 } while (ret < 0 && errno == EINTR);
1211 DBG("Consumer mmap write() ret %zd (len %lu)", ret, len);
1212 if (ret < 0) {
1213 PERROR("Error in file write");
1214 if (written == 0) {
1215 written = ret;
1216 }
1217 goto end;
1218 } else if (ret > len) {
1219 PERROR("Error in file write (ret %zd > len %lu)", ret, len);
1220 written += ret;
1221 goto end;
1222 } else {
1223 len -= ret;
1224 mmap_offset += ret;
1225 }
1226
1227 /* This call is useless on a socket so better save a syscall. */
1228 if (!relayd) {
1229 /* This won't block, but will start writeout asynchronously */
1230 lttng_sync_file_range(outfd, stream->out_fd_offset, ret,
1231 SYNC_FILE_RANGE_WRITE);
1232 stream->out_fd_offset += ret;
1233 }
1234 written += ret;
1235 }
1236 lttng_consumer_sync_trace_file(stream, orig_offset);
1237
1238 end:
1239 /* Unlock only if ctrl socket used */
1240 if (relayd && stream->metadata_flag) {
1241 pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
1242 }
1243
1244 rcu_read_unlock();
1245 return written;
1246 }
1247
1248 /*
1249 * Splice the data from the ring buffer to the tracefile.
1250 *
1251 * Returns the number of bytes spliced.
1252 */
1253 ssize_t lttng_consumer_on_read_subbuffer_splice(
1254 struct lttng_consumer_local_data *ctx,
1255 struct lttng_consumer_stream *stream, unsigned long len,
1256 unsigned long padding)
1257 {
1258 ssize_t ret = 0, written = 0, ret_splice = 0;
1259 loff_t offset = 0;
1260 off_t orig_offset = stream->out_fd_offset;
1261 int fd = stream->wait_fd;
1262 /* Default is on the disk */
1263 int outfd = stream->out_fd;
1264 struct consumer_relayd_sock_pair *relayd = NULL;
1265 int *splice_pipe;
1266
1267 switch (consumer_data.type) {
1268 case LTTNG_CONSUMER_KERNEL:
1269 break;
1270 case LTTNG_CONSUMER32_UST:
1271 case LTTNG_CONSUMER64_UST:
1272 /* Not supported for user space tracing */
1273 return -ENOSYS;
1274 default:
1275 ERR("Unknown consumer_data type");
1276 assert(0);
1277 }
1278
1279 /* RCU lock for the relayd pointer */
1280 rcu_read_lock();
1281
1282 /* Flag that the current stream if set for network streaming. */
1283 if (stream->net_seq_idx != -1) {
1284 relayd = consumer_find_relayd(stream->net_seq_idx);
1285 if (relayd == NULL) {
1286 goto end;
1287 }
1288 }
1289
1290 /*
1291 * Choose right pipe for splice. Metadata and trace data are handled by
1292 * different threads hence the use of two pipes in order not to race or
1293 * corrupt the written data.
1294 */
1295 if (stream->metadata_flag) {
1296 splice_pipe = ctx->consumer_splice_metadata_pipe;
1297 } else {
1298 splice_pipe = ctx->consumer_thread_pipe;
1299 }
1300
1301 /* Write metadata stream id before payload */
1302 if (relayd) {
1303 int total_len = len;
1304
1305 if (stream->metadata_flag) {
1306 /*
1307 * Lock the control socket for the complete duration of the function
1308 * since from this point on we will use the socket.
1309 */
1310 pthread_mutex_lock(&relayd->ctrl_sock_mutex);
1311
1312 ret = write_relayd_metadata_id(splice_pipe[1], stream, relayd,
1313 padding);
1314 if (ret < 0) {
1315 written = ret;
1316 goto end;
1317 }
1318
1319 total_len += sizeof(struct lttcomm_relayd_metadata_payload);
1320 }
1321
1322 ret = write_relayd_stream_header(stream, total_len, padding, relayd);
1323 if (ret >= 0) {
1324 /* Use the returned socket. */
1325 outfd = ret;
1326 } else {
1327 ERR("Remote relayd disconnected. Stopping");
1328 goto end;
1329 }
1330 } else {
1331 /* No streaming, we have to set the len with the full padding */
1332 len += padding;
1333 }
1334
1335 while (len > 0) {
1336 DBG("splice chan to pipe offset %lu of len %lu (fd : %d, pipe: %d)",
1337 (unsigned long)offset, len, fd, splice_pipe[1]);
1338 ret_splice = splice(fd, &offset, splice_pipe[1], NULL, len,
1339 SPLICE_F_MOVE | SPLICE_F_MORE);
1340 DBG("splice chan to pipe, ret %zd", ret_splice);
1341 if (ret_splice < 0) {
1342 PERROR("Error in relay splice");
1343 if (written == 0) {
1344 written = ret_splice;
1345 }
1346 ret = errno;
1347 goto splice_error;
1348 }
1349
1350 /* Handle stream on the relayd if the output is on the network */
1351 if (relayd) {
1352 if (stream->metadata_flag) {
1353 size_t metadata_payload_size =
1354 sizeof(struct lttcomm_relayd_metadata_payload);
1355
1356 /* Update counter to fit the spliced data */
1357 ret_splice += metadata_payload_size;
1358 len += metadata_payload_size;
1359 /*
1360 * We do this so the return value can match the len passed as
1361 * argument to this function.
1362 */
1363 written -= metadata_payload_size;
1364 }
1365 }
1366
1367 /* Splice data out */
1368 ret_splice = splice(splice_pipe[0], NULL, outfd, NULL,
1369 ret_splice, SPLICE_F_MOVE | SPLICE_F_MORE);
1370 DBG("Consumer splice pipe to file, ret %zd", ret_splice);
1371 if (ret_splice < 0) {
1372 PERROR("Error in file splice");
1373 if (written == 0) {
1374 written = ret_splice;
1375 }
1376 ret = errno;
1377 goto splice_error;
1378 } else if (ret_splice > len) {
1379 errno = EINVAL;
1380 PERROR("Wrote more data than requested %zd (len: %lu)",
1381 ret_splice, len);
1382 written += ret_splice;
1383 ret = errno;
1384 goto splice_error;
1385 }
1386 len -= ret_splice;
1387
1388 /* This call is useless on a socket so better save a syscall. */
1389 if (!relayd) {
1390 /* This won't block, but will start writeout asynchronously */
1391 lttng_sync_file_range(outfd, stream->out_fd_offset, ret_splice,
1392 SYNC_FILE_RANGE_WRITE);
1393 stream->out_fd_offset += ret_splice;
1394 }
1395 written += ret_splice;
1396 }
1397 lttng_consumer_sync_trace_file(stream, orig_offset);
1398
1399 ret = ret_splice;
1400
1401 goto end;
1402
1403 splice_error:
1404 /* send the appropriate error description to sessiond */
1405 switch (ret) {
1406 case EBADF:
1407 lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_SPLICE_EBADF);
1408 break;
1409 case EINVAL:
1410 lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_SPLICE_EINVAL);
1411 break;
1412 case ENOMEM:
1413 lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_SPLICE_ENOMEM);
1414 break;
1415 case ESPIPE:
1416 lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_SPLICE_ESPIPE);
1417 break;
1418 }
1419
1420 end:
1421 if (relayd && stream->metadata_flag) {
1422 pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
1423 }
1424
1425 rcu_read_unlock();
1426 return written;
1427 }
1428
1429 /*
1430 * Take a snapshot for a specific fd
1431 *
1432 * Returns 0 on success, < 0 on error
1433 */
1434 int lttng_consumer_take_snapshot(struct lttng_consumer_local_data *ctx,
1435 struct lttng_consumer_stream *stream)
1436 {
1437 switch (consumer_data.type) {
1438 case LTTNG_CONSUMER_KERNEL:
1439 return lttng_kconsumer_take_snapshot(ctx, stream);
1440 case LTTNG_CONSUMER32_UST:
1441 case LTTNG_CONSUMER64_UST:
1442 return lttng_ustconsumer_take_snapshot(ctx, stream);
1443 default:
1444 ERR("Unknown consumer_data type");
1445 assert(0);
1446 return -ENOSYS;
1447 }
1448
1449 }
1450
1451 /*
1452 * Get the produced position
1453 *
1454 * Returns 0 on success, < 0 on error
1455 */
1456 int lttng_consumer_get_produced_snapshot(
1457 struct lttng_consumer_local_data *ctx,
1458 struct lttng_consumer_stream *stream,
1459 unsigned long *pos)
1460 {
1461 switch (consumer_data.type) {
1462 case LTTNG_CONSUMER_KERNEL:
1463 return lttng_kconsumer_get_produced_snapshot(ctx, stream, pos);
1464 case LTTNG_CONSUMER32_UST:
1465 case LTTNG_CONSUMER64_UST:
1466 return lttng_ustconsumer_get_produced_snapshot(ctx, stream, pos);
1467 default:
1468 ERR("Unknown consumer_data type");
1469 assert(0);
1470 return -ENOSYS;
1471 }
1472 }
1473
1474 int lttng_consumer_recv_cmd(struct lttng_consumer_local_data *ctx,
1475 int sock, struct pollfd *consumer_sockpoll)
1476 {
1477 switch (consumer_data.type) {
1478 case LTTNG_CONSUMER_KERNEL:
1479 return lttng_kconsumer_recv_cmd(ctx, sock, consumer_sockpoll);
1480 case LTTNG_CONSUMER32_UST:
1481 case LTTNG_CONSUMER64_UST:
1482 return lttng_ustconsumer_recv_cmd(ctx, sock, consumer_sockpoll);
1483 default:
1484 ERR("Unknown consumer_data type");
1485 assert(0);
1486 return -ENOSYS;
1487 }
1488 }
1489
1490 /*
1491 * Iterate over all stream element of the hashtable and free them. This is race
1492 * free since the hashtable received MUST be in a race free synchronization
1493 * state. It's the caller responsability to make sure of that.
1494 */
1495 static void destroy_stream_ht(struct lttng_ht *ht)
1496 {
1497 int ret;
1498 struct lttng_ht_iter iter;
1499 struct lttng_consumer_stream *stream;
1500
1501 if (ht == NULL) {
1502 return;
1503 }
1504
1505 rcu_read_lock();
1506 cds_lfht_for_each_entry(ht->ht, &iter.iter, stream, node.node) {
1507 ret = lttng_ht_del(ht, &iter);
1508 assert(!ret);
1509
1510 free(stream);
1511 }
1512 rcu_read_unlock();
1513
1514 lttng_ht_destroy(ht);
1515 }
1516
1517 /*
1518 * Clean up a metadata stream and free its memory.
1519 */
1520 static void consumer_del_metadata_stream(struct lttng_consumer_stream *stream)
1521 {
1522 int ret;
1523 struct lttng_consumer_channel *free_chan = NULL;
1524 struct consumer_relayd_sock_pair *relayd;
1525
1526 assert(stream);
1527 /*
1528 * This call should NEVER receive regular stream. It must always be
1529 * metadata stream and this is crucial for data structure synchronization.
1530 */
1531 assert(stream->metadata_flag);
1532
1533 pthread_mutex_lock(&consumer_data.lock);
1534 switch (consumer_data.type) {
1535 case LTTNG_CONSUMER_KERNEL:
1536 if (stream->mmap_base != NULL) {
1537 ret = munmap(stream->mmap_base, stream->mmap_len);
1538 if (ret != 0) {
1539 PERROR("munmap metadata stream");
1540 }
1541 }
1542 break;
1543 case LTTNG_CONSUMER32_UST:
1544 case LTTNG_CONSUMER64_UST:
1545 lttng_ustconsumer_del_stream(stream);
1546 break;
1547 default:
1548 ERR("Unknown consumer_data type");
1549 assert(0);
1550 }
1551 pthread_mutex_unlock(&consumer_data.lock);
1552
1553 if (stream->out_fd >= 0) {
1554 ret = close(stream->out_fd);
1555 if (ret) {
1556 PERROR("close");
1557 }
1558 }
1559
1560 if (stream->wait_fd >= 0 && !stream->wait_fd_is_copy) {
1561 ret = close(stream->wait_fd);
1562 if (ret) {
1563 PERROR("close");
1564 }
1565 }
1566
1567 if (stream->shm_fd >= 0 && stream->wait_fd != stream->shm_fd) {
1568 ret = close(stream->shm_fd);
1569 if (ret) {
1570 PERROR("close");
1571 }
1572 }
1573
1574 /* Check and cleanup relayd */
1575 rcu_read_lock();
1576 relayd = consumer_find_relayd(stream->net_seq_idx);
1577 if (relayd != NULL) {
1578 uatomic_dec(&relayd->refcount);
1579 assert(uatomic_read(&relayd->refcount) >= 0);
1580
1581 /* Closing streams requires to lock the control socket. */
1582 pthread_mutex_lock(&relayd->ctrl_sock_mutex);
1583 ret = relayd_send_close_stream(&relayd->control_sock,
1584 stream->relayd_stream_id, stream->next_net_seq_num - 1);
1585 pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
1586 if (ret < 0) {
1587 DBG("Unable to close stream on the relayd. Continuing");
1588 /*
1589 * Continue here. There is nothing we can do for the relayd.
1590 * Chances are that the relayd has closed the socket so we just
1591 * continue cleaning up.
1592 */
1593 }
1594
1595 /* Both conditions are met, we destroy the relayd. */
1596 if (uatomic_read(&relayd->refcount) == 0 &&
1597 uatomic_read(&relayd->destroy_flag)) {
1598 destroy_relayd(relayd);
1599 }
1600 }
1601 rcu_read_unlock();
1602
1603 /* Atomically decrement channel refcount since other threads can use it. */
1604 uatomic_dec(&stream->chan->refcount);
1605 if (!uatomic_read(&stream->chan->refcount)) {
1606 free_chan = stream->chan;
1607 }
1608
1609 if (free_chan) {
1610 consumer_del_channel(free_chan);
1611 }
1612
1613 free(stream);
1614 }
1615
1616 /*
1617 * Action done with the metadata stream when adding it to the consumer internal
1618 * data structures to handle it.
1619 */
1620 static void consumer_add_metadata_stream(struct lttng_consumer_stream *stream)
1621 {
1622 struct consumer_relayd_sock_pair *relayd;
1623
1624 /* Find relayd and, if one is found, increment refcount. */
1625 rcu_read_lock();
1626 relayd = consumer_find_relayd(stream->net_seq_idx);
1627 if (relayd != NULL) {
1628 uatomic_inc(&relayd->refcount);
1629 }
1630 rcu_read_unlock();
1631 }
1632
1633 /*
1634 * Thread polls on metadata file descriptor and write them on disk or on the
1635 * network.
1636 */
1637 void *lttng_consumer_thread_poll_metadata(void *data)
1638 {
1639 int ret, i, pollfd;
1640 uint32_t revents, nb_fd;
1641 struct lttng_consumer_stream *stream;
1642 struct lttng_ht_iter iter;
1643 struct lttng_ht_node_ulong *node;
1644 struct lttng_ht *metadata_ht = NULL;
1645 struct lttng_poll_event events;
1646 struct lttng_consumer_local_data *ctx = data;
1647 ssize_t len;
1648
1649 rcu_register_thread();
1650
1651 DBG("Thread metadata poll started");
1652
1653 metadata_ht = lttng_ht_new(0, LTTNG_HT_TYPE_ULONG);
1654 if (metadata_ht == NULL) {
1655 goto end;
1656 }
1657
1658 /* Size is set to 1 for the consumer_metadata pipe */
1659 ret = lttng_poll_create(&events, 2, LTTNG_CLOEXEC);
1660 if (ret < 0) {
1661 ERR("Poll set creation failed");
1662 goto end;
1663 }
1664
1665 ret = lttng_poll_add(&events, ctx->consumer_metadata_pipe[0], LPOLLIN);
1666 if (ret < 0) {
1667 goto end;
1668 }
1669
1670 /* Main loop */
1671 DBG("Metadata main loop started");
1672
1673 while (1) {
1674 lttng_poll_reset(&events);
1675
1676 nb_fd = LTTNG_POLL_GETNB(&events);
1677
1678 /* Only the metadata pipe is set */
1679 if (nb_fd == 0 && consumer_quit == 1) {
1680 goto end;
1681 }
1682
1683 restart:
1684 DBG("Metadata poll wait with %d fd(s)", nb_fd);
1685 ret = lttng_poll_wait(&events, -1);
1686 DBG("Metadata event catched in thread");
1687 if (ret < 0) {
1688 if (errno == EINTR) {
1689 goto restart;
1690 }
1691 goto error;
1692 }
1693
1694 for (i = 0; i < nb_fd; i++) {
1695 revents = LTTNG_POLL_GETEV(&events, i);
1696 pollfd = LTTNG_POLL_GETFD(&events, i);
1697
1698 /* Check the metadata pipe for incoming metadata. */
1699 if (pollfd == ctx->consumer_metadata_pipe[0]) {
1700 if (revents & (LPOLLERR | LPOLLHUP )) {
1701 DBG("Metadata thread pipe hung up");
1702 /*
1703 * Remove the pipe from the poll set and continue the loop
1704 * since their might be data to consume.
1705 */
1706 lttng_poll_del(&events, ctx->consumer_metadata_pipe[0]);
1707 close(ctx->consumer_metadata_pipe[0]);
1708 continue;
1709 } else if (revents & LPOLLIN) {
1710 stream = zmalloc(sizeof(struct lttng_consumer_stream));
1711 if (stream == NULL) {
1712 PERROR("zmalloc metadata consumer stream");
1713 goto error;
1714 }
1715
1716 do {
1717 /* Get the stream and add it to the local hash table */
1718 ret = read(pollfd, stream,
1719 sizeof(struct lttng_consumer_stream));
1720 } while (ret < 0 && errno == EINTR);
1721 if (ret < 0 || ret < sizeof(struct lttng_consumer_stream)) {
1722 PERROR("read metadata stream");
1723 free(stream);
1724 /*
1725 * Let's continue here and hope we can still work
1726 * without stopping the consumer. XXX: Should we?
1727 */
1728 continue;
1729 }
1730
1731 DBG("Adding metadata stream %d to poll set",
1732 stream->wait_fd);
1733
1734 rcu_read_lock();
1735 /* The node should be init at this point */
1736 lttng_ht_add_unique_ulong(metadata_ht,
1737 &stream->waitfd_node);
1738 rcu_read_unlock();
1739
1740 /* Add metadata stream to the global poll events list */
1741 lttng_poll_add(&events, stream->wait_fd,
1742 LPOLLIN | LPOLLPRI);
1743
1744 consumer_add_metadata_stream(stream);
1745 }
1746
1747 /* Metadata pipe handled. Continue handling the others */
1748 continue;
1749 }
1750
1751 /* From here, the event is a metadata wait fd */
1752
1753 rcu_read_lock();
1754 lttng_ht_lookup(metadata_ht, (void *)((unsigned long) pollfd),
1755 &iter);
1756 node = lttng_ht_iter_get_node_ulong(&iter);
1757 if (node == NULL) {
1758 /* FD not found, continue loop */
1759 rcu_read_unlock();
1760 continue;
1761 }
1762
1763 stream = caa_container_of(node, struct lttng_consumer_stream,
1764 waitfd_node);
1765
1766 /* Get the data out of the metadata file descriptor */
1767 if (revents & (LPOLLIN | LPOLLPRI)) {
1768 DBG("Metadata available on fd %d", pollfd);
1769 assert(stream->wait_fd == pollfd);
1770
1771 len = ctx->on_buffer_ready(stream, ctx);
1772 /* It's ok to have an unavailable sub-buffer */
1773 if (len < 0 && len != -EAGAIN) {
1774 rcu_read_unlock();
1775 goto end;
1776 } else if (len > 0) {
1777 stream->data_read = 1;
1778 }
1779 }
1780
1781 /*
1782 * Remove the stream from the hash table since there is no data
1783 * left on the fd because we previously did a read on the buffer.
1784 */
1785 if (revents & (LPOLLERR | LPOLLHUP)) {
1786 DBG("Metadata fd %d is hup|err|nval.", pollfd);
1787 if (!stream->hangup_flush_done
1788 && (consumer_data.type == LTTNG_CONSUMER32_UST
1789 || consumer_data.type == LTTNG_CONSUMER64_UST)) {
1790 DBG("Attempting to flush and consume the UST buffers");
1791 lttng_ustconsumer_on_stream_hangup(stream);
1792
1793 /* We just flushed the stream now read it. */
1794 len = ctx->on_buffer_ready(stream, ctx);
1795 /* It's ok to have an unavailable sub-buffer */
1796 if (len < 0 && len != -EAGAIN) {
1797 rcu_read_unlock();
1798 goto end;
1799 }
1800 }
1801
1802 /* Removing it from hash table, poll set and free memory */
1803 lttng_ht_del(metadata_ht, &iter);
1804
1805 lttng_poll_del(&events, stream->wait_fd);
1806 consumer_del_metadata_stream(stream);
1807 }
1808 rcu_read_unlock();
1809 }
1810 }
1811
1812 error:
1813 end:
1814 DBG("Metadata poll thread exiting");
1815 lttng_poll_clean(&events);
1816
1817 if (metadata_ht) {
1818 destroy_stream_ht(metadata_ht);
1819 }
1820
1821 rcu_unregister_thread();
1822 return NULL;
1823 }
1824
1825 /*
1826 * This thread polls the fds in the set to consume the data and write
1827 * it to tracefile if necessary.
1828 */
1829 void *lttng_consumer_thread_poll_fds(void *data)
1830 {
1831 int num_rdy, num_hup, high_prio, ret, i;
1832 struct pollfd *pollfd = NULL;
1833 /* local view of the streams */
1834 struct lttng_consumer_stream **local_stream = NULL;
1835 /* local view of consumer_data.fds_count */
1836 int nb_fd = 0;
1837 struct lttng_consumer_local_data *ctx = data;
1838 ssize_t len;
1839 pthread_t metadata_thread;
1840 void *status;
1841
1842 rcu_register_thread();
1843
1844 /* Start metadata polling thread */
1845 ret = pthread_create(&metadata_thread, NULL,
1846 lttng_consumer_thread_poll_metadata, (void *) ctx);
1847 if (ret < 0) {
1848 PERROR("pthread_create metadata thread");
1849 goto end;
1850 }
1851
1852 local_stream = zmalloc(sizeof(struct lttng_consumer_stream));
1853
1854 while (1) {
1855 high_prio = 0;
1856 num_hup = 0;
1857
1858 /*
1859 * the fds set has been updated, we need to update our
1860 * local array as well
1861 */
1862 pthread_mutex_lock(&consumer_data.lock);
1863 if (consumer_data.need_update) {
1864 if (pollfd != NULL) {
1865 free(pollfd);
1866 pollfd = NULL;
1867 }
1868 if (local_stream != NULL) {
1869 free(local_stream);
1870 local_stream = NULL;
1871 }
1872
1873 /* allocate for all fds + 1 for the consumer_poll_pipe */
1874 pollfd = zmalloc((consumer_data.stream_count + 1) * sizeof(struct pollfd));
1875 if (pollfd == NULL) {
1876 perror("pollfd malloc");
1877 pthread_mutex_unlock(&consumer_data.lock);
1878 goto end;
1879 }
1880
1881 /* allocate for all fds + 1 for the consumer_poll_pipe */
1882 local_stream = zmalloc((consumer_data.stream_count + 1) *
1883 sizeof(struct lttng_consumer_stream));
1884 if (local_stream == NULL) {
1885 perror("local_stream malloc");
1886 pthread_mutex_unlock(&consumer_data.lock);
1887 goto end;
1888 }
1889 ret = consumer_update_poll_array(ctx, &pollfd, local_stream);
1890 if (ret < 0) {
1891 ERR("Error in allocating pollfd or local_outfds");
1892 lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_POLL_ERROR);
1893 pthread_mutex_unlock(&consumer_data.lock);
1894 goto end;
1895 }
1896 nb_fd = ret;
1897 consumer_data.need_update = 0;
1898 }
1899 pthread_mutex_unlock(&consumer_data.lock);
1900
1901 /* No FDs and consumer_quit, consumer_cleanup the thread */
1902 if (nb_fd == 0 && consumer_quit == 1) {
1903 goto end;
1904 }
1905 /* poll on the array of fds */
1906 restart:
1907 DBG("polling on %d fd", nb_fd + 1);
1908 num_rdy = poll(pollfd, nb_fd + 1, consumer_poll_timeout);
1909 DBG("poll num_rdy : %d", num_rdy);
1910 if (num_rdy == -1) {
1911 /*
1912 * Restart interrupted system call.
1913 */
1914 if (errno == EINTR) {
1915 goto restart;
1916 }
1917 perror("Poll error");
1918 lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_POLL_ERROR);
1919 goto end;
1920 } else if (num_rdy == 0) {
1921 DBG("Polling thread timed out");
1922 goto end;
1923 }
1924
1925 /*
1926 * If the consumer_poll_pipe triggered poll go directly to the
1927 * beginning of the loop to update the array. We want to prioritize
1928 * array update over low-priority reads.
1929 */
1930 if (pollfd[nb_fd].revents & (POLLIN | POLLPRI)) {
1931 size_t pipe_readlen;
1932 char tmp;
1933
1934 DBG("consumer_poll_pipe wake up");
1935 /* Consume 1 byte of pipe data */
1936 do {
1937 pipe_readlen = read(ctx->consumer_poll_pipe[0], &tmp, 1);
1938 } while (pipe_readlen == -1 && errno == EINTR);
1939 continue;
1940 }
1941
1942 /* Take care of high priority channels first. */
1943 for (i = 0; i < nb_fd; i++) {
1944 if (pollfd[i].revents & POLLPRI) {
1945 DBG("Urgent read on fd %d", pollfd[i].fd);
1946 high_prio = 1;
1947 len = ctx->on_buffer_ready(local_stream[i], ctx);
1948 /* it's ok to have an unavailable sub-buffer */
1949 if (len < 0 && len != -EAGAIN) {
1950 goto end;
1951 } else if (len > 0) {
1952 local_stream[i]->data_read = 1;
1953 }
1954 }
1955 }
1956
1957 /*
1958 * If we read high prio channel in this loop, try again
1959 * for more high prio data.
1960 */
1961 if (high_prio) {
1962 continue;
1963 }
1964
1965 /* Take care of low priority channels. */
1966 for (i = 0; i < nb_fd; i++) {
1967 if ((pollfd[i].revents & POLLIN) ||
1968 local_stream[i]->hangup_flush_done) {
1969 DBG("Normal read on fd %d", pollfd[i].fd);
1970 len = ctx->on_buffer_ready(local_stream[i], ctx);
1971 /* it's ok to have an unavailable sub-buffer */
1972 if (len < 0 && len != -EAGAIN) {
1973 goto end;
1974 } else if (len > 0) {
1975 local_stream[i]->data_read = 1;
1976 }
1977 }
1978 }
1979
1980 /* Handle hangup and errors */
1981 for (i = 0; i < nb_fd; i++) {
1982 if (!local_stream[i]->hangup_flush_done
1983 && (pollfd[i].revents & (POLLHUP | POLLERR | POLLNVAL))
1984 && (consumer_data.type == LTTNG_CONSUMER32_UST
1985 || consumer_data.type == LTTNG_CONSUMER64_UST)) {
1986 DBG("fd %d is hup|err|nval. Attempting flush and read.",
1987 pollfd[i].fd);
1988 lttng_ustconsumer_on_stream_hangup(local_stream[i]);
1989 /* Attempt read again, for the data we just flushed. */
1990 local_stream[i]->data_read = 1;
1991 }
1992 /*
1993 * If the poll flag is HUP/ERR/NVAL and we have
1994 * read no data in this pass, we can remove the
1995 * stream from its hash table.
1996 */
1997 if ((pollfd[i].revents & POLLHUP)) {
1998 DBG("Polling fd %d tells it has hung up.", pollfd[i].fd);
1999 if (!local_stream[i]->data_read) {
2000 consumer_del_stream(local_stream[i]);
2001 num_hup++;
2002 }
2003 } else if (pollfd[i].revents & POLLERR) {
2004 ERR("Error returned in polling fd %d.", pollfd[i].fd);
2005 if (!local_stream[i]->data_read) {
2006 consumer_del_stream(local_stream[i]);
2007 num_hup++;
2008 }
2009 } else if (pollfd[i].revents & POLLNVAL) {
2010 ERR("Polling fd %d tells fd is not open.", pollfd[i].fd);
2011 if (!local_stream[i]->data_read) {
2012 consumer_del_stream(local_stream[i]);
2013 num_hup++;
2014 }
2015 }
2016 local_stream[i]->data_read = 0;
2017 }
2018 }
2019 end:
2020 DBG("polling thread exiting");
2021 if (pollfd != NULL) {
2022 free(pollfd);
2023 pollfd = NULL;
2024 }
2025 if (local_stream != NULL) {
2026 free(local_stream);
2027 local_stream = NULL;
2028 }
2029
2030 /*
2031 * Close the write side of the pipe so epoll_wait() in
2032 * lttng_consumer_thread_poll_metadata can catch it. The thread is
2033 * monitoring the read side of the pipe. If we close them both, epoll_wait
2034 * strangely does not return and could create a endless wait period if the
2035 * pipe is the only tracked fd in the poll set. The thread will take care
2036 * of closing the read side.
2037 */
2038 close(ctx->consumer_metadata_pipe[1]);
2039 if (ret) {
2040 ret = pthread_join(metadata_thread, &status);
2041 if (ret < 0) {
2042 PERROR("pthread_join metadata thread");
2043 }
2044 }
2045
2046 rcu_unregister_thread();
2047 return NULL;
2048 }
2049
2050 /*
2051 * This thread listens on the consumerd socket and receives the file
2052 * descriptors from the session daemon.
2053 */
2054 void *lttng_consumer_thread_receive_fds(void *data)
2055 {
2056 int sock, client_socket, ret;
2057 /*
2058 * structure to poll for incoming data on communication socket avoids
2059 * making blocking sockets.
2060 */
2061 struct pollfd consumer_sockpoll[2];
2062 struct lttng_consumer_local_data *ctx = data;
2063
2064 rcu_register_thread();
2065
2066 DBG("Creating command socket %s", ctx->consumer_command_sock_path);
2067 unlink(ctx->consumer_command_sock_path);
2068 client_socket = lttcomm_create_unix_sock(ctx->consumer_command_sock_path);
2069 if (client_socket < 0) {
2070 ERR("Cannot create command socket");
2071 goto end;
2072 }
2073
2074 ret = lttcomm_listen_unix_sock(client_socket);
2075 if (ret < 0) {
2076 goto end;
2077 }
2078
2079 DBG("Sending ready command to lttng-sessiond");
2080 ret = lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_COMMAND_SOCK_READY);
2081 /* return < 0 on error, but == 0 is not fatal */
2082 if (ret < 0) {
2083 ERR("Error sending ready command to lttng-sessiond");
2084 goto end;
2085 }
2086
2087 ret = fcntl(client_socket, F_SETFL, O_NONBLOCK);
2088 if (ret < 0) {
2089 perror("fcntl O_NONBLOCK");
2090 goto end;
2091 }
2092
2093 /* prepare the FDs to poll : to client socket and the should_quit pipe */
2094 consumer_sockpoll[0].fd = ctx->consumer_should_quit[0];
2095 consumer_sockpoll[0].events = POLLIN | POLLPRI;
2096 consumer_sockpoll[1].fd = client_socket;
2097 consumer_sockpoll[1].events = POLLIN | POLLPRI;
2098
2099 if (lttng_consumer_poll_socket(consumer_sockpoll) < 0) {
2100 goto end;
2101 }
2102 DBG("Connection on client_socket");
2103
2104 /* Blocking call, waiting for transmission */
2105 sock = lttcomm_accept_unix_sock(client_socket);
2106 if (sock <= 0) {
2107 WARN("On accept");
2108 goto end;
2109 }
2110 ret = fcntl(sock, F_SETFL, O_NONBLOCK);
2111 if (ret < 0) {
2112 perror("fcntl O_NONBLOCK");
2113 goto end;
2114 }
2115
2116 /* update the polling structure to poll on the established socket */
2117 consumer_sockpoll[1].fd = sock;
2118 consumer_sockpoll[1].events = POLLIN | POLLPRI;
2119
2120 while (1) {
2121 if (lttng_consumer_poll_socket(consumer_sockpoll) < 0) {
2122 goto end;
2123 }
2124 DBG("Incoming command on sock");
2125 ret = lttng_consumer_recv_cmd(ctx, sock, consumer_sockpoll);
2126 if (ret == -ENOENT) {
2127 DBG("Received STOP command");
2128 goto end;
2129 }
2130 if (ret <= 0) {
2131 /*
2132 * This could simply be a session daemon quitting. Don't output
2133 * ERR() here.
2134 */
2135 DBG("Communication interrupted on command socket");
2136 goto end;
2137 }
2138 if (consumer_quit) {
2139 DBG("consumer_thread_receive_fds received quit from signal");
2140 goto end;
2141 }
2142 DBG("received fds on sock");
2143 }
2144 end:
2145 DBG("consumer_thread_receive_fds exiting");
2146
2147 /*
2148 * when all fds have hung up, the polling thread
2149 * can exit cleanly
2150 */
2151 consumer_quit = 1;
2152
2153 /*
2154 * 2s of grace period, if no polling events occur during
2155 * this period, the polling thread will exit even if there
2156 * are still open FDs (should not happen, but safety mechanism).
2157 */
2158 consumer_poll_timeout = LTTNG_CONSUMER_POLL_TIMEOUT;
2159
2160 /*
2161 * Wake-up the other end by writing a null byte in the pipe
2162 * (non-blocking). Important note: Because writing into the
2163 * pipe is non-blocking (and therefore we allow dropping wakeup
2164 * data, as long as there is wakeup data present in the pipe
2165 * buffer to wake up the other end), the other end should
2166 * perform the following sequence for waiting:
2167 * 1) empty the pipe (reads).
2168 * 2) perform update operation.
2169 * 3) wait on the pipe (poll).
2170 */
2171 do {
2172 ret = write(ctx->consumer_poll_pipe[1], "", 1);
2173 } while (ret < 0 && errno == EINTR);
2174 rcu_unregister_thread();
2175 return NULL;
2176 }
2177
2178 ssize_t lttng_consumer_read_subbuffer(struct lttng_consumer_stream *stream,
2179 struct lttng_consumer_local_data *ctx)
2180 {
2181 switch (consumer_data.type) {
2182 case LTTNG_CONSUMER_KERNEL:
2183 return lttng_kconsumer_read_subbuffer(stream, ctx);
2184 case LTTNG_CONSUMER32_UST:
2185 case LTTNG_CONSUMER64_UST:
2186 return lttng_ustconsumer_read_subbuffer(stream, ctx);
2187 default:
2188 ERR("Unknown consumer_data type");
2189 assert(0);
2190 return -ENOSYS;
2191 }
2192 }
2193
2194 int lttng_consumer_on_recv_stream(struct lttng_consumer_stream *stream)
2195 {
2196 switch (consumer_data.type) {
2197 case LTTNG_CONSUMER_KERNEL:
2198 return lttng_kconsumer_on_recv_stream(stream);
2199 case LTTNG_CONSUMER32_UST:
2200 case LTTNG_CONSUMER64_UST:
2201 return lttng_ustconsumer_on_recv_stream(stream);
2202 default:
2203 ERR("Unknown consumer_data type");
2204 assert(0);
2205 return -ENOSYS;
2206 }
2207 }
2208
2209 /*
2210 * Allocate and set consumer data hash tables.
2211 */
2212 void lttng_consumer_init(void)
2213 {
2214 consumer_data.stream_ht = lttng_ht_new(0, LTTNG_HT_TYPE_ULONG);
2215 consumer_data.channel_ht = lttng_ht_new(0, LTTNG_HT_TYPE_ULONG);
2216 consumer_data.relayd_ht = lttng_ht_new(0, LTTNG_HT_TYPE_ULONG);
2217 }
2218
2219 /*
2220 * Process the ADD_RELAYD command receive by a consumer.
2221 *
2222 * This will create a relayd socket pair and add it to the relayd hash table.
2223 * The caller MUST acquire a RCU read side lock before calling it.
2224 */
2225 int consumer_add_relayd_socket(int net_seq_idx, int sock_type,
2226 struct lttng_consumer_local_data *ctx, int sock,
2227 struct pollfd *consumer_sockpoll, struct lttcomm_sock *relayd_sock)
2228 {
2229 int fd, ret = -1;
2230 struct consumer_relayd_sock_pair *relayd;
2231
2232 DBG("Consumer adding relayd socket (idx: %d)", net_seq_idx);
2233
2234 /* Get relayd reference if exists. */
2235 relayd = consumer_find_relayd(net_seq_idx);
2236 if (relayd == NULL) {
2237 /* Not found. Allocate one. */
2238 relayd = consumer_allocate_relayd_sock_pair(net_seq_idx);
2239 if (relayd == NULL) {
2240 lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_OUTFD_ERROR);
2241 goto error;
2242 }
2243 }
2244
2245 /* Poll on consumer socket. */
2246 if (lttng_consumer_poll_socket(consumer_sockpoll) < 0) {
2247 ret = -EINTR;
2248 goto error;
2249 }
2250
2251 /* Get relayd socket from session daemon */
2252 ret = lttcomm_recv_fds_unix_sock(sock, &fd, 1);
2253 if (ret != sizeof(fd)) {
2254 lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_ERROR_RECV_FD);
2255 ret = -1;
2256 goto error;
2257 }
2258
2259 /* Copy socket information and received FD */
2260 switch (sock_type) {
2261 case LTTNG_STREAM_CONTROL:
2262 /* Copy received lttcomm socket */
2263 lttcomm_copy_sock(&relayd->control_sock, relayd_sock);
2264 ret = lttcomm_create_sock(&relayd->control_sock);
2265 if (ret < 0) {
2266 goto error;
2267 }
2268
2269 /* Close the created socket fd which is useless */
2270 close(relayd->control_sock.fd);
2271
2272 /* Assign new file descriptor */
2273 relayd->control_sock.fd = fd;
2274 break;
2275 case LTTNG_STREAM_DATA:
2276 /* Copy received lttcomm socket */
2277 lttcomm_copy_sock(&relayd->data_sock, relayd_sock);
2278 ret = lttcomm_create_sock(&relayd->data_sock);
2279 if (ret < 0) {
2280 goto error;
2281 }
2282
2283 /* Close the created socket fd which is useless */
2284 close(relayd->data_sock.fd);
2285
2286 /* Assign new file descriptor */
2287 relayd->data_sock.fd = fd;
2288 break;
2289 default:
2290 ERR("Unknown relayd socket type (%d)", sock_type);
2291 goto error;
2292 }
2293
2294 DBG("Consumer %s socket created successfully with net idx %d (fd: %d)",
2295 sock_type == LTTNG_STREAM_CONTROL ? "control" : "data",
2296 relayd->net_seq_idx, fd);
2297
2298 /*
2299 * Add relayd socket pair to consumer data hashtable. If object already
2300 * exists or on error, the function gracefully returns.
2301 */
2302 add_relayd(relayd);
2303
2304 /* All good! */
2305 ret = 0;
2306
2307 error:
2308 return ret;
2309 }
This page took 0.136948 seconds and 4 git commands to generate.