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.
21 #include <common/common.h>
22 #include <common/index/index.h>
24 #include "lttng-relayd.h"
25 #include "viewer-stream.h"
27 static void free_stream(struct relay_viewer_stream
*stream
)
31 free(stream
->path_name
);
32 free(stream
->channel_name
);
36 static void deferred_free_viewer_stream(struct rcu_head
*head
)
38 struct relay_viewer_stream
*stream
=
39 caa_container_of(head
, struct relay_viewer_stream
, rcu_node
);
44 struct relay_viewer_stream
*viewer_stream_create(struct relay_stream
*stream
,
45 enum lttng_viewer_seek seek_t
, struct ctf_trace
*ctf_trace
)
47 struct relay_viewer_stream
*vstream
;
52 vstream
= zmalloc(sizeof(*vstream
));
54 PERROR("relay viewer stream zmalloc");
58 vstream
->session_id
= stream
->session_id
;
59 vstream
->stream_handle
= stream
->stream_handle
;
60 vstream
->path_name
= strndup(stream
->path_name
, LTTNG_VIEWER_PATH_MAX
);
61 vstream
->channel_name
= strndup(stream
->channel_name
,
62 LTTNG_VIEWER_NAME_MAX
);
63 vstream
->tracefile_count
= stream
->tracefile_count
;
64 vstream
->metadata_flag
= stream
->metadata_flag
;
65 vstream
->tracefile_count_last
= -1ULL;
68 case LTTNG_VIEWER_SEEK_BEGINNING
:
69 vstream
->tracefile_count_current
= stream
->oldest_tracefile_id
;
71 case LTTNG_VIEWER_SEEK_LAST
:
72 vstream
->tracefile_count_current
= stream
->tracefile_count_current
;
79 if (vstream
->metadata_flag
) {
80 ctf_trace
->viewer_metadata_stream
= vstream
;
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
== LTTNG_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 ctf_trace
*ctf_trace
,
148 struct relay_viewer_stream
*stream
)
155 ctf_trace_put_ref(ctf_trace
);
158 if (stream
->read_fd
>= 0) {
159 ret
= close(stream
->read_fd
);
161 PERROR("close read_fd");
164 if (stream
->index_read_fd
>= 0) {
165 ret
= close(stream
->index_read_fd
);
167 PERROR("close index_read_fd");
171 call_rcu(&stream
->rcu_node
, deferred_free_viewer_stream
);
175 * Find viewer stream by id. RCU read side lock MUST be acquired.
177 * Return stream if found else NULL.
179 struct relay_viewer_stream
*viewer_stream_find_by_id(uint64_t id
)
181 struct lttng_ht_node_u64
*node
;
182 struct lttng_ht_iter iter
;
183 struct relay_viewer_stream
*stream
= NULL
;
185 lttng_ht_lookup(viewer_streams_ht
, &id
, &iter
);
186 node
= lttng_ht_iter_get_node_u64(&iter
);
188 DBG("Relay viewer stream %" PRIu64
" not found", id
);
191 stream
= caa_container_of(node
, struct relay_viewer_stream
, stream_n
);
198 * Rotate a stream to the next tracefile.
200 * Must be called with viewer_stream_rotation_lock held.
201 * Returns 0 on success, 1 on EOF, a negative value on error.
203 int viewer_stream_rotate(struct relay_viewer_stream
*vstream
,
204 struct relay_stream
*stream
)
207 uint64_t tracefile_id
;
212 if (vstream
->tracefile_count
== 0) {
213 /* Ignore rotation, there is none to do. */
218 tracefile_id
= (vstream
->tracefile_count_current
+ 1) %
219 vstream
->tracefile_count
;
221 /* Detect the last tracefile to open. */
222 if (vstream
->tracefile_count_last
!= -1ULL &&
223 vstream
->tracefile_count_last
==
224 vstream
->tracefile_count_current
) {
230 * The writer and the reader are not working in the same tracefile, we can
231 * read up to EOF, we don't care about the total_index_received.
233 if (stream
->close_flag
|| (stream
->tracefile_count_current
!= tracefile_id
)) {
234 vstream
->close_write_flag
= 1;
237 * We are opening a file that is still open in write, make sure we
238 * limit our reading to the number of indexes received.
240 vstream
->close_write_flag
= 0;
241 if (stream
->close_flag
) {
242 vstream
->total_index_received
= stream
->total_index_received
;
245 vstream
->tracefile_count_current
= tracefile_id
;
247 ret
= close(vstream
->index_read_fd
);
249 PERROR("close index file %d", vstream
->index_read_fd
);
251 vstream
->index_read_fd
= -1;
253 ret
= close(vstream
->read_fd
);
255 PERROR("close tracefile %d", vstream
->read_fd
);
257 vstream
->read_fd
= -1;
259 pthread_mutex_lock(&vstream
->overwrite_lock
);
260 vstream
->abort_flag
= 0;
261 pthread_mutex_unlock(&vstream
->overwrite_lock
);
263 ret
= index_open(vstream
->path_name
, vstream
->channel_name
,
264 vstream
->tracefile_count
, vstream
->tracefile_count_current
);
268 vstream
->index_read_fd
= ret
;