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 ret
= lttng_ustconsumer_allocate_stream(stream
);
191 ERR("Unknown consumer_data type");
195 DBG("Allocated stream %s (key %d, shm_fd %d, wait_fd %d, mmap_len %llu, out_fd %d)",
196 stream
->path_name
, stream
->key
,
199 (unsigned long long) stream
->mmap_len
,
206 * Add a stream to the global list protected by a mutex.
208 int consumer_add_stream(struct lttng_consumer_stream
*stream
)
212 pthread_mutex_lock(&consumer_data
.lock
);
213 /* Check if already exist */
214 if (consumer_find_stream(stream
->key
)) {
218 cds_list_add(&stream
->list
, &consumer_data
.stream_list
.head
);
219 consumer_data
.stream_count
++;
220 consumer_data
.need_update
= 1;
222 switch (consumer_data
.type
) {
223 case LTTNG_CONSUMER_KERNEL
:
225 case LTTNG_CONSUMER_UST
:
226 /* Streams are in CPU number order (we rely on this) */
227 stream
->cpu
= stream
->chan
->nr_streams
++;
230 ERR("Unknown consumer_data type");
236 pthread_mutex_unlock(&consumer_data
.lock
);
241 * Update a stream according to what we just received.
243 void consumer_change_stream_state(int stream_key
,
244 enum lttng_consumer_stream_state state
)
246 struct lttng_consumer_stream
*stream
;
248 pthread_mutex_lock(&consumer_data
.lock
);
249 stream
= consumer_find_stream(stream_key
);
251 stream
->state
= state
;
253 consumer_data
.need_update
= 1;
254 pthread_mutex_unlock(&consumer_data
.lock
);
258 * Remove a channel from the global list protected by a mutex. This
259 * function is also responsible for freeing its data structures.
261 void consumer_del_channel(struct lttng_consumer_channel
*channel
)
265 pthread_mutex_lock(&consumer_data
.lock
);
267 switch (consumer_data
.type
) {
268 case LTTNG_CONSUMER_KERNEL
:
270 case LTTNG_CONSUMER_UST
:
271 lttng_ustconsumer_del_channel(channel
);
274 ERR("Unknown consumer_data type");
279 cds_list_del(&channel
->list
);
280 if (channel
->mmap_base
!= NULL
) {
281 ret
= munmap(channel
->mmap_base
, channel
->mmap_len
);
286 if (channel
->wait_fd
>= 0 && !channel
->wait_fd_is_copy
) {
287 close(channel
->wait_fd
);
289 if (channel
->shm_fd
>= 0 && channel
->wait_fd
!= channel
->shm_fd
290 && !channel
->shm_fd_is_copy
) {
291 close(channel
->shm_fd
);
295 pthread_mutex_unlock(&consumer_data
.lock
);
298 struct lttng_consumer_channel
*consumer_allocate_channel(
300 int shm_fd
, int wait_fd
,
302 uint64_t max_sb_size
)
304 struct lttng_consumer_channel
*channel
;
307 channel
= malloc(sizeof(*channel
));
308 if (channel
== NULL
) {
309 perror("malloc struct lttng_consumer_channel");
312 channel
->key
= channel_key
;
313 channel
->shm_fd
= shm_fd
;
314 channel
->wait_fd
= wait_fd
;
315 channel
->mmap_len
= mmap_len
;
316 channel
->max_sb_size
= max_sb_size
;
317 channel
->refcount
= 0;
318 channel
->nr_streams
= 0;
320 switch (consumer_data
.type
) {
321 case LTTNG_CONSUMER_KERNEL
:
322 channel
->mmap_base
= NULL
;
323 channel
->mmap_len
= 0;
325 case LTTNG_CONSUMER_UST
:
326 ret
= lttng_ustconsumer_allocate_channel(channel
);
333 ERR("Unknown consumer_data type");
337 DBG("Allocated channel (key %d, shm_fd %d, wait_fd %d, mmap_len %llu, max_sb_size %llu)",
341 (unsigned long long) channel
->mmap_len
,
342 (unsigned long long) channel
->max_sb_size
);
348 * Add a channel to the global list protected by a mutex.
350 int consumer_add_channel(struct lttng_consumer_channel
*channel
)
354 pthread_mutex_lock(&consumer_data
.lock
);
355 /* Check if already exist */
356 if (consumer_find_channel(channel
->key
)) {
360 cds_list_add(&channel
->list
, &consumer_data
.channel_list
.head
);
362 pthread_mutex_unlock(&consumer_data
.lock
);
367 * Allocate the pollfd structure and the local view of the out fds to avoid
368 * doing a lookup in the linked list and concurrency issues when writing is
369 * needed. Called with consumer_data.lock held.
371 * Returns the number of fds in the structures.
373 int consumer_update_poll_array(
374 struct lttng_consumer_local_data
*ctx
, struct pollfd
**pollfd
,
375 struct lttng_consumer_stream
**local_stream
)
377 struct lttng_consumer_stream
*iter
;
380 DBG("Updating poll fd array");
381 cds_list_for_each_entry(iter
, &consumer_data
.stream_list
.head
, list
) {
382 if (iter
->state
!= LTTNG_CONSUMER_ACTIVE_STREAM
) {
385 DBG("Active FD %d", iter
->wait_fd
);
386 (*pollfd
)[i
].fd
= iter
->wait_fd
;
387 (*pollfd
)[i
].events
= POLLIN
| POLLPRI
;
388 local_stream
[i
] = iter
;
393 * Insert the consumer_poll_pipe at the end of the array and don't
394 * increment i so nb_fd is the number of real FD.
396 (*pollfd
)[i
].fd
= ctx
->consumer_poll_pipe
[0];
397 (*pollfd
)[i
].events
= POLLIN
;
402 * Poll on the should_quit pipe and the command socket return -1 on error and
403 * should exit, 0 if data is available on the command socket
405 int lttng_consumer_poll_socket(struct pollfd
*consumer_sockpoll
)
409 num_rdy
= poll(consumer_sockpoll
, 2, -1);
411 perror("Poll error");
414 if (consumer_sockpoll
[0].revents
== POLLIN
) {
415 DBG("consumer_should_quit wake up");
425 * Set the error socket.
427 void lttng_consumer_set_error_sock(
428 struct lttng_consumer_local_data
*ctx
, int sock
)
430 ctx
->consumer_error_socket
= sock
;
434 * Set the command socket path.
437 void lttng_consumer_set_command_sock_path(
438 struct lttng_consumer_local_data
*ctx
, char *sock
)
440 ctx
->consumer_command_sock_path
= sock
;
444 * Send return code to the session daemon.
445 * If the socket is not defined, we return 0, it is not a fatal error
447 int lttng_consumer_send_error(
448 struct lttng_consumer_local_data
*ctx
, int cmd
)
450 if (ctx
->consumer_error_socket
> 0) {
451 return lttcomm_send_unix_sock(ctx
->consumer_error_socket
, &cmd
,
452 sizeof(enum lttcomm_sessiond_command
));
459 * Close all the tracefiles and stream fds, should be called when all instances
462 void lttng_consumer_cleanup(void)
464 struct lttng_consumer_stream
*iter
, *tmp
;
465 struct lttng_consumer_channel
*citer
, *ctmp
;
468 * close all outfd. Called when there are no more threads
469 * running (after joining on the threads), no need to protect
470 * list iteration with mutex.
472 cds_list_for_each_entry_safe(iter
, tmp
,
473 &consumer_data
.stream_list
.head
, list
) {
474 consumer_del_stream(iter
);
476 cds_list_for_each_entry_safe(citer
, ctmp
,
477 &consumer_data
.channel_list
.head
, list
) {
478 consumer_del_channel(citer
);
483 * Called from signal handler.
485 void lttng_consumer_should_exit(struct lttng_consumer_local_data
*ctx
)
489 ret
= write(ctx
->consumer_should_quit
[1], "4", 1);
491 perror("write consumer quit");
495 void lttng_consumer_sync_trace_file(
496 struct lttng_consumer_stream
*stream
, off_t orig_offset
)
498 int outfd
= stream
->out_fd
;
501 * This does a blocking write-and-wait on any page that belongs to the
502 * subbuffer prior to the one we just wrote.
503 * Don't care about error values, as these are just hints and ways to
504 * limit the amount of page cache used.
506 if (orig_offset
< stream
->chan
->max_sb_size
) {
509 sync_file_range(outfd
, orig_offset
- stream
->chan
->max_sb_size
,
510 stream
->chan
->max_sb_size
,
511 SYNC_FILE_RANGE_WAIT_BEFORE
512 | SYNC_FILE_RANGE_WRITE
513 | SYNC_FILE_RANGE_WAIT_AFTER
);
515 * Give hints to the kernel about how we access the file:
516 * POSIX_FADV_DONTNEED : we won't re-access data in a near future after
519 * We need to call fadvise again after the file grows because the
520 * kernel does not seem to apply fadvise to non-existing parts of the
523 * Call fadvise _after_ having waited for the page writeback to
524 * complete because the dirty page writeback semantic is not well
525 * defined. So it can be expected to lead to lower throughput in
528 posix_fadvise(outfd
, orig_offset
- stream
->chan
->max_sb_size
,
529 stream
->chan
->max_sb_size
, POSIX_FADV_DONTNEED
);
533 * Initialise the necessary environnement :
534 * - create a new context
535 * - create the poll_pipe
536 * - create the should_quit pipe (for signal handler)
537 * - create the thread pipe (for splice)
539 * Takes a function pointer as argument, this function is called when data is
540 * available on a buffer. This function is responsible to do the
541 * kernctl_get_next_subbuf, read the data with mmap or splice depending on the
542 * buffer configuration and then kernctl_put_next_subbuf at the end.
544 * Returns a pointer to the new context or NULL on error.
546 struct lttng_consumer_local_data
*lttng_consumer_create(
547 enum lttng_consumer_type type
,
548 int (*buffer_ready
)(struct lttng_consumer_stream
*stream
),
549 int (*recv_channel
)(struct lttng_consumer_channel
*channel
),
550 int (*recv_stream
)(struct lttng_consumer_stream
*stream
),
551 int (*update_stream
)(int stream_key
, uint32_t state
))
554 struct lttng_consumer_local_data
*ctx
;
556 assert(consumer_data
.type
== LTTNG_CONSUMER_UNKNOWN
||
557 consumer_data
.type
== type
);
558 consumer_data
.type
= type
;
560 ctx
= malloc(sizeof(struct lttng_consumer_local_data
));
562 perror("allocating context");
566 ctx
->consumer_error_socket
= -1;
567 /* assign the callbacks */
568 ctx
->on_buffer_ready
= buffer_ready
;
569 ctx
->on_recv_channel
= recv_channel
;
570 ctx
->on_recv_stream
= recv_stream
;
571 ctx
->on_update_stream
= update_stream
;
573 ret
= pipe(ctx
->consumer_poll_pipe
);
575 perror("Error creating poll pipe");
576 goto error_poll_pipe
;
579 ret
= pipe(ctx
->consumer_should_quit
);
581 perror("Error creating recv pipe");
582 goto error_quit_pipe
;
585 ret
= pipe(ctx
->consumer_thread_pipe
);
587 perror("Error creating thread pipe");
588 goto error_thread_pipe
;
595 for (i
= 0; i
< 2; i
++) {
598 err
= close(ctx
->consumer_should_quit
[i
]);
602 for (i
= 0; i
< 2; i
++) {
605 err
= close(ctx
->consumer_poll_pipe
[i
]);
615 * Close all fds associated with the instance and free the context.
617 void lttng_consumer_destroy(struct lttng_consumer_local_data
*ctx
)
619 close(ctx
->consumer_error_socket
);
620 close(ctx
->consumer_thread_pipe
[0]);
621 close(ctx
->consumer_thread_pipe
[1]);
622 close(ctx
->consumer_poll_pipe
[0]);
623 close(ctx
->consumer_poll_pipe
[1]);
624 close(ctx
->consumer_should_quit
[0]);
625 close(ctx
->consumer_should_quit
[1]);
626 unlink(ctx
->consumer_command_sock_path
);
631 * Mmap the ring buffer, read it and write the data to the tracefile.
633 * Returns the number of bytes written
635 int lttng_consumer_on_read_subbuffer_mmap(
636 struct lttng_consumer_local_data
*ctx
,
637 struct lttng_consumer_stream
*stream
, unsigned long len
)
639 switch (consumer_data
.type
) {
640 case LTTNG_CONSUMER_KERNEL
:
641 return lttng_kconsumer_on_read_subbuffer_mmap(ctx
, stream
, len
);
642 case LTTNG_CONSUMER_UST
:
643 return lttng_ustconsumer_on_read_subbuffer_mmap(ctx
, stream
, len
);
645 ERR("Unknown consumer_data type");
651 * Splice the data from the ring buffer to the tracefile.
653 * Returns the number of bytes spliced.
655 int lttng_consumer_on_read_subbuffer_splice(
656 struct lttng_consumer_local_data
*ctx
,
657 struct lttng_consumer_stream
*stream
, unsigned long len
)
659 switch (consumer_data
.type
) {
660 case LTTNG_CONSUMER_KERNEL
:
661 return lttng_kconsumer_on_read_subbuffer_splice(ctx
, stream
, len
);
662 case LTTNG_CONSUMER_UST
:
665 ERR("Unknown consumer_data type");
673 * Take a snapshot for a specific fd
675 * Returns 0 on success, < 0 on error
677 int lttng_consumer_take_snapshot(struct lttng_consumer_local_data
*ctx
,
678 struct lttng_consumer_stream
*stream
)
680 switch (consumer_data
.type
) {
681 case LTTNG_CONSUMER_KERNEL
:
682 return lttng_kconsumer_take_snapshot(ctx
, stream
);
683 case LTTNG_CONSUMER_UST
:
684 return lttng_ustconsumer_take_snapshot(ctx
, stream
);
686 ERR("Unknown consumer_data type");
694 * Get the produced position
696 * Returns 0 on success, < 0 on error
698 int lttng_consumer_get_produced_snapshot(
699 struct lttng_consumer_local_data
*ctx
,
700 struct lttng_consumer_stream
*stream
,
703 switch (consumer_data
.type
) {
704 case LTTNG_CONSUMER_KERNEL
:
705 return lttng_kconsumer_get_produced_snapshot(ctx
, stream
, pos
);
706 case LTTNG_CONSUMER_UST
:
707 return lttng_ustconsumer_get_produced_snapshot(ctx
, stream
, pos
);
709 ERR("Unknown consumer_data type");
715 int lttng_consumer_recv_cmd(struct lttng_consumer_local_data
*ctx
,
716 int sock
, struct pollfd
*consumer_sockpoll
)
718 switch (consumer_data
.type
) {
719 case LTTNG_CONSUMER_KERNEL
:
720 return lttng_kconsumer_recv_cmd(ctx
, sock
, consumer_sockpoll
);
721 case LTTNG_CONSUMER_UST
:
722 return lttng_ustconsumer_recv_cmd(ctx
, sock
, consumer_sockpoll
);
724 ERR("Unknown consumer_data type");
731 * This thread polls the fds in the ltt_fd_list to consume the data and write
732 * it to tracefile if necessary.
734 void *lttng_consumer_thread_poll_fds(void *data
)
736 int num_rdy
, num_hup
, high_prio
, ret
, i
;
737 struct pollfd
*pollfd
= NULL
;
738 /* local view of the streams */
739 struct lttng_consumer_stream
**local_stream
= NULL
;
740 /* local view of consumer_data.fds_count */
744 struct lttng_consumer_local_data
*ctx
= data
;
746 local_stream
= malloc(sizeof(struct lttng_consumer_stream
));
753 * the ltt_fd_list has been updated, we need to update our
754 * local array as well
756 pthread_mutex_lock(&consumer_data
.lock
);
757 if (consumer_data
.need_update
) {
758 if (pollfd
!= NULL
) {
762 if (local_stream
!= NULL
) {
767 /* allocate for all fds + 1 for the consumer_poll_pipe */
768 pollfd
= malloc((consumer_data
.stream_count
+ 1) * sizeof(struct pollfd
));
769 if (pollfd
== NULL
) {
770 perror("pollfd malloc");
771 pthread_mutex_unlock(&consumer_data
.lock
);
775 /* allocate for all fds + 1 for the consumer_poll_pipe */
776 local_stream
= malloc((consumer_data
.stream_count
+ 1) *
777 sizeof(struct lttng_consumer_stream
));
778 if (local_stream
== NULL
) {
779 perror("local_stream malloc");
780 pthread_mutex_unlock(&consumer_data
.lock
);
783 ret
= consumer_update_poll_array(ctx
, &pollfd
, local_stream
);
785 ERR("Error in allocating pollfd or local_outfds");
786 lttng_consumer_send_error(ctx
, CONSUMERD_POLL_ERROR
);
787 pthread_mutex_unlock(&consumer_data
.lock
);
791 consumer_data
.need_update
= 0;
793 pthread_mutex_unlock(&consumer_data
.lock
);
795 /* poll on the array of fds */
796 DBG("polling on %d fd", nb_fd
+ 1);
797 num_rdy
= poll(pollfd
, nb_fd
+ 1, consumer_poll_timeout
);
798 DBG("poll num_rdy : %d", num_rdy
);
800 perror("Poll error");
801 lttng_consumer_send_error(ctx
, CONSUMERD_POLL_ERROR
);
803 } else if (num_rdy
== 0) {
804 DBG("Polling thread timed out");
808 /* No FDs and consumer_quit, kconsumer_cleanup the thread */
809 if (nb_fd
== 0 && consumer_quit
== 1) {
814 * If the consumer_poll_pipe triggered poll go
815 * directly to the beginning of the loop to update the
816 * array. We want to prioritize array update over
817 * low-priority reads.
819 if (pollfd
[nb_fd
].revents
== POLLIN
) {
820 DBG("consumer_poll_pipe wake up");
821 tmp2
= read(ctx
->consumer_poll_pipe
[0], &tmp
, 1);
823 perror("read kconsumer poll");
828 /* Take care of high priority channels first. */
829 for (i
= 0; i
< nb_fd
; i
++) {
830 switch(pollfd
[i
].revents
) {
832 ERR("Error returned in polling fd %d.", pollfd
[i
].fd
);
833 consumer_del_stream(local_stream
[i
]);
837 DBG("Polling fd %d tells it has hung up.", pollfd
[i
].fd
);
838 consumer_del_stream(local_stream
[i
]);
842 ERR("Polling fd %d tells fd is not open.", pollfd
[i
].fd
);
843 consumer_del_stream(local_stream
[i
]);
847 DBG("Urgent read on fd %d", pollfd
[i
].fd
);
849 ret
= ctx
->on_buffer_ready(local_stream
[i
]);
850 /* it's ok to have an unavailable sub-buffer */
858 /* If every buffer FD has hung up, we end the read loop here */
859 if (nb_fd
> 0 && num_hup
== nb_fd
) {
860 DBG("every buffer FD has hung up\n");
861 if (consumer_quit
== 1) {
867 /* Take care of low priority channels. */
868 if (high_prio
== 0) {
869 for (i
= 0; i
< nb_fd
; i
++) {
870 if (pollfd
[i
].revents
== POLLIN
) {
871 DBG("Normal read on fd %d", pollfd
[i
].fd
);
872 ret
= ctx
->on_buffer_ready(local_stream
[i
]);
873 /* it's ok to have an unavailable subbuffer */
882 DBG("polling thread exiting");
883 if (pollfd
!= NULL
) {
887 if (local_stream
!= NULL
) {
895 * This thread listens on the consumerd socket and receives the file
896 * descriptors from the session daemon.
898 void *lttng_consumer_thread_receive_fds(void *data
)
900 int sock
, client_socket
, ret
;
902 * structure to poll for incoming data on communication socket avoids
903 * making blocking sockets.
905 struct pollfd consumer_sockpoll
[2];
906 struct lttng_consumer_local_data
*ctx
= data
;
908 DBG("Creating command socket %s", ctx
->consumer_command_sock_path
);
909 unlink(ctx
->consumer_command_sock_path
);
910 client_socket
= lttcomm_create_unix_sock(ctx
->consumer_command_sock_path
);
911 if (client_socket
< 0) {
912 ERR("Cannot create command socket");
916 ret
= lttcomm_listen_unix_sock(client_socket
);
921 DBG("Sending ready command to lttng-sessiond");
922 ret
= lttng_consumer_send_error(ctx
, CONSUMERD_COMMAND_SOCK_READY
);
923 /* return < 0 on error, but == 0 is not fatal */
925 ERR("Error sending ready command to lttng-sessiond");
929 ret
= fcntl(client_socket
, F_SETFL
, O_NONBLOCK
);
931 perror("fcntl O_NONBLOCK");
935 /* prepare the FDs to poll : to client socket and the should_quit pipe */
936 consumer_sockpoll
[0].fd
= ctx
->consumer_should_quit
[0];
937 consumer_sockpoll
[0].events
= POLLIN
| POLLPRI
;
938 consumer_sockpoll
[1].fd
= client_socket
;
939 consumer_sockpoll
[1].events
= POLLIN
| POLLPRI
;
941 if (lttng_consumer_poll_socket(consumer_sockpoll
) < 0) {
944 DBG("Connection on client_socket");
946 /* Blocking call, waiting for transmission */
947 sock
= lttcomm_accept_unix_sock(client_socket
);
952 ret
= fcntl(sock
, F_SETFL
, O_NONBLOCK
);
954 perror("fcntl O_NONBLOCK");
958 /* update the polling structure to poll on the established socket */
959 consumer_sockpoll
[1].fd
= sock
;
960 consumer_sockpoll
[1].events
= POLLIN
| POLLPRI
;
963 if (lttng_consumer_poll_socket(consumer_sockpoll
) < 0) {
966 DBG("Incoming command on sock");
967 ret
= lttng_consumer_recv_cmd(ctx
, sock
, consumer_sockpoll
);
968 if (ret
== -ENOENT
) {
969 DBG("Received STOP command");
973 ERR("Communication interrupted on command socket");
977 DBG("consumer_thread_receive_fds received quit from signal");
980 DBG("received fds on sock");
983 DBG("consumer_thread_receive_fds exiting");
986 * when all fds have hung up, the polling thread
992 * 2s of grace period, if no polling events occur during
993 * this period, the polling thread will exit even if there
994 * are still open FDs (should not happen, but safety mechanism).
996 consumer_poll_timeout
= LTTNG_CONSUMER_POLL_TIMEOUT
;
998 /* wake up the polling thread */
999 ret
= write(ctx
->consumer_poll_pipe
[1], "4", 1);
1001 perror("poll pipe write");