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 int (*recv_channel
)(struct lttng_consumer_channel
*channel
),
551 int (*recv_stream
)(struct lttng_consumer_stream
*stream
),
552 int (*update_stream
)(int stream_key
, uint32_t state
))
555 struct lttng_consumer_local_data
*ctx
;
557 assert(consumer_data
.type
== LTTNG_CONSUMER_UNKNOWN
||
558 consumer_data
.type
== type
);
559 consumer_data
.type
= type
;
561 ctx
= malloc(sizeof(struct lttng_consumer_local_data
));
563 perror("allocating context");
567 ctx
->consumer_error_socket
= -1;
568 /* assign the callbacks */
569 ctx
->on_buffer_ready
= buffer_ready
;
570 ctx
->on_recv_channel
= recv_channel
;
571 ctx
->on_recv_stream
= recv_stream
;
572 ctx
->on_update_stream
= update_stream
;
574 ret
= pipe(ctx
->consumer_poll_pipe
);
576 perror("Error creating poll pipe");
577 goto error_poll_pipe
;
580 ret
= pipe(ctx
->consumer_should_quit
);
582 perror("Error creating recv pipe");
583 goto error_quit_pipe
;
586 ret
= pipe(ctx
->consumer_thread_pipe
);
588 perror("Error creating thread pipe");
589 goto error_thread_pipe
;
596 for (i
= 0; i
< 2; i
++) {
599 err
= close(ctx
->consumer_should_quit
[i
]);
603 for (i
= 0; i
< 2; i
++) {
606 err
= close(ctx
->consumer_poll_pipe
[i
]);
616 * Close all fds associated with the instance and free the context.
618 void lttng_consumer_destroy(struct lttng_consumer_local_data
*ctx
)
620 close(ctx
->consumer_error_socket
);
621 close(ctx
->consumer_thread_pipe
[0]);
622 close(ctx
->consumer_thread_pipe
[1]);
623 close(ctx
->consumer_poll_pipe
[0]);
624 close(ctx
->consumer_poll_pipe
[1]);
625 close(ctx
->consumer_should_quit
[0]);
626 close(ctx
->consumer_should_quit
[1]);
627 unlink(ctx
->consumer_command_sock_path
);
632 * Mmap the ring buffer, read it and write the data to the tracefile.
634 * Returns the number of bytes written
636 int lttng_consumer_on_read_subbuffer_mmap(
637 struct lttng_consumer_local_data
*ctx
,
638 struct lttng_consumer_stream
*stream
, unsigned long len
)
640 switch (consumer_data
.type
) {
641 case LTTNG_CONSUMER_KERNEL
:
642 return lttng_kconsumer_on_read_subbuffer_mmap(ctx
, stream
, len
);
643 case LTTNG_CONSUMER_UST
:
644 return lttng_ustconsumer_on_read_subbuffer_mmap(ctx
, stream
, len
);
646 ERR("Unknown consumer_data type");
652 * Splice the data from the ring buffer to the tracefile.
654 * Returns the number of bytes spliced.
656 int lttng_consumer_on_read_subbuffer_splice(
657 struct lttng_consumer_local_data
*ctx
,
658 struct lttng_consumer_stream
*stream
, unsigned long len
)
660 switch (consumer_data
.type
) {
661 case LTTNG_CONSUMER_KERNEL
:
662 return lttng_kconsumer_on_read_subbuffer_splice(ctx
, stream
, len
);
663 case LTTNG_CONSUMER_UST
:
666 ERR("Unknown consumer_data type");
674 * Take a snapshot for a specific fd
676 * Returns 0 on success, < 0 on error
678 int lttng_consumer_take_snapshot(struct lttng_consumer_local_data
*ctx
,
679 struct lttng_consumer_stream
*stream
)
681 switch (consumer_data
.type
) {
682 case LTTNG_CONSUMER_KERNEL
:
683 return lttng_kconsumer_take_snapshot(ctx
, stream
);
684 case LTTNG_CONSUMER_UST
:
685 return lttng_ustconsumer_take_snapshot(ctx
, stream
);
687 ERR("Unknown consumer_data type");
695 * Get the produced position
697 * Returns 0 on success, < 0 on error
699 int lttng_consumer_get_produced_snapshot(
700 struct lttng_consumer_local_data
*ctx
,
701 struct lttng_consumer_stream
*stream
,
704 switch (consumer_data
.type
) {
705 case LTTNG_CONSUMER_KERNEL
:
706 return lttng_kconsumer_get_produced_snapshot(ctx
, stream
, pos
);
707 case LTTNG_CONSUMER_UST
:
708 return lttng_ustconsumer_get_produced_snapshot(ctx
, stream
, pos
);
710 ERR("Unknown consumer_data type");
716 int lttng_consumer_recv_cmd(struct lttng_consumer_local_data
*ctx
,
717 int sock
, struct pollfd
*consumer_sockpoll
)
719 switch (consumer_data
.type
) {
720 case LTTNG_CONSUMER_KERNEL
:
721 return lttng_kconsumer_recv_cmd(ctx
, sock
, consumer_sockpoll
);
722 case LTTNG_CONSUMER_UST
:
723 return lttng_ustconsumer_recv_cmd(ctx
, sock
, consumer_sockpoll
);
725 ERR("Unknown consumer_data type");
732 * This thread polls the fds in the ltt_fd_list to consume the data and write
733 * it to tracefile if necessary.
735 void *lttng_consumer_thread_poll_fds(void *data
)
737 int num_rdy
, num_hup
, high_prio
, ret
, i
;
738 struct pollfd
*pollfd
= NULL
;
739 /* local view of the streams */
740 struct lttng_consumer_stream
**local_stream
= NULL
;
741 /* local view of consumer_data.fds_count */
745 struct lttng_consumer_local_data
*ctx
= data
;
747 local_stream
= malloc(sizeof(struct lttng_consumer_stream
));
754 * the ltt_fd_list has been updated, we need to update our
755 * local array as well
757 pthread_mutex_lock(&consumer_data
.lock
);
758 if (consumer_data
.need_update
) {
759 if (pollfd
!= NULL
) {
763 if (local_stream
!= NULL
) {
768 /* allocate for all fds + 1 for the consumer_poll_pipe */
769 pollfd
= malloc((consumer_data
.stream_count
+ 1) * sizeof(struct pollfd
));
770 if (pollfd
== NULL
) {
771 perror("pollfd malloc");
772 pthread_mutex_unlock(&consumer_data
.lock
);
776 /* allocate for all fds + 1 for the consumer_poll_pipe */
777 local_stream
= malloc((consumer_data
.stream_count
+ 1) *
778 sizeof(struct lttng_consumer_stream
));
779 if (local_stream
== NULL
) {
780 perror("local_stream malloc");
781 pthread_mutex_unlock(&consumer_data
.lock
);
784 ret
= consumer_update_poll_array(ctx
, &pollfd
, local_stream
);
786 ERR("Error in allocating pollfd or local_outfds");
787 lttng_consumer_send_error(ctx
, CONSUMERD_POLL_ERROR
);
788 pthread_mutex_unlock(&consumer_data
.lock
);
792 consumer_data
.need_update
= 0;
794 pthread_mutex_unlock(&consumer_data
.lock
);
796 /* poll on the array of fds */
797 DBG("polling on %d fd", nb_fd
+ 1);
798 num_rdy
= poll(pollfd
, nb_fd
+ 1, consumer_poll_timeout
);
799 DBG("poll num_rdy : %d", num_rdy
);
801 perror("Poll error");
802 lttng_consumer_send_error(ctx
, CONSUMERD_POLL_ERROR
);
804 } else if (num_rdy
== 0) {
805 DBG("Polling thread timed out");
809 /* No FDs and consumer_quit, kconsumer_cleanup the thread */
810 if (nb_fd
== 0 && consumer_quit
== 1) {
815 * If the consumer_poll_pipe triggered poll go
816 * directly to the beginning of the loop to update the
817 * array. We want to prioritize array update over
818 * low-priority reads.
820 if (pollfd
[nb_fd
].revents
== POLLIN
) {
821 DBG("consumer_poll_pipe wake up");
822 tmp2
= read(ctx
->consumer_poll_pipe
[0], &tmp
, 1);
824 perror("read kconsumer poll");
829 /* Take care of high priority channels first. */
830 for (i
= 0; i
< nb_fd
; i
++) {
831 switch(pollfd
[i
].revents
) {
833 ERR("Error returned in polling fd %d.", pollfd
[i
].fd
);
834 consumer_del_stream(local_stream
[i
]);
838 DBG("Polling fd %d tells it has hung up.", pollfd
[i
].fd
);
839 consumer_del_stream(local_stream
[i
]);
843 ERR("Polling fd %d tells fd is not open.", pollfd
[i
].fd
);
844 consumer_del_stream(local_stream
[i
]);
848 DBG("Urgent read on fd %d", pollfd
[i
].fd
);
850 ret
= ctx
->on_buffer_ready(local_stream
[i
]);
851 /* it's ok to have an unavailable sub-buffer */
859 /* If every buffer FD has hung up, we end the read loop here */
860 if (nb_fd
> 0 && num_hup
== nb_fd
) {
861 DBG("every buffer FD has hung up\n");
862 if (consumer_quit
== 1) {
868 /* Take care of low priority channels. */
869 if (high_prio
== 0) {
870 for (i
= 0; i
< nb_fd
; i
++) {
871 if (pollfd
[i
].revents
== POLLIN
) {
872 DBG("Normal read on fd %d", pollfd
[i
].fd
);
873 ret
= ctx
->on_buffer_ready(local_stream
[i
]);
874 /* it's ok to have an unavailable subbuffer */
883 DBG("polling thread exiting");
884 if (pollfd
!= NULL
) {
888 if (local_stream
!= NULL
) {
896 * This thread listens on the consumerd socket and receives the file
897 * descriptors from the session daemon.
899 void *lttng_consumer_thread_receive_fds(void *data
)
901 int sock
, client_socket
, ret
;
903 * structure to poll for incoming data on communication socket avoids
904 * making blocking sockets.
906 struct pollfd consumer_sockpoll
[2];
907 struct lttng_consumer_local_data
*ctx
= data
;
909 DBG("Creating command socket %s", ctx
->consumer_command_sock_path
);
910 unlink(ctx
->consumer_command_sock_path
);
911 client_socket
= lttcomm_create_unix_sock(ctx
->consumer_command_sock_path
);
912 if (client_socket
< 0) {
913 ERR("Cannot create command socket");
917 ret
= lttcomm_listen_unix_sock(client_socket
);
922 DBG("Sending ready command to lttng-sessiond");
923 ret
= lttng_consumer_send_error(ctx
, CONSUMERD_COMMAND_SOCK_READY
);
924 /* return < 0 on error, but == 0 is not fatal */
926 ERR("Error sending ready command to lttng-sessiond");
930 ret
= fcntl(client_socket
, F_SETFL
, O_NONBLOCK
);
932 perror("fcntl O_NONBLOCK");
936 /* prepare the FDs to poll : to client socket and the should_quit pipe */
937 consumer_sockpoll
[0].fd
= ctx
->consumer_should_quit
[0];
938 consumer_sockpoll
[0].events
= POLLIN
| POLLPRI
;
939 consumer_sockpoll
[1].fd
= client_socket
;
940 consumer_sockpoll
[1].events
= POLLIN
| POLLPRI
;
942 if (lttng_consumer_poll_socket(consumer_sockpoll
) < 0) {
945 DBG("Connection on client_socket");
947 /* Blocking call, waiting for transmission */
948 sock
= lttcomm_accept_unix_sock(client_socket
);
953 ret
= fcntl(sock
, F_SETFL
, O_NONBLOCK
);
955 perror("fcntl O_NONBLOCK");
959 /* update the polling structure to poll on the established socket */
960 consumer_sockpoll
[1].fd
= sock
;
961 consumer_sockpoll
[1].events
= POLLIN
| POLLPRI
;
964 if (lttng_consumer_poll_socket(consumer_sockpoll
) < 0) {
967 DBG("Incoming command on sock");
968 ret
= lttng_consumer_recv_cmd(ctx
, sock
, consumer_sockpoll
);
969 if (ret
== -ENOENT
) {
970 DBG("Received STOP command");
974 ERR("Communication interrupted on command socket");
978 DBG("consumer_thread_receive_fds received quit from signal");
981 DBG("received fds on sock");
984 DBG("consumer_thread_receive_fds exiting");
987 * when all fds have hung up, the polling thread
993 * 2s of grace period, if no polling events occur during
994 * this period, the polling thread will exit even if there
995 * are still open FDs (should not happen, but safety mechanism).
997 consumer_poll_timeout
= LTTNG_CONSUMER_POLL_TIMEOUT
;
999 /* wake up the polling thread */
1000 ret
= write(ctx
->consumer_poll_pipe
[1], "4", 1);
1002 perror("poll pipe write");