Change malloc to zmalloc on UST metadata creation
[lttng-tools.git] / liblttng-consumer / lttng-consumer.c
CommitLineData
3bd1e081
MD
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
6 * modify it under the terms of the GNU General Public License
7 * as published by the Free Software Foundation; only version 2
8 * of the License.
9 *
10 * This program is distributed in the hope that it will be useful,
11 * but WITHOUT ANY WARRANTY; without even the implied warranty of
12 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
13 * GNU General Public License for more details.
14 *
15 * You should have received a copy of the GNU General Public License
16 * along with this program; if not, write to the Free Software
17 * Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
18 */
19
20#define _GNU_SOURCE
21#include <assert.h>
22#include <fcntl.h>
23#include <poll.h>
24#include <pthread.h>
25#include <stdlib.h>
26#include <string.h>
27#include <sys/mman.h>
28#include <sys/socket.h>
29#include <sys/types.h>
30#include <unistd.h>
31
32#include <lttng-kernel-ctl.h>
33#include <lttng-sessiond-comm.h>
34#include <lttng/lttng-consumer.h>
35#include <lttng/lttng-kconsumer.h>
36#include <lttng/lttng-ustconsumer.h>
37#include <lttngerr.h>
38
39struct lttng_consumer_global_data consumer_data = {
40 .stream_list.head = CDS_LIST_HEAD_INIT(consumer_data.stream_list.head),
41 .channel_list.head = CDS_LIST_HEAD_INIT(consumer_data.channel_list.head),
42 .stream_count = 0,
43 .need_update = 1,
44 .type = LTTNG_CONSUMER_UNKNOWN,
45};
46
47/* timeout parameter, to control the polling thread grace period. */
48int consumer_poll_timeout = -1;
49
50/*
51 * Flag to inform the polling thread to quit when all fd hung up. Updated by
52 * the consumer_thread_receive_fds when it notices that all fds has hung up.
53 * Also updated by the signal handler (consumer_should_exit()). Read by the
54 * polling threads.
55 */
56volatile int consumer_quit = 0;
57
58/*
59 * Find a stream. The consumer_data.lock must be locked during this
60 * call.
61 */
62static struct lttng_consumer_stream *consumer_find_stream(int key)
63{
64 struct lttng_consumer_stream *iter;
65
66 cds_list_for_each_entry(iter, &consumer_data.stream_list.head, list) {
67 if (iter->key == key) {
68 DBG("Found stream key %d", key);
69 return iter;
70 }
71 }
72 return NULL;
73}
74
75static struct lttng_consumer_channel *consumer_find_channel(int key)
76{
77 struct lttng_consumer_channel *iter;
78
79 cds_list_for_each_entry(iter, &consumer_data.channel_list.head, list) {
80 if (iter->key == key) {
81 DBG("Found channel key %d", key);
82 return iter;
83 }
84 }
85 return NULL;
86}
87
88/*
89 * Remove a stream from the global list protected by a mutex. This
90 * function is also responsible for freeing its data structures.
91 */
92void consumer_del_stream(struct lttng_consumer_stream *stream)
93{
94 int ret;
95 struct lttng_consumer_channel *free_chan = NULL;
96
97 pthread_mutex_lock(&consumer_data.lock);
98
99 switch (consumer_data.type) {
100 case LTTNG_CONSUMER_KERNEL:
101 if (stream->mmap_base != NULL) {
102 ret = munmap(stream->mmap_base, stream->mmap_len);
103 if (ret != 0) {
104 perror("munmap");
105 }
106 }
107 break;
108 case LTTNG_CONSUMER_UST:
109 lttng_ustconsumer_del_stream(stream);
110 break;
111 default:
112 ERR("Unknown consumer_data type");
113 assert(0);
114 goto end;
115 }
116
117 cds_list_del(&stream->list);
118 if (consumer_data.stream_count <= 0) {
119 goto end;
120 }
121 consumer_data.stream_count--;
122 if (!stream) {
123 goto end;
124 }
125 if (stream->out_fd >= 0) {
126 close(stream->out_fd);
127 }
b5c5fc29 128 if (stream->wait_fd >= 0 && !stream->wait_fd_is_copy) {
3bd1e081
MD
129 close(stream->wait_fd);
130 }
b5c5fc29
MD
131 if (stream->shm_fd >= 0 && stream->wait_fd != stream->shm_fd
132 && !stream->shm_fd_is_copy) {
3bd1e081
MD
133 close(stream->shm_fd);
134 }
135 if (!--stream->chan->refcount)
136 free_chan = stream->chan;
137 free(stream);
138end:
139 consumer_data.need_update = 1;
140 pthread_mutex_unlock(&consumer_data.lock);
141
142 if (free_chan)
143 consumer_del_channel(free_chan);
144}
145
146struct lttng_consumer_stream *consumer_allocate_stream(
147 int channel_key, int stream_key,
148 int shm_fd, int wait_fd,
149 enum lttng_consumer_stream_state state,
150 uint64_t mmap_len,
151 enum lttng_event_output output,
152 const char *path_name)
153{
154 struct lttng_consumer_stream *stream;
155 int ret;
156
157 stream = malloc(sizeof(*stream));
158 if (stream == NULL) {
159 perror("malloc struct lttng_consumer_stream");
160 goto end;
161 }
162 stream->chan = consumer_find_channel(channel_key);
163 if (!stream->chan) {
164 perror("Unable to find channel key");
165 goto end;
166 }
167 stream->chan->refcount++;
168 stream->key = stream_key;
169 stream->shm_fd = shm_fd;
170 stream->wait_fd = wait_fd;
171 stream->out_fd = -1;
172 stream->out_fd_offset = 0;
173 stream->state = state;
174 stream->mmap_len = mmap_len;
175 stream->mmap_base = NULL;
176 stream->output = output;
177 strncpy(stream->path_name, path_name, PATH_MAX - 1);
178 stream->path_name[PATH_MAX - 1] = '\0';
179
180 switch (consumer_data.type) {
181 case LTTNG_CONSUMER_KERNEL:
182 break;
183 case LTTNG_CONSUMER_UST:
5af2f756 184 stream->cpu = stream->chan->cpucount++;
3bd1e081
MD
185 ret = lttng_ustconsumer_allocate_stream(stream);
186 if (ret) {
187 free(stream);
188 return NULL;
189 }
190 break;
191 default:
192 ERR("Unknown consumer_data type");
193 assert(0);
194 goto end;
195 }
196 DBG("Allocated stream %s (key %d, shm_fd %d, wait_fd %d, mmap_len %llu, out_fd %d)",
197 stream->path_name, stream->key,
198 stream->shm_fd,
199 stream->wait_fd,
200 (unsigned long long) stream->mmap_len,
201 stream->out_fd);
202end:
203 return stream;
204}
205
206/*
207 * Add a stream to the global list protected by a mutex.
208 */
209int consumer_add_stream(struct lttng_consumer_stream *stream)
210{
211 int ret = 0;
212
213 pthread_mutex_lock(&consumer_data.lock);
214 /* Check if already exist */
215 if (consumer_find_stream(stream->key)) {
216 ret = -1;
217 goto end;
218 }
219 cds_list_add(&stream->list, &consumer_data.stream_list.head);
220 consumer_data.stream_count++;
221 consumer_data.need_update = 1;
222
223 switch (consumer_data.type) {
224 case LTTNG_CONSUMER_KERNEL:
225 break;
226 case LTTNG_CONSUMER_UST:
227 /* Streams are in CPU number order (we rely on this) */
228 stream->cpu = stream->chan->nr_streams++;
229 break;
230 default:
231 ERR("Unknown consumer_data type");
232 assert(0);
233 goto end;
234 }
235
236end:
237 pthread_mutex_unlock(&consumer_data.lock);
238 return ret;
239}
240
241/*
242 * Update a stream according to what we just received.
243 */
244void consumer_change_stream_state(int stream_key,
245 enum lttng_consumer_stream_state state)
246{
247 struct lttng_consumer_stream *stream;
248
249 pthread_mutex_lock(&consumer_data.lock);
250 stream = consumer_find_stream(stream_key);
251 if (stream) {
252 stream->state = state;
253 }
254 consumer_data.need_update = 1;
255 pthread_mutex_unlock(&consumer_data.lock);
256}
257
258/*
259 * Remove a channel from the global list protected by a mutex. This
260 * function is also responsible for freeing its data structures.
261 */
262void consumer_del_channel(struct lttng_consumer_channel *channel)
263{
264 int ret;
265
266 pthread_mutex_lock(&consumer_data.lock);
267
268 switch (consumer_data.type) {
269 case LTTNG_CONSUMER_KERNEL:
270 break;
271 case LTTNG_CONSUMER_UST:
272 lttng_ustconsumer_del_channel(channel);
273 break;
274 default:
275 ERR("Unknown consumer_data type");
276 assert(0);
277 goto end;
278 }
279
280 cds_list_del(&channel->list);
281 if (channel->mmap_base != NULL) {
282 ret = munmap(channel->mmap_base, channel->mmap_len);
283 if (ret != 0) {
284 perror("munmap");
285 }
286 }
b5c5fc29 287 if (channel->wait_fd >= 0 && !channel->wait_fd_is_copy) {
3bd1e081
MD
288 close(channel->wait_fd);
289 }
b5c5fc29
MD
290 if (channel->shm_fd >= 0 && channel->wait_fd != channel->shm_fd
291 && !channel->shm_fd_is_copy) {
3bd1e081
MD
292 close(channel->shm_fd);
293 }
294 free(channel);
295end:
296 pthread_mutex_unlock(&consumer_data.lock);
297}
298
299struct lttng_consumer_channel *consumer_allocate_channel(
300 int channel_key,
301 int shm_fd, int wait_fd,
302 uint64_t mmap_len,
303 uint64_t max_sb_size)
304{
305 struct lttng_consumer_channel *channel;
306 int ret;
307
276b26d1 308 channel = zmalloc(sizeof(*channel));
3bd1e081
MD
309 if (channel == NULL) {
310 perror("malloc struct lttng_consumer_channel");
311 goto end;
312 }
313 channel->key = channel_key;
314 channel->shm_fd = shm_fd;
315 channel->wait_fd = wait_fd;
316 channel->mmap_len = mmap_len;
317 channel->max_sb_size = max_sb_size;
318 channel->refcount = 0;
319 channel->nr_streams = 0;
320
321 switch (consumer_data.type) {
322 case LTTNG_CONSUMER_KERNEL:
323 channel->mmap_base = NULL;
324 channel->mmap_len = 0;
325 break;
326 case LTTNG_CONSUMER_UST:
327 ret = lttng_ustconsumer_allocate_channel(channel);
328 if (ret) {
329 free(channel);
330 return NULL;
331 }
332 break;
333 default:
334 ERR("Unknown consumer_data type");
335 assert(0);
336 goto end;
337 }
338 DBG("Allocated channel (key %d, shm_fd %d, wait_fd %d, mmap_len %llu, max_sb_size %llu)",
339 channel->key,
340 channel->shm_fd,
341 channel->wait_fd,
342 (unsigned long long) channel->mmap_len,
343 (unsigned long long) channel->max_sb_size);
344end:
345 return channel;
346}
347
348/*
349 * Add a channel to the global list protected by a mutex.
350 */
351int consumer_add_channel(struct lttng_consumer_channel *channel)
352{
353 int ret = 0;
354
355 pthread_mutex_lock(&consumer_data.lock);
356 /* Check if already exist */
357 if (consumer_find_channel(channel->key)) {
358 ret = -1;
359 goto end;
360 }
361 cds_list_add(&channel->list, &consumer_data.channel_list.head);
362end:
363 pthread_mutex_unlock(&consumer_data.lock);
364 return ret;
365}
366
367/*
368 * Allocate the pollfd structure and the local view of the out fds to avoid
369 * doing a lookup in the linked list and concurrency issues when writing is
370 * needed. Called with consumer_data.lock held.
371 *
372 * Returns the number of fds in the structures.
373 */
374int consumer_update_poll_array(
375 struct lttng_consumer_local_data *ctx, struct pollfd **pollfd,
376 struct lttng_consumer_stream **local_stream)
377{
378 struct lttng_consumer_stream *iter;
379 int i = 0;
380
381 DBG("Updating poll fd array");
382 cds_list_for_each_entry(iter, &consumer_data.stream_list.head, list) {
383 if (iter->state != LTTNG_CONSUMER_ACTIVE_STREAM) {
384 continue;
385 }
386 DBG("Active FD %d", iter->wait_fd);
387 (*pollfd)[i].fd = iter->wait_fd;
388 (*pollfd)[i].events = POLLIN | POLLPRI;
389 local_stream[i] = iter;
390 i++;
391 }
392
393 /*
394 * Insert the consumer_poll_pipe at the end of the array and don't
395 * increment i so nb_fd is the number of real FD.
396 */
397 (*pollfd)[i].fd = ctx->consumer_poll_pipe[0];
398 (*pollfd)[i].events = POLLIN;
399 return i;
400}
401
402/*
403 * Poll on the should_quit pipe and the command socket return -1 on error and
404 * should exit, 0 if data is available on the command socket
405 */
406int lttng_consumer_poll_socket(struct pollfd *consumer_sockpoll)
407{
408 int num_rdy;
409
410 num_rdy = poll(consumer_sockpoll, 2, -1);
411 if (num_rdy == -1) {
412 perror("Poll error");
413 goto exit;
414 }
415 if (consumer_sockpoll[0].revents == POLLIN) {
416 DBG("consumer_should_quit wake up");
417 goto exit;
418 }
419 return 0;
420
421exit:
422 return -1;
423}
424
425/*
426 * Set the error socket.
427 */
428void lttng_consumer_set_error_sock(
429 struct lttng_consumer_local_data *ctx, int sock)
430{
431 ctx->consumer_error_socket = sock;
432}
433
434/*
435 * Set the command socket path.
436 */
437
438void lttng_consumer_set_command_sock_path(
439 struct lttng_consumer_local_data *ctx, char *sock)
440{
441 ctx->consumer_command_sock_path = sock;
442}
443
444/*
445 * Send return code to the session daemon.
446 * If the socket is not defined, we return 0, it is not a fatal error
447 */
448int lttng_consumer_send_error(
449 struct lttng_consumer_local_data *ctx, int cmd)
450{
451 if (ctx->consumer_error_socket > 0) {
452 return lttcomm_send_unix_sock(ctx->consumer_error_socket, &cmd,
453 sizeof(enum lttcomm_sessiond_command));
454 }
455
456 return 0;
457}
458
459/*
460 * Close all the tracefiles and stream fds, should be called when all instances
461 * are destroyed.
462 */
463void lttng_consumer_cleanup(void)
464{
465 struct lttng_consumer_stream *iter, *tmp;
466 struct lttng_consumer_channel *citer, *ctmp;
467
468 /*
469 * close all outfd. Called when there are no more threads
470 * running (after joining on the threads), no need to protect
471 * list iteration with mutex.
472 */
473 cds_list_for_each_entry_safe(iter, tmp,
474 &consumer_data.stream_list.head, list) {
475 consumer_del_stream(iter);
476 }
477 cds_list_for_each_entry_safe(citer, ctmp,
478 &consumer_data.channel_list.head, list) {
479 consumer_del_channel(citer);
480 }
481}
482
483/*
484 * Called from signal handler.
485 */
486void lttng_consumer_should_exit(struct lttng_consumer_local_data *ctx)
487{
488 int ret;
489 consumer_quit = 1;
490 ret = write(ctx->consumer_should_quit[1], "4", 1);
491 if (ret < 0) {
492 perror("write consumer quit");
493 }
494}
495
496void lttng_consumer_sync_trace_file(
497 struct lttng_consumer_stream *stream, off_t orig_offset)
498{
499 int outfd = stream->out_fd;
500
501 /*
502 * This does a blocking write-and-wait on any page that belongs to the
503 * subbuffer prior to the one we just wrote.
504 * Don't care about error values, as these are just hints and ways to
505 * limit the amount of page cache used.
506 */
507 if (orig_offset < stream->chan->max_sb_size) {
508 return;
509 }
510 sync_file_range(outfd, orig_offset - stream->chan->max_sb_size,
511 stream->chan->max_sb_size,
512 SYNC_FILE_RANGE_WAIT_BEFORE
513 | SYNC_FILE_RANGE_WRITE
514 | SYNC_FILE_RANGE_WAIT_AFTER);
515 /*
516 * Give hints to the kernel about how we access the file:
517 * POSIX_FADV_DONTNEED : we won't re-access data in a near future after
518 * we write it.
519 *
520 * We need to call fadvise again after the file grows because the
521 * kernel does not seem to apply fadvise to non-existing parts of the
522 * file.
523 *
524 * Call fadvise _after_ having waited for the page writeback to
525 * complete because the dirty page writeback semantic is not well
526 * defined. So it can be expected to lead to lower throughput in
527 * streaming.
528 */
529 posix_fadvise(outfd, orig_offset - stream->chan->max_sb_size,
530 stream->chan->max_sb_size, POSIX_FADV_DONTNEED);
531}
532
533/*
534 * Initialise the necessary environnement :
535 * - create a new context
536 * - create the poll_pipe
537 * - create the should_quit pipe (for signal handler)
538 * - create the thread pipe (for splice)
539 *
540 * Takes a function pointer as argument, this function is called when data is
541 * available on a buffer. This function is responsible to do the
542 * kernctl_get_next_subbuf, read the data with mmap or splice depending on the
543 * buffer configuration and then kernctl_put_next_subbuf at the end.
544 *
545 * Returns a pointer to the new context or NULL on error.
546 */
547struct lttng_consumer_local_data *lttng_consumer_create(
548 enum lttng_consumer_type type,
d41f73b7
MD
549 int (*buffer_ready)(struct lttng_consumer_stream *stream,
550 struct lttng_consumer_local_data *ctx),
3bd1e081
MD
551 int (*recv_channel)(struct lttng_consumer_channel *channel),
552 int (*recv_stream)(struct lttng_consumer_stream *stream),
553 int (*update_stream)(int stream_key, uint32_t state))
554{
555 int ret, i;
556 struct lttng_consumer_local_data *ctx;
557
558 assert(consumer_data.type == LTTNG_CONSUMER_UNKNOWN ||
559 consumer_data.type == type);
560 consumer_data.type = type;
561
562 ctx = malloc(sizeof(struct lttng_consumer_local_data));
563 if (ctx == NULL) {
564 perror("allocating context");
565 goto error;
566 }
567
568 ctx->consumer_error_socket = -1;
569 /* assign the callbacks */
570 ctx->on_buffer_ready = buffer_ready;
571 ctx->on_recv_channel = recv_channel;
572 ctx->on_recv_stream = recv_stream;
573 ctx->on_update_stream = update_stream;
574
575 ret = pipe(ctx->consumer_poll_pipe);
576 if (ret < 0) {
577 perror("Error creating poll pipe");
578 goto error_poll_pipe;
579 }
580
581 ret = pipe(ctx->consumer_should_quit);
582 if (ret < 0) {
583 perror("Error creating recv pipe");
584 goto error_quit_pipe;
585 }
586
587 ret = pipe(ctx->consumer_thread_pipe);
588 if (ret < 0) {
589 perror("Error creating thread pipe");
590 goto error_thread_pipe;
591 }
592
593 return ctx;
594
595
596error_thread_pipe:
597 for (i = 0; i < 2; i++) {
598 int err;
599
600 err = close(ctx->consumer_should_quit[i]);
601 assert(!err);
602 }
603error_quit_pipe:
604 for (i = 0; i < 2; i++) {
605 int err;
606
607 err = close(ctx->consumer_poll_pipe[i]);
608 assert(!err);
609 }
610error_poll_pipe:
611 free(ctx);
612error:
613 return NULL;
614}
615
616/*
617 * Close all fds associated with the instance and free the context.
618 */
619void lttng_consumer_destroy(struct lttng_consumer_local_data *ctx)
620{
621 close(ctx->consumer_error_socket);
622 close(ctx->consumer_thread_pipe[0]);
623 close(ctx->consumer_thread_pipe[1]);
624 close(ctx->consumer_poll_pipe[0]);
625 close(ctx->consumer_poll_pipe[1]);
626 close(ctx->consumer_should_quit[0]);
627 close(ctx->consumer_should_quit[1]);
628 unlink(ctx->consumer_command_sock_path);
629 free(ctx);
630}
631
632/*
633 * Mmap the ring buffer, read it and write the data to the tracefile.
634 *
635 * Returns the number of bytes written
636 */
637int lttng_consumer_on_read_subbuffer_mmap(
638 struct lttng_consumer_local_data *ctx,
639 struct lttng_consumer_stream *stream, unsigned long len)
640{
641 switch (consumer_data.type) {
642 case LTTNG_CONSUMER_KERNEL:
643 return lttng_kconsumer_on_read_subbuffer_mmap(ctx, stream, len);
644 case LTTNG_CONSUMER_UST:
645 return lttng_ustconsumer_on_read_subbuffer_mmap(ctx, stream, len);
646 default:
647 ERR("Unknown consumer_data type");
648 assert(0);
649 }
650}
651
652/*
653 * Splice the data from the ring buffer to the tracefile.
654 *
655 * Returns the number of bytes spliced.
656 */
657int lttng_consumer_on_read_subbuffer_splice(
658 struct lttng_consumer_local_data *ctx,
659 struct lttng_consumer_stream *stream, unsigned long len)
660{
661 switch (consumer_data.type) {
662 case LTTNG_CONSUMER_KERNEL:
663 return lttng_kconsumer_on_read_subbuffer_splice(ctx, stream, len);
664 case LTTNG_CONSUMER_UST:
665 return -ENOSYS;
666 default:
667 ERR("Unknown consumer_data type");
668 assert(0);
669 return -ENOSYS;
670 }
671
672}
673
674/*
675 * Take a snapshot for a specific fd
676 *
677 * Returns 0 on success, < 0 on error
678 */
679int lttng_consumer_take_snapshot(struct lttng_consumer_local_data *ctx,
680 struct lttng_consumer_stream *stream)
681{
682 switch (consumer_data.type) {
683 case LTTNG_CONSUMER_KERNEL:
684 return lttng_kconsumer_take_snapshot(ctx, stream);
685 case LTTNG_CONSUMER_UST:
686 return lttng_ustconsumer_take_snapshot(ctx, stream);
687 default:
688 ERR("Unknown consumer_data type");
689 assert(0);
690 return -ENOSYS;
691 }
692
693}
694
695/*
696 * Get the produced position
697 *
698 * Returns 0 on success, < 0 on error
699 */
700int lttng_consumer_get_produced_snapshot(
701 struct lttng_consumer_local_data *ctx,
702 struct lttng_consumer_stream *stream,
703 unsigned long *pos)
704{
705 switch (consumer_data.type) {
706 case LTTNG_CONSUMER_KERNEL:
707 return lttng_kconsumer_get_produced_snapshot(ctx, stream, pos);
708 case LTTNG_CONSUMER_UST:
709 return lttng_ustconsumer_get_produced_snapshot(ctx, stream, pos);
710 default:
711 ERR("Unknown consumer_data type");
712 assert(0);
713 return -ENOSYS;
714 }
715}
716
717int lttng_consumer_recv_cmd(struct lttng_consumer_local_data *ctx,
718 int sock, struct pollfd *consumer_sockpoll)
719{
720 switch (consumer_data.type) {
721 case LTTNG_CONSUMER_KERNEL:
722 return lttng_kconsumer_recv_cmd(ctx, sock, consumer_sockpoll);
723 case LTTNG_CONSUMER_UST:
724 return lttng_ustconsumer_recv_cmd(ctx, sock, consumer_sockpoll);
725 default:
726 ERR("Unknown consumer_data type");
727 assert(0);
728 return -ENOSYS;
729 }
730}
731
732/*
733 * This thread polls the fds in the ltt_fd_list to consume the data and write
734 * it to tracefile if necessary.
735 */
736void *lttng_consumer_thread_poll_fds(void *data)
737{
738 int num_rdy, num_hup, high_prio, ret, i;
739 struct pollfd *pollfd = NULL;
740 /* local view of the streams */
741 struct lttng_consumer_stream **local_stream = NULL;
742 /* local view of consumer_data.fds_count */
743 int nb_fd = 0;
744 char tmp;
745 int tmp2;
746 struct lttng_consumer_local_data *ctx = data;
747
748 local_stream = malloc(sizeof(struct lttng_consumer_stream));
749
750 while (1) {
751 high_prio = 0;
752 num_hup = 0;
753
754 /*
755 * the ltt_fd_list has been updated, we need to update our
756 * local array as well
757 */
758 pthread_mutex_lock(&consumer_data.lock);
759 if (consumer_data.need_update) {
760 if (pollfd != NULL) {
761 free(pollfd);
762 pollfd = NULL;
763 }
764 if (local_stream != NULL) {
765 free(local_stream);
766 local_stream = NULL;
767 }
768
769 /* allocate for all fds + 1 for the consumer_poll_pipe */
770 pollfd = malloc((consumer_data.stream_count + 1) * sizeof(struct pollfd));
771 if (pollfd == NULL) {
772 perror("pollfd malloc");
773 pthread_mutex_unlock(&consumer_data.lock);
774 goto end;
775 }
776
777 /* allocate for all fds + 1 for the consumer_poll_pipe */
778 local_stream = malloc((consumer_data.stream_count + 1) *
779 sizeof(struct lttng_consumer_stream));
780 if (local_stream == NULL) {
781 perror("local_stream malloc");
782 pthread_mutex_unlock(&consumer_data.lock);
783 goto end;
784 }
785 ret = consumer_update_poll_array(ctx, &pollfd, local_stream);
786 if (ret < 0) {
787 ERR("Error in allocating pollfd or local_outfds");
788 lttng_consumer_send_error(ctx, CONSUMERD_POLL_ERROR);
789 pthread_mutex_unlock(&consumer_data.lock);
790 goto end;
791 }
792 nb_fd = ret;
793 consumer_data.need_update = 0;
794 }
795 pthread_mutex_unlock(&consumer_data.lock);
796
797 /* poll on the array of fds */
798 DBG("polling on %d fd", nb_fd + 1);
799 num_rdy = poll(pollfd, nb_fd + 1, consumer_poll_timeout);
800 DBG("poll num_rdy : %d", num_rdy);
801 if (num_rdy == -1) {
802 perror("Poll error");
803 lttng_consumer_send_error(ctx, CONSUMERD_POLL_ERROR);
804 goto end;
805 } else if (num_rdy == 0) {
806 DBG("Polling thread timed out");
807 goto end;
808 }
809
d41f73b7 810 /* No FDs and consumer_quit, consumer_cleanup the thread */
3bd1e081
MD
811 if (nb_fd == 0 && consumer_quit == 1) {
812 goto end;
813 }
814
815 /*
816 * If the consumer_poll_pipe triggered poll go
817 * directly to the beginning of the loop to update the
818 * array. We want to prioritize array update over
819 * low-priority reads.
820 */
d41f73b7 821 if (pollfd[nb_fd].revents & POLLIN) {
3bd1e081
MD
822 DBG("consumer_poll_pipe wake up");
823 tmp2 = read(ctx->consumer_poll_pipe[0], &tmp, 1);
824 if (tmp2 < 0) {
d41f73b7 825 perror("read consumer poll");
3bd1e081
MD
826 }
827 continue;
828 }
829
830 /* Take care of high priority channels first. */
831 for (i = 0; i < nb_fd; i++) {
d41f73b7
MD
832 if (pollfd[i].revents & POLLPRI) {
833 DBG("Urgent read on fd %d", pollfd[i].fd);
834 high_prio = 1;
835 ret = ctx->on_buffer_ready(local_stream[i], ctx);
836 /* it's ok to have an unavailable sub-buffer */
837 if (ret == EAGAIN) {
838 ret = 0;
839 }
840 } else if (pollfd[i].revents & POLLERR) {
3bd1e081
MD
841 ERR("Error returned in polling fd %d.", pollfd[i].fd);
842 consumer_del_stream(local_stream[i]);
843 num_hup++;
d41f73b7
MD
844 } else if (pollfd[i].revents & POLLNVAL) {
845 ERR("Polling fd %d tells fd is not open.", pollfd[i].fd);
3bd1e081
MD
846 consumer_del_stream(local_stream[i]);
847 num_hup++;
1c3c14ac
MD
848 } else if ((pollfd[i].revents & POLLHUP) &&
849 !(pollfd[i].revents & POLLIN)) {
d056b477
MD
850 if (consumer_data.type == LTTNG_CONSUMER_UST) {
851 DBG("Polling fd %d tells it has hung up. Attempting flush and read.",
852 pollfd[i].fd);
853 if (!local_stream[i]->hangup_flush_done) {
854 lttng_ustconsumer_on_stream_hangup(local_stream[i]);
855 /* try reading after flush */
856 ret = ctx->on_buffer_ready(local_stream[i], ctx);
857 /* it's ok to have an unavailable sub-buffer */
858 if (ret == EAGAIN) {
859 ret = 0;
860 }
861 }
862 } else {
863 DBG("Polling fd %d tells it has hung up.", pollfd[i].fd);
864 }
3bd1e081
MD
865 consumer_del_stream(local_stream[i]);
866 num_hup++;
3bd1e081
MD
867 }
868 }
869
870 /* If every buffer FD has hung up, we end the read loop here */
871 if (nb_fd > 0 && num_hup == nb_fd) {
872 DBG("every buffer FD has hung up\n");
873 if (consumer_quit == 1) {
874 goto end;
875 }
876 continue;
877 }
878
879 /* Take care of low priority channels. */
880 if (high_prio == 0) {
881 for (i = 0; i < nb_fd; i++) {
d41f73b7 882 if (pollfd[i].revents & POLLIN) {
3bd1e081 883 DBG("Normal read on fd %d", pollfd[i].fd);
d41f73b7 884 ret = ctx->on_buffer_ready(local_stream[i], ctx);
3bd1e081
MD
885 /* it's ok to have an unavailable subbuffer */
886 if (ret == EAGAIN) {
887 ret = 0;
888 }
889 }
890 }
891 }
892 }
893end:
894 DBG("polling thread exiting");
895 if (pollfd != NULL) {
896 free(pollfd);
897 pollfd = NULL;
898 }
899 if (local_stream != NULL) {
900 free(local_stream);
901 local_stream = NULL;
902 }
903 return NULL;
904}
905
906/*
907 * This thread listens on the consumerd socket and receives the file
908 * descriptors from the session daemon.
909 */
910void *lttng_consumer_thread_receive_fds(void *data)
911{
912 int sock, client_socket, ret;
913 /*
914 * structure to poll for incoming data on communication socket avoids
915 * making blocking sockets.
916 */
917 struct pollfd consumer_sockpoll[2];
918 struct lttng_consumer_local_data *ctx = data;
919
920 DBG("Creating command socket %s", ctx->consumer_command_sock_path);
921 unlink(ctx->consumer_command_sock_path);
922 client_socket = lttcomm_create_unix_sock(ctx->consumer_command_sock_path);
923 if (client_socket < 0) {
924 ERR("Cannot create command socket");
925 goto end;
926 }
927
928 ret = lttcomm_listen_unix_sock(client_socket);
929 if (ret < 0) {
930 goto end;
931 }
932
32258573 933 DBG("Sending ready command to lttng-sessiond");
3bd1e081
MD
934 ret = lttng_consumer_send_error(ctx, CONSUMERD_COMMAND_SOCK_READY);
935 /* return < 0 on error, but == 0 is not fatal */
936 if (ret < 0) {
32258573 937 ERR("Error sending ready command to lttng-sessiond");
3bd1e081
MD
938 goto end;
939 }
940
941 ret = fcntl(client_socket, F_SETFL, O_NONBLOCK);
942 if (ret < 0) {
943 perror("fcntl O_NONBLOCK");
944 goto end;
945 }
946
947 /* prepare the FDs to poll : to client socket and the should_quit pipe */
948 consumer_sockpoll[0].fd = ctx->consumer_should_quit[0];
949 consumer_sockpoll[0].events = POLLIN | POLLPRI;
950 consumer_sockpoll[1].fd = client_socket;
951 consumer_sockpoll[1].events = POLLIN | POLLPRI;
952
953 if (lttng_consumer_poll_socket(consumer_sockpoll) < 0) {
954 goto end;
955 }
956 DBG("Connection on client_socket");
957
958 /* Blocking call, waiting for transmission */
959 sock = lttcomm_accept_unix_sock(client_socket);
960 if (sock <= 0) {
961 WARN("On accept");
962 goto end;
963 }
964 ret = fcntl(sock, F_SETFL, O_NONBLOCK);
965 if (ret < 0) {
966 perror("fcntl O_NONBLOCK");
967 goto end;
968 }
969
970 /* update the polling structure to poll on the established socket */
971 consumer_sockpoll[1].fd = sock;
972 consumer_sockpoll[1].events = POLLIN | POLLPRI;
973
974 while (1) {
975 if (lttng_consumer_poll_socket(consumer_sockpoll) < 0) {
976 goto end;
977 }
978 DBG("Incoming command on sock");
979 ret = lttng_consumer_recv_cmd(ctx, sock, consumer_sockpoll);
980 if (ret == -ENOENT) {
981 DBG("Received STOP command");
982 goto end;
983 }
984 if (ret < 0) {
985 ERR("Communication interrupted on command socket");
986 goto end;
987 }
988 if (consumer_quit) {
989 DBG("consumer_thread_receive_fds received quit from signal");
990 goto end;
991 }
992 DBG("received fds on sock");
993 }
994end:
995 DBG("consumer_thread_receive_fds exiting");
996
997 /*
998 * when all fds have hung up, the polling thread
999 * can exit cleanly
1000 */
1001 consumer_quit = 1;
1002
1003 /*
1004 * 2s of grace period, if no polling events occur during
1005 * this period, the polling thread will exit even if there
1006 * are still open FDs (should not happen, but safety mechanism).
1007 */
1008 consumer_poll_timeout = LTTNG_CONSUMER_POLL_TIMEOUT;
1009
1010 /* wake up the polling thread */
1011 ret = write(ctx->consumer_poll_pipe[1], "4", 1);
1012 if (ret < 0) {
1013 perror("poll pipe write");
1014 }
1015 return NULL;
1016}
d41f73b7
MD
1017
1018int lttng_consumer_read_subbuffer(struct lttng_consumer_stream *stream,
1019 struct lttng_consumer_local_data *ctx)
1020{
1021 switch (consumer_data.type) {
1022 case LTTNG_CONSUMER_KERNEL:
1023 return lttng_kconsumer_read_subbuffer(stream, ctx);
1024 case LTTNG_CONSUMER_UST:
1025 return lttng_ustconsumer_read_subbuffer(stream, ctx);
1026 default:
1027 ERR("Unknown consumer_data type");
1028 assert(0);
1029 return -ENOSYS;
1030 }
1031}
1032
1033int lttng_consumer_on_recv_stream(struct lttng_consumer_stream *stream)
1034{
1035 switch (consumer_data.type) {
1036 case LTTNG_CONSUMER_KERNEL:
1037 return lttng_kconsumer_on_recv_stream(stream);
1038 case LTTNG_CONSUMER_UST:
1039 return lttng_ustconsumer_on_recv_stream(stream);
1040 default:
1041 ERR("Unknown consumer_data type");
1042 assert(0);
1043 return -ENOSYS;
1044 }
1045}
This page took 0.099784 seconds and 4 git commands to generate.