dd8c621f177d56968306fb500ce4546870b5bd2d
[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 *
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, but WITHOUT
10 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
11 * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
12 * 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 <poll.h>
22 #include <pthread.h>
23 #include <stdlib.h>
24 #include <string.h>
25 #include <sys/mman.h>
26 #include <sys/socket.h>
27 #include <sys/types.h>
28 #include <unistd.h>
29
30 #include <common/common.h>
31 #include <common/kernel-ctl/kernel-ctl.h>
32 #include <common/sessiond-comm/sessiond-comm.h>
33 #include <common/kernel-consumer/kernel-consumer.h>
34 #include <common/ust-consumer/ust-consumer.h>
35
36 #include "consumer.h"
37
38 struct lttng_consumer_global_data consumer_data = {
39 .stream_count = 0,
40 .need_update = 1,
41 .type = LTTNG_CONSUMER_UNKNOWN,
42 };
43
44 /* timeout parameter, to control the polling thread grace period. */
45 int consumer_poll_timeout = -1;
46
47 /*
48 * Flag to inform the polling thread to quit when all fd hung up. Updated by
49 * the consumer_thread_receive_fds when it notices that all fds has hung up.
50 * Also updated by the signal handler (consumer_should_exit()). Read by the
51 * polling threads.
52 */
53 volatile int consumer_quit = 0;
54
55 /*
56 * Find a stream. The consumer_data.lock must be locked during this
57 * call.
58 */
59 static struct lttng_consumer_stream *consumer_find_stream(int key)
60 {
61 struct lttng_ht_iter iter;
62 struct lttng_ht_node_ulong *node;
63 struct lttng_consumer_stream *stream = NULL;
64
65 /* Negative keys are lookup failures */
66 if (key < 0)
67 return NULL;
68
69 rcu_read_lock();
70
71 lttng_ht_lookup(consumer_data.stream_ht, (void *)((unsigned long) key),
72 &iter);
73 node = lttng_ht_iter_get_node_ulong(&iter);
74 if (node != NULL) {
75 stream = caa_container_of(node, struct lttng_consumer_stream, node);
76 }
77
78 rcu_read_unlock();
79
80 return stream;
81 }
82
83 static void consumer_steal_stream_key(int key)
84 {
85 struct lttng_consumer_stream *stream;
86
87 rcu_read_lock();
88 stream = consumer_find_stream(key);
89 if (stream) {
90 stream->key = -1;
91 /*
92 * We don't want the lookup to match, but we still need
93 * to iterate on this stream when iterating over the hash table. Just
94 * change the node key.
95 */
96 stream->node.key = -1;
97 }
98 rcu_read_unlock();
99 }
100
101 static struct lttng_consumer_channel *consumer_find_channel(int key)
102 {
103 struct lttng_ht_iter iter;
104 struct lttng_ht_node_ulong *node;
105 struct lttng_consumer_channel *channel = NULL;
106
107 /* Negative keys are lookup failures */
108 if (key < 0)
109 return NULL;
110
111 rcu_read_lock();
112
113 lttng_ht_lookup(consumer_data.channel_ht, (void *)((unsigned long) key),
114 &iter);
115 node = lttng_ht_iter_get_node_ulong(&iter);
116 if (node != NULL) {
117 channel = caa_container_of(node, struct lttng_consumer_channel, node);
118 }
119
120 rcu_read_unlock();
121
122 return channel;
123 }
124
125 static void consumer_steal_channel_key(int key)
126 {
127 struct lttng_consumer_channel *channel;
128
129 rcu_read_lock();
130 channel = consumer_find_channel(key);
131 if (channel) {
132 channel->key = -1;
133 /*
134 * We don't want the lookup to match, but we still need
135 * to iterate on this channel when iterating over the hash table. Just
136 * change the node key.
137 */
138 channel->node.key = -1;
139 }
140 rcu_read_unlock();
141 }
142
143 static
144 void consumer_free_stream(struct rcu_head *head)
145 {
146 struct lttng_ht_node_ulong *node =
147 caa_container_of(head, struct lttng_ht_node_ulong, head);
148 struct lttng_consumer_stream *stream =
149 caa_container_of(node, struct lttng_consumer_stream, node);
150
151 free(stream);
152 }
153
154 /*
155 * Remove a stream from the global list protected by a mutex. This
156 * function is also responsible for freeing its data structures.
157 */
158 void consumer_del_stream(struct lttng_consumer_stream *stream)
159 {
160 int ret;
161 struct lttng_ht_iter iter;
162 struct lttng_consumer_channel *free_chan = NULL;
163
164 pthread_mutex_lock(&consumer_data.lock);
165
166 switch (consumer_data.type) {
167 case LTTNG_CONSUMER_KERNEL:
168 if (stream->mmap_base != NULL) {
169 ret = munmap(stream->mmap_base, stream->mmap_len);
170 if (ret != 0) {
171 perror("munmap");
172 }
173 }
174 break;
175 case LTTNG_CONSUMER32_UST:
176 case LTTNG_CONSUMER64_UST:
177 lttng_ustconsumer_del_stream(stream);
178 break;
179 default:
180 ERR("Unknown consumer_data type");
181 assert(0);
182 goto end;
183 }
184
185 rcu_read_lock();
186 iter.iter.node = &stream->node.node;
187 ret = lttng_ht_del(consumer_data.stream_ht, &iter);
188 assert(!ret);
189
190 rcu_read_unlock();
191
192 if (consumer_data.stream_count <= 0) {
193 goto end;
194 }
195 consumer_data.stream_count--;
196 if (!stream) {
197 goto end;
198 }
199 if (stream->out_fd >= 0) {
200 ret = close(stream->out_fd);
201 if (ret) {
202 PERROR("close");
203 }
204 }
205 if (stream->wait_fd >= 0 && !stream->wait_fd_is_copy) {
206 ret = close(stream->wait_fd);
207 if (ret) {
208 PERROR("close");
209 }
210 }
211 if (stream->shm_fd >= 0 && stream->wait_fd != stream->shm_fd) {
212 ret = close(stream->shm_fd);
213 if (ret) {
214 PERROR("close");
215 }
216 }
217 if (!--stream->chan->refcount)
218 free_chan = stream->chan;
219
220 call_rcu(&stream->node.head, consumer_free_stream);
221 end:
222 consumer_data.need_update = 1;
223 pthread_mutex_unlock(&consumer_data.lock);
224
225 if (free_chan)
226 consumer_del_channel(free_chan);
227 }
228
229 struct lttng_consumer_stream *consumer_allocate_stream(
230 int channel_key, int stream_key,
231 int shm_fd, int wait_fd,
232 enum lttng_consumer_stream_state state,
233 uint64_t mmap_len,
234 enum lttng_event_output output,
235 const char *path_name,
236 uid_t uid,
237 gid_t gid)
238 {
239 struct lttng_consumer_stream *stream;
240 int ret;
241
242 stream = zmalloc(sizeof(*stream));
243 if (stream == NULL) {
244 perror("malloc struct lttng_consumer_stream");
245 goto end;
246 }
247 stream->chan = consumer_find_channel(channel_key);
248 if (!stream->chan) {
249 perror("Unable to find channel key");
250 goto end;
251 }
252 stream->chan->refcount++;
253 stream->key = stream_key;
254 stream->shm_fd = shm_fd;
255 stream->wait_fd = wait_fd;
256 stream->out_fd = -1;
257 stream->out_fd_offset = 0;
258 stream->state = state;
259 stream->mmap_len = mmap_len;
260 stream->mmap_base = NULL;
261 stream->output = output;
262 stream->uid = uid;
263 stream->gid = gid;
264 strncpy(stream->path_name, path_name, PATH_MAX - 1);
265 stream->path_name[PATH_MAX - 1] = '\0';
266 lttng_ht_node_init_ulong(&stream->node, stream->key);
267
268 switch (consumer_data.type) {
269 case LTTNG_CONSUMER_KERNEL:
270 break;
271 case LTTNG_CONSUMER32_UST:
272 case LTTNG_CONSUMER64_UST:
273 stream->cpu = stream->chan->cpucount++;
274 ret = lttng_ustconsumer_allocate_stream(stream);
275 if (ret) {
276 free(stream);
277 return NULL;
278 }
279 break;
280 default:
281 ERR("Unknown consumer_data type");
282 assert(0);
283 goto end;
284 }
285 DBG("Allocated stream %s (key %d, shm_fd %d, wait_fd %d, mmap_len %llu, out_fd %d)",
286 stream->path_name, stream->key,
287 stream->shm_fd,
288 stream->wait_fd,
289 (unsigned long long) stream->mmap_len,
290 stream->out_fd);
291 end:
292 return stream;
293 }
294
295 /*
296 * Add a stream to the global list protected by a mutex.
297 */
298 int consumer_add_stream(struct lttng_consumer_stream *stream)
299 {
300 int ret = 0;
301 struct lttng_ht_node_ulong *node;
302 struct lttng_ht_iter iter;
303
304 pthread_mutex_lock(&consumer_data.lock);
305 /* Steal stream identifier, for UST */
306 consumer_steal_stream_key(stream->key);
307 rcu_read_lock();
308
309 lttng_ht_lookup(consumer_data.stream_ht,
310 (void *)((unsigned long) stream->key), &iter);
311 node = lttng_ht_iter_get_node_ulong(&iter);
312 if (node != NULL) {
313 rcu_read_unlock();
314 /* Stream already exist. Ignore the insertion */
315 goto end;
316 }
317
318 lttng_ht_add_unique_ulong(consumer_data.stream_ht, &stream->node);
319 rcu_read_unlock();
320 consumer_data.stream_count++;
321 consumer_data.need_update = 1;
322
323 switch (consumer_data.type) {
324 case LTTNG_CONSUMER_KERNEL:
325 break;
326 case LTTNG_CONSUMER32_UST:
327 case LTTNG_CONSUMER64_UST:
328 /* Streams are in CPU number order (we rely on this) */
329 stream->cpu = stream->chan->nr_streams++;
330 break;
331 default:
332 ERR("Unknown consumer_data type");
333 assert(0);
334 goto end;
335 }
336
337 end:
338 pthread_mutex_unlock(&consumer_data.lock);
339
340 return ret;
341 }
342
343 /*
344 * Update a stream according to what we just received.
345 */
346 void consumer_change_stream_state(int stream_key,
347 enum lttng_consumer_stream_state state)
348 {
349 struct lttng_consumer_stream *stream;
350
351 pthread_mutex_lock(&consumer_data.lock);
352 stream = consumer_find_stream(stream_key);
353 if (stream) {
354 stream->state = state;
355 }
356 consumer_data.need_update = 1;
357 pthread_mutex_unlock(&consumer_data.lock);
358 }
359
360 static
361 void consumer_free_channel(struct rcu_head *head)
362 {
363 struct lttng_ht_node_ulong *node =
364 caa_container_of(head, struct lttng_ht_node_ulong, head);
365 struct lttng_consumer_channel *channel =
366 caa_container_of(node, struct lttng_consumer_channel, node);
367
368 free(channel);
369 }
370
371 /*
372 * Remove a channel from the global list protected by a mutex. This
373 * function is also responsible for freeing its data structures.
374 */
375 void consumer_del_channel(struct lttng_consumer_channel *channel)
376 {
377 int ret;
378 struct lttng_ht_iter iter;
379
380 pthread_mutex_lock(&consumer_data.lock);
381
382 switch (consumer_data.type) {
383 case LTTNG_CONSUMER_KERNEL:
384 break;
385 case LTTNG_CONSUMER32_UST:
386 case LTTNG_CONSUMER64_UST:
387 lttng_ustconsumer_del_channel(channel);
388 break;
389 default:
390 ERR("Unknown consumer_data type");
391 assert(0);
392 goto end;
393 }
394
395 rcu_read_lock();
396 iter.iter.node = &channel->node.node;
397 ret = lttng_ht_del(consumer_data.channel_ht, &iter);
398 assert(!ret);
399 rcu_read_unlock();
400
401 if (channel->mmap_base != NULL) {
402 ret = munmap(channel->mmap_base, channel->mmap_len);
403 if (ret != 0) {
404 perror("munmap");
405 }
406 }
407 if (channel->wait_fd >= 0 && !channel->wait_fd_is_copy) {
408 ret = close(channel->wait_fd);
409 if (ret) {
410 PERROR("close");
411 }
412 }
413 if (channel->shm_fd >= 0 && channel->wait_fd != channel->shm_fd) {
414 ret = close(channel->shm_fd);
415 if (ret) {
416 PERROR("close");
417 }
418 }
419
420 call_rcu(&channel->node.head, consumer_free_channel);
421 end:
422 pthread_mutex_unlock(&consumer_data.lock);
423 }
424
425 struct lttng_consumer_channel *consumer_allocate_channel(
426 int channel_key,
427 int shm_fd, int wait_fd,
428 uint64_t mmap_len,
429 uint64_t max_sb_size)
430 {
431 struct lttng_consumer_channel *channel;
432 int ret;
433
434 channel = zmalloc(sizeof(*channel));
435 if (channel == NULL) {
436 perror("malloc struct lttng_consumer_channel");
437 goto end;
438 }
439 channel->key = channel_key;
440 channel->shm_fd = shm_fd;
441 channel->wait_fd = wait_fd;
442 channel->mmap_len = mmap_len;
443 channel->max_sb_size = max_sb_size;
444 channel->refcount = 0;
445 channel->nr_streams = 0;
446 lttng_ht_node_init_ulong(&channel->node, channel->key);
447
448 switch (consumer_data.type) {
449 case LTTNG_CONSUMER_KERNEL:
450 channel->mmap_base = NULL;
451 channel->mmap_len = 0;
452 break;
453 case LTTNG_CONSUMER32_UST:
454 case LTTNG_CONSUMER64_UST:
455 ret = lttng_ustconsumer_allocate_channel(channel);
456 if (ret) {
457 free(channel);
458 return NULL;
459 }
460 break;
461 default:
462 ERR("Unknown consumer_data type");
463 assert(0);
464 goto end;
465 }
466 DBG("Allocated channel (key %d, shm_fd %d, wait_fd %d, mmap_len %llu, max_sb_size %llu)",
467 channel->key,
468 channel->shm_fd,
469 channel->wait_fd,
470 (unsigned long long) channel->mmap_len,
471 (unsigned long long) channel->max_sb_size);
472 end:
473 return channel;
474 }
475
476 /*
477 * Add a channel to the global list protected by a mutex.
478 */
479 int consumer_add_channel(struct lttng_consumer_channel *channel)
480 {
481 struct lttng_ht_node_ulong *node;
482 struct lttng_ht_iter iter;
483
484 pthread_mutex_lock(&consumer_data.lock);
485 /* Steal channel identifier, for UST */
486 consumer_steal_channel_key(channel->key);
487 rcu_read_lock();
488
489 lttng_ht_lookup(consumer_data.channel_ht,
490 (void *)((unsigned long) channel->key), &iter);
491 node = lttng_ht_iter_get_node_ulong(&iter);
492 if (node != NULL) {
493 /* Channel already exist. Ignore the insertion */
494 goto end;
495 }
496
497 lttng_ht_add_unique_ulong(consumer_data.channel_ht, &channel->node);
498
499 end:
500 rcu_read_unlock();
501 pthread_mutex_unlock(&consumer_data.lock);
502
503 return 0;
504 }
505
506 /*
507 * Allocate the pollfd structure and the local view of the out fds to avoid
508 * doing a lookup in the linked list and concurrency issues when writing is
509 * needed. Called with consumer_data.lock held.
510 *
511 * Returns the number of fds in the structures.
512 */
513 int consumer_update_poll_array(
514 struct lttng_consumer_local_data *ctx, struct pollfd **pollfd,
515 struct lttng_consumer_stream **local_stream)
516 {
517 int i = 0;
518 struct lttng_ht_iter iter;
519 struct lttng_consumer_stream *stream;
520
521 DBG("Updating poll fd array");
522 rcu_read_lock();
523 cds_lfht_for_each_entry(consumer_data.stream_ht->ht, &iter.iter, stream,
524 node.node) {
525 if (stream->state != LTTNG_CONSUMER_ACTIVE_STREAM) {
526 continue;
527 }
528 DBG("Active FD %d", stream->wait_fd);
529 (*pollfd)[i].fd = stream->wait_fd;
530 (*pollfd)[i].events = POLLIN | POLLPRI;
531 local_stream[i] = stream;
532 i++;
533 }
534 rcu_read_unlock();
535
536 /*
537 * Insert the consumer_poll_pipe at the end of the array and don't
538 * increment i so nb_fd is the number of real FD.
539 */
540 (*pollfd)[i].fd = ctx->consumer_poll_pipe[0];
541 (*pollfd)[i].events = POLLIN | POLLPRI;
542 return i;
543 }
544
545 /*
546 * Poll on the should_quit pipe and the command socket return -1 on error and
547 * should exit, 0 if data is available on the command socket
548 */
549 int lttng_consumer_poll_socket(struct pollfd *consumer_sockpoll)
550 {
551 int num_rdy;
552
553 restart:
554 num_rdy = poll(consumer_sockpoll, 2, -1);
555 if (num_rdy == -1) {
556 /*
557 * Restart interrupted system call.
558 */
559 if (errno == EINTR) {
560 goto restart;
561 }
562 perror("Poll error");
563 goto exit;
564 }
565 if (consumer_sockpoll[0].revents & (POLLIN | POLLPRI)) {
566 DBG("consumer_should_quit wake up");
567 goto exit;
568 }
569 return 0;
570
571 exit:
572 return -1;
573 }
574
575 /*
576 * Set the error socket.
577 */
578 void lttng_consumer_set_error_sock(
579 struct lttng_consumer_local_data *ctx, int sock)
580 {
581 ctx->consumer_error_socket = sock;
582 }
583
584 /*
585 * Set the command socket path.
586 */
587
588 void lttng_consumer_set_command_sock_path(
589 struct lttng_consumer_local_data *ctx, char *sock)
590 {
591 ctx->consumer_command_sock_path = sock;
592 }
593
594 /*
595 * Send return code to the session daemon.
596 * If the socket is not defined, we return 0, it is not a fatal error
597 */
598 int lttng_consumer_send_error(
599 struct lttng_consumer_local_data *ctx, int cmd)
600 {
601 if (ctx->consumer_error_socket > 0) {
602 return lttcomm_send_unix_sock(ctx->consumer_error_socket, &cmd,
603 sizeof(enum lttcomm_sessiond_command));
604 }
605
606 return 0;
607 }
608
609 /*
610 * Close all the tracefiles and stream fds, should be called when all instances
611 * are destroyed.
612 */
613 void lttng_consumer_cleanup(void)
614 {
615 struct lttng_ht_iter iter;
616 struct lttng_ht_node_ulong *node;
617
618 rcu_read_lock();
619
620 /*
621 * close all outfd. Called when there are no more threads running (after
622 * joining on the threads), no need to protect list iteration with mutex.
623 */
624 cds_lfht_for_each_entry(consumer_data.stream_ht->ht, &iter.iter, node,
625 node) {
626 struct lttng_consumer_stream *stream =
627 caa_container_of(node, struct lttng_consumer_stream, node);
628 consumer_del_stream(stream);
629 }
630
631 cds_lfht_for_each_entry(consumer_data.channel_ht->ht, &iter.iter, node,
632 node) {
633 struct lttng_consumer_channel *channel =
634 caa_container_of(node, struct lttng_consumer_channel, node);
635 consumer_del_channel(channel);
636 }
637
638 rcu_read_unlock();
639
640 lttng_ht_destroy(consumer_data.stream_ht);
641 lttng_ht_destroy(consumer_data.channel_ht);
642 }
643
644 /*
645 * Called from signal handler.
646 */
647 void lttng_consumer_should_exit(struct lttng_consumer_local_data *ctx)
648 {
649 int ret;
650 consumer_quit = 1;
651 ret = write(ctx->consumer_should_quit[1], "4", 1);
652 if (ret < 0) {
653 perror("write consumer quit");
654 }
655 }
656
657 void lttng_consumer_sync_trace_file(
658 struct lttng_consumer_stream *stream, off_t orig_offset)
659 {
660 int outfd = stream->out_fd;
661
662 /*
663 * This does a blocking write-and-wait on any page that belongs to the
664 * subbuffer prior to the one we just wrote.
665 * Don't care about error values, as these are just hints and ways to
666 * limit the amount of page cache used.
667 */
668 if (orig_offset < stream->chan->max_sb_size) {
669 return;
670 }
671 lttng_sync_file_range(outfd, orig_offset - stream->chan->max_sb_size,
672 stream->chan->max_sb_size,
673 SYNC_FILE_RANGE_WAIT_BEFORE
674 | SYNC_FILE_RANGE_WRITE
675 | SYNC_FILE_RANGE_WAIT_AFTER);
676 /*
677 * Give hints to the kernel about how we access the file:
678 * POSIX_FADV_DONTNEED : we won't re-access data in a near future after
679 * we write it.
680 *
681 * We need to call fadvise again after the file grows because the
682 * kernel does not seem to apply fadvise to non-existing parts of the
683 * file.
684 *
685 * Call fadvise _after_ having waited for the page writeback to
686 * complete because the dirty page writeback semantic is not well
687 * defined. So it can be expected to lead to lower throughput in
688 * streaming.
689 */
690 posix_fadvise(outfd, orig_offset - stream->chan->max_sb_size,
691 stream->chan->max_sb_size, POSIX_FADV_DONTNEED);
692 }
693
694 /*
695 * Initialise the necessary environnement :
696 * - create a new context
697 * - create the poll_pipe
698 * - create the should_quit pipe (for signal handler)
699 * - create the thread pipe (for splice)
700 *
701 * Takes a function pointer as argument, this function is called when data is
702 * available on a buffer. This function is responsible to do the
703 * kernctl_get_next_subbuf, read the data with mmap or splice depending on the
704 * buffer configuration and then kernctl_put_next_subbuf at the end.
705 *
706 * Returns a pointer to the new context or NULL on error.
707 */
708 struct lttng_consumer_local_data *lttng_consumer_create(
709 enum lttng_consumer_type type,
710 ssize_t (*buffer_ready)(struct lttng_consumer_stream *stream,
711 struct lttng_consumer_local_data *ctx),
712 int (*recv_channel)(struct lttng_consumer_channel *channel),
713 int (*recv_stream)(struct lttng_consumer_stream *stream),
714 int (*update_stream)(int stream_key, uint32_t state))
715 {
716 int ret, i;
717 struct lttng_consumer_local_data *ctx;
718
719 assert(consumer_data.type == LTTNG_CONSUMER_UNKNOWN ||
720 consumer_data.type == type);
721 consumer_data.type = type;
722
723 ctx = zmalloc(sizeof(struct lttng_consumer_local_data));
724 if (ctx == NULL) {
725 perror("allocating context");
726 goto error;
727 }
728
729 ctx->consumer_error_socket = -1;
730 /* assign the callbacks */
731 ctx->on_buffer_ready = buffer_ready;
732 ctx->on_recv_channel = recv_channel;
733 ctx->on_recv_stream = recv_stream;
734 ctx->on_update_stream = update_stream;
735
736 ret = pipe(ctx->consumer_poll_pipe);
737 if (ret < 0) {
738 perror("Error creating poll pipe");
739 goto error_poll_pipe;
740 }
741
742 /* set read end of the pipe to non-blocking */
743 ret = fcntl(ctx->consumer_poll_pipe[0], F_SETFL, O_NONBLOCK);
744 if (ret < 0) {
745 perror("fcntl O_NONBLOCK");
746 goto error_poll_fcntl;
747 }
748
749 /* set write end of the pipe to non-blocking */
750 ret = fcntl(ctx->consumer_poll_pipe[1], F_SETFL, O_NONBLOCK);
751 if (ret < 0) {
752 perror("fcntl O_NONBLOCK");
753 goto error_poll_fcntl;
754 }
755
756 ret = pipe(ctx->consumer_should_quit);
757 if (ret < 0) {
758 perror("Error creating recv pipe");
759 goto error_quit_pipe;
760 }
761
762 ret = pipe(ctx->consumer_thread_pipe);
763 if (ret < 0) {
764 perror("Error creating thread pipe");
765 goto error_thread_pipe;
766 }
767
768 return ctx;
769
770
771 error_thread_pipe:
772 for (i = 0; i < 2; i++) {
773 int err;
774
775 err = close(ctx->consumer_should_quit[i]);
776 if (err) {
777 PERROR("close");
778 }
779 }
780 error_poll_fcntl:
781 error_quit_pipe:
782 for (i = 0; i < 2; i++) {
783 int err;
784
785 err = close(ctx->consumer_poll_pipe[i]);
786 if (err) {
787 PERROR("close");
788 }
789 }
790 error_poll_pipe:
791 free(ctx);
792 error:
793 return NULL;
794 }
795
796 /*
797 * Close all fds associated with the instance and free the context.
798 */
799 void lttng_consumer_destroy(struct lttng_consumer_local_data *ctx)
800 {
801 int ret;
802
803 ret = close(ctx->consumer_error_socket);
804 if (ret) {
805 PERROR("close");
806 }
807 ret = close(ctx->consumer_thread_pipe[0]);
808 if (ret) {
809 PERROR("close");
810 }
811 ret = close(ctx->consumer_thread_pipe[1]);
812 if (ret) {
813 PERROR("close");
814 }
815 ret = close(ctx->consumer_poll_pipe[0]);
816 if (ret) {
817 PERROR("close");
818 }
819 ret = close(ctx->consumer_poll_pipe[1]);
820 if (ret) {
821 PERROR("close");
822 }
823 ret = close(ctx->consumer_should_quit[0]);
824 if (ret) {
825 PERROR("close");
826 }
827 ret = close(ctx->consumer_should_quit[1]);
828 if (ret) {
829 PERROR("close");
830 }
831 unlink(ctx->consumer_command_sock_path);
832 free(ctx);
833 }
834
835 /*
836 * Mmap the ring buffer, read it and write the data to the tracefile.
837 *
838 * Returns the number of bytes written
839 */
840 ssize_t lttng_consumer_on_read_subbuffer_mmap(
841 struct lttng_consumer_local_data *ctx,
842 struct lttng_consumer_stream *stream, unsigned long len)
843 {
844 switch (consumer_data.type) {
845 case LTTNG_CONSUMER_KERNEL:
846 return lttng_kconsumer_on_read_subbuffer_mmap(ctx, stream, len);
847 case LTTNG_CONSUMER32_UST:
848 case LTTNG_CONSUMER64_UST:
849 return lttng_ustconsumer_on_read_subbuffer_mmap(ctx, stream, len);
850 default:
851 ERR("Unknown consumer_data type");
852 assert(0);
853 }
854
855 return 0;
856 }
857
858 /*
859 * Splice the data from the ring buffer to the tracefile.
860 *
861 * Returns the number of bytes spliced.
862 */
863 ssize_t lttng_consumer_on_read_subbuffer_splice(
864 struct lttng_consumer_local_data *ctx,
865 struct lttng_consumer_stream *stream, unsigned long len)
866 {
867 switch (consumer_data.type) {
868 case LTTNG_CONSUMER_KERNEL:
869 return lttng_kconsumer_on_read_subbuffer_splice(ctx, stream, len);
870 case LTTNG_CONSUMER32_UST:
871 case LTTNG_CONSUMER64_UST:
872 return -ENOSYS;
873 default:
874 ERR("Unknown consumer_data type");
875 assert(0);
876 return -ENOSYS;
877 }
878
879 }
880
881 /*
882 * Take a snapshot for a specific fd
883 *
884 * Returns 0 on success, < 0 on error
885 */
886 int lttng_consumer_take_snapshot(struct lttng_consumer_local_data *ctx,
887 struct lttng_consumer_stream *stream)
888 {
889 switch (consumer_data.type) {
890 case LTTNG_CONSUMER_KERNEL:
891 return lttng_kconsumer_take_snapshot(ctx, stream);
892 case LTTNG_CONSUMER32_UST:
893 case LTTNG_CONSUMER64_UST:
894 return lttng_ustconsumer_take_snapshot(ctx, stream);
895 default:
896 ERR("Unknown consumer_data type");
897 assert(0);
898 return -ENOSYS;
899 }
900
901 }
902
903 /*
904 * Get the produced position
905 *
906 * Returns 0 on success, < 0 on error
907 */
908 int lttng_consumer_get_produced_snapshot(
909 struct lttng_consumer_local_data *ctx,
910 struct lttng_consumer_stream *stream,
911 unsigned long *pos)
912 {
913 switch (consumer_data.type) {
914 case LTTNG_CONSUMER_KERNEL:
915 return lttng_kconsumer_get_produced_snapshot(ctx, stream, pos);
916 case LTTNG_CONSUMER32_UST:
917 case LTTNG_CONSUMER64_UST:
918 return lttng_ustconsumer_get_produced_snapshot(ctx, stream, pos);
919 default:
920 ERR("Unknown consumer_data type");
921 assert(0);
922 return -ENOSYS;
923 }
924 }
925
926 int lttng_consumer_recv_cmd(struct lttng_consumer_local_data *ctx,
927 int sock, struct pollfd *consumer_sockpoll)
928 {
929 switch (consumer_data.type) {
930 case LTTNG_CONSUMER_KERNEL:
931 return lttng_kconsumer_recv_cmd(ctx, sock, consumer_sockpoll);
932 case LTTNG_CONSUMER32_UST:
933 case LTTNG_CONSUMER64_UST:
934 return lttng_ustconsumer_recv_cmd(ctx, sock, consumer_sockpoll);
935 default:
936 ERR("Unknown consumer_data type");
937 assert(0);
938 return -ENOSYS;
939 }
940 }
941
942 /*
943 * This thread polls the fds in the set to consume the data and write
944 * it to tracefile if necessary.
945 */
946 void *lttng_consumer_thread_poll_fds(void *data)
947 {
948 int num_rdy, num_hup, high_prio, ret, i;
949 struct pollfd *pollfd = NULL;
950 /* local view of the streams */
951 struct lttng_consumer_stream **local_stream = NULL;
952 /* local view of consumer_data.fds_count */
953 int nb_fd = 0;
954 struct lttng_consumer_local_data *ctx = data;
955
956 rcu_register_thread();
957
958 local_stream = zmalloc(sizeof(struct lttng_consumer_stream));
959
960 while (1) {
961 high_prio = 0;
962 num_hup = 0;
963
964 /*
965 * the fds set has been updated, we need to update our
966 * local array as well
967 */
968 pthread_mutex_lock(&consumer_data.lock);
969 if (consumer_data.need_update) {
970 if (pollfd != NULL) {
971 free(pollfd);
972 pollfd = NULL;
973 }
974 if (local_stream != NULL) {
975 free(local_stream);
976 local_stream = NULL;
977 }
978
979 /* allocate for all fds + 1 for the consumer_poll_pipe */
980 pollfd = zmalloc((consumer_data.stream_count + 1) * sizeof(struct pollfd));
981 if (pollfd == NULL) {
982 perror("pollfd malloc");
983 pthread_mutex_unlock(&consumer_data.lock);
984 goto end;
985 }
986
987 /* allocate for all fds + 1 for the consumer_poll_pipe */
988 local_stream = zmalloc((consumer_data.stream_count + 1) *
989 sizeof(struct lttng_consumer_stream));
990 if (local_stream == NULL) {
991 perror("local_stream malloc");
992 pthread_mutex_unlock(&consumer_data.lock);
993 goto end;
994 }
995 ret = consumer_update_poll_array(ctx, &pollfd, local_stream);
996 if (ret < 0) {
997 ERR("Error in allocating pollfd or local_outfds");
998 lttng_consumer_send_error(ctx, CONSUMERD_POLL_ERROR);
999 pthread_mutex_unlock(&consumer_data.lock);
1000 goto end;
1001 }
1002 nb_fd = ret;
1003 consumer_data.need_update = 0;
1004 }
1005 pthread_mutex_unlock(&consumer_data.lock);
1006
1007 /* No FDs and consumer_quit, consumer_cleanup the thread */
1008 if (nb_fd == 0 && consumer_quit == 1) {
1009 goto end;
1010 }
1011 /* poll on the array of fds */
1012 restart:
1013 DBG("polling on %d fd", nb_fd + 1);
1014 num_rdy = poll(pollfd, nb_fd + 1, consumer_poll_timeout);
1015 DBG("poll num_rdy : %d", num_rdy);
1016 if (num_rdy == -1) {
1017 /*
1018 * Restart interrupted system call.
1019 */
1020 if (errno == EINTR) {
1021 goto restart;
1022 }
1023 perror("Poll error");
1024 lttng_consumer_send_error(ctx, CONSUMERD_POLL_ERROR);
1025 goto end;
1026 } else if (num_rdy == 0) {
1027 DBG("Polling thread timed out");
1028 goto end;
1029 }
1030
1031 /*
1032 * If the consumer_poll_pipe triggered poll go
1033 * directly to the beginning of the loop to update the
1034 * array. We want to prioritize array update over
1035 * low-priority reads.
1036 */
1037 if (pollfd[nb_fd].revents & (POLLIN | POLLPRI)) {
1038 size_t pipe_readlen;
1039 char tmp;
1040
1041 DBG("consumer_poll_pipe wake up");
1042 /* Consume 1 byte of pipe data */
1043 do {
1044 pipe_readlen = read(ctx->consumer_poll_pipe[0], &tmp, 1);
1045 } while (pipe_readlen == -1 && errno == EINTR);
1046 continue;
1047 }
1048
1049 /* Take care of high priority channels first. */
1050 for (i = 0; i < nb_fd; i++) {
1051 if (pollfd[i].revents & POLLPRI) {
1052 ssize_t len;
1053
1054 DBG("Urgent read on fd %d", pollfd[i].fd);
1055 high_prio = 1;
1056 len = ctx->on_buffer_ready(local_stream[i], ctx);
1057 /* it's ok to have an unavailable sub-buffer */
1058 if (len < 0 && len != -EAGAIN) {
1059 goto end;
1060 } else if (len > 0) {
1061 local_stream[i]->data_read = 1;
1062 }
1063 }
1064 }
1065
1066 /*
1067 * If we read high prio channel in this loop, try again
1068 * for more high prio data.
1069 */
1070 if (high_prio) {
1071 continue;
1072 }
1073
1074 /* Take care of low priority channels. */
1075 for (i = 0; i < nb_fd; i++) {
1076 if ((pollfd[i].revents & POLLIN) ||
1077 local_stream[i]->hangup_flush_done) {
1078 ssize_t len;
1079
1080 DBG("Normal read on fd %d", pollfd[i].fd);
1081 len = ctx->on_buffer_ready(local_stream[i], ctx);
1082 /* it's ok to have an unavailable sub-buffer */
1083 if (len < 0 && len != -EAGAIN) {
1084 goto end;
1085 } else if (len > 0) {
1086 local_stream[i]->data_read = 1;
1087 }
1088 }
1089 }
1090
1091 /* Handle hangup and errors */
1092 for (i = 0; i < nb_fd; i++) {
1093 if (!local_stream[i]->hangup_flush_done
1094 && (pollfd[i].revents & (POLLHUP | POLLERR | POLLNVAL))
1095 && (consumer_data.type == LTTNG_CONSUMER32_UST
1096 || consumer_data.type == LTTNG_CONSUMER64_UST)) {
1097 DBG("fd %d is hup|err|nval. Attempting flush and read.",
1098 pollfd[i].fd);
1099 lttng_ustconsumer_on_stream_hangup(local_stream[i]);
1100 /* Attempt read again, for the data we just flushed. */
1101 local_stream[i]->data_read = 1;
1102 }
1103 /*
1104 * If the poll flag is HUP/ERR/NVAL and we have
1105 * read no data in this pass, we can remove the
1106 * stream from its hash table.
1107 */
1108 if ((pollfd[i].revents & POLLHUP)) {
1109 DBG("Polling fd %d tells it has hung up.", pollfd[i].fd);
1110 if (!local_stream[i]->data_read) {
1111 consumer_del_stream(local_stream[i]);
1112 num_hup++;
1113 }
1114 } else if (pollfd[i].revents & POLLERR) {
1115 ERR("Error returned in polling fd %d.", pollfd[i].fd);
1116 if (!local_stream[i]->data_read) {
1117 consumer_del_stream(local_stream[i]);
1118 num_hup++;
1119 }
1120 } else if (pollfd[i].revents & POLLNVAL) {
1121 ERR("Polling fd %d tells fd is not open.", pollfd[i].fd);
1122 if (!local_stream[i]->data_read) {
1123 consumer_del_stream(local_stream[i]);
1124 num_hup++;
1125 }
1126 }
1127 local_stream[i]->data_read = 0;
1128 }
1129 }
1130 end:
1131 DBG("polling thread exiting");
1132 if (pollfd != NULL) {
1133 free(pollfd);
1134 pollfd = NULL;
1135 }
1136 if (local_stream != NULL) {
1137 free(local_stream);
1138 local_stream = NULL;
1139 }
1140 rcu_unregister_thread();
1141 return NULL;
1142 }
1143
1144 /*
1145 * This thread listens on the consumerd socket and receives the file
1146 * descriptors from the session daemon.
1147 */
1148 void *lttng_consumer_thread_receive_fds(void *data)
1149 {
1150 int sock, client_socket, ret;
1151 /*
1152 * structure to poll for incoming data on communication socket avoids
1153 * making blocking sockets.
1154 */
1155 struct pollfd consumer_sockpoll[2];
1156 struct lttng_consumer_local_data *ctx = data;
1157
1158 rcu_register_thread();
1159
1160 DBG("Creating command socket %s", ctx->consumer_command_sock_path);
1161 unlink(ctx->consumer_command_sock_path);
1162 client_socket = lttcomm_create_unix_sock(ctx->consumer_command_sock_path);
1163 if (client_socket < 0) {
1164 ERR("Cannot create command socket");
1165 goto end;
1166 }
1167
1168 ret = lttcomm_listen_unix_sock(client_socket);
1169 if (ret < 0) {
1170 goto end;
1171 }
1172
1173 DBG("Sending ready command to lttng-sessiond");
1174 ret = lttng_consumer_send_error(ctx, CONSUMERD_COMMAND_SOCK_READY);
1175 /* return < 0 on error, but == 0 is not fatal */
1176 if (ret < 0) {
1177 ERR("Error sending ready command to lttng-sessiond");
1178 goto end;
1179 }
1180
1181 ret = fcntl(client_socket, F_SETFL, O_NONBLOCK);
1182 if (ret < 0) {
1183 perror("fcntl O_NONBLOCK");
1184 goto end;
1185 }
1186
1187 /* prepare the FDs to poll : to client socket and the should_quit pipe */
1188 consumer_sockpoll[0].fd = ctx->consumer_should_quit[0];
1189 consumer_sockpoll[0].events = POLLIN | POLLPRI;
1190 consumer_sockpoll[1].fd = client_socket;
1191 consumer_sockpoll[1].events = POLLIN | POLLPRI;
1192
1193 if (lttng_consumer_poll_socket(consumer_sockpoll) < 0) {
1194 goto end;
1195 }
1196 DBG("Connection on client_socket");
1197
1198 /* Blocking call, waiting for transmission */
1199 sock = lttcomm_accept_unix_sock(client_socket);
1200 if (sock <= 0) {
1201 WARN("On accept");
1202 goto end;
1203 }
1204 ret = fcntl(sock, F_SETFL, O_NONBLOCK);
1205 if (ret < 0) {
1206 perror("fcntl O_NONBLOCK");
1207 goto end;
1208 }
1209
1210 /* update the polling structure to poll on the established socket */
1211 consumer_sockpoll[1].fd = sock;
1212 consumer_sockpoll[1].events = POLLIN | POLLPRI;
1213
1214 while (1) {
1215 if (lttng_consumer_poll_socket(consumer_sockpoll) < 0) {
1216 goto end;
1217 }
1218 DBG("Incoming command on sock");
1219 ret = lttng_consumer_recv_cmd(ctx, sock, consumer_sockpoll);
1220 if (ret == -ENOENT) {
1221 DBG("Received STOP command");
1222 goto end;
1223 }
1224 if (ret < 0) {
1225 ERR("Communication interrupted on command socket");
1226 goto end;
1227 }
1228 if (consumer_quit) {
1229 DBG("consumer_thread_receive_fds received quit from signal");
1230 goto end;
1231 }
1232 DBG("received fds on sock");
1233 }
1234 end:
1235 DBG("consumer_thread_receive_fds exiting");
1236
1237 /*
1238 * when all fds have hung up, the polling thread
1239 * can exit cleanly
1240 */
1241 consumer_quit = 1;
1242
1243 /*
1244 * 2s of grace period, if no polling events occur during
1245 * this period, the polling thread will exit even if there
1246 * are still open FDs (should not happen, but safety mechanism).
1247 */
1248 consumer_poll_timeout = LTTNG_CONSUMER_POLL_TIMEOUT;
1249
1250 /*
1251 * Wake-up the other end by writing a null byte in the pipe
1252 * (non-blocking). Important note: Because writing into the
1253 * pipe is non-blocking (and therefore we allow dropping wakeup
1254 * data, as long as there is wakeup data present in the pipe
1255 * buffer to wake up the other end), the other end should
1256 * perform the following sequence for waiting:
1257 * 1) empty the pipe (reads).
1258 * 2) perform update operation.
1259 * 3) wait on the pipe (poll).
1260 */
1261 do {
1262 ret = write(ctx->consumer_poll_pipe[1], "", 1);
1263 } while (ret == -1UL && errno == EINTR);
1264 rcu_unregister_thread();
1265 return NULL;
1266 }
1267
1268 ssize_t lttng_consumer_read_subbuffer(struct lttng_consumer_stream *stream,
1269 struct lttng_consumer_local_data *ctx)
1270 {
1271 switch (consumer_data.type) {
1272 case LTTNG_CONSUMER_KERNEL:
1273 return lttng_kconsumer_read_subbuffer(stream, ctx);
1274 case LTTNG_CONSUMER32_UST:
1275 case LTTNG_CONSUMER64_UST:
1276 return lttng_ustconsumer_read_subbuffer(stream, ctx);
1277 default:
1278 ERR("Unknown consumer_data type");
1279 assert(0);
1280 return -ENOSYS;
1281 }
1282 }
1283
1284 int lttng_consumer_on_recv_stream(struct lttng_consumer_stream *stream)
1285 {
1286 switch (consumer_data.type) {
1287 case LTTNG_CONSUMER_KERNEL:
1288 return lttng_kconsumer_on_recv_stream(stream);
1289 case LTTNG_CONSUMER32_UST:
1290 case LTTNG_CONSUMER64_UST:
1291 return lttng_ustconsumer_on_recv_stream(stream);
1292 default:
1293 ERR("Unknown consumer_data type");
1294 assert(0);
1295 return -ENOSYS;
1296 }
1297 }
1298
1299 /*
1300 * Allocate and set consumer data hash tables.
1301 */
1302 void lttng_consumer_init(void)
1303 {
1304 consumer_data.stream_ht = lttng_ht_new(0, LTTNG_HT_TYPE_ULONG);
1305 consumer_data.channel_ht = lttng_ht_new(0, LTTNG_HT_TYPE_ULONG);
1306 }
1307
This page took 0.089923 seconds and 3 git commands to generate.