2 * Copyright (C) 2011 - Julien Desfossez <julien.desfossez@polymtl.ca>
3 * Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
4 * 2012 - David Goulet <dgoulet@efficios.com>
6 * This program is free software; you can redistribute it and/or modify
7 * it under the terms of the GNU General Public License, version 2 only,
8 * as published by the Free Software Foundation.
10 * This program is distributed in the hope that it will be useful, but WITHOUT
11 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
12 * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
15 * You should have received a copy of the GNU General Public License along
16 * with this program; if not, write to the Free Software Foundation, Inc.,
17 * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
27 #include <sys/socket.h>
28 #include <sys/types.h>
32 #include <common/common.h>
33 #include <common/utils.h>
34 #include <common/compat/poll.h>
35 #include <common/kernel-ctl/kernel-ctl.h>
36 #include <common/sessiond-comm/relayd.h>
37 #include <common/sessiond-comm/sessiond-comm.h>
38 #include <common/kernel-consumer/kernel-consumer.h>
39 #include <common/relayd/relayd.h>
40 #include <common/ust-consumer/ust-consumer.h>
44 struct lttng_consumer_global_data consumer_data
= {
47 .type
= LTTNG_CONSUMER_UNKNOWN
,
50 /* timeout parameter, to control the polling thread grace period. */
51 int consumer_poll_timeout
= -1;
54 * Flag to inform the polling thread to quit when all fd hung up. Updated by
55 * the consumer_thread_receive_fds when it notices that all fds has hung up.
56 * Also updated by the signal handler (consumer_should_exit()). Read by the
59 volatile int consumer_quit
;
62 * The following two hash tables are visible by all threads which are separated
63 * in different source files.
65 * Global hash table containing respectively metadata and data streams. The
66 * stream element in this ht should only be updated by the metadata poll thread
67 * for the metadata and the data poll thread for the data.
69 struct lttng_ht
*metadata_ht
;
70 struct lttng_ht
*data_ht
;
73 * Notify a thread pipe to poll back again. This usually means that some global
74 * state has changed so we just send back the thread in a poll wait call.
76 static void notify_thread_pipe(int wpipe
)
81 struct lttng_consumer_stream
*null_stream
= NULL
;
83 ret
= write(wpipe
, &null_stream
, sizeof(null_stream
));
84 } while (ret
< 0 && errno
== EINTR
);
88 * Find a stream. The consumer_data.lock must be locked during this
91 static struct lttng_consumer_stream
*consumer_find_stream(int key
,
94 struct lttng_ht_iter iter
;
95 struct lttng_ht_node_ulong
*node
;
96 struct lttng_consumer_stream
*stream
= NULL
;
100 /* Negative keys are lookup failures */
107 lttng_ht_lookup(ht
, (void *)((unsigned long) key
), &iter
);
108 node
= lttng_ht_iter_get_node_ulong(&iter
);
110 stream
= caa_container_of(node
, struct lttng_consumer_stream
, node
);
118 void consumer_steal_stream_key(int key
, struct lttng_ht
*ht
)
120 struct lttng_consumer_stream
*stream
;
123 stream
= consumer_find_stream(key
, ht
);
127 * We don't want the lookup to match, but we still need
128 * to iterate on this stream when iterating over the hash table. Just
129 * change the node key.
131 stream
->node
.key
= -1;
136 static struct lttng_consumer_channel
*consumer_find_channel(int key
)
138 struct lttng_ht_iter iter
;
139 struct lttng_ht_node_ulong
*node
;
140 struct lttng_consumer_channel
*channel
= NULL
;
142 /* Negative keys are lookup failures */
149 lttng_ht_lookup(consumer_data
.channel_ht
, (void *)((unsigned long) key
),
151 node
= lttng_ht_iter_get_node_ulong(&iter
);
153 channel
= caa_container_of(node
, struct lttng_consumer_channel
, node
);
161 static void consumer_steal_channel_key(int key
)
163 struct lttng_consumer_channel
*channel
;
166 channel
= consumer_find_channel(key
);
170 * We don't want the lookup to match, but we still need
171 * to iterate on this channel when iterating over the hash table. Just
172 * change the node key.
174 channel
->node
.key
= -1;
180 void consumer_free_stream(struct rcu_head
*head
)
182 struct lttng_ht_node_ulong
*node
=
183 caa_container_of(head
, struct lttng_ht_node_ulong
, head
);
184 struct lttng_consumer_stream
*stream
=
185 caa_container_of(node
, struct lttng_consumer_stream
, node
);
191 * RCU protected relayd socket pair free.
193 static void consumer_rcu_free_relayd(struct rcu_head
*head
)
195 struct lttng_ht_node_ulong
*node
=
196 caa_container_of(head
, struct lttng_ht_node_ulong
, head
);
197 struct consumer_relayd_sock_pair
*relayd
=
198 caa_container_of(node
, struct consumer_relayd_sock_pair
, node
);
201 * Close all sockets. This is done in the call RCU since we don't want the
202 * socket fds to be reassigned thus potentially creating bad state of the
205 * We do not have to lock the control socket mutex here since at this stage
206 * there is no one referencing to this relayd object.
208 (void) relayd_close(&relayd
->control_sock
);
209 (void) relayd_close(&relayd
->data_sock
);
215 * Destroy and free relayd socket pair object.
217 * This function MUST be called with the consumer_data lock acquired.
219 static void destroy_relayd(struct consumer_relayd_sock_pair
*relayd
)
222 struct lttng_ht_iter iter
;
224 if (relayd
== NULL
) {
228 DBG("Consumer destroy and close relayd socket pair");
230 iter
.iter
.node
= &relayd
->node
.node
;
231 ret
= lttng_ht_del(consumer_data
.relayd_ht
, &iter
);
233 /* We assume the relayd is being or is destroyed */
237 /* RCU free() call */
238 call_rcu(&relayd
->node
.head
, consumer_rcu_free_relayd
);
242 * Update the end point status of all streams having the given network sequence
243 * index (relayd index).
245 * It's atomically set without having the stream mutex locked which is fine
246 * because we handle the write/read race with a pipe wakeup for each thread.
248 static void update_endpoint_status_by_netidx(int net_seq_idx
,
249 enum consumer_endpoint_status status
)
251 struct lttng_ht_iter iter
;
252 struct lttng_consumer_stream
*stream
;
254 DBG("Consumer set delete flag on stream by idx %d", net_seq_idx
);
258 /* Let's begin with metadata */
259 cds_lfht_for_each_entry(metadata_ht
->ht
, &iter
.iter
, stream
, node
.node
) {
260 if (stream
->net_seq_idx
== net_seq_idx
) {
261 uatomic_set(&stream
->endpoint_status
, status
);
262 DBG("Delete flag set to metadata stream %d", stream
->wait_fd
);
266 /* Follow up by the data streams */
267 cds_lfht_for_each_entry(data_ht
->ht
, &iter
.iter
, stream
, node
.node
) {
268 if (stream
->net_seq_idx
== net_seq_idx
) {
269 uatomic_set(&stream
->endpoint_status
, status
);
270 DBG("Delete flag set to data stream %d", stream
->wait_fd
);
277 * Cleanup a relayd object by flagging every associated streams for deletion,
278 * destroying the object meaning removing it from the relayd hash table,
279 * closing the sockets and freeing the memory in a RCU call.
281 * If a local data context is available, notify the threads that the streams'
282 * state have changed.
284 static void cleanup_relayd(struct consumer_relayd_sock_pair
*relayd
,
285 struct lttng_consumer_local_data
*ctx
)
291 DBG("Cleaning up relayd sockets");
293 /* Save the net sequence index before destroying the object */
294 netidx
= relayd
->net_seq_idx
;
297 * Delete the relayd from the relayd hash table, close the sockets and free
298 * the object in a RCU call.
300 destroy_relayd(relayd
);
302 /* Set inactive endpoint to all streams */
303 update_endpoint_status_by_netidx(netidx
, CONSUMER_ENDPOINT_INACTIVE
);
306 * With a local data context, notify the threads that the streams' state
307 * have changed. The write() action on the pipe acts as an "implicit"
308 * memory barrier ordering the updates of the end point status from the
309 * read of this status which happens AFTER receiving this notify.
312 notify_thread_pipe(ctx
->consumer_data_pipe
[1]);
313 notify_thread_pipe(ctx
->consumer_metadata_pipe
[1]);
318 * Flag a relayd socket pair for destruction. Destroy it if the refcount
321 * RCU read side lock MUST be aquired before calling this function.
323 void consumer_flag_relayd_for_destroy(struct consumer_relayd_sock_pair
*relayd
)
327 /* Set destroy flag for this object */
328 uatomic_set(&relayd
->destroy_flag
, 1);
330 /* Destroy the relayd if refcount is 0 */
331 if (uatomic_read(&relayd
->refcount
) == 0) {
332 destroy_relayd(relayd
);
337 * Remove a stream from the global list protected by a mutex. This
338 * function is also responsible for freeing its data structures.
340 void consumer_del_stream(struct lttng_consumer_stream
*stream
,
344 struct lttng_ht_iter iter
;
345 struct lttng_consumer_channel
*free_chan
= NULL
;
346 struct consumer_relayd_sock_pair
*relayd
;
350 DBG("Consumer del stream %d", stream
->wait_fd
);
353 /* Means the stream was allocated but not successfully added */
357 pthread_mutex_lock(&stream
->lock
);
358 pthread_mutex_lock(&consumer_data
.lock
);
360 switch (consumer_data
.type
) {
361 case LTTNG_CONSUMER_KERNEL
:
362 if (stream
->mmap_base
!= NULL
) {
363 ret
= munmap(stream
->mmap_base
, stream
->mmap_len
);
369 case LTTNG_CONSUMER32_UST
:
370 case LTTNG_CONSUMER64_UST
:
371 lttng_ustconsumer_del_stream(stream
);
374 ERR("Unknown consumer_data type");
380 iter
.iter
.node
= &stream
->node
.node
;
381 ret
= lttng_ht_del(ht
, &iter
);
384 /* Remove node session id from the consumer_data stream ht */
385 iter
.iter
.node
= &stream
->node_session_id
.node
;
386 ret
= lttng_ht_del(consumer_data
.stream_list_ht
, &iter
);
390 assert(consumer_data
.stream_count
> 0);
391 consumer_data
.stream_count
--;
393 if (stream
->out_fd
>= 0) {
394 ret
= close(stream
->out_fd
);
399 if (stream
->wait_fd
>= 0 && !stream
->wait_fd_is_copy
) {
400 ret
= close(stream
->wait_fd
);
405 if (stream
->shm_fd
>= 0 && stream
->wait_fd
!= stream
->shm_fd
) {
406 ret
= close(stream
->shm_fd
);
412 /* Check and cleanup relayd */
414 relayd
= consumer_find_relayd(stream
->net_seq_idx
);
415 if (relayd
!= NULL
) {
416 uatomic_dec(&relayd
->refcount
);
417 assert(uatomic_read(&relayd
->refcount
) >= 0);
419 /* Closing streams requires to lock the control socket. */
420 pthread_mutex_lock(&relayd
->ctrl_sock_mutex
);
421 ret
= relayd_send_close_stream(&relayd
->control_sock
,
422 stream
->relayd_stream_id
,
423 stream
->next_net_seq_num
- 1);
424 pthread_mutex_unlock(&relayd
->ctrl_sock_mutex
);
426 DBG("Unable to close stream on the relayd. Continuing");
428 * Continue here. There is nothing we can do for the relayd.
429 * Chances are that the relayd has closed the socket so we just
430 * continue cleaning up.
434 /* Both conditions are met, we destroy the relayd. */
435 if (uatomic_read(&relayd
->refcount
) == 0 &&
436 uatomic_read(&relayd
->destroy_flag
)) {
437 destroy_relayd(relayd
);
442 uatomic_dec(&stream
->chan
->refcount
);
443 if (!uatomic_read(&stream
->chan
->refcount
)
444 && !uatomic_read(&stream
->chan
->nb_init_streams
)) {
445 free_chan
= stream
->chan
;
449 consumer_data
.need_update
= 1;
450 pthread_mutex_unlock(&consumer_data
.lock
);
451 pthread_mutex_unlock(&stream
->lock
);
454 consumer_del_channel(free_chan
);
458 call_rcu(&stream
->node
.head
, consumer_free_stream
);
461 struct lttng_consumer_stream
*consumer_allocate_stream(
462 int channel_key
, int stream_key
,
463 int shm_fd
, int wait_fd
,
464 enum lttng_consumer_stream_state state
,
466 enum lttng_event_output output
,
467 const char *path_name
,
475 struct lttng_consumer_stream
*stream
;
477 stream
= zmalloc(sizeof(*stream
));
478 if (stream
== NULL
) {
479 PERROR("malloc struct lttng_consumer_stream");
480 *alloc_ret
= -ENOMEM
;
485 * Get stream's channel reference. Needed when adding the stream to the
488 stream
->chan
= consumer_find_channel(channel_key
);
490 *alloc_ret
= -ENOENT
;
491 ERR("Unable to find channel for stream %d", stream_key
);
495 stream
->key
= stream_key
;
496 stream
->shm_fd
= shm_fd
;
497 stream
->wait_fd
= wait_fd
;
499 stream
->out_fd_offset
= 0;
500 stream
->state
= state
;
501 stream
->mmap_len
= mmap_len
;
502 stream
->mmap_base
= NULL
;
503 stream
->output
= output
;
506 stream
->net_seq_idx
= net_index
;
507 stream
->metadata_flag
= metadata_flag
;
508 stream
->session_id
= session_id
;
509 strncpy(stream
->path_name
, path_name
, sizeof(stream
->path_name
));
510 stream
->path_name
[sizeof(stream
->path_name
) - 1] = '\0';
511 pthread_mutex_init(&stream
->lock
, NULL
);
514 * Index differently the metadata node because the thread is using an
515 * internal hash table to match streams in the metadata_ht to the epoll set
519 lttng_ht_node_init_ulong(&stream
->node
, stream
->wait_fd
);
521 lttng_ht_node_init_ulong(&stream
->node
, stream
->key
);
524 /* Init session id node with the stream session id */
525 lttng_ht_node_init_ulong(&stream
->node_session_id
, stream
->session_id
);
528 * The cpu number is needed before using any ustctl_* actions. Ignored for
529 * the kernel so the value does not matter.
531 pthread_mutex_lock(&consumer_data
.lock
);
532 stream
->cpu
= stream
->chan
->cpucount
++;
533 pthread_mutex_unlock(&consumer_data
.lock
);
535 DBG3("Allocated stream %s (key %d, shm_fd %d, wait_fd %d, mmap_len %llu,"
536 " out_fd %d, net_seq_idx %d, session_id %" PRIu64
,
537 stream
->path_name
, stream
->key
, stream
->shm_fd
, stream
->wait_fd
,
538 (unsigned long long) stream
->mmap_len
, stream
->out_fd
,
539 stream
->net_seq_idx
, stream
->session_id
);
549 * Add a stream to the global list protected by a mutex.
551 static int consumer_add_stream(struct lttng_consumer_stream
*stream
,
555 struct consumer_relayd_sock_pair
*relayd
;
560 DBG3("Adding consumer stream %d", stream
->key
);
562 pthread_mutex_lock(&consumer_data
.lock
);
565 /* Steal stream identifier to avoid having streams with the same key */
566 consumer_steal_stream_key(stream
->key
, ht
);
568 lttng_ht_add_unique_ulong(ht
, &stream
->node
);
571 * Add stream to the stream_list_ht of the consumer data. No need to steal
572 * the key since the HT does not use it and we allow to add redundant keys
575 lttng_ht_add_ulong(consumer_data
.stream_list_ht
, &stream
->node_session_id
);
577 /* Check and cleanup relayd */
578 relayd
= consumer_find_relayd(stream
->net_seq_idx
);
579 if (relayd
!= NULL
) {
580 uatomic_inc(&relayd
->refcount
);
583 /* Update channel refcount once added without error(s). */
584 uatomic_inc(&stream
->chan
->refcount
);
587 * When nb_init_streams reaches 0, we don't need to trigger any action in
588 * terms of destroying the associated channel, because the action that
589 * causes the count to become 0 also causes a stream to be added. The
590 * channel deletion will thus be triggered by the following removal of this
593 if (uatomic_read(&stream
->chan
->nb_init_streams
) > 0) {
594 uatomic_dec(&stream
->chan
->nb_init_streams
);
597 /* Update consumer data once the node is inserted. */
598 consumer_data
.stream_count
++;
599 consumer_data
.need_update
= 1;
602 pthread_mutex_unlock(&consumer_data
.lock
);
608 * Add relayd socket to global consumer data hashtable. RCU read side lock MUST
609 * be acquired before calling this.
611 static int add_relayd(struct consumer_relayd_sock_pair
*relayd
)
614 struct lttng_ht_node_ulong
*node
;
615 struct lttng_ht_iter iter
;
617 if (relayd
== NULL
) {
622 lttng_ht_lookup(consumer_data
.relayd_ht
,
623 (void *)((unsigned long) relayd
->net_seq_idx
), &iter
);
624 node
= lttng_ht_iter_get_node_ulong(&iter
);
626 /* Relayd already exist. Ignore the insertion */
629 lttng_ht_add_unique_ulong(consumer_data
.relayd_ht
, &relayd
->node
);
636 * Allocate and return a consumer relayd socket.
638 struct consumer_relayd_sock_pair
*consumer_allocate_relayd_sock_pair(
641 struct consumer_relayd_sock_pair
*obj
= NULL
;
643 /* Negative net sequence index is a failure */
644 if (net_seq_idx
< 0) {
648 obj
= zmalloc(sizeof(struct consumer_relayd_sock_pair
));
650 PERROR("zmalloc relayd sock");
654 obj
->net_seq_idx
= net_seq_idx
;
656 obj
->destroy_flag
= 0;
657 lttng_ht_node_init_ulong(&obj
->node
, obj
->net_seq_idx
);
658 pthread_mutex_init(&obj
->ctrl_sock_mutex
, NULL
);
665 * Find a relayd socket pair in the global consumer data.
667 * Return the object if found else NULL.
668 * RCU read-side lock must be held across this call and while using the
671 struct consumer_relayd_sock_pair
*consumer_find_relayd(int key
)
673 struct lttng_ht_iter iter
;
674 struct lttng_ht_node_ulong
*node
;
675 struct consumer_relayd_sock_pair
*relayd
= NULL
;
677 /* Negative keys are lookup failures */
682 lttng_ht_lookup(consumer_data
.relayd_ht
, (void *)((unsigned long) key
),
684 node
= lttng_ht_iter_get_node_ulong(&iter
);
686 relayd
= caa_container_of(node
, struct consumer_relayd_sock_pair
, node
);
694 * Handle stream for relayd transmission if the stream applies for network
695 * streaming where the net sequence index is set.
697 * Return destination file descriptor or negative value on error.
699 static int write_relayd_stream_header(struct lttng_consumer_stream
*stream
,
700 size_t data_size
, unsigned long padding
,
701 struct consumer_relayd_sock_pair
*relayd
)
704 struct lttcomm_relayd_data_hdr data_hdr
;
710 /* Reset data header */
711 memset(&data_hdr
, 0, sizeof(data_hdr
));
713 if (stream
->metadata_flag
) {
714 /* Caller MUST acquire the relayd control socket lock */
715 ret
= relayd_send_metadata(&relayd
->control_sock
, data_size
);
720 /* Metadata are always sent on the control socket. */
721 outfd
= relayd
->control_sock
.fd
;
723 /* Set header with stream information */
724 data_hdr
.stream_id
= htobe64(stream
->relayd_stream_id
);
725 data_hdr
.data_size
= htobe32(data_size
);
726 data_hdr
.padding_size
= htobe32(padding
);
727 data_hdr
.net_seq_num
= htobe64(stream
->next_net_seq_num
++);
728 /* Other fields are zeroed previously */
730 ret
= relayd_send_data_hdr(&relayd
->data_sock
, &data_hdr
,
736 /* Set to go on data socket */
737 outfd
= relayd
->data_sock
.fd
;
745 void consumer_free_channel(struct rcu_head
*head
)
747 struct lttng_ht_node_ulong
*node
=
748 caa_container_of(head
, struct lttng_ht_node_ulong
, head
);
749 struct lttng_consumer_channel
*channel
=
750 caa_container_of(node
, struct lttng_consumer_channel
, node
);
756 * Remove a channel from the global list protected by a mutex. This
757 * function is also responsible for freeing its data structures.
759 void consumer_del_channel(struct lttng_consumer_channel
*channel
)
762 struct lttng_ht_iter iter
;
764 pthread_mutex_lock(&consumer_data
.lock
);
766 switch (consumer_data
.type
) {
767 case LTTNG_CONSUMER_KERNEL
:
769 case LTTNG_CONSUMER32_UST
:
770 case LTTNG_CONSUMER64_UST
:
771 lttng_ustconsumer_del_channel(channel
);
774 ERR("Unknown consumer_data type");
780 iter
.iter
.node
= &channel
->node
.node
;
781 ret
= lttng_ht_del(consumer_data
.channel_ht
, &iter
);
785 if (channel
->mmap_base
!= NULL
) {
786 ret
= munmap(channel
->mmap_base
, channel
->mmap_len
);
791 if (channel
->wait_fd
>= 0 && !channel
->wait_fd_is_copy
) {
792 ret
= close(channel
->wait_fd
);
797 if (channel
->shm_fd
>= 0 && channel
->wait_fd
!= channel
->shm_fd
) {
798 ret
= close(channel
->shm_fd
);
804 call_rcu(&channel
->node
.head
, consumer_free_channel
);
806 pthread_mutex_unlock(&consumer_data
.lock
);
809 struct lttng_consumer_channel
*consumer_allocate_channel(
811 int shm_fd
, int wait_fd
,
813 uint64_t max_sb_size
,
814 unsigned int nb_init_streams
)
816 struct lttng_consumer_channel
*channel
;
819 channel
= zmalloc(sizeof(*channel
));
820 if (channel
== NULL
) {
821 PERROR("malloc struct lttng_consumer_channel");
824 channel
->key
= channel_key
;
825 channel
->shm_fd
= shm_fd
;
826 channel
->wait_fd
= wait_fd
;
827 channel
->mmap_len
= mmap_len
;
828 channel
->max_sb_size
= max_sb_size
;
829 channel
->refcount
= 0;
830 channel
->nb_init_streams
= nb_init_streams
;
831 lttng_ht_node_init_ulong(&channel
->node
, channel
->key
);
833 switch (consumer_data
.type
) {
834 case LTTNG_CONSUMER_KERNEL
:
835 channel
->mmap_base
= NULL
;
836 channel
->mmap_len
= 0;
838 case LTTNG_CONSUMER32_UST
:
839 case LTTNG_CONSUMER64_UST
:
840 ret
= lttng_ustconsumer_allocate_channel(channel
);
847 ERR("Unknown consumer_data type");
851 DBG("Allocated channel (key %d, shm_fd %d, wait_fd %d, mmap_len %llu, max_sb_size %llu)",
852 channel
->key
, channel
->shm_fd
, channel
->wait_fd
,
853 (unsigned long long) channel
->mmap_len
,
854 (unsigned long long) channel
->max_sb_size
);
860 * Add a channel to the global list protected by a mutex.
862 int consumer_add_channel(struct lttng_consumer_channel
*channel
)
864 struct lttng_ht_node_ulong
*node
;
865 struct lttng_ht_iter iter
;
867 pthread_mutex_lock(&consumer_data
.lock
);
868 /* Steal channel identifier, for UST */
869 consumer_steal_channel_key(channel
->key
);
872 lttng_ht_lookup(consumer_data
.channel_ht
,
873 (void *)((unsigned long) channel
->key
), &iter
);
874 node
= lttng_ht_iter_get_node_ulong(&iter
);
876 /* Channel already exist. Ignore the insertion */
880 lttng_ht_add_unique_ulong(consumer_data
.channel_ht
, &channel
->node
);
884 pthread_mutex_unlock(&consumer_data
.lock
);
890 * Allocate the pollfd structure and the local view of the out fds to avoid
891 * doing a lookup in the linked list and concurrency issues when writing is
892 * needed. Called with consumer_data.lock held.
894 * Returns the number of fds in the structures.
896 static int consumer_update_poll_array(
897 struct lttng_consumer_local_data
*ctx
, struct pollfd
**pollfd
,
898 struct lttng_consumer_stream
**local_stream
, struct lttng_ht
*ht
)
901 struct lttng_ht_iter iter
;
902 struct lttng_consumer_stream
*stream
;
904 DBG("Updating poll fd array");
906 cds_lfht_for_each_entry(ht
->ht
, &iter
.iter
, stream
, node
.node
) {
908 * Only active streams with an active end point can be added to the
909 * poll set and local stream storage of the thread.
911 * There is a potential race here for endpoint_status to be updated
912 * just after the check. However, this is OK since the stream(s) will
913 * be deleted once the thread is notified that the end point state has
914 * changed where this function will be called back again.
916 if (stream
->state
!= LTTNG_CONSUMER_ACTIVE_STREAM
||
917 stream
->endpoint_status
) {
920 DBG("Active FD %d", stream
->wait_fd
);
921 (*pollfd
)[i
].fd
= stream
->wait_fd
;
922 (*pollfd
)[i
].events
= POLLIN
| POLLPRI
;
923 local_stream
[i
] = stream
;
929 * Insert the consumer_data_pipe at the end of the array and don't
930 * increment i so nb_fd is the number of real FD.
932 (*pollfd
)[i
].fd
= ctx
->consumer_data_pipe
[0];
933 (*pollfd
)[i
].events
= POLLIN
| POLLPRI
;
938 * Poll on the should_quit pipe and the command socket return -1 on error and
939 * should exit, 0 if data is available on the command socket
941 int lttng_consumer_poll_socket(struct pollfd
*consumer_sockpoll
)
946 num_rdy
= poll(consumer_sockpoll
, 2, -1);
949 * Restart interrupted system call.
951 if (errno
== EINTR
) {
954 PERROR("Poll error");
957 if (consumer_sockpoll
[0].revents
& (POLLIN
| POLLPRI
)) {
958 DBG("consumer_should_quit wake up");
968 * Set the error socket.
970 void lttng_consumer_set_error_sock(
971 struct lttng_consumer_local_data
*ctx
, int sock
)
973 ctx
->consumer_error_socket
= sock
;
977 * Set the command socket path.
979 void lttng_consumer_set_command_sock_path(
980 struct lttng_consumer_local_data
*ctx
, char *sock
)
982 ctx
->consumer_command_sock_path
= sock
;
986 * Send return code to the session daemon.
987 * If the socket is not defined, we return 0, it is not a fatal error
989 int lttng_consumer_send_error(
990 struct lttng_consumer_local_data
*ctx
, int cmd
)
992 if (ctx
->consumer_error_socket
> 0) {
993 return lttcomm_send_unix_sock(ctx
->consumer_error_socket
, &cmd
,
994 sizeof(enum lttcomm_sessiond_command
));
1001 * Close all the tracefiles and stream fds, should be called when all instances
1004 void lttng_consumer_cleanup(void)
1006 struct lttng_ht_iter iter
;
1007 struct lttng_ht_node_ulong
*node
;
1011 cds_lfht_for_each_entry(consumer_data
.channel_ht
->ht
, &iter
.iter
, node
,
1013 struct lttng_consumer_channel
*channel
=
1014 caa_container_of(node
, struct lttng_consumer_channel
, node
);
1015 consumer_del_channel(channel
);
1020 lttng_ht_destroy(consumer_data
.channel_ht
);
1024 * Called from signal handler.
1026 void lttng_consumer_should_exit(struct lttng_consumer_local_data
*ctx
)
1031 ret
= write(ctx
->consumer_should_quit
[1], "4", 1);
1032 } while (ret
< 0 && errno
== EINTR
);
1034 PERROR("write consumer quit");
1037 DBG("Consumer flag that it should quit");
1040 void lttng_consumer_sync_trace_file(struct lttng_consumer_stream
*stream
,
1043 int outfd
= stream
->out_fd
;
1046 * This does a blocking write-and-wait on any page that belongs to the
1047 * subbuffer prior to the one we just wrote.
1048 * Don't care about error values, as these are just hints and ways to
1049 * limit the amount of page cache used.
1051 if (orig_offset
< stream
->chan
->max_sb_size
) {
1054 lttng_sync_file_range(outfd
, orig_offset
- stream
->chan
->max_sb_size
,
1055 stream
->chan
->max_sb_size
,
1056 SYNC_FILE_RANGE_WAIT_BEFORE
1057 | SYNC_FILE_RANGE_WRITE
1058 | SYNC_FILE_RANGE_WAIT_AFTER
);
1060 * Give hints to the kernel about how we access the file:
1061 * POSIX_FADV_DONTNEED : we won't re-access data in a near future after
1064 * We need to call fadvise again after the file grows because the
1065 * kernel does not seem to apply fadvise to non-existing parts of the
1068 * Call fadvise _after_ having waited for the page writeback to
1069 * complete because the dirty page writeback semantic is not well
1070 * defined. So it can be expected to lead to lower throughput in
1073 posix_fadvise(outfd
, orig_offset
- stream
->chan
->max_sb_size
,
1074 stream
->chan
->max_sb_size
, POSIX_FADV_DONTNEED
);
1078 * Initialise the necessary environnement :
1079 * - create a new context
1080 * - create the poll_pipe
1081 * - create the should_quit pipe (for signal handler)
1082 * - create the thread pipe (for splice)
1084 * Takes a function pointer as argument, this function is called when data is
1085 * available on a buffer. This function is responsible to do the
1086 * kernctl_get_next_subbuf, read the data with mmap or splice depending on the
1087 * buffer configuration and then kernctl_put_next_subbuf at the end.
1089 * Returns a pointer to the new context or NULL on error.
1091 struct lttng_consumer_local_data
*lttng_consumer_create(
1092 enum lttng_consumer_type type
,
1093 ssize_t (*buffer_ready
)(struct lttng_consumer_stream
*stream
,
1094 struct lttng_consumer_local_data
*ctx
),
1095 int (*recv_channel
)(struct lttng_consumer_channel
*channel
),
1096 int (*recv_stream
)(struct lttng_consumer_stream
*stream
),
1097 int (*update_stream
)(int stream_key
, uint32_t state
))
1100 struct lttng_consumer_local_data
*ctx
;
1102 assert(consumer_data
.type
== LTTNG_CONSUMER_UNKNOWN
||
1103 consumer_data
.type
== type
);
1104 consumer_data
.type
= type
;
1106 ctx
= zmalloc(sizeof(struct lttng_consumer_local_data
));
1108 PERROR("allocating context");
1112 ctx
->consumer_error_socket
= -1;
1113 /* assign the callbacks */
1114 ctx
->on_buffer_ready
= buffer_ready
;
1115 ctx
->on_recv_channel
= recv_channel
;
1116 ctx
->on_recv_stream
= recv_stream
;
1117 ctx
->on_update_stream
= update_stream
;
1119 ret
= pipe(ctx
->consumer_data_pipe
);
1121 PERROR("Error creating poll pipe");
1122 goto error_poll_pipe
;
1125 /* set read end of the pipe to non-blocking */
1126 ret
= fcntl(ctx
->consumer_data_pipe
[0], F_SETFL
, O_NONBLOCK
);
1128 PERROR("fcntl O_NONBLOCK");
1129 goto error_poll_fcntl
;
1132 /* set write end of the pipe to non-blocking */
1133 ret
= fcntl(ctx
->consumer_data_pipe
[1], F_SETFL
, O_NONBLOCK
);
1135 PERROR("fcntl O_NONBLOCK");
1136 goto error_poll_fcntl
;
1139 ret
= pipe(ctx
->consumer_should_quit
);
1141 PERROR("Error creating recv pipe");
1142 goto error_quit_pipe
;
1145 ret
= pipe(ctx
->consumer_thread_pipe
);
1147 PERROR("Error creating thread pipe");
1148 goto error_thread_pipe
;
1151 ret
= utils_create_pipe(ctx
->consumer_metadata_pipe
);
1153 goto error_metadata_pipe
;
1156 ret
= utils_create_pipe(ctx
->consumer_splice_metadata_pipe
);
1158 goto error_splice_pipe
;
1164 utils_close_pipe(ctx
->consumer_metadata_pipe
);
1165 error_metadata_pipe
:
1166 utils_close_pipe(ctx
->consumer_thread_pipe
);
1168 for (i
= 0; i
< 2; i
++) {
1171 err
= close(ctx
->consumer_should_quit
[i
]);
1178 for (i
= 0; i
< 2; i
++) {
1181 err
= close(ctx
->consumer_data_pipe
[i
]);
1193 * Close all fds associated with the instance and free the context.
1195 void lttng_consumer_destroy(struct lttng_consumer_local_data
*ctx
)
1199 DBG("Consumer destroying it. Closing everything.");
1201 ret
= close(ctx
->consumer_error_socket
);
1205 ret
= close(ctx
->consumer_thread_pipe
[0]);
1209 ret
= close(ctx
->consumer_thread_pipe
[1]);
1213 ret
= close(ctx
->consumer_data_pipe
[0]);
1217 ret
= close(ctx
->consumer_data_pipe
[1]);
1221 ret
= close(ctx
->consumer_should_quit
[0]);
1225 ret
= close(ctx
->consumer_should_quit
[1]);
1229 utils_close_pipe(ctx
->consumer_splice_metadata_pipe
);
1231 unlink(ctx
->consumer_command_sock_path
);
1236 * Write the metadata stream id on the specified file descriptor.
1238 static int write_relayd_metadata_id(int fd
,
1239 struct lttng_consumer_stream
*stream
,
1240 struct consumer_relayd_sock_pair
*relayd
,
1241 unsigned long padding
)
1244 struct lttcomm_relayd_metadata_payload hdr
;
1246 hdr
.stream_id
= htobe64(stream
->relayd_stream_id
);
1247 hdr
.padding_size
= htobe32(padding
);
1249 ret
= write(fd
, (void *) &hdr
, sizeof(hdr
));
1250 } while (ret
< 0 && errno
== EINTR
);
1252 PERROR("write metadata stream id");
1255 DBG("Metadata stream id %" PRIu64
" with padding %lu written before data",
1256 stream
->relayd_stream_id
, padding
);
1263 * Mmap the ring buffer, read it and write the data to the tracefile. This is a
1264 * core function for writing trace buffers to either the local filesystem or
1267 * Careful review MUST be put if any changes occur!
1269 * Returns the number of bytes written
1271 ssize_t
lttng_consumer_on_read_subbuffer_mmap(
1272 struct lttng_consumer_local_data
*ctx
,
1273 struct lttng_consumer_stream
*stream
, unsigned long len
,
1274 unsigned long padding
)
1276 unsigned long mmap_offset
;
1277 ssize_t ret
= 0, written
= 0;
1278 off_t orig_offset
= stream
->out_fd_offset
;
1279 /* Default is on the disk */
1280 int outfd
= stream
->out_fd
;
1281 struct consumer_relayd_sock_pair
*relayd
= NULL
;
1282 unsigned int relayd_hang_up
= 0;
1284 /* RCU lock for the relayd pointer */
1287 pthread_mutex_lock(&stream
->lock
);
1289 /* Flag that the current stream if set for network streaming. */
1290 if (stream
->net_seq_idx
!= -1) {
1291 relayd
= consumer_find_relayd(stream
->net_seq_idx
);
1292 if (relayd
== NULL
) {
1297 /* get the offset inside the fd to mmap */
1298 switch (consumer_data
.type
) {
1299 case LTTNG_CONSUMER_KERNEL
:
1300 ret
= kernctl_get_mmap_read_offset(stream
->wait_fd
, &mmap_offset
);
1302 case LTTNG_CONSUMER32_UST
:
1303 case LTTNG_CONSUMER64_UST
:
1304 ret
= lttng_ustctl_get_mmap_read_offset(stream
->chan
->handle
,
1305 stream
->buf
, &mmap_offset
);
1308 ERR("Unknown consumer_data type");
1313 PERROR("tracer ctl get_mmap_read_offset");
1318 /* Handle stream on the relayd if the output is on the network */
1320 unsigned long netlen
= len
;
1323 * Lock the control socket for the complete duration of the function
1324 * since from this point on we will use the socket.
1326 if (stream
->metadata_flag
) {
1327 /* Metadata requires the control socket. */
1328 pthread_mutex_lock(&relayd
->ctrl_sock_mutex
);
1329 netlen
+= sizeof(struct lttcomm_relayd_metadata_payload
);
1332 ret
= write_relayd_stream_header(stream
, netlen
, padding
, relayd
);
1334 /* Use the returned socket. */
1337 /* Write metadata stream id before payload */
1338 if (stream
->metadata_flag
) {
1339 ret
= write_relayd_metadata_id(outfd
, stream
, relayd
, padding
);
1342 /* Socket operation failed. We consider the relayd dead */
1343 if (ret
== -EPIPE
|| ret
== -EINVAL
) {
1351 /* Socket operation failed. We consider the relayd dead */
1352 if (ret
== -EPIPE
|| ret
== -EINVAL
) {
1356 /* Else, use the default set before which is the filesystem. */
1359 /* No streaming, we have to set the len with the full padding */
1365 ret
= write(outfd
, stream
->mmap_base
+ mmap_offset
, len
);
1366 } while (ret
< 0 && errno
== EINTR
);
1367 DBG("Consumer mmap write() ret %zd (len %lu)", ret
, len
);
1369 PERROR("Error in file write");
1373 /* Socket operation failed. We consider the relayd dead */
1374 if (errno
== EPIPE
|| errno
== EINVAL
) {
1379 } else if (ret
> len
) {
1380 PERROR("Error in file write (ret %zd > len %lu)", ret
, len
);
1388 /* This call is useless on a socket so better save a syscall. */
1390 /* This won't block, but will start writeout asynchronously */
1391 lttng_sync_file_range(outfd
, stream
->out_fd_offset
, ret
,
1392 SYNC_FILE_RANGE_WRITE
);
1393 stream
->out_fd_offset
+= ret
;
1397 lttng_consumer_sync_trace_file(stream
, orig_offset
);
1401 * This is a special case that the relayd has closed its socket. Let's
1402 * cleanup the relayd object and all associated streams.
1404 if (relayd
&& relayd_hang_up
) {
1405 cleanup_relayd(relayd
, ctx
);
1409 /* Unlock only if ctrl socket used */
1410 if (relayd
&& stream
->metadata_flag
) {
1411 pthread_mutex_unlock(&relayd
->ctrl_sock_mutex
);
1413 pthread_mutex_unlock(&stream
->lock
);
1420 * Splice the data from the ring buffer to the tracefile.
1422 * Returns the number of bytes spliced.
1424 ssize_t
lttng_consumer_on_read_subbuffer_splice(
1425 struct lttng_consumer_local_data
*ctx
,
1426 struct lttng_consumer_stream
*stream
, unsigned long len
,
1427 unsigned long padding
)
1429 ssize_t ret
= 0, written
= 0, ret_splice
= 0;
1431 off_t orig_offset
= stream
->out_fd_offset
;
1432 int fd
= stream
->wait_fd
;
1433 /* Default is on the disk */
1434 int outfd
= stream
->out_fd
;
1435 struct consumer_relayd_sock_pair
*relayd
= NULL
;
1437 unsigned int relayd_hang_up
= 0;
1439 switch (consumer_data
.type
) {
1440 case LTTNG_CONSUMER_KERNEL
:
1442 case LTTNG_CONSUMER32_UST
:
1443 case LTTNG_CONSUMER64_UST
:
1444 /* Not supported for user space tracing */
1447 ERR("Unknown consumer_data type");
1451 /* RCU lock for the relayd pointer */
1454 pthread_mutex_lock(&stream
->lock
);
1456 /* Flag that the current stream if set for network streaming. */
1457 if (stream
->net_seq_idx
!= -1) {
1458 relayd
= consumer_find_relayd(stream
->net_seq_idx
);
1459 if (relayd
== NULL
) {
1465 * Choose right pipe for splice. Metadata and trace data are handled by
1466 * different threads hence the use of two pipes in order not to race or
1467 * corrupt the written data.
1469 if (stream
->metadata_flag
) {
1470 splice_pipe
= ctx
->consumer_splice_metadata_pipe
;
1472 splice_pipe
= ctx
->consumer_thread_pipe
;
1475 /* Write metadata stream id before payload */
1477 int total_len
= len
;
1479 if (stream
->metadata_flag
) {
1481 * Lock the control socket for the complete duration of the function
1482 * since from this point on we will use the socket.
1484 pthread_mutex_lock(&relayd
->ctrl_sock_mutex
);
1486 ret
= write_relayd_metadata_id(splice_pipe
[1], stream
, relayd
,
1490 /* Socket operation failed. We consider the relayd dead */
1491 if (ret
== -EBADF
) {
1492 WARN("Remote relayd disconnected. Stopping");
1499 total_len
+= sizeof(struct lttcomm_relayd_metadata_payload
);
1502 ret
= write_relayd_stream_header(stream
, total_len
, padding
, relayd
);
1504 /* Use the returned socket. */
1507 /* Socket operation failed. We consider the relayd dead */
1508 if (ret
== -EBADF
) {
1509 WARN("Remote relayd disconnected. Stopping");
1516 /* No streaming, we have to set the len with the full padding */
1521 DBG("splice chan to pipe offset %lu of len %lu (fd : %d, pipe: %d)",
1522 (unsigned long)offset
, len
, fd
, splice_pipe
[1]);
1523 ret_splice
= splice(fd
, &offset
, splice_pipe
[1], NULL
, len
,
1524 SPLICE_F_MOVE
| SPLICE_F_MORE
);
1525 DBG("splice chan to pipe, ret %zd", ret_splice
);
1526 if (ret_splice
< 0) {
1527 PERROR("Error in relay splice");
1529 written
= ret_splice
;
1535 /* Handle stream on the relayd if the output is on the network */
1537 if (stream
->metadata_flag
) {
1538 size_t metadata_payload_size
=
1539 sizeof(struct lttcomm_relayd_metadata_payload
);
1541 /* Update counter to fit the spliced data */
1542 ret_splice
+= metadata_payload_size
;
1543 len
+= metadata_payload_size
;
1545 * We do this so the return value can match the len passed as
1546 * argument to this function.
1548 written
-= metadata_payload_size
;
1552 /* Splice data out */
1553 ret_splice
= splice(splice_pipe
[0], NULL
, outfd
, NULL
,
1554 ret_splice
, SPLICE_F_MOVE
| SPLICE_F_MORE
);
1555 DBG("Consumer splice pipe to file, ret %zd", ret_splice
);
1556 if (ret_splice
< 0) {
1557 PERROR("Error in file splice");
1559 written
= ret_splice
;
1561 /* Socket operation failed. We consider the relayd dead */
1562 if (errno
== EBADF
|| errno
== EPIPE
) {
1563 WARN("Remote relayd disconnected. Stopping");
1569 } else if (ret_splice
> len
) {
1571 PERROR("Wrote more data than requested %zd (len: %lu)",
1573 written
+= ret_splice
;
1579 /* This call is useless on a socket so better save a syscall. */
1581 /* This won't block, but will start writeout asynchronously */
1582 lttng_sync_file_range(outfd
, stream
->out_fd_offset
, ret_splice
,
1583 SYNC_FILE_RANGE_WRITE
);
1584 stream
->out_fd_offset
+= ret_splice
;
1586 written
+= ret_splice
;
1588 lttng_consumer_sync_trace_file(stream
, orig_offset
);
1596 * This is a special case that the relayd has closed its socket. Let's
1597 * cleanup the relayd object and all associated streams.
1599 if (relayd
&& relayd_hang_up
) {
1600 cleanup_relayd(relayd
, ctx
);
1601 /* Skip splice error so the consumer does not fail */
1606 /* send the appropriate error description to sessiond */
1609 lttng_consumer_send_error(ctx
, LTTCOMM_CONSUMERD_SPLICE_EINVAL
);
1612 lttng_consumer_send_error(ctx
, LTTCOMM_CONSUMERD_SPLICE_ENOMEM
);
1615 lttng_consumer_send_error(ctx
, LTTCOMM_CONSUMERD_SPLICE_ESPIPE
);
1620 if (relayd
&& stream
->metadata_flag
) {
1621 pthread_mutex_unlock(&relayd
->ctrl_sock_mutex
);
1623 pthread_mutex_unlock(&stream
->lock
);
1630 * Take a snapshot for a specific fd
1632 * Returns 0 on success, < 0 on error
1634 int lttng_consumer_take_snapshot(struct lttng_consumer_local_data
*ctx
,
1635 struct lttng_consumer_stream
*stream
)
1637 switch (consumer_data
.type
) {
1638 case LTTNG_CONSUMER_KERNEL
:
1639 return lttng_kconsumer_take_snapshot(ctx
, stream
);
1640 case LTTNG_CONSUMER32_UST
:
1641 case LTTNG_CONSUMER64_UST
:
1642 return lttng_ustconsumer_take_snapshot(ctx
, stream
);
1644 ERR("Unknown consumer_data type");
1652 * Get the produced position
1654 * Returns 0 on success, < 0 on error
1656 int lttng_consumer_get_produced_snapshot(
1657 struct lttng_consumer_local_data
*ctx
,
1658 struct lttng_consumer_stream
*stream
,
1661 switch (consumer_data
.type
) {
1662 case LTTNG_CONSUMER_KERNEL
:
1663 return lttng_kconsumer_get_produced_snapshot(ctx
, stream
, pos
);
1664 case LTTNG_CONSUMER32_UST
:
1665 case LTTNG_CONSUMER64_UST
:
1666 return lttng_ustconsumer_get_produced_snapshot(ctx
, stream
, pos
);
1668 ERR("Unknown consumer_data type");
1674 int lttng_consumer_recv_cmd(struct lttng_consumer_local_data
*ctx
,
1675 int sock
, struct pollfd
*consumer_sockpoll
)
1677 switch (consumer_data
.type
) {
1678 case LTTNG_CONSUMER_KERNEL
:
1679 return lttng_kconsumer_recv_cmd(ctx
, sock
, consumer_sockpoll
);
1680 case LTTNG_CONSUMER32_UST
:
1681 case LTTNG_CONSUMER64_UST
:
1682 return lttng_ustconsumer_recv_cmd(ctx
, sock
, consumer_sockpoll
);
1684 ERR("Unknown consumer_data type");
1691 * Iterate over all streams of the hashtable and free them properly.
1693 * WARNING: *MUST* be used with data stream only.
1695 static void destroy_data_stream_ht(struct lttng_ht
*ht
)
1698 struct lttng_ht_iter iter
;
1699 struct lttng_consumer_stream
*stream
;
1706 cds_lfht_for_each_entry(ht
->ht
, &iter
.iter
, stream
, node
.node
) {
1707 ret
= lttng_ht_del(ht
, &iter
);
1710 call_rcu(&stream
->node
.head
, consumer_free_stream
);
1714 lttng_ht_destroy(ht
);
1718 * Iterate over all streams of the hashtable and free them properly.
1720 * XXX: Should not be only for metadata stream or else use an other name.
1722 static void destroy_stream_ht(struct lttng_ht
*ht
)
1725 struct lttng_ht_iter iter
;
1726 struct lttng_consumer_stream
*stream
;
1733 cds_lfht_for_each_entry(ht
->ht
, &iter
.iter
, stream
, node
.node
) {
1734 ret
= lttng_ht_del(ht
, &iter
);
1737 call_rcu(&stream
->node
.head
, consumer_free_stream
);
1741 lttng_ht_destroy(ht
);
1745 * Clean up a metadata stream and free its memory.
1747 void consumer_del_metadata_stream(struct lttng_consumer_stream
*stream
,
1748 struct lttng_ht
*ht
)
1751 struct lttng_ht_iter iter
;
1752 struct lttng_consumer_channel
*free_chan
= NULL
;
1753 struct consumer_relayd_sock_pair
*relayd
;
1757 * This call should NEVER receive regular stream. It must always be
1758 * metadata stream and this is crucial for data structure synchronization.
1760 assert(stream
->metadata_flag
);
1762 DBG3("Consumer delete metadata stream %d", stream
->wait_fd
);
1765 /* Means the stream was allocated but not successfully added */
1769 pthread_mutex_lock(&stream
->lock
);
1771 pthread_mutex_lock(&consumer_data
.lock
);
1772 switch (consumer_data
.type
) {
1773 case LTTNG_CONSUMER_KERNEL
:
1774 if (stream
->mmap_base
!= NULL
) {
1775 ret
= munmap(stream
->mmap_base
, stream
->mmap_len
);
1777 PERROR("munmap metadata stream");
1781 case LTTNG_CONSUMER32_UST
:
1782 case LTTNG_CONSUMER64_UST
:
1783 lttng_ustconsumer_del_stream(stream
);
1786 ERR("Unknown consumer_data type");
1792 iter
.iter
.node
= &stream
->node
.node
;
1793 ret
= lttng_ht_del(ht
, &iter
);
1796 /* Remove node session id from the consumer_data stream ht */
1797 iter
.iter
.node
= &stream
->node_session_id
.node
;
1798 ret
= lttng_ht_del(consumer_data
.stream_list_ht
, &iter
);
1802 if (stream
->out_fd
>= 0) {
1803 ret
= close(stream
->out_fd
);
1809 if (stream
->wait_fd
>= 0 && !stream
->wait_fd_is_copy
) {
1810 ret
= close(stream
->wait_fd
);
1816 if (stream
->shm_fd
>= 0 && stream
->wait_fd
!= stream
->shm_fd
) {
1817 ret
= close(stream
->shm_fd
);
1823 /* Check and cleanup relayd */
1825 relayd
= consumer_find_relayd(stream
->net_seq_idx
);
1826 if (relayd
!= NULL
) {
1827 uatomic_dec(&relayd
->refcount
);
1828 assert(uatomic_read(&relayd
->refcount
) >= 0);
1830 /* Closing streams requires to lock the control socket. */
1831 pthread_mutex_lock(&relayd
->ctrl_sock_mutex
);
1832 ret
= relayd_send_close_stream(&relayd
->control_sock
,
1833 stream
->relayd_stream_id
, stream
->next_net_seq_num
- 1);
1834 pthread_mutex_unlock(&relayd
->ctrl_sock_mutex
);
1836 DBG("Unable to close stream on the relayd. Continuing");
1838 * Continue here. There is nothing we can do for the relayd.
1839 * Chances are that the relayd has closed the socket so we just
1840 * continue cleaning up.
1844 /* Both conditions are met, we destroy the relayd. */
1845 if (uatomic_read(&relayd
->refcount
) == 0 &&
1846 uatomic_read(&relayd
->destroy_flag
)) {
1847 destroy_relayd(relayd
);
1852 /* Atomically decrement channel refcount since other threads can use it. */
1853 uatomic_dec(&stream
->chan
->refcount
);
1854 if (!uatomic_read(&stream
->chan
->refcount
)
1855 && !uatomic_read(&stream
->chan
->nb_init_streams
)) {
1856 /* Go for channel deletion! */
1857 free_chan
= stream
->chan
;
1861 pthread_mutex_unlock(&consumer_data
.lock
);
1862 pthread_mutex_unlock(&stream
->lock
);
1865 consumer_del_channel(free_chan
);
1869 call_rcu(&stream
->node
.head
, consumer_free_stream
);
1873 * Action done with the metadata stream when adding it to the consumer internal
1874 * data structures to handle it.
1876 static int consumer_add_metadata_stream(struct lttng_consumer_stream
*stream
,
1877 struct lttng_ht
*ht
)
1880 struct consumer_relayd_sock_pair
*relayd
;
1885 DBG3("Adding metadata stream %d to hash table", stream
->wait_fd
);
1887 pthread_mutex_lock(&consumer_data
.lock
);
1890 * From here, refcounts are updated so be _careful_ when returning an error
1895 /* Find relayd and, if one is found, increment refcount. */
1896 relayd
= consumer_find_relayd(stream
->net_seq_idx
);
1897 if (relayd
!= NULL
) {
1898 uatomic_inc(&relayd
->refcount
);
1901 /* Update channel refcount once added without error(s). */
1902 uatomic_inc(&stream
->chan
->refcount
);
1905 * When nb_init_streams reaches 0, we don't need to trigger any action in
1906 * terms of destroying the associated channel, because the action that
1907 * causes the count to become 0 also causes a stream to be added. The
1908 * channel deletion will thus be triggered by the following removal of this
1911 if (uatomic_read(&stream
->chan
->nb_init_streams
) > 0) {
1912 uatomic_dec(&stream
->chan
->nb_init_streams
);
1915 /* Steal stream identifier to avoid having streams with the same key */
1916 consumer_steal_stream_key(stream
->key
, ht
);
1918 lttng_ht_add_unique_ulong(ht
, &stream
->node
);
1921 * Add stream to the stream_list_ht of the consumer data. No need to steal
1922 * the key since the HT does not use it and we allow to add redundant keys
1925 lttng_ht_add_ulong(consumer_data
.stream_list_ht
, &stream
->node_session_id
);
1929 pthread_mutex_unlock(&consumer_data
.lock
);
1934 * Delete data stream that are flagged for deletion (endpoint_status).
1936 static void validate_endpoint_status_data_stream(void)
1938 struct lttng_ht_iter iter
;
1939 struct lttng_consumer_stream
*stream
;
1941 DBG("Consumer delete flagged data stream");
1944 cds_lfht_for_each_entry(data_ht
->ht
, &iter
.iter
, stream
, node
.node
) {
1945 /* Validate delete flag of the stream */
1946 if (stream
->endpoint_status
!= CONSUMER_ENDPOINT_INACTIVE
) {
1949 /* Delete it right now */
1950 consumer_del_stream(stream
, data_ht
);
1956 * Delete metadata stream that are flagged for deletion (endpoint_status).
1958 static void validate_endpoint_status_metadata_stream(
1959 struct lttng_poll_event
*pollset
)
1961 struct lttng_ht_iter iter
;
1962 struct lttng_consumer_stream
*stream
;
1964 DBG("Consumer delete flagged metadata stream");
1969 cds_lfht_for_each_entry(metadata_ht
->ht
, &iter
.iter
, stream
, node
.node
) {
1970 /* Validate delete flag of the stream */
1971 if (!stream
->endpoint_status
) {
1975 * Remove from pollset so the metadata thread can continue without
1976 * blocking on a deleted stream.
1978 lttng_poll_del(pollset
, stream
->wait_fd
);
1980 /* Delete it right now */
1981 consumer_del_metadata_stream(stream
, metadata_ht
);
1987 * Thread polls on metadata file descriptor and write them on disk or on the
1990 void *consumer_thread_metadata_poll(void *data
)
1993 uint32_t revents
, nb_fd
;
1994 struct lttng_consumer_stream
*stream
= NULL
;
1995 struct lttng_ht_iter iter
;
1996 struct lttng_ht_node_ulong
*node
;
1997 struct lttng_poll_event events
;
1998 struct lttng_consumer_local_data
*ctx
= data
;
2001 rcu_register_thread();
2003 DBG("Thread metadata poll started");
2005 /* Size is set to 1 for the consumer_metadata pipe */
2006 ret
= lttng_poll_create(&events
, 2, LTTNG_CLOEXEC
);
2008 ERR("Poll set creation failed");
2012 ret
= lttng_poll_add(&events
, ctx
->consumer_metadata_pipe
[0], LPOLLIN
);
2018 DBG("Metadata main loop started");
2021 lttng_poll_reset(&events
);
2023 nb_fd
= LTTNG_POLL_GETNB(&events
);
2025 /* Only the metadata pipe is set */
2026 if (nb_fd
== 0 && consumer_quit
== 1) {
2031 DBG("Metadata poll wait with %d fd(s)", nb_fd
);
2032 ret
= lttng_poll_wait(&events
, -1);
2033 DBG("Metadata event catched in thread");
2035 if (errno
== EINTR
) {
2036 ERR("Poll EINTR catched");
2042 /* From here, the event is a metadata wait fd */
2043 for (i
= 0; i
< nb_fd
; i
++) {
2044 revents
= LTTNG_POLL_GETEV(&events
, i
);
2045 pollfd
= LTTNG_POLL_GETFD(&events
, i
);
2047 /* Just don't waste time if no returned events for the fd */
2052 if (pollfd
== ctx
->consumer_metadata_pipe
[0]) {
2053 if (revents
& (LPOLLERR
| LPOLLHUP
)) {
2054 DBG("Metadata thread pipe hung up");
2056 * Remove the pipe from the poll set and continue the loop
2057 * since their might be data to consume.
2059 lttng_poll_del(&events
, ctx
->consumer_metadata_pipe
[0]);
2060 close(ctx
->consumer_metadata_pipe
[0]);
2062 } else if (revents
& LPOLLIN
) {
2064 /* Get the stream pointer received */
2065 ret
= read(pollfd
, &stream
, sizeof(stream
));
2066 } while (ret
< 0 && errno
== EINTR
);
2068 ret
< sizeof(struct lttng_consumer_stream
*)) {
2069 PERROR("read metadata stream");
2071 * Let's continue here and hope we can still work
2072 * without stopping the consumer. XXX: Should we?
2077 /* A NULL stream means that the state has changed. */
2078 if (stream
== NULL
) {
2079 /* Check for deleted streams. */
2080 validate_endpoint_status_metadata_stream(&events
);
2084 DBG("Adding metadata stream %d to poll set",
2087 ret
= consumer_add_metadata_stream(stream
, metadata_ht
);
2089 ERR("Unable to add metadata stream");
2090 /* Stream was not setup properly. Continuing. */
2091 consumer_del_metadata_stream(stream
, NULL
);
2095 /* Add metadata stream to the global poll events list */
2096 lttng_poll_add(&events
, stream
->wait_fd
,
2097 LPOLLIN
| LPOLLPRI
);
2100 /* Handle other stream */
2105 lttng_ht_lookup(metadata_ht
, (void *)((unsigned long) pollfd
),
2107 node
= lttng_ht_iter_get_node_ulong(&iter
);
2110 stream
= caa_container_of(node
, struct lttng_consumer_stream
,
2113 /* Check for error event */
2114 if (revents
& (LPOLLERR
| LPOLLHUP
)) {
2115 DBG("Metadata fd %d is hup|err.", pollfd
);
2116 if (!stream
->hangup_flush_done
2117 && (consumer_data
.type
== LTTNG_CONSUMER32_UST
2118 || consumer_data
.type
== LTTNG_CONSUMER64_UST
)) {
2119 DBG("Attempting to flush and consume the UST buffers");
2120 lttng_ustconsumer_on_stream_hangup(stream
);
2122 /* We just flushed the stream now read it. */
2124 len
= ctx
->on_buffer_ready(stream
, ctx
);
2126 * We don't check the return value here since if we get
2127 * a negative len, it means an error occured thus we
2128 * simply remove it from the poll set and free the
2134 lttng_poll_del(&events
, stream
->wait_fd
);
2136 * This call update the channel states, closes file descriptors
2137 * and securely free the stream.
2139 consumer_del_metadata_stream(stream
, metadata_ht
);
2140 } else if (revents
& (LPOLLIN
| LPOLLPRI
)) {
2141 /* Get the data out of the metadata file descriptor */
2142 DBG("Metadata available on fd %d", pollfd
);
2143 assert(stream
->wait_fd
== pollfd
);
2145 len
= ctx
->on_buffer_ready(stream
, ctx
);
2146 /* It's ok to have an unavailable sub-buffer */
2147 if (len
< 0 && len
!= -EAGAIN
&& len
!= -ENODATA
) {
2148 /* Clean up stream from consumer and free it. */
2149 lttng_poll_del(&events
, stream
->wait_fd
);
2150 consumer_del_metadata_stream(stream
, metadata_ht
);
2151 } else if (len
> 0) {
2152 stream
->data_read
= 1;
2156 /* Release RCU lock for the stream looked up */
2163 DBG("Metadata poll thread exiting");
2164 lttng_poll_clean(&events
);
2167 destroy_stream_ht(metadata_ht
);
2170 rcu_unregister_thread();
2175 * This thread polls the fds in the set to consume the data and write
2176 * it to tracefile if necessary.
2178 void *consumer_thread_data_poll(void *data
)
2180 int num_rdy
, num_hup
, high_prio
, ret
, i
;
2181 struct pollfd
*pollfd
= NULL
;
2182 /* local view of the streams */
2183 struct lttng_consumer_stream
**local_stream
= NULL
, *new_stream
= NULL
;
2184 /* local view of consumer_data.fds_count */
2186 struct lttng_consumer_local_data
*ctx
= data
;
2189 rcu_register_thread();
2191 data_ht
= lttng_ht_new(0, LTTNG_HT_TYPE_ULONG
);
2192 if (data_ht
== NULL
) {
2196 local_stream
= zmalloc(sizeof(struct lttng_consumer_stream
));
2203 * the fds set has been updated, we need to update our
2204 * local array as well
2206 pthread_mutex_lock(&consumer_data
.lock
);
2207 if (consumer_data
.need_update
) {
2208 if (pollfd
!= NULL
) {
2212 if (local_stream
!= NULL
) {
2214 local_stream
= NULL
;
2217 /* allocate for all fds + 1 for the consumer_data_pipe */
2218 pollfd
= zmalloc((consumer_data
.stream_count
+ 1) * sizeof(struct pollfd
));
2219 if (pollfd
== NULL
) {
2220 PERROR("pollfd malloc");
2221 pthread_mutex_unlock(&consumer_data
.lock
);
2225 /* allocate for all fds + 1 for the consumer_data_pipe */
2226 local_stream
= zmalloc((consumer_data
.stream_count
+ 1) *
2227 sizeof(struct lttng_consumer_stream
));
2228 if (local_stream
== NULL
) {
2229 PERROR("local_stream malloc");
2230 pthread_mutex_unlock(&consumer_data
.lock
);
2233 ret
= consumer_update_poll_array(ctx
, &pollfd
, local_stream
,
2236 ERR("Error in allocating pollfd or local_outfds");
2237 lttng_consumer_send_error(ctx
, LTTCOMM_CONSUMERD_POLL_ERROR
);
2238 pthread_mutex_unlock(&consumer_data
.lock
);
2242 consumer_data
.need_update
= 0;
2244 pthread_mutex_unlock(&consumer_data
.lock
);
2246 /* No FDs and consumer_quit, consumer_cleanup the thread */
2247 if (nb_fd
== 0 && consumer_quit
== 1) {
2250 /* poll on the array of fds */
2252 DBG("polling on %d fd", nb_fd
+ 1);
2253 num_rdy
= poll(pollfd
, nb_fd
+ 1, consumer_poll_timeout
);
2254 DBG("poll num_rdy : %d", num_rdy
);
2255 if (num_rdy
== -1) {
2257 * Restart interrupted system call.
2259 if (errno
== EINTR
) {
2262 PERROR("Poll error");
2263 lttng_consumer_send_error(ctx
, LTTCOMM_CONSUMERD_POLL_ERROR
);
2265 } else if (num_rdy
== 0) {
2266 DBG("Polling thread timed out");
2271 * If the consumer_data_pipe triggered poll go directly to the
2272 * beginning of the loop to update the array. We want to prioritize
2273 * array update over low-priority reads.
2275 if (pollfd
[nb_fd
].revents
& (POLLIN
| POLLPRI
)) {
2276 size_t pipe_readlen
;
2278 DBG("consumer_data_pipe wake up");
2279 /* Consume 1 byte of pipe data */
2281 pipe_readlen
= read(ctx
->consumer_data_pipe
[0], &new_stream
,
2282 sizeof(new_stream
));
2283 } while (pipe_readlen
== -1 && errno
== EINTR
);
2286 * If the stream is NULL, just ignore it. It's also possible that
2287 * the sessiond poll thread changed the consumer_quit state and is
2288 * waking us up to test it.
2290 if (new_stream
== NULL
) {
2291 validate_endpoint_status_data_stream();
2295 ret
= consumer_add_stream(new_stream
, data_ht
);
2297 ERR("Consumer add stream %d failed. Continuing",
2300 * At this point, if the add_stream fails, it is not in the
2301 * hash table thus passing the NULL value here.
2303 consumer_del_stream(new_stream
, NULL
);
2306 /* Continue to update the local streams and handle prio ones */
2310 /* Take care of high priority channels first. */
2311 for (i
= 0; i
< nb_fd
; i
++) {
2312 if (local_stream
[i
] == NULL
) {
2315 if (pollfd
[i
].revents
& POLLPRI
) {
2316 DBG("Urgent read on fd %d", pollfd
[i
].fd
);
2318 len
= ctx
->on_buffer_ready(local_stream
[i
], ctx
);
2319 /* it's ok to have an unavailable sub-buffer */
2320 if (len
< 0 && len
!= -EAGAIN
&& len
!= -ENODATA
) {
2321 /* Clean the stream and free it. */
2322 consumer_del_stream(local_stream
[i
], data_ht
);
2323 local_stream
[i
] = NULL
;
2324 } else if (len
> 0) {
2325 local_stream
[i
]->data_read
= 1;
2331 * If we read high prio channel in this loop, try again
2332 * for more high prio data.
2338 /* Take care of low priority channels. */
2339 for (i
= 0; i
< nb_fd
; i
++) {
2340 if (local_stream
[i
] == NULL
) {
2343 if ((pollfd
[i
].revents
& POLLIN
) ||
2344 local_stream
[i
]->hangup_flush_done
) {
2345 DBG("Normal read on fd %d", pollfd
[i
].fd
);
2346 len
= ctx
->on_buffer_ready(local_stream
[i
], ctx
);
2347 /* it's ok to have an unavailable sub-buffer */
2348 if (len
< 0 && len
!= -EAGAIN
&& len
!= -ENODATA
) {
2349 /* Clean the stream and free it. */
2350 consumer_del_stream(local_stream
[i
], data_ht
);
2351 local_stream
[i
] = NULL
;
2352 } else if (len
> 0) {
2353 local_stream
[i
]->data_read
= 1;
2358 /* Handle hangup and errors */
2359 for (i
= 0; i
< nb_fd
; i
++) {
2360 if (local_stream
[i
] == NULL
) {
2363 if (!local_stream
[i
]->hangup_flush_done
2364 && (pollfd
[i
].revents
& (POLLHUP
| POLLERR
| POLLNVAL
))
2365 && (consumer_data
.type
== LTTNG_CONSUMER32_UST
2366 || consumer_data
.type
== LTTNG_CONSUMER64_UST
)) {
2367 DBG("fd %d is hup|err|nval. Attempting flush and read.",
2369 lttng_ustconsumer_on_stream_hangup(local_stream
[i
]);
2370 /* Attempt read again, for the data we just flushed. */
2371 local_stream
[i
]->data_read
= 1;
2374 * If the poll flag is HUP/ERR/NVAL and we have
2375 * read no data in this pass, we can remove the
2376 * stream from its hash table.
2378 if ((pollfd
[i
].revents
& POLLHUP
)) {
2379 DBG("Polling fd %d tells it has hung up.", pollfd
[i
].fd
);
2380 if (!local_stream
[i
]->data_read
) {
2381 consumer_del_stream(local_stream
[i
], data_ht
);
2382 local_stream
[i
] = NULL
;
2385 } else if (pollfd
[i
].revents
& POLLERR
) {
2386 ERR("Error returned in polling fd %d.", pollfd
[i
].fd
);
2387 if (!local_stream
[i
]->data_read
) {
2388 consumer_del_stream(local_stream
[i
], data_ht
);
2389 local_stream
[i
] = NULL
;
2392 } else if (pollfd
[i
].revents
& POLLNVAL
) {
2393 ERR("Polling fd %d tells fd is not open.", pollfd
[i
].fd
);
2394 if (!local_stream
[i
]->data_read
) {
2395 consumer_del_stream(local_stream
[i
], data_ht
);
2396 local_stream
[i
] = NULL
;
2400 if (local_stream
[i
] != NULL
) {
2401 local_stream
[i
]->data_read
= 0;
2406 DBG("polling thread exiting");
2407 if (pollfd
!= NULL
) {
2411 if (local_stream
!= NULL
) {
2413 local_stream
= NULL
;
2417 * Close the write side of the pipe so epoll_wait() in
2418 * consumer_thread_metadata_poll can catch it. The thread is monitoring the
2419 * read side of the pipe. If we close them both, epoll_wait strangely does
2420 * not return and could create a endless wait period if the pipe is the
2421 * only tracked fd in the poll set. The thread will take care of closing
2424 close(ctx
->consumer_metadata_pipe
[1]);
2427 destroy_data_stream_ht(data_ht
);
2430 rcu_unregister_thread();
2435 * This thread listens on the consumerd socket and receives the file
2436 * descriptors from the session daemon.
2438 void *consumer_thread_sessiond_poll(void *data
)
2440 int sock
, client_socket
, ret
;
2442 * structure to poll for incoming data on communication socket avoids
2443 * making blocking sockets.
2445 struct pollfd consumer_sockpoll
[2];
2446 struct lttng_consumer_local_data
*ctx
= data
;
2448 rcu_register_thread();
2450 DBG("Creating command socket %s", ctx
->consumer_command_sock_path
);
2451 unlink(ctx
->consumer_command_sock_path
);
2452 client_socket
= lttcomm_create_unix_sock(ctx
->consumer_command_sock_path
);
2453 if (client_socket
< 0) {
2454 ERR("Cannot create command socket");
2458 ret
= lttcomm_listen_unix_sock(client_socket
);
2463 DBG("Sending ready command to lttng-sessiond");
2464 ret
= lttng_consumer_send_error(ctx
, LTTCOMM_CONSUMERD_COMMAND_SOCK_READY
);
2465 /* return < 0 on error, but == 0 is not fatal */
2467 ERR("Error sending ready command to lttng-sessiond");
2471 ret
= fcntl(client_socket
, F_SETFL
, O_NONBLOCK
);
2473 PERROR("fcntl O_NONBLOCK");
2477 /* prepare the FDs to poll : to client socket and the should_quit pipe */
2478 consumer_sockpoll
[0].fd
= ctx
->consumer_should_quit
[0];
2479 consumer_sockpoll
[0].events
= POLLIN
| POLLPRI
;
2480 consumer_sockpoll
[1].fd
= client_socket
;
2481 consumer_sockpoll
[1].events
= POLLIN
| POLLPRI
;
2483 if (lttng_consumer_poll_socket(consumer_sockpoll
) < 0) {
2486 DBG("Connection on client_socket");
2488 /* Blocking call, waiting for transmission */
2489 sock
= lttcomm_accept_unix_sock(client_socket
);
2494 ret
= fcntl(sock
, F_SETFL
, O_NONBLOCK
);
2496 PERROR("fcntl O_NONBLOCK");
2500 /* update the polling structure to poll on the established socket */
2501 consumer_sockpoll
[1].fd
= sock
;
2502 consumer_sockpoll
[1].events
= POLLIN
| POLLPRI
;
2505 if (lttng_consumer_poll_socket(consumer_sockpoll
) < 0) {
2508 DBG("Incoming command on sock");
2509 ret
= lttng_consumer_recv_cmd(ctx
, sock
, consumer_sockpoll
);
2510 if (ret
== -ENOENT
) {
2511 DBG("Received STOP command");
2516 * This could simply be a session daemon quitting. Don't output
2519 DBG("Communication interrupted on command socket");
2522 if (consumer_quit
) {
2523 DBG("consumer_thread_receive_fds received quit from signal");
2526 DBG("received fds on sock");
2529 DBG("consumer_thread_receive_fds exiting");
2532 * when all fds have hung up, the polling thread
2538 * 2s of grace period, if no polling events occur during
2539 * this period, the polling thread will exit even if there
2540 * are still open FDs (should not happen, but safety mechanism).
2542 consumer_poll_timeout
= LTTNG_CONSUMER_POLL_TIMEOUT
;
2545 * Notify the data poll thread to poll back again and test the
2546 * consumer_quit state that we just set so to quit gracefully.
2548 notify_thread_pipe(ctx
->consumer_data_pipe
[1]);
2550 rcu_unregister_thread();
2554 ssize_t
lttng_consumer_read_subbuffer(struct lttng_consumer_stream
*stream
,
2555 struct lttng_consumer_local_data
*ctx
)
2557 switch (consumer_data
.type
) {
2558 case LTTNG_CONSUMER_KERNEL
:
2559 return lttng_kconsumer_read_subbuffer(stream
, ctx
);
2560 case LTTNG_CONSUMER32_UST
:
2561 case LTTNG_CONSUMER64_UST
:
2562 return lttng_ustconsumer_read_subbuffer(stream
, ctx
);
2564 ERR("Unknown consumer_data type");
2570 int lttng_consumer_on_recv_stream(struct lttng_consumer_stream
*stream
)
2572 switch (consumer_data
.type
) {
2573 case LTTNG_CONSUMER_KERNEL
:
2574 return lttng_kconsumer_on_recv_stream(stream
);
2575 case LTTNG_CONSUMER32_UST
:
2576 case LTTNG_CONSUMER64_UST
:
2577 return lttng_ustconsumer_on_recv_stream(stream
);
2579 ERR("Unknown consumer_data type");
2586 * Allocate and set consumer data hash tables.
2588 void lttng_consumer_init(void)
2590 consumer_data
.channel_ht
= lttng_ht_new(0, LTTNG_HT_TYPE_ULONG
);
2591 consumer_data
.relayd_ht
= lttng_ht_new(0, LTTNG_HT_TYPE_ULONG
);
2592 consumer_data
.stream_list_ht
= lttng_ht_new(0, LTTNG_HT_TYPE_ULONG
);
2594 metadata_ht
= lttng_ht_new(0, LTTNG_HT_TYPE_ULONG
);
2595 assert(metadata_ht
);
2596 data_ht
= lttng_ht_new(0, LTTNG_HT_TYPE_ULONG
);
2601 * Process the ADD_RELAYD command receive by a consumer.
2603 * This will create a relayd socket pair and add it to the relayd hash table.
2604 * The caller MUST acquire a RCU read side lock before calling it.
2606 int consumer_add_relayd_socket(int net_seq_idx
, int sock_type
,
2607 struct lttng_consumer_local_data
*ctx
, int sock
,
2608 struct pollfd
*consumer_sockpoll
, struct lttcomm_sock
*relayd_sock
)
2611 struct consumer_relayd_sock_pair
*relayd
;
2613 DBG("Consumer adding relayd socket (idx: %d)", net_seq_idx
);
2615 /* Get relayd reference if exists. */
2616 relayd
= consumer_find_relayd(net_seq_idx
);
2617 if (relayd
== NULL
) {
2618 /* Not found. Allocate one. */
2619 relayd
= consumer_allocate_relayd_sock_pair(net_seq_idx
);
2620 if (relayd
== NULL
) {
2621 lttng_consumer_send_error(ctx
, LTTCOMM_CONSUMERD_OUTFD_ERROR
);
2626 /* Poll on consumer socket. */
2627 if (lttng_consumer_poll_socket(consumer_sockpoll
) < 0) {
2632 /* Get relayd socket from session daemon */
2633 ret
= lttcomm_recv_fds_unix_sock(sock
, &fd
, 1);
2634 if (ret
!= sizeof(fd
)) {
2635 lttng_consumer_send_error(ctx
, LTTCOMM_CONSUMERD_ERROR_RECV_FD
);
2640 /* Copy socket information and received FD */
2641 switch (sock_type
) {
2642 case LTTNG_STREAM_CONTROL
:
2643 /* Copy received lttcomm socket */
2644 lttcomm_copy_sock(&relayd
->control_sock
, relayd_sock
);
2645 ret
= lttcomm_create_sock(&relayd
->control_sock
);
2650 /* Close the created socket fd which is useless */
2651 close(relayd
->control_sock
.fd
);
2653 /* Assign new file descriptor */
2654 relayd
->control_sock
.fd
= fd
;
2656 case LTTNG_STREAM_DATA
:
2657 /* Copy received lttcomm socket */
2658 lttcomm_copy_sock(&relayd
->data_sock
, relayd_sock
);
2659 ret
= lttcomm_create_sock(&relayd
->data_sock
);
2664 /* Close the created socket fd which is useless */
2665 close(relayd
->data_sock
.fd
);
2667 /* Assign new file descriptor */
2668 relayd
->data_sock
.fd
= fd
;
2671 ERR("Unknown relayd socket type (%d)", sock_type
);
2675 DBG("Consumer %s socket created successfully with net idx %d (fd: %d)",
2676 sock_type
== LTTNG_STREAM_CONTROL
? "control" : "data",
2677 relayd
->net_seq_idx
, fd
);
2680 * Add relayd socket pair to consumer data hashtable. If object already
2681 * exists or on error, the function gracefully returns.
2693 * Try to lock the stream mutex.
2695 * On success, 1 is returned else 0 indicating that the mutex is NOT lock.
2697 static int stream_try_lock(struct lttng_consumer_stream
*stream
)
2704 * Try to lock the stream mutex. On failure, we know that the stream is
2705 * being used else where hence there is data still being extracted.
2707 ret
= pthread_mutex_trylock(&stream
->lock
);
2709 /* For both EBUSY and EINVAL error, the mutex is NOT locked. */
2721 * Check if for a given session id there is still data needed to be extract
2724 * Return 1 if data is pending or else 0 meaning ready to be read.
2726 int consumer_data_pending(uint64_t id
)
2729 struct lttng_ht_iter iter
;
2730 struct lttng_ht
*ht
;
2731 struct lttng_consumer_stream
*stream
;
2732 struct consumer_relayd_sock_pair
*relayd
;
2733 int (*data_pending
)(struct lttng_consumer_stream
*);
2735 DBG("Consumer data pending command on session id %" PRIu64
, id
);
2738 pthread_mutex_lock(&consumer_data
.lock
);
2740 switch (consumer_data
.type
) {
2741 case LTTNG_CONSUMER_KERNEL
:
2742 data_pending
= lttng_kconsumer_data_pending
;
2744 case LTTNG_CONSUMER32_UST
:
2745 case LTTNG_CONSUMER64_UST
:
2746 data_pending
= lttng_ustconsumer_data_pending
;
2749 ERR("Unknown consumer data type");
2753 /* Ease our life a bit */
2754 ht
= consumer_data
.stream_list_ht
;
2756 cds_lfht_for_each_entry_duplicate(ht
->ht
,
2757 ht
->hash_fct((void *)((unsigned long) id
), lttng_ht_seed
),
2758 ht
->match_fct
, (void *)((unsigned long) id
),
2759 &iter
.iter
, stream
, node_session_id
.node
) {
2760 /* If this call fails, the stream is being used hence data pending. */
2761 ret
= stream_try_lock(stream
);
2763 goto data_not_pending
;
2767 * A removed node from the hash table indicates that the stream has
2768 * been deleted thus having a guarantee that the buffers are closed
2769 * on the consumer side. However, data can still be transmitted
2770 * over the network so don't skip the relayd check.
2772 ret
= cds_lfht_is_node_deleted(&stream
->node
.node
);
2774 /* Check the stream if there is data in the buffers. */
2775 ret
= data_pending(stream
);
2777 pthread_mutex_unlock(&stream
->lock
);
2778 goto data_not_pending
;
2783 if (stream
->net_seq_idx
!= -1) {
2784 relayd
= consumer_find_relayd(stream
->net_seq_idx
);
2787 * At this point, if the relayd object is not available for the
2788 * given stream, it is because the relayd is being cleaned up
2789 * so every stream associated with it (for a session id value)
2790 * are or will be marked for deletion hence no data pending.
2792 pthread_mutex_unlock(&stream
->lock
);
2793 goto data_not_pending
;
2796 pthread_mutex_lock(&relayd
->ctrl_sock_mutex
);
2797 if (stream
->metadata_flag
) {
2798 ret
= relayd_quiescent_control(&relayd
->control_sock
);
2800 ret
= relayd_data_pending(&relayd
->control_sock
,
2801 stream
->relayd_stream_id
, stream
->next_net_seq_num
);
2803 pthread_mutex_unlock(&relayd
->ctrl_sock_mutex
);
2805 pthread_mutex_unlock(&stream
->lock
);
2806 goto data_not_pending
;
2809 pthread_mutex_unlock(&stream
->lock
);
2813 * Finding _no_ node in the hash table means that the stream(s) have been
2814 * removed thus data is guaranteed to be available for analysis from the
2815 * trace files. This is *only* true for local consumer and not network
2819 /* Data is available to be read by a viewer. */
2820 pthread_mutex_unlock(&consumer_data
.lock
);
2825 /* Data is still being extracted from buffers. */
2826 pthread_mutex_unlock(&consumer_data
.lock
);