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 /* Negative keys are lookup failures */
69 cds_list_for_each_entry(iter
, &consumer_data
.stream_list
.head
, list
) {
70 if (iter
->key
== key
) {
71 DBG("Found stream key %d", key
);
78 static void consumer_steal_stream_key(int key
)
80 struct lttng_consumer_stream
*stream
;
82 stream
= consumer_find_stream(key
);
87 static struct lttng_consumer_channel
*consumer_find_channel(int key
)
89 struct lttng_consumer_channel
*iter
;
91 /* Negative keys are lookup failures */
94 cds_list_for_each_entry(iter
, &consumer_data
.channel_list
.head
, list
) {
95 if (iter
->key
== key
) {
96 DBG("Found channel key %d", key
);
103 static void consumer_steal_channel_key(int key
)
105 struct lttng_consumer_channel
*channel
;
107 channel
= consumer_find_channel(key
);
113 * Remove a stream from the global list protected by a mutex. This
114 * function is also responsible for freeing its data structures.
116 void consumer_del_stream(struct lttng_consumer_stream
*stream
)
119 struct lttng_consumer_channel
*free_chan
= NULL
;
121 pthread_mutex_lock(&consumer_data
.lock
);
123 switch (consumer_data
.type
) {
124 case LTTNG_CONSUMER_KERNEL
:
125 if (stream
->mmap_base
!= NULL
) {
126 ret
= munmap(stream
->mmap_base
, stream
->mmap_len
);
132 case LTTNG_CONSUMER32_UST
:
133 case LTTNG_CONSUMER64_UST
:
134 lttng_ustconsumer_del_stream(stream
);
137 ERR("Unknown consumer_data type");
142 cds_list_del(&stream
->list
);
143 if (consumer_data
.stream_count
<= 0) {
146 consumer_data
.stream_count
--;
150 if (stream
->out_fd
>= 0) {
151 close(stream
->out_fd
);
153 if (stream
->wait_fd
>= 0 && !stream
->wait_fd_is_copy
) {
154 close(stream
->wait_fd
);
156 if (stream
->shm_fd
>= 0 && stream
->wait_fd
!= stream
->shm_fd
157 && !stream
->shm_fd_is_copy
) {
158 close(stream
->shm_fd
);
160 if (!--stream
->chan
->refcount
)
161 free_chan
= stream
->chan
;
164 consumer_data
.need_update
= 1;
165 pthread_mutex_unlock(&consumer_data
.lock
);
168 consumer_del_channel(free_chan
);
171 struct lttng_consumer_stream
*consumer_allocate_stream(
172 int channel_key
, int stream_key
,
173 int shm_fd
, int wait_fd
,
174 enum lttng_consumer_stream_state state
,
176 enum lttng_event_output output
,
177 const char *path_name
,
181 struct lttng_consumer_stream
*stream
;
184 stream
= zmalloc(sizeof(*stream
));
185 if (stream
== NULL
) {
186 perror("malloc struct lttng_consumer_stream");
189 stream
->chan
= consumer_find_channel(channel_key
);
191 perror("Unable to find channel key");
194 stream
->chan
->refcount
++;
195 stream
->key
= stream_key
;
196 stream
->shm_fd
= shm_fd
;
197 stream
->wait_fd
= wait_fd
;
199 stream
->out_fd_offset
= 0;
200 stream
->state
= state
;
201 stream
->mmap_len
= mmap_len
;
202 stream
->mmap_base
= NULL
;
203 stream
->output
= output
;
206 strncpy(stream
->path_name
, path_name
, PATH_MAX
- 1);
207 stream
->path_name
[PATH_MAX
- 1] = '\0';
209 switch (consumer_data
.type
) {
210 case LTTNG_CONSUMER_KERNEL
:
212 case LTTNG_CONSUMER32_UST
:
213 case LTTNG_CONSUMER64_UST
:
214 stream
->cpu
= stream
->chan
->cpucount
++;
215 ret
= lttng_ustconsumer_allocate_stream(stream
);
222 ERR("Unknown consumer_data type");
226 DBG("Allocated stream %s (key %d, shm_fd %d, wait_fd %d, mmap_len %llu, out_fd %d)",
227 stream
->path_name
, stream
->key
,
230 (unsigned long long) stream
->mmap_len
,
237 * Add a stream to the global list protected by a mutex.
239 int consumer_add_stream(struct lttng_consumer_stream
*stream
)
243 pthread_mutex_lock(&consumer_data
.lock
);
244 /* Steal stream identifier, for UST */
245 consumer_steal_stream_key(stream
->key
);
246 cds_list_add(&stream
->list
, &consumer_data
.stream_list
.head
);
247 consumer_data
.stream_count
++;
248 consumer_data
.need_update
= 1;
250 switch (consumer_data
.type
) {
251 case LTTNG_CONSUMER_KERNEL
:
253 case LTTNG_CONSUMER32_UST
:
254 case LTTNG_CONSUMER64_UST
:
255 /* Streams are in CPU number order (we rely on this) */
256 stream
->cpu
= stream
->chan
->nr_streams
++;
259 ERR("Unknown consumer_data type");
265 pthread_mutex_unlock(&consumer_data
.lock
);
270 * Update a stream according to what we just received.
272 void consumer_change_stream_state(int stream_key
,
273 enum lttng_consumer_stream_state state
)
275 struct lttng_consumer_stream
*stream
;
277 pthread_mutex_lock(&consumer_data
.lock
);
278 stream
= consumer_find_stream(stream_key
);
280 stream
->state
= state
;
282 consumer_data
.need_update
= 1;
283 pthread_mutex_unlock(&consumer_data
.lock
);
287 * Remove a channel from the global list protected by a mutex. This
288 * function is also responsible for freeing its data structures.
290 void consumer_del_channel(struct lttng_consumer_channel
*channel
)
294 pthread_mutex_lock(&consumer_data
.lock
);
296 switch (consumer_data
.type
) {
297 case LTTNG_CONSUMER_KERNEL
:
299 case LTTNG_CONSUMER32_UST
:
300 case LTTNG_CONSUMER64_UST
:
301 lttng_ustconsumer_del_channel(channel
);
304 ERR("Unknown consumer_data type");
309 cds_list_del(&channel
->list
);
310 if (channel
->mmap_base
!= NULL
) {
311 ret
= munmap(channel
->mmap_base
, channel
->mmap_len
);
316 if (channel
->wait_fd
>= 0 && !channel
->wait_fd_is_copy
) {
317 close(channel
->wait_fd
);
319 if (channel
->shm_fd
>= 0 && channel
->wait_fd
!= channel
->shm_fd
320 && !channel
->shm_fd_is_copy
) {
321 close(channel
->shm_fd
);
325 pthread_mutex_unlock(&consumer_data
.lock
);
328 struct lttng_consumer_channel
*consumer_allocate_channel(
330 int shm_fd
, int wait_fd
,
332 uint64_t max_sb_size
)
334 struct lttng_consumer_channel
*channel
;
337 channel
= zmalloc(sizeof(*channel
));
338 if (channel
== NULL
) {
339 perror("malloc struct lttng_consumer_channel");
342 channel
->key
= channel_key
;
343 channel
->shm_fd
= shm_fd
;
344 channel
->wait_fd
= wait_fd
;
345 channel
->mmap_len
= mmap_len
;
346 channel
->max_sb_size
= max_sb_size
;
347 channel
->refcount
= 0;
348 channel
->nr_streams
= 0;
350 switch (consumer_data
.type
) {
351 case LTTNG_CONSUMER_KERNEL
:
352 channel
->mmap_base
= NULL
;
353 channel
->mmap_len
= 0;
355 case LTTNG_CONSUMER32_UST
:
356 case LTTNG_CONSUMER64_UST
:
357 ret
= lttng_ustconsumer_allocate_channel(channel
);
364 ERR("Unknown consumer_data type");
368 DBG("Allocated channel (key %d, shm_fd %d, wait_fd %d, mmap_len %llu, max_sb_size %llu)",
372 (unsigned long long) channel
->mmap_len
,
373 (unsigned long long) channel
->max_sb_size
);
379 * Add a channel to the global list protected by a mutex.
381 int consumer_add_channel(struct lttng_consumer_channel
*channel
)
383 pthread_mutex_lock(&consumer_data
.lock
);
384 /* Steal channel identifier, for UST */
385 consumer_steal_channel_key(channel
->key
);
386 cds_list_add(&channel
->list
, &consumer_data
.channel_list
.head
);
387 pthread_mutex_unlock(&consumer_data
.lock
);
392 * Allocate the pollfd structure and the local view of the out fds to avoid
393 * doing a lookup in the linked list and concurrency issues when writing is
394 * needed. Called with consumer_data.lock held.
396 * Returns the number of fds in the structures.
398 int consumer_update_poll_array(
399 struct lttng_consumer_local_data
*ctx
, struct pollfd
**pollfd
,
400 struct lttng_consumer_stream
**local_stream
)
402 struct lttng_consumer_stream
*iter
;
405 DBG("Updating poll fd array");
406 cds_list_for_each_entry(iter
, &consumer_data
.stream_list
.head
, list
) {
407 if (iter
->state
!= LTTNG_CONSUMER_ACTIVE_STREAM
) {
410 DBG("Active FD %d", iter
->wait_fd
);
411 (*pollfd
)[i
].fd
= iter
->wait_fd
;
412 (*pollfd
)[i
].events
= POLLIN
| POLLPRI
;
413 local_stream
[i
] = iter
;
418 * Insert the consumer_poll_pipe at the end of the array and don't
419 * increment i so nb_fd is the number of real FD.
421 (*pollfd
)[i
].fd
= ctx
->consumer_poll_pipe
[0];
422 (*pollfd
)[i
].events
= POLLIN
;
427 * Poll on the should_quit pipe and the command socket return -1 on error and
428 * should exit, 0 if data is available on the command socket
430 int lttng_consumer_poll_socket(struct pollfd
*consumer_sockpoll
)
434 num_rdy
= poll(consumer_sockpoll
, 2, -1);
436 perror("Poll error");
439 if (consumer_sockpoll
[0].revents
== POLLIN
) {
440 DBG("consumer_should_quit wake up");
450 * Set the error socket.
452 void lttng_consumer_set_error_sock(
453 struct lttng_consumer_local_data
*ctx
, int sock
)
455 ctx
->consumer_error_socket
= sock
;
459 * Set the command socket path.
462 void lttng_consumer_set_command_sock_path(
463 struct lttng_consumer_local_data
*ctx
, char *sock
)
465 ctx
->consumer_command_sock_path
= sock
;
469 * Send return code to the session daemon.
470 * If the socket is not defined, we return 0, it is not a fatal error
472 int lttng_consumer_send_error(
473 struct lttng_consumer_local_data
*ctx
, int cmd
)
475 if (ctx
->consumer_error_socket
> 0) {
476 return lttcomm_send_unix_sock(ctx
->consumer_error_socket
, &cmd
,
477 sizeof(enum lttcomm_sessiond_command
));
484 * Close all the tracefiles and stream fds, should be called when all instances
487 void lttng_consumer_cleanup(void)
489 struct lttng_consumer_stream
*iter
, *tmp
;
490 struct lttng_consumer_channel
*citer
, *ctmp
;
493 * close all outfd. Called when there are no more threads
494 * running (after joining on the threads), no need to protect
495 * list iteration with mutex.
497 cds_list_for_each_entry_safe(iter
, tmp
,
498 &consumer_data
.stream_list
.head
, list
) {
499 consumer_del_stream(iter
);
501 cds_list_for_each_entry_safe(citer
, ctmp
,
502 &consumer_data
.channel_list
.head
, list
) {
503 consumer_del_channel(citer
);
508 * Called from signal handler.
510 void lttng_consumer_should_exit(struct lttng_consumer_local_data
*ctx
)
514 ret
= write(ctx
->consumer_should_quit
[1], "4", 1);
516 perror("write consumer quit");
520 void lttng_consumer_sync_trace_file(
521 struct lttng_consumer_stream
*stream
, off_t orig_offset
)
523 int outfd
= stream
->out_fd
;
526 * This does a blocking write-and-wait on any page that belongs to the
527 * subbuffer prior to the one we just wrote.
528 * Don't care about error values, as these are just hints and ways to
529 * limit the amount of page cache used.
531 if (orig_offset
< stream
->chan
->max_sb_size
) {
534 sync_file_range(outfd
, orig_offset
- stream
->chan
->max_sb_size
,
535 stream
->chan
->max_sb_size
,
536 SYNC_FILE_RANGE_WAIT_BEFORE
537 | SYNC_FILE_RANGE_WRITE
538 | SYNC_FILE_RANGE_WAIT_AFTER
);
540 * Give hints to the kernel about how we access the file:
541 * POSIX_FADV_DONTNEED : we won't re-access data in a near future after
544 * We need to call fadvise again after the file grows because the
545 * kernel does not seem to apply fadvise to non-existing parts of the
548 * Call fadvise _after_ having waited for the page writeback to
549 * complete because the dirty page writeback semantic is not well
550 * defined. So it can be expected to lead to lower throughput in
553 posix_fadvise(outfd
, orig_offset
- stream
->chan
->max_sb_size
,
554 stream
->chan
->max_sb_size
, POSIX_FADV_DONTNEED
);
558 * Initialise the necessary environnement :
559 * - create a new context
560 * - create the poll_pipe
561 * - create the should_quit pipe (for signal handler)
562 * - create the thread pipe (for splice)
564 * Takes a function pointer as argument, this function is called when data is
565 * available on a buffer. This function is responsible to do the
566 * kernctl_get_next_subbuf, read the data with mmap or splice depending on the
567 * buffer configuration and then kernctl_put_next_subbuf at the end.
569 * Returns a pointer to the new context or NULL on error.
571 struct lttng_consumer_local_data
*lttng_consumer_create(
572 enum lttng_consumer_type type
,
573 int (*buffer_ready
)(struct lttng_consumer_stream
*stream
,
574 struct lttng_consumer_local_data
*ctx
),
575 int (*recv_channel
)(struct lttng_consumer_channel
*channel
),
576 int (*recv_stream
)(struct lttng_consumer_stream
*stream
),
577 int (*update_stream
)(int stream_key
, uint32_t state
))
580 struct lttng_consumer_local_data
*ctx
;
582 assert(consumer_data
.type
== LTTNG_CONSUMER_UNKNOWN
||
583 consumer_data
.type
== type
);
584 consumer_data
.type
= type
;
586 ctx
= zmalloc(sizeof(struct lttng_consumer_local_data
));
588 perror("allocating context");
592 ctx
->consumer_error_socket
= -1;
593 /* assign the callbacks */
594 ctx
->on_buffer_ready
= buffer_ready
;
595 ctx
->on_recv_channel
= recv_channel
;
596 ctx
->on_recv_stream
= recv_stream
;
597 ctx
->on_update_stream
= update_stream
;
599 ret
= pipe(ctx
->consumer_poll_pipe
);
601 perror("Error creating poll pipe");
602 goto error_poll_pipe
;
605 ret
= pipe(ctx
->consumer_should_quit
);
607 perror("Error creating recv pipe");
608 goto error_quit_pipe
;
611 ret
= pipe(ctx
->consumer_thread_pipe
);
613 perror("Error creating thread pipe");
614 goto error_thread_pipe
;
621 for (i
= 0; i
< 2; i
++) {
624 err
= close(ctx
->consumer_should_quit
[i
]);
628 for (i
= 0; i
< 2; i
++) {
631 err
= close(ctx
->consumer_poll_pipe
[i
]);
641 * Close all fds associated with the instance and free the context.
643 void lttng_consumer_destroy(struct lttng_consumer_local_data
*ctx
)
645 close(ctx
->consumer_error_socket
);
646 close(ctx
->consumer_thread_pipe
[0]);
647 close(ctx
->consumer_thread_pipe
[1]);
648 close(ctx
->consumer_poll_pipe
[0]);
649 close(ctx
->consumer_poll_pipe
[1]);
650 close(ctx
->consumer_should_quit
[0]);
651 close(ctx
->consumer_should_quit
[1]);
652 unlink(ctx
->consumer_command_sock_path
);
657 * Mmap the ring buffer, read it and write the data to the tracefile.
659 * Returns the number of bytes written
661 int lttng_consumer_on_read_subbuffer_mmap(
662 struct lttng_consumer_local_data
*ctx
,
663 struct lttng_consumer_stream
*stream
, unsigned long len
)
665 switch (consumer_data
.type
) {
666 case LTTNG_CONSUMER_KERNEL
:
667 return lttng_kconsumer_on_read_subbuffer_mmap(ctx
, stream
, len
);
668 case LTTNG_CONSUMER32_UST
:
669 case LTTNG_CONSUMER64_UST
:
670 return lttng_ustconsumer_on_read_subbuffer_mmap(ctx
, stream
, len
);
672 ERR("Unknown consumer_data type");
678 * Splice the data from the ring buffer to the tracefile.
680 * Returns the number of bytes spliced.
682 int lttng_consumer_on_read_subbuffer_splice(
683 struct lttng_consumer_local_data
*ctx
,
684 struct lttng_consumer_stream
*stream
, unsigned long len
)
686 switch (consumer_data
.type
) {
687 case LTTNG_CONSUMER_KERNEL
:
688 return lttng_kconsumer_on_read_subbuffer_splice(ctx
, stream
, len
);
689 case LTTNG_CONSUMER32_UST
:
690 case LTTNG_CONSUMER64_UST
:
693 ERR("Unknown consumer_data type");
701 * Take a snapshot for a specific fd
703 * Returns 0 on success, < 0 on error
705 int lttng_consumer_take_snapshot(struct lttng_consumer_local_data
*ctx
,
706 struct lttng_consumer_stream
*stream
)
708 switch (consumer_data
.type
) {
709 case LTTNG_CONSUMER_KERNEL
:
710 return lttng_kconsumer_take_snapshot(ctx
, stream
);
711 case LTTNG_CONSUMER32_UST
:
712 case LTTNG_CONSUMER64_UST
:
713 return lttng_ustconsumer_take_snapshot(ctx
, stream
);
715 ERR("Unknown consumer_data type");
723 * Get the produced position
725 * Returns 0 on success, < 0 on error
727 int lttng_consumer_get_produced_snapshot(
728 struct lttng_consumer_local_data
*ctx
,
729 struct lttng_consumer_stream
*stream
,
732 switch (consumer_data
.type
) {
733 case LTTNG_CONSUMER_KERNEL
:
734 return lttng_kconsumer_get_produced_snapshot(ctx
, stream
, pos
);
735 case LTTNG_CONSUMER32_UST
:
736 case LTTNG_CONSUMER64_UST
:
737 return lttng_ustconsumer_get_produced_snapshot(ctx
, stream
, pos
);
739 ERR("Unknown consumer_data type");
745 int lttng_consumer_recv_cmd(struct lttng_consumer_local_data
*ctx
,
746 int sock
, struct pollfd
*consumer_sockpoll
)
748 switch (consumer_data
.type
) {
749 case LTTNG_CONSUMER_KERNEL
:
750 return lttng_kconsumer_recv_cmd(ctx
, sock
, consumer_sockpoll
);
751 case LTTNG_CONSUMER32_UST
:
752 case LTTNG_CONSUMER64_UST
:
753 return lttng_ustconsumer_recv_cmd(ctx
, sock
, consumer_sockpoll
);
755 ERR("Unknown consumer_data type");
762 * This thread polls the fds in the ltt_fd_list to consume the data and write
763 * it to tracefile if necessary.
765 void *lttng_consumer_thread_poll_fds(void *data
)
767 int num_rdy
, num_hup
, high_prio
, ret
, i
;
768 struct pollfd
*pollfd
= NULL
;
769 /* local view of the streams */
770 struct lttng_consumer_stream
**local_stream
= NULL
;
771 /* local view of consumer_data.fds_count */
775 struct lttng_consumer_local_data
*ctx
= data
;
777 local_stream
= zmalloc(sizeof(struct lttng_consumer_stream
));
784 * the ltt_fd_list has been updated, we need to update our
785 * local array as well
787 pthread_mutex_lock(&consumer_data
.lock
);
788 if (consumer_data
.need_update
) {
789 if (pollfd
!= NULL
) {
793 if (local_stream
!= NULL
) {
798 /* allocate for all fds + 1 for the consumer_poll_pipe */
799 pollfd
= zmalloc((consumer_data
.stream_count
+ 1) * sizeof(struct pollfd
));
800 if (pollfd
== NULL
) {
801 perror("pollfd malloc");
802 pthread_mutex_unlock(&consumer_data
.lock
);
806 /* allocate for all fds + 1 for the consumer_poll_pipe */
807 local_stream
= zmalloc((consumer_data
.stream_count
+ 1) *
808 sizeof(struct lttng_consumer_stream
));
809 if (local_stream
== NULL
) {
810 perror("local_stream malloc");
811 pthread_mutex_unlock(&consumer_data
.lock
);
814 ret
= consumer_update_poll_array(ctx
, &pollfd
, local_stream
);
816 ERR("Error in allocating pollfd or local_outfds");
817 lttng_consumer_send_error(ctx
, CONSUMERD_POLL_ERROR
);
818 pthread_mutex_unlock(&consumer_data
.lock
);
822 consumer_data
.need_update
= 0;
824 pthread_mutex_unlock(&consumer_data
.lock
);
826 /* poll on the array of fds */
827 DBG("polling on %d fd", nb_fd
+ 1);
828 num_rdy
= poll(pollfd
, nb_fd
+ 1, consumer_poll_timeout
);
829 DBG("poll num_rdy : %d", num_rdy
);
831 perror("Poll error");
832 lttng_consumer_send_error(ctx
, CONSUMERD_POLL_ERROR
);
834 } else if (num_rdy
== 0) {
835 DBG("Polling thread timed out");
839 /* No FDs and consumer_quit, consumer_cleanup the thread */
840 if (nb_fd
== 0 && consumer_quit
== 1) {
845 * If the consumer_poll_pipe triggered poll go
846 * directly to the beginning of the loop to update the
847 * array. We want to prioritize array update over
848 * low-priority reads.
850 if (pollfd
[nb_fd
].revents
& POLLIN
) {
851 DBG("consumer_poll_pipe wake up");
852 tmp2
= read(ctx
->consumer_poll_pipe
[0], &tmp
, 1);
854 perror("read consumer poll");
859 /* Take care of high priority channels first. */
860 for (i
= 0; i
< nb_fd
; i
++) {
861 if (pollfd
[i
].revents
& POLLPRI
) {
862 DBG("Urgent read on fd %d", pollfd
[i
].fd
);
864 ret
= ctx
->on_buffer_ready(local_stream
[i
], ctx
);
865 /* it's ok to have an unavailable sub-buffer */
869 } else if (pollfd
[i
].revents
& POLLERR
) {
870 ERR("Error returned in polling fd %d.", pollfd
[i
].fd
);
871 consumer_del_stream(local_stream
[i
]);
873 } else if (pollfd
[i
].revents
& POLLNVAL
) {
874 ERR("Polling fd %d tells fd is not open.", pollfd
[i
].fd
);
875 consumer_del_stream(local_stream
[i
]);
877 } else if ((pollfd
[i
].revents
& POLLHUP
) &&
878 !(pollfd
[i
].revents
& POLLIN
)) {
879 if (consumer_data
.type
== LTTNG_CONSUMER32_UST
880 || consumer_data
.type
== LTTNG_CONSUMER64_UST
) {
881 DBG("Polling fd %d tells it has hung up. Attempting flush and read.",
883 if (!local_stream
[i
]->hangup_flush_done
) {
884 lttng_ustconsumer_on_stream_hangup(local_stream
[i
]);
885 /* read after flush */
887 ret
= ctx
->on_buffer_ready(local_stream
[i
], ctx
);
888 } while (ret
== EAGAIN
);
891 DBG("Polling fd %d tells it has hung up.", pollfd
[i
].fd
);
893 consumer_del_stream(local_stream
[i
]);
898 /* If every buffer FD has hung up, we end the read loop here */
899 if (nb_fd
> 0 && num_hup
== nb_fd
) {
900 DBG("every buffer FD has hung up\n");
901 if (consumer_quit
== 1) {
907 /* Take care of low priority channels. */
908 if (high_prio
== 0) {
909 for (i
= 0; i
< nb_fd
; i
++) {
910 if (pollfd
[i
].revents
& POLLIN
) {
911 DBG("Normal read on fd %d", pollfd
[i
].fd
);
912 ret
= ctx
->on_buffer_ready(local_stream
[i
], ctx
);
913 /* it's ok to have an unavailable subbuffer */
922 DBG("polling thread exiting");
923 if (pollfd
!= NULL
) {
927 if (local_stream
!= NULL
) {
935 * This thread listens on the consumerd socket and receives the file
936 * descriptors from the session daemon.
938 void *lttng_consumer_thread_receive_fds(void *data
)
940 int sock
, client_socket
, ret
;
942 * structure to poll for incoming data on communication socket avoids
943 * making blocking sockets.
945 struct pollfd consumer_sockpoll
[2];
946 struct lttng_consumer_local_data
*ctx
= data
;
948 DBG("Creating command socket %s", ctx
->consumer_command_sock_path
);
949 unlink(ctx
->consumer_command_sock_path
);
950 client_socket
= lttcomm_create_unix_sock(ctx
->consumer_command_sock_path
);
951 if (client_socket
< 0) {
952 ERR("Cannot create command socket");
956 ret
= lttcomm_listen_unix_sock(client_socket
);
961 DBG("Sending ready command to lttng-sessiond");
962 ret
= lttng_consumer_send_error(ctx
, CONSUMERD_COMMAND_SOCK_READY
);
963 /* return < 0 on error, but == 0 is not fatal */
965 ERR("Error sending ready command to lttng-sessiond");
969 ret
= fcntl(client_socket
, F_SETFL
, O_NONBLOCK
);
971 perror("fcntl O_NONBLOCK");
975 /* prepare the FDs to poll : to client socket and the should_quit pipe */
976 consumer_sockpoll
[0].fd
= ctx
->consumer_should_quit
[0];
977 consumer_sockpoll
[0].events
= POLLIN
| POLLPRI
;
978 consumer_sockpoll
[1].fd
= client_socket
;
979 consumer_sockpoll
[1].events
= POLLIN
| POLLPRI
;
981 if (lttng_consumer_poll_socket(consumer_sockpoll
) < 0) {
984 DBG("Connection on client_socket");
986 /* Blocking call, waiting for transmission */
987 sock
= lttcomm_accept_unix_sock(client_socket
);
992 ret
= fcntl(sock
, F_SETFL
, O_NONBLOCK
);
994 perror("fcntl O_NONBLOCK");
998 /* update the polling structure to poll on the established socket */
999 consumer_sockpoll
[1].fd
= sock
;
1000 consumer_sockpoll
[1].events
= POLLIN
| POLLPRI
;
1003 if (lttng_consumer_poll_socket(consumer_sockpoll
) < 0) {
1006 DBG("Incoming command on sock");
1007 ret
= lttng_consumer_recv_cmd(ctx
, sock
, consumer_sockpoll
);
1008 if (ret
== -ENOENT
) {
1009 DBG("Received STOP command");
1013 ERR("Communication interrupted on command socket");
1016 if (consumer_quit
) {
1017 DBG("consumer_thread_receive_fds received quit from signal");
1020 DBG("received fds on sock");
1023 DBG("consumer_thread_receive_fds exiting");
1026 * when all fds have hung up, the polling thread
1032 * 2s of grace period, if no polling events occur during
1033 * this period, the polling thread will exit even if there
1034 * are still open FDs (should not happen, but safety mechanism).
1036 consumer_poll_timeout
= LTTNG_CONSUMER_POLL_TIMEOUT
;
1038 /* wake up the polling thread */
1039 ret
= write(ctx
->consumer_poll_pipe
[1], "4", 1);
1041 perror("poll pipe write");
1046 int lttng_consumer_read_subbuffer(struct lttng_consumer_stream
*stream
,
1047 struct lttng_consumer_local_data
*ctx
)
1049 switch (consumer_data
.type
) {
1050 case LTTNG_CONSUMER_KERNEL
:
1051 return lttng_kconsumer_read_subbuffer(stream
, ctx
);
1052 case LTTNG_CONSUMER32_UST
:
1053 case LTTNG_CONSUMER64_UST
:
1054 return lttng_ustconsumer_read_subbuffer(stream
, ctx
);
1056 ERR("Unknown consumer_data type");
1062 int lttng_consumer_on_recv_stream(struct lttng_consumer_stream
*stream
)
1064 switch (consumer_data
.type
) {
1065 case LTTNG_CONSUMER_KERNEL
:
1066 return lttng_kconsumer_on_recv_stream(stream
);
1067 case LTTNG_CONSUMER32_UST
:
1068 case LTTNG_CONSUMER64_UST
:
1069 return lttng_ustconsumer_on_recv_stream(stream
);
1071 ERR("Unknown consumer_data type");