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
, struct ctf_trace
*ctf_trace
)
46 struct relay_viewer_stream
*vstream
;
51 vstream
= zmalloc(sizeof(*vstream
));
53 PERROR("relay viewer stream zmalloc");
57 vstream
->session_id
= stream
->session_id
;
58 vstream
->stream_handle
= stream
->stream_handle
;
59 vstream
->path_name
= strndup(stream
->path_name
, LTTNG_VIEWER_PATH_MAX
);
60 if (vstream
->path_name
== NULL
) {
61 PERROR("relay viewer path_name alloc");
64 vstream
->channel_name
= strndup(stream
->channel_name
,
65 LTTNG_VIEWER_NAME_MAX
);
66 if (vstream
->channel_name
== NULL
) {
67 PERROR("relay viewer channel_name alloc");
70 vstream
->tracefile_count
= stream
->tracefile_count
;
71 vstream
->metadata_flag
= stream
->metadata_flag
;
72 vstream
->tracefile_count_last
= -1ULL;
75 case LTTNG_VIEWER_SEEK_BEGINNING
:
76 vstream
->tracefile_count_current
= stream
->oldest_tracefile_id
;
78 case LTTNG_VIEWER_SEEK_LAST
:
79 vstream
->tracefile_count_current
= stream
->tracefile_count_current
;
86 if (vstream
->metadata_flag
) {
87 ctf_trace
->viewer_metadata_stream
= vstream
;
90 /* Globally visible after the add unique. */
91 lttng_ht_node_init_u64(&vstream
->stream_n
, stream
->stream_handle
);
92 lttng_ht_add_unique_u64(viewer_streams_ht
, &vstream
->stream_n
);
94 vstream
->index_read_fd
= -1;
95 vstream
->read_fd
= -1;
98 * This is to avoid a race between the initialization of this object and
99 * the close of the given stream. If the stream is unable to find this
100 * viewer stream when closing, this copy will at least take the latest
101 * value. We also need that for the seek_last.
103 vstream
->total_index_received
= stream
->total_index_received
;
106 * If we never received an index for the current stream, delay the opening
107 * of the index, otherwise open it right now.
109 if (vstream
->tracefile_count_current
== stream
->tracefile_count_current
110 && vstream
->total_index_received
== 0) {
111 vstream
->index_read_fd
= -1;
115 read_fd
= index_open(vstream
->path_name
, vstream
->channel_name
,
116 vstream
->tracefile_count
, vstream
->tracefile_count_current
);
120 vstream
->index_read_fd
= read_fd
;
123 if (seek_t
== LTTNG_VIEWER_SEEK_LAST
&& vstream
->index_read_fd
>= 0) {
126 lseek_ret
= lseek(vstream
->index_read_fd
,
127 vstream
->total_index_received
* sizeof(struct ctf_packet_index
),
132 vstream
->last_sent_index
= vstream
->total_index_received
;
139 free_stream(vstream
);
144 void viewer_stream_delete(struct relay_viewer_stream
*stream
)
147 struct lttng_ht_iter iter
;
149 iter
.iter
.node
= &stream
->stream_n
.node
;
150 ret
= lttng_ht_del(viewer_streams_ht
, &iter
);
154 void viewer_stream_destroy(struct ctf_trace
*ctf_trace
,
155 struct relay_viewer_stream
*stream
)
162 ctf_trace_put_ref(ctf_trace
);
165 if (stream
->read_fd
>= 0) {
166 ret
= close(stream
->read_fd
);
168 PERROR("close read_fd");
171 if (stream
->index_read_fd
>= 0) {
172 ret
= close(stream
->index_read_fd
);
174 PERROR("close index_read_fd");
178 call_rcu(&stream
->rcu_node
, deferred_free_viewer_stream
);
182 * Find viewer stream by id. RCU read side lock MUST be acquired.
184 * Return stream if found else NULL.
186 struct relay_viewer_stream
*viewer_stream_find_by_id(uint64_t id
)
188 struct lttng_ht_node_u64
*node
;
189 struct lttng_ht_iter iter
;
190 struct relay_viewer_stream
*stream
= NULL
;
192 lttng_ht_lookup(viewer_streams_ht
, &id
, &iter
);
193 node
= lttng_ht_iter_get_node_u64(&iter
);
195 DBG("Relay viewer stream %" PRIu64
" not found", id
);
198 stream
= caa_container_of(node
, struct relay_viewer_stream
, stream_n
);
205 * Rotate a stream to the next tracefile.
207 * Must be called with viewer_stream_rotation_lock held.
208 * Returns 0 on success, 1 on EOF, a negative value on error.
210 int viewer_stream_rotate(struct relay_viewer_stream
*vstream
,
211 struct relay_stream
*stream
)
214 uint64_t tracefile_id
;
219 if (vstream
->tracefile_count
== 0) {
220 /* Ignore rotation, there is none to do. */
225 tracefile_id
= (vstream
->tracefile_count_current
+ 1) %
226 vstream
->tracefile_count
;
228 /* Detect the last tracefile to open. */
229 if (vstream
->tracefile_count_last
!= -1ULL &&
230 vstream
->tracefile_count_last
==
231 vstream
->tracefile_count_current
) {
237 * The writer and the reader are not working in the same tracefile, we can
238 * read up to EOF, we don't care about the total_index_received.
240 if (stream
->close_flag
|| (stream
->tracefile_count_current
!= tracefile_id
)) {
241 vstream
->close_write_flag
= 1;
244 * We are opening a file that is still open in write, make sure we
245 * limit our reading to the number of indexes received.
247 vstream
->close_write_flag
= 0;
248 if (stream
->close_flag
) {
249 vstream
->total_index_received
= stream
->total_index_received
;
252 vstream
->tracefile_count_current
= tracefile_id
;
254 ret
= close(vstream
->index_read_fd
);
256 PERROR("close index file %d", vstream
->index_read_fd
);
258 vstream
->index_read_fd
= -1;
260 ret
= close(vstream
->read_fd
);
262 PERROR("close tracefile %d", vstream
->read_fd
);
264 vstream
->read_fd
= -1;
266 pthread_mutex_lock(&vstream
->overwrite_lock
);
267 vstream
->abort_flag
= 0;
268 pthread_mutex_unlock(&vstream
->overwrite_lock
);
270 ret
= index_open(vstream
->path_name
, vstream
->channel_name
,
271 vstream
->tracefile_count
, vstream
->tracefile_count_current
);
275 vstream
->index_read_fd
= ret
;