2 * Copyright (C) 2013 - Julien Desfossez <jdesfossez@efficios.com>
3 * David Goulet <dgoulet@efficios.com>
5 * This program is free software; you can redistribute it and/or modify it
6 * under the terms of the GNU General Public License, version 2 only, as
7 * published by the Free Software Foundation.
9 * This program is distributed in the hope that it will be useful, but WITHOUT
10 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
11 * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
14 * You should have received a copy of the GNU General Public License along with
15 * this program; if not, write to the Free Software Foundation, Inc., 51
16 * Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
20 #include <common/common.h>
21 #include <common/index/index.h>
23 #include "lttng-relayd.h"
24 #include "viewer-stream.h"
26 static void free_stream(struct relay_viewer_stream
*stream
)
30 free(stream
->path_name
);
31 free(stream
->channel_name
);
35 static void deferred_free_viewer_stream(struct rcu_head
*head
)
37 struct relay_viewer_stream
*stream
=
38 caa_container_of(head
, struct relay_viewer_stream
, rcu_node
);
43 struct relay_viewer_stream
*viewer_stream_create(struct relay_stream
*stream
,
44 enum lttng_viewer_seek seek_t
)
46 struct relay_viewer_stream
*vstream
;
50 vstream
= zmalloc(sizeof(*vstream
));
52 PERROR("relay viewer stream zmalloc");
56 vstream
->session_id
= stream
->session
->id
;
57 vstream
->stream_handle
= stream
->stream_handle
;
58 vstream
->path_name
= strndup(stream
->path_name
, LTTNG_VIEWER_PATH_MAX
);
59 vstream
->channel_name
= strndup(stream
->channel_name
,
60 LTTNG_VIEWER_NAME_MAX
);
61 vstream
->tracefile_count
= stream
->tracefile_count
;
62 vstream
->metadata_flag
= stream
->metadata_flag
;
63 vstream
->tracefile_count_last
= -1ULL;
66 case VIEWER_SEEK_BEGINNING
:
67 vstream
->tracefile_count_current
= stream
->oldest_tracefile_id
;
69 case VIEWER_SEEK_LAST
:
70 vstream
->tracefile_count_current
= stream
->tracefile_count_current
;
77 vstream
->ctf_trace
= stream
->ctf_trace
;
78 if (vstream
->metadata_flag
) {
79 vstream
->ctf_trace
->viewer_metadata_stream
= vstream
;
81 uatomic_inc(&vstream
->ctf_trace
->refcount
);
83 /* Globally visible after the add unique. */
84 lttng_ht_node_init_u64(&vstream
->stream_n
, stream
->stream_handle
);
85 lttng_ht_add_unique_u64(viewer_streams_ht
, &vstream
->stream_n
);
87 vstream
->index_read_fd
= -1;
88 vstream
->read_fd
= -1;
91 * This is to avoid a race between the initialization of this object and
92 * the close of the given stream. If the stream is unable to find this
93 * viewer stream when closing, this copy will at least take the latest
94 * value. We also need that for the seek_last.
96 vstream
->total_index_received
= stream
->total_index_received
;
99 * If we never received an index for the current stream, delay the opening
100 * of the index, otherwise open it right now.
102 if (vstream
->tracefile_count_current
== stream
->tracefile_count_current
103 && vstream
->total_index_received
== 0) {
104 vstream
->index_read_fd
= -1;
108 read_fd
= index_open(vstream
->path_name
, vstream
->channel_name
,
109 vstream
->tracefile_count
, vstream
->tracefile_count_current
);
113 vstream
->index_read_fd
= read_fd
;
116 if (seek_t
== VIEWER_SEEK_LAST
&& vstream
->index_read_fd
>= 0) {
119 lseek_ret
= lseek(vstream
->index_read_fd
,
120 vstream
->total_index_received
* sizeof(struct ctf_packet_index
),
125 vstream
->last_sent_index
= vstream
->total_index_received
;
132 free_stream(vstream
);
137 void viewer_stream_delete(struct relay_viewer_stream
*stream
)
140 struct lttng_ht_iter iter
;
142 iter
.iter
.node
= &stream
->stream_n
.node
;
143 ret
= lttng_ht_del(viewer_streams_ht
, &iter
);
147 void viewer_stream_destroy(struct relay_viewer_stream
*stream
)
150 unsigned long ret_ref
;
153 ret_ref
= uatomic_add_return(&stream
->ctf_trace
->refcount
, -1);
154 assert(ret_ref
>= 0);
156 if (stream
->read_fd
>= 0) {
157 ret
= close(stream
->read_fd
);
159 PERROR("close read_fd");
162 if (stream
->index_read_fd
>= 0) {
163 ret
= close(stream
->index_read_fd
);
165 PERROR("close index_read_fd");
170 * If the only stream left in the HT is the metadata stream,
171 * we need to remove it because we won't detect a EOF for this
174 if (ret_ref
== 1 && stream
->ctf_trace
->viewer_metadata_stream
) {
175 viewer_stream_delete(stream
->ctf_trace
->viewer_metadata_stream
);
176 viewer_stream_destroy(stream
->ctf_trace
->viewer_metadata_stream
);
177 stream
->ctf_trace
->metadata_stream
= NULL
;
178 DBG("Freeing ctf_trace %" PRIu64
, stream
->ctf_trace
->id
);
180 * The streaming-side is already closed and we can't receive a new
181 * stream concurrently at this point (since the session is being
182 * destroyed), so when we detect the refcount equals 0, we are the
183 * only owners of the ctf_trace and we can free it ourself.
185 free(stream
->ctf_trace
);
188 call_rcu(&stream
->rcu_node
, deferred_free_viewer_stream
);
192 * Find viewer stream by id. RCU read side lock MUST be acquired.
194 * Return stream if found else NULL.
196 struct relay_viewer_stream
*viewer_stream_find_by_id(uint64_t id
)
198 struct lttng_ht_node_u64
*node
;
199 struct lttng_ht_iter iter
;
200 struct relay_viewer_stream
*stream
= NULL
;
202 lttng_ht_lookup(viewer_streams_ht
, &id
, &iter
);
203 node
= lttng_ht_iter_get_node_u64(&iter
);
205 DBG("Relay viewer stream %" PRIu64
" not found", id
);
208 stream
= caa_container_of(node
, struct relay_viewer_stream
, stream_n
);
215 * Rotate a stream to the next tracefile.
217 * Returns 0 on success, 1 on EOF, a negative value on error.
219 int viewer_stream_rotate(struct relay_viewer_stream
*vstream
,
220 struct relay_stream
*stream
)
223 uint64_t tracefile_id
;
227 tracefile_id
= (vstream
->tracefile_count_current
+ 1) %
228 vstream
->tracefile_count
;
230 /* Detect the last tracefile to open. */
231 if (vstream
->tracefile_count_last
!= -1ULL &&
232 vstream
->tracefile_count_last
==
233 vstream
->tracefile_count_current
) {
239 * If the stream on the streaming side still exists, lock to execute
240 * rotation in order to avoid races between a modification on the index
244 pthread_mutex_lock(&stream
->viewer_stream_rotation_lock
);
248 * The writer and the reader are not working in the same tracefile, we can
249 * read up to EOF, we don't care about the total_index_received.
251 if (!stream
|| (stream
->tracefile_count_current
!= tracefile_id
)) {
252 vstream
->close_write_flag
= 1;
255 * We are opening a file that is still open in write, make sure we
256 * limit our reading to the number of indexes received.
258 vstream
->close_write_flag
= 0;
260 vstream
->total_index_received
= stream
->total_index_received
;
263 vstream
->tracefile_count_current
= tracefile_id
;
265 ret
= close(vstream
->index_read_fd
);
267 PERROR("close index file %d", vstream
->index_read_fd
);
269 vstream
->index_read_fd
= -1;
271 ret
= close(vstream
->read_fd
);
273 PERROR("close tracefile %d", vstream
->read_fd
);
275 vstream
->read_fd
= -1;
277 pthread_mutex_lock(&vstream
->overwrite_lock
);
278 vstream
->abort_flag
= 0;
279 pthread_mutex_unlock(&vstream
->overwrite_lock
);
282 pthread_mutex_unlock(&stream
->viewer_stream_rotation_lock
);
285 ret
= index_open(vstream
->path_name
, vstream
->channel_name
,
286 vstream
->tracefile_count
, vstream
->tracefile_count_current
);
290 vstream
->index_read_fd
= ret
;