2 * Copyright (C) 2011 - Julien Desfossez <julien.desfossez@polymtl.ca>
3 * Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
4 * Copyright (C) 2013 - David Goulet <dgoulet@efficios.com>
6 * This program is free software; you can redistribute it and/or modify it
7 * under the terms of the GNU General Public License, version 2 only, as
8 * 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 with
16 * this program; if not, write to the Free Software Foundation, Inc., 51
17 * Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
25 #include <common/common.h>
26 #include <common/relayd/relayd.h>
27 #include <common/ust-consumer/ust-consumer.h>
29 #include "consumer-stream.h"
32 * RCU call to free stream. MUST only be used with call_rcu().
34 static void free_stream_rcu(struct rcu_head
*head
)
36 struct lttng_ht_node_u64
*node
=
37 caa_container_of(head
, struct lttng_ht_node_u64
, head
);
38 struct lttng_consumer_stream
*stream
=
39 caa_container_of(node
, struct lttng_consumer_stream
, node
);
41 pthread_mutex_destroy(&stream
->lock
);
46 * Close stream on the relayd side. This call can destroy a relayd if the
49 * A RCU read side lock MUST be acquired if the relayd object was looked up in
50 * a hash table before calling this.
52 void consumer_stream_relayd_close(struct lttng_consumer_stream
*stream
,
53 struct consumer_relayd_sock_pair
*relayd
)
60 uatomic_dec(&relayd
->refcount
);
61 assert(uatomic_read(&relayd
->refcount
) >= 0);
63 /* Closing streams requires to lock the control socket. */
64 pthread_mutex_lock(&relayd
->ctrl_sock_mutex
);
65 ret
= relayd_send_close_stream(&relayd
->control_sock
,
66 stream
->relayd_stream_id
,
67 stream
->next_net_seq_num
- 1);
68 pthread_mutex_unlock(&relayd
->ctrl_sock_mutex
);
70 DBG("Unable to close stream on the relayd. Continuing");
72 * Continue here. There is nothing we can do for the relayd.
73 * Chances are that the relayd has closed the socket so we just
74 * continue cleaning up.
78 /* Both conditions are met, we destroy the relayd. */
79 if (uatomic_read(&relayd
->refcount
) == 0 &&
80 uatomic_read(&relayd
->destroy_flag
)) {
81 consumer_destroy_relayd(relayd
);
86 * Close stream's file descriptors and, if needed, close stream also on the
89 * The consumer data lock MUST be acquired.
90 * The stream lock MUST be acquired.
92 void consumer_stream_close(struct lttng_consumer_stream
*stream
)
95 struct consumer_relayd_sock_pair
*relayd
;
99 switch (consumer_data
.type
) {
100 case LTTNG_CONSUMER_KERNEL
:
101 if (stream
->mmap_base
!= NULL
) {
102 ret
= munmap(stream
->mmap_base
, stream
->mmap_len
);
108 if (stream
->wait_fd
>= 0) {
109 ret
= close(stream
->wait_fd
);
115 case LTTNG_CONSUMER32_UST
:
116 case LTTNG_CONSUMER64_UST
:
117 lttng_ustconsumer_del_stream(stream
);
120 ERR("Unknown consumer_data type");
124 /* Close output fd. Could be a socket or local file at this point. */
125 if (stream
->out_fd
>= 0) {
126 ret
= close(stream
->out_fd
);
132 /* Check and cleanup relayd if needed. */
134 relayd
= consumer_find_relayd(stream
->net_seq_idx
);
135 if (relayd
!= NULL
) {
136 consumer_stream_relayd_close(stream
, relayd
);
142 * Delete the stream from all possible hash tables.
144 * The consumer data lock MUST be acquired.
145 * The stream lock MUST be acquired.
147 void consumer_stream_delete(struct lttng_consumer_stream
*stream
,
151 struct lttng_ht_iter iter
;
158 iter
.iter
.node
= &stream
->node
.node
;
159 ret
= lttng_ht_del(ht
, &iter
);
163 /* Delete from stream per channel ID hash table. */
164 iter
.iter
.node
= &stream
->node_channel_id
.node
;
166 * The returned value is of no importance. Even if the node is NOT in the
167 * hash table, we continue since we may have been called by a code path
168 * that did not add the stream to a (all) hash table. Same goes for the
169 * next call ht del call.
171 (void) lttng_ht_del(consumer_data
.stream_per_chan_id_ht
, &iter
);
173 /* Delete from the global stream list. */
174 iter
.iter
.node
= &stream
->node_session_id
.node
;
175 /* See the previous ht del on why we ignore the returned value. */
176 (void) lttng_ht_del(consumer_data
.stream_list_ht
, &iter
);
180 /* Decrement the stream count of the global consumer data. */
181 assert(consumer_data
.stream_count
> 0);
182 consumer_data
.stream_count
--;
186 * Free the given stream within a RCU call.
188 void consumer_stream_free(struct lttng_consumer_stream
*stream
)
192 call_rcu(&stream
->node
.head
, free_stream_rcu
);
196 * Destroy a stream completely. This will delete, close and free the stream.
197 * Once return, the stream is NO longer usable. Its channel may get destroyed
198 * if conditions are met.
200 * This MUST be called WITHOUT the consumer data and stream lock acquired.
202 void consumer_stream_destroy(struct lttng_consumer_stream
*stream
,
205 struct lttng_consumer_channel
*free_chan
= NULL
;
209 DBG("Consumer stream destroy - wait_fd: %d", stream
->wait_fd
);
211 pthread_mutex_lock(&consumer_data
.lock
);
212 pthread_mutex_lock(&stream
->lock
);
214 /* Remove every reference of the stream in the consumer. */
215 consumer_stream_delete(stream
, ht
);
217 /* Close down everything including the relayd if one. */
218 consumer_stream_close(stream
);
220 /* Update refcount of channel and see if we need to destroy it. */
221 if (!uatomic_sub_return(&stream
->chan
->refcount
, 1)
222 && !uatomic_read(&stream
->chan
->nb_init_stream_left
)) {
223 free_chan
= stream
->chan
;
226 /* Indicates that the consumer data state MUST be updated after this. */
227 consumer_data
.need_update
= 1;
229 pthread_mutex_unlock(&stream
->lock
);
230 pthread_mutex_unlock(&consumer_data
.lock
);
233 consumer_del_channel(free_chan
);
236 /* Free stream within a RCU call. */
237 consumer_stream_free(stream
);