2 * Copyright (C) 2011 - Julien Desfossez <julien.desfossez@polymtl.ca>
3 * Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
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
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.
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.
28 #include <sys/socket.h>
29 #include <sys/types.h>
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>
39 struct 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
),
44 .type
= LTTNG_CONSUMER_UNKNOWN
,
47 /* timeout parameter, to control the polling thread grace period. */
48 int consumer_poll_timeout
= -1;
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
56 volatile int consumer_quit
= 0;
59 * Find a stream. The consumer_data.lock must be locked during this
62 static struct lttng_consumer_stream
*consumer_find_stream(int key
)
64 struct lttng_consumer_stream
*iter
;
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
);
75 static struct lttng_consumer_channel
*consumer_find_channel(int key
)
77 struct lttng_consumer_channel
*iter
;
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
);
89 * Remove a stream from the global list protected by a mutex. This
90 * function is also responsible for freeing its data structures.
92 void consumer_del_stream(struct lttng_consumer_stream
*stream
)
95 struct lttng_consumer_channel
*free_chan
= NULL
;
97 pthread_mutex_lock(&consumer_data
.lock
);
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
);
108 case LTTNG_CONSUMER_UST
:
109 lttng_ustconsumer_del_stream(stream
);
112 ERR("Unknown consumer_data type");
117 cds_list_del(&stream
->list
);
118 if (consumer_data
.stream_count
<= 0) {
121 consumer_data
.stream_count
--;
125 if (stream
->out_fd
>= 0) {
126 close(stream
->out_fd
);
128 if (stream
->wait_fd
>= 0 && !stream
->wait_fd_is_copy
) {
129 close(stream
->wait_fd
);
131 if (stream
->shm_fd
>= 0 && stream
->wait_fd
!= stream
->shm_fd
132 && !stream
->shm_fd_is_copy
) {
133 close(stream
->shm_fd
);
135 if (!--stream
->chan
->refcount
)
136 free_chan
= stream
->chan
;
139 consumer_data
.need_update
= 1;
140 pthread_mutex_unlock(&consumer_data
.lock
);
143 consumer_del_channel(free_chan
);
146 struct 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
,
151 enum lttng_event_output output
,
152 const char *path_name
)
154 struct lttng_consumer_stream
*stream
;
157 stream
= malloc(sizeof(*stream
));
158 if (stream
== NULL
) {
159 perror("malloc struct lttng_consumer_stream");
162 stream
->chan
= consumer_find_channel(channel_key
);
164 perror("Unable to find channel key");
167 stream
->chan
->refcount
++;
168 stream
->key
= stream_key
;
169 stream
->shm_fd
= shm_fd
;
170 stream
->wait_fd
= wait_fd
;
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';
180 switch (consumer_data
.type
) {
181 case LTTNG_CONSUMER_KERNEL
:
183 case LTTNG_CONSUMER_UST
:
184 stream
->cpu
= stream
->chan
->cpucount
++;
185 ret
= lttng_ustconsumer_allocate_stream(stream
);
192 ERR("Unknown consumer_data type");
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
,
200 (unsigned long long) stream
->mmap_len
,
207 * Add a stream to the global list protected by a mutex.
209 int consumer_add_stream(struct lttng_consumer_stream
*stream
)
213 pthread_mutex_lock(&consumer_data
.lock
);
214 /* Check if already exist */
215 if (consumer_find_stream(stream
->key
)) {
219 cds_list_add(&stream
->list
, &consumer_data
.stream_list
.head
);
220 consumer_data
.stream_count
++;
221 consumer_data
.need_update
= 1;
223 switch (consumer_data
.type
) {
224 case LTTNG_CONSUMER_KERNEL
:
226 case LTTNG_CONSUMER_UST
:
227 /* Streams are in CPU number order (we rely on this) */
228 stream
->cpu
= stream
->chan
->nr_streams
++;
231 ERR("Unknown consumer_data type");
237 pthread_mutex_unlock(&consumer_data
.lock
);
242 * Update a stream according to what we just received.
244 void consumer_change_stream_state(int stream_key
,
245 enum lttng_consumer_stream_state state
)
247 struct lttng_consumer_stream
*stream
;
249 pthread_mutex_lock(&consumer_data
.lock
);
250 stream
= consumer_find_stream(stream_key
);
252 stream
->state
= state
;
254 consumer_data
.need_update
= 1;
255 pthread_mutex_unlock(&consumer_data
.lock
);
259 * Remove a channel from the global list protected by a mutex. This
260 * function is also responsible for freeing its data structures.
262 void consumer_del_channel(struct lttng_consumer_channel
*channel
)
266 pthread_mutex_lock(&consumer_data
.lock
);
268 switch (consumer_data
.type
) {
269 case LTTNG_CONSUMER_KERNEL
:
271 case LTTNG_CONSUMER_UST
:
272 lttng_ustconsumer_del_channel(channel
);
275 ERR("Unknown consumer_data type");
280 cds_list_del(&channel
->list
);
281 if (channel
->mmap_base
!= NULL
) {
282 ret
= munmap(channel
->mmap_base
, channel
->mmap_len
);
287 if (channel
->wait_fd
>= 0 && !channel
->wait_fd_is_copy
) {
288 close(channel
->wait_fd
);
290 if (channel
->shm_fd
>= 0 && channel
->wait_fd
!= channel
->shm_fd
291 && !channel
->shm_fd_is_copy
) {
292 close(channel
->shm_fd
);
296 pthread_mutex_unlock(&consumer_data
.lock
);
299 struct lttng_consumer_channel
*consumer_allocate_channel(
301 int shm_fd
, int wait_fd
,
303 uint64_t max_sb_size
)
305 struct lttng_consumer_channel
*channel
;
308 channel
= malloc(sizeof(*channel
));
309 if (channel
== NULL
) {
310 perror("malloc struct lttng_consumer_channel");
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;
321 switch (consumer_data
.type
) {
322 case LTTNG_CONSUMER_KERNEL
:
323 channel
->mmap_base
= NULL
;
324 channel
->mmap_len
= 0;
326 case LTTNG_CONSUMER_UST
:
327 ret
= lttng_ustconsumer_allocate_channel(channel
);
334 ERR("Unknown consumer_data type");
338 DBG("Allocated channel (key %d, shm_fd %d, wait_fd %d, mmap_len %llu, max_sb_size %llu)",
342 (unsigned long long) channel
->mmap_len
,
343 (unsigned long long) channel
->max_sb_size
);
349 * Add a channel to the global list protected by a mutex.
351 int consumer_add_channel(struct lttng_consumer_channel
*channel
)
355 pthread_mutex_lock(&consumer_data
.lock
);
356 /* Check if already exist */
357 if (consumer_find_channel(channel
->key
)) {
361 cds_list_add(&channel
->list
, &consumer_data
.channel_list
.head
);
363 pthread_mutex_unlock(&consumer_data
.lock
);
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.
372 * Returns the number of fds in the structures.
374 int consumer_update_poll_array(
375 struct lttng_consumer_local_data
*ctx
, struct pollfd
**pollfd
,
376 struct lttng_consumer_stream
**local_stream
)
378 struct lttng_consumer_stream
*iter
;
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
) {
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
;
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.
397 (*pollfd
)[i
].fd
= ctx
->consumer_poll_pipe
[0];
398 (*pollfd
)[i
].events
= POLLIN
;
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
406 int lttng_consumer_poll_socket(struct pollfd
*consumer_sockpoll
)
410 num_rdy
= poll(consumer_sockpoll
, 2, -1);
412 perror("Poll error");
415 if (consumer_sockpoll
[0].revents
== POLLIN
) {
416 DBG("consumer_should_quit wake up");
426 * Set the error socket.
428 void lttng_consumer_set_error_sock(
429 struct lttng_consumer_local_data
*ctx
, int sock
)
431 ctx
->consumer_error_socket
= sock
;
435 * Set the command socket path.
438 void lttng_consumer_set_command_sock_path(
439 struct lttng_consumer_local_data
*ctx
, char *sock
)
441 ctx
->consumer_command_sock_path
= sock
;
445 * Send return code to the session daemon.
446 * If the socket is not defined, we return 0, it is not a fatal error
448 int lttng_consumer_send_error(
449 struct lttng_consumer_local_data
*ctx
, int cmd
)
451 if (ctx
->consumer_error_socket
> 0) {
452 return lttcomm_send_unix_sock(ctx
->consumer_error_socket
, &cmd
,
453 sizeof(enum lttcomm_sessiond_command
));
460 * Close all the tracefiles and stream fds, should be called when all instances
463 void lttng_consumer_cleanup(void)
465 struct lttng_consumer_stream
*iter
, *tmp
;
466 struct lttng_consumer_channel
*citer
, *ctmp
;
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.
473 cds_list_for_each_entry_safe(iter
, tmp
,
474 &consumer_data
.stream_list
.head
, list
) {
475 consumer_del_stream(iter
);
477 cds_list_for_each_entry_safe(citer
, ctmp
,
478 &consumer_data
.channel_list
.head
, list
) {
479 consumer_del_channel(citer
);
484 * Called from signal handler.
486 void lttng_consumer_should_exit(struct lttng_consumer_local_data
*ctx
)
490 ret
= write(ctx
->consumer_should_quit
[1], "4", 1);
492 perror("write consumer quit");
496 void lttng_consumer_sync_trace_file(
497 struct lttng_consumer_stream
*stream
, off_t orig_offset
)
499 int outfd
= stream
->out_fd
;
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.
507 if (orig_offset
< stream
->chan
->max_sb_size
) {
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
);
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
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
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
529 posix_fadvise(outfd
, orig_offset
- stream
->chan
->max_sb_size
,
530 stream
->chan
->max_sb_size
, POSIX_FADV_DONTNEED
);
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)
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.
545 * Returns a pointer to the new context or NULL on error.
547 struct lttng_consumer_local_data
*lttng_consumer_create(
548 enum lttng_consumer_type type
,
549 int (*buffer_ready
)(struct lttng_consumer_stream
*stream
,
550 struct lttng_consumer_local_data
*ctx
),
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
))
556 struct lttng_consumer_local_data
*ctx
;
558 assert(consumer_data
.type
== LTTNG_CONSUMER_UNKNOWN
||
559 consumer_data
.type
== type
);
560 consumer_data
.type
= type
;
562 ctx
= malloc(sizeof(struct lttng_consumer_local_data
));
564 perror("allocating context");
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
;
575 ret
= pipe(ctx
->consumer_poll_pipe
);
577 perror("Error creating poll pipe");
578 goto error_poll_pipe
;
581 ret
= pipe(ctx
->consumer_should_quit
);
583 perror("Error creating recv pipe");
584 goto error_quit_pipe
;
587 ret
= pipe(ctx
->consumer_thread_pipe
);
589 perror("Error creating thread pipe");
590 goto error_thread_pipe
;
597 for (i
= 0; i
< 2; i
++) {
600 err
= close(ctx
->consumer_should_quit
[i
]);
604 for (i
= 0; i
< 2; i
++) {
607 err
= close(ctx
->consumer_poll_pipe
[i
]);
617 * Close all fds associated with the instance and free the context.
619 void lttng_consumer_destroy(struct lttng_consumer_local_data
*ctx
)
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
);
633 * Mmap the ring buffer, read it and write the data to the tracefile.
635 * Returns the number of bytes written
637 int lttng_consumer_on_read_subbuffer_mmap(
638 struct lttng_consumer_local_data
*ctx
,
639 struct lttng_consumer_stream
*stream
, unsigned long len
)
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
);
647 ERR("Unknown consumer_data type");
653 * Splice the data from the ring buffer to the tracefile.
655 * Returns the number of bytes spliced.
657 int lttng_consumer_on_read_subbuffer_splice(
658 struct lttng_consumer_local_data
*ctx
,
659 struct lttng_consumer_stream
*stream
, unsigned long len
)
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
:
667 ERR("Unknown consumer_data type");
675 * Take a snapshot for a specific fd
677 * Returns 0 on success, < 0 on error
679 int lttng_consumer_take_snapshot(struct lttng_consumer_local_data
*ctx
,
680 struct lttng_consumer_stream
*stream
)
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
);
688 ERR("Unknown consumer_data type");
696 * Get the produced position
698 * Returns 0 on success, < 0 on error
700 int lttng_consumer_get_produced_snapshot(
701 struct lttng_consumer_local_data
*ctx
,
702 struct lttng_consumer_stream
*stream
,
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
);
711 ERR("Unknown consumer_data type");
717 int lttng_consumer_recv_cmd(struct lttng_consumer_local_data
*ctx
,
718 int sock
, struct pollfd
*consumer_sockpoll
)
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
);
726 ERR("Unknown consumer_data type");
733 * This thread polls the fds in the ltt_fd_list to consume the data and write
734 * it to tracefile if necessary.
736 void *lttng_consumer_thread_poll_fds(void *data
)
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 */
746 struct lttng_consumer_local_data
*ctx
= data
;
748 local_stream
= malloc(sizeof(struct lttng_consumer_stream
));
755 * the ltt_fd_list has been updated, we need to update our
756 * local array as well
758 pthread_mutex_lock(&consumer_data
.lock
);
759 if (consumer_data
.need_update
) {
760 if (pollfd
!= NULL
) {
764 if (local_stream
!= NULL
) {
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
);
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
);
785 ret
= consumer_update_poll_array(ctx
, &pollfd
, local_stream
);
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
);
793 consumer_data
.need_update
= 0;
795 pthread_mutex_unlock(&consumer_data
.lock
);
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
);
802 perror("Poll error");
803 lttng_consumer_send_error(ctx
, CONSUMERD_POLL_ERROR
);
805 } else if (num_rdy
== 0) {
806 DBG("Polling thread timed out");
810 /* No FDs and consumer_quit, consumer_cleanup the thread */
811 if (nb_fd
== 0 && consumer_quit
== 1) {
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.
821 if (pollfd
[nb_fd
].revents
& POLLIN
) {
822 DBG("consumer_poll_pipe wake up");
823 tmp2
= read(ctx
->consumer_poll_pipe
[0], &tmp
, 1);
825 perror("read consumer poll");
830 /* Take care of high priority channels first. */
831 for (i
= 0; i
< nb_fd
; i
++) {
832 if (pollfd
[i
].revents
& POLLPRI
) {
833 DBG("Urgent read on fd %d", pollfd
[i
].fd
);
835 ret
= ctx
->on_buffer_ready(local_stream
[i
], ctx
);
836 /* it's ok to have an unavailable sub-buffer */
840 } else if (pollfd
[i
].revents
& POLLERR
) {
841 ERR("Error returned in polling fd %d.", pollfd
[i
].fd
);
842 consumer_del_stream(local_stream
[i
]);
844 } else if (pollfd
[i
].revents
& POLLNVAL
) {
845 ERR("Polling fd %d tells fd is not open.", pollfd
[i
].fd
);
846 consumer_del_stream(local_stream
[i
]);
848 } else if ((pollfd
[i
].revents
& POLLHUP
) &&
849 !(pollfd
[i
].revents
& POLLIN
)) {
850 DBG("Polling fd %d tells it has hung up.", pollfd
[i
].fd
);
851 consumer_del_stream(local_stream
[i
]);
856 /* If every buffer FD has hung up, we end the read loop here */
857 if (nb_fd
> 0 && num_hup
== nb_fd
) {
858 DBG("every buffer FD has hung up\n");
859 if (consumer_quit
== 1) {
865 /* Take care of low priority channels. */
866 if (high_prio
== 0) {
867 for (i
= 0; i
< nb_fd
; i
++) {
868 if (pollfd
[i
].revents
& POLLIN
) {
869 DBG("Normal read on fd %d", pollfd
[i
].fd
);
870 ret
= ctx
->on_buffer_ready(local_stream
[i
], ctx
);
871 /* it's ok to have an unavailable subbuffer */
880 DBG("polling thread exiting");
881 if (pollfd
!= NULL
) {
885 if (local_stream
!= NULL
) {
893 * This thread listens on the consumerd socket and receives the file
894 * descriptors from the session daemon.
896 void *lttng_consumer_thread_receive_fds(void *data
)
898 int sock
, client_socket
, ret
;
900 * structure to poll for incoming data on communication socket avoids
901 * making blocking sockets.
903 struct pollfd consumer_sockpoll
[2];
904 struct lttng_consumer_local_data
*ctx
= data
;
906 DBG("Creating command socket %s", ctx
->consumer_command_sock_path
);
907 unlink(ctx
->consumer_command_sock_path
);
908 client_socket
= lttcomm_create_unix_sock(ctx
->consumer_command_sock_path
);
909 if (client_socket
< 0) {
910 ERR("Cannot create command socket");
914 ret
= lttcomm_listen_unix_sock(client_socket
);
919 DBG("Sending ready command to lttng-sessiond");
920 ret
= lttng_consumer_send_error(ctx
, CONSUMERD_COMMAND_SOCK_READY
);
921 /* return < 0 on error, but == 0 is not fatal */
923 ERR("Error sending ready command to lttng-sessiond");
927 ret
= fcntl(client_socket
, F_SETFL
, O_NONBLOCK
);
929 perror("fcntl O_NONBLOCK");
933 /* prepare the FDs to poll : to client socket and the should_quit pipe */
934 consumer_sockpoll
[0].fd
= ctx
->consumer_should_quit
[0];
935 consumer_sockpoll
[0].events
= POLLIN
| POLLPRI
;
936 consumer_sockpoll
[1].fd
= client_socket
;
937 consumer_sockpoll
[1].events
= POLLIN
| POLLPRI
;
939 if (lttng_consumer_poll_socket(consumer_sockpoll
) < 0) {
942 DBG("Connection on client_socket");
944 /* Blocking call, waiting for transmission */
945 sock
= lttcomm_accept_unix_sock(client_socket
);
950 ret
= fcntl(sock
, F_SETFL
, O_NONBLOCK
);
952 perror("fcntl O_NONBLOCK");
956 /* update the polling structure to poll on the established socket */
957 consumer_sockpoll
[1].fd
= sock
;
958 consumer_sockpoll
[1].events
= POLLIN
| POLLPRI
;
961 if (lttng_consumer_poll_socket(consumer_sockpoll
) < 0) {
964 DBG("Incoming command on sock");
965 ret
= lttng_consumer_recv_cmd(ctx
, sock
, consumer_sockpoll
);
966 if (ret
== -ENOENT
) {
967 DBG("Received STOP command");
971 ERR("Communication interrupted on command socket");
975 DBG("consumer_thread_receive_fds received quit from signal");
978 DBG("received fds on sock");
981 DBG("consumer_thread_receive_fds exiting");
984 * when all fds have hung up, the polling thread
990 * 2s of grace period, if no polling events occur during
991 * this period, the polling thread will exit even if there
992 * are still open FDs (should not happen, but safety mechanism).
994 consumer_poll_timeout
= LTTNG_CONSUMER_POLL_TIMEOUT
;
996 /* wake up the polling thread */
997 ret
= write(ctx
->consumer_poll_pipe
[1], "4", 1);
999 perror("poll pipe write");
1004 int lttng_consumer_read_subbuffer(struct lttng_consumer_stream
*stream
,
1005 struct lttng_consumer_local_data
*ctx
)
1007 switch (consumer_data
.type
) {
1008 case LTTNG_CONSUMER_KERNEL
:
1009 return lttng_kconsumer_read_subbuffer(stream
, ctx
);
1010 case LTTNG_CONSUMER_UST
:
1011 return lttng_ustconsumer_read_subbuffer(stream
, ctx
);
1013 ERR("Unknown consumer_data type");
1019 int lttng_consumer_on_recv_stream(struct lttng_consumer_stream
*stream
)
1021 switch (consumer_data
.type
) {
1022 case LTTNG_CONSUMER_KERNEL
:
1023 return lttng_kconsumer_on_recv_stream(stream
);
1024 case LTTNG_CONSUMER_UST
:
1025 return lttng_ustconsumer_on_recv_stream(stream
);
1027 ERR("Unknown consumer_data type");