Port: Use LTTNG_NAME_MAX instead of NAME_MAX
[lttng-tools.git] / src / bin / lttng-relayd / stream.c
1 /*
2 * Copyright (C) 2013 - Julien Desfossez <jdesfossez@efficios.com>
3 * David Goulet <dgoulet@efficios.com>
4 * 2015 - Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
5 *
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.
9 *
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
13 * more details.
14 *
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.
18 */
19
20 #define _GNU_SOURCE
21 #define _LGPL_SOURCE
22 #include <common/common.h>
23 #include <common/utils.h>
24 #include <common/defaults.h>
25 #include <urcu/rculist.h>
26 #include <sys/stat.h>
27
28 #include "lttng-relayd.h"
29 #include "index.h"
30 #include "stream.h"
31 #include "viewer-stream.h"
32
33 /* Should be called with RCU read-side lock held. */
34 bool stream_get(struct relay_stream *stream)
35 {
36 bool has_ref = false;
37
38 pthread_mutex_lock(&stream->reflock);
39 if (stream->ref.refcount != 0) {
40 has_ref = true;
41 urcu_ref_get(&stream->ref);
42 }
43 pthread_mutex_unlock(&stream->reflock);
44
45 return has_ref;
46 }
47
48 /*
49 * Get stream from stream id from the streams hash table. Return stream
50 * if found else NULL. A stream reference is taken when a stream is
51 * returned. stream_put() must be called on that stream.
52 */
53 struct relay_stream *stream_get_by_id(uint64_t stream_id)
54 {
55 struct lttng_ht_node_u64 *node;
56 struct lttng_ht_iter iter;
57 struct relay_stream *stream = NULL;
58
59 rcu_read_lock();
60 lttng_ht_lookup(relay_streams_ht, &stream_id, &iter);
61 node = lttng_ht_iter_get_node_u64(&iter);
62 if (!node) {
63 DBG("Relay stream %" PRIu64 " not found", stream_id);
64 goto end;
65 }
66 stream = caa_container_of(node, struct relay_stream, node);
67 if (!stream_get(stream)) {
68 stream = NULL;
69 }
70 end:
71 rcu_read_unlock();
72 return stream;
73 }
74
75 /*
76 * We keep ownership of path_name and channel_name.
77 */
78 struct relay_stream *stream_create(struct ctf_trace *trace,
79 uint64_t stream_handle, char *path_name,
80 char *channel_name, uint64_t tracefile_size,
81 uint64_t tracefile_count)
82 {
83 int ret;
84 struct relay_stream *stream = NULL;
85 struct relay_session *session = trace->session;
86
87 stream = zmalloc(sizeof(struct relay_stream));
88 if (stream == NULL) {
89 PERROR("relay stream zmalloc");
90 ret = -1;
91 goto error_no_alloc;
92 }
93
94 stream->stream_handle = stream_handle;
95 stream->prev_seq = -1ULL;
96 stream->last_net_seq_num = -1ULL;
97 stream->ctf_stream_id = -1ULL;
98 stream->tracefile_size = tracefile_size;
99 stream->tracefile_count = tracefile_count;
100 stream->path_name = path_name;
101 stream->channel_name = channel_name;
102 lttng_ht_node_init_u64(&stream->node, stream->stream_handle);
103 pthread_mutex_init(&stream->lock, NULL);
104 pthread_mutex_init(&stream->reflock, NULL);
105 urcu_ref_init(&stream->ref);
106 ctf_trace_get(trace);
107 stream->trace = trace;
108
109 stream->indexes_ht = lttng_ht_new(0, LTTNG_HT_TYPE_U64);
110 if (!stream->indexes_ht) {
111 ERR("Cannot created indexes_ht");
112 ret = -1;
113 goto end;
114 }
115
116 ret = utils_mkdir_recursive(stream->path_name, S_IRWXU | S_IRWXG,
117 -1, -1);
118 if (ret < 0) {
119 ERR("relay creating output directory");
120 goto end;
121 }
122
123 /*
124 * No need to use run_as API here because whatever we receive,
125 * the relayd uses its own credentials for the stream files.
126 */
127 ret = utils_create_stream_file(stream->path_name, stream->channel_name,
128 stream->tracefile_size, 0, -1, -1, NULL);
129 if (ret < 0) {
130 ERR("Create output file");
131 goto end;
132 }
133 stream->stream_fd = stream_fd_create(ret);
134 if (!stream->stream_fd) {
135 if (close(ret)) {
136 PERROR("Error closing file %d", ret);
137 }
138 ret = -1;
139 goto end;
140 }
141 stream->tfa = tracefile_array_create(stream->tracefile_count);
142 if (!stream->tfa) {
143 ret = -1;
144 goto end;
145 }
146 if (stream->tracefile_size) {
147 DBG("Tracefile %s/%s_0 created", stream->path_name, stream->channel_name);
148 } else {
149 DBG("Tracefile %s/%s created", stream->path_name, stream->channel_name);
150 }
151
152 if (!strncmp(stream->channel_name, DEFAULT_METADATA_NAME, LTTNG_NAME_MAX)) {
153 stream->is_metadata = 1;
154 }
155
156 stream->in_recv_list = true;
157
158 /*
159 * Add the stream in the recv list of the session. Once the end stream
160 * message is received, all session streams are published.
161 */
162 pthread_mutex_lock(&session->recv_list_lock);
163 cds_list_add_rcu(&stream->recv_node, &session->recv_list);
164 session->stream_count++;
165 pthread_mutex_unlock(&session->recv_list_lock);
166
167 /*
168 * Both in the ctf_trace object and the global stream ht since the data
169 * side of the relayd does not have the concept of session.
170 */
171 lttng_ht_add_unique_u64(relay_streams_ht, &stream->node);
172 stream->in_stream_ht = true;
173
174 DBG("Relay new stream added %s with ID %" PRIu64, stream->channel_name,
175 stream->stream_handle);
176 ret = 0;
177
178 end:
179 if (ret) {
180 if (stream->stream_fd) {
181 stream_fd_put(stream->stream_fd);
182 stream->stream_fd = NULL;
183 }
184 stream_put(stream);
185 stream = NULL;
186 }
187 return stream;
188
189 error_no_alloc:
190 /*
191 * path_name and channel_name need to be freed explicitly here
192 * because we cannot rely on stream_put().
193 */
194 free(path_name);
195 free(channel_name);
196 return NULL;
197 }
198
199 /*
200 * Called with the session lock held.
201 */
202 void stream_publish(struct relay_stream *stream)
203 {
204 struct relay_session *session;
205
206 pthread_mutex_lock(&stream->lock);
207 if (stream->published) {
208 goto unlock;
209 }
210
211 session = stream->trace->session;
212
213 pthread_mutex_lock(&session->recv_list_lock);
214 if (stream->in_recv_list) {
215 cds_list_del_rcu(&stream->recv_node);
216 stream->in_recv_list = false;
217 }
218 pthread_mutex_unlock(&session->recv_list_lock);
219
220 pthread_mutex_lock(&stream->trace->stream_list_lock);
221 cds_list_add_rcu(&stream->stream_node, &stream->trace->stream_list);
222 pthread_mutex_unlock(&stream->trace->stream_list_lock);
223
224 stream->published = true;
225 unlock:
226 pthread_mutex_unlock(&stream->lock);
227 }
228
229 /*
230 * Stream must be protected by holding the stream lock or by virtue of being
231 * called from stream_destroy, in which case it is guaranteed to be accessed
232 * from a single thread by the reflock.
233 */
234 static void stream_unpublish(struct relay_stream *stream)
235 {
236 if (stream->in_stream_ht) {
237 struct lttng_ht_iter iter;
238 int ret;
239
240 iter.iter.node = &stream->node.node;
241 ret = lttng_ht_del(relay_streams_ht, &iter);
242 assert(!ret);
243 stream->in_stream_ht = false;
244 }
245 if (stream->published) {
246 pthread_mutex_lock(&stream->trace->stream_list_lock);
247 cds_list_del_rcu(&stream->stream_node);
248 pthread_mutex_unlock(&stream->trace->stream_list_lock);
249 stream->published = false;
250 }
251 }
252
253 static void stream_destroy(struct relay_stream *stream)
254 {
255 if (stream->indexes_ht) {
256 /*
257 * Calling lttng_ht_destroy in call_rcu worker thread so
258 * we don't hold the RCU read-side lock while calling
259 * it.
260 */
261 lttng_ht_destroy(stream->indexes_ht);
262 }
263 if (stream->tfa) {
264 tracefile_array_destroy(stream->tfa);
265 }
266 free(stream->path_name);
267 free(stream->channel_name);
268 free(stream);
269 }
270
271 static void stream_destroy_rcu(struct rcu_head *rcu_head)
272 {
273 struct relay_stream *stream =
274 caa_container_of(rcu_head, struct relay_stream, rcu_node);
275
276 stream_destroy(stream);
277 }
278
279 /*
280 * No need to take stream->lock since this is only called on the final
281 * stream_put which ensures that a single thread may act on the stream.
282 *
283 * At that point, the object is also protected by the reflock which
284 * guarantees that no other thread may share ownership of this stream.
285 */
286 static void stream_release(struct urcu_ref *ref)
287 {
288 struct relay_stream *stream =
289 caa_container_of(ref, struct relay_stream, ref);
290 struct relay_session *session;
291
292 session = stream->trace->session;
293
294 DBG("Releasing stream id %" PRIu64, stream->stream_handle);
295
296 pthread_mutex_lock(&session->recv_list_lock);
297 session->stream_count--;
298 if (stream->in_recv_list) {
299 cds_list_del_rcu(&stream->recv_node);
300 stream->in_recv_list = false;
301 }
302 pthread_mutex_unlock(&session->recv_list_lock);
303
304 stream_unpublish(stream);
305
306 if (stream->stream_fd) {
307 stream_fd_put(stream->stream_fd);
308 stream->stream_fd = NULL;
309 }
310 if (stream->index_fd) {
311 stream_fd_put(stream->index_fd);
312 stream->index_fd = NULL;
313 }
314 if (stream->trace) {
315 ctf_trace_put(stream->trace);
316 stream->trace = NULL;
317 }
318
319 call_rcu(&stream->rcu_node, stream_destroy_rcu);
320 }
321
322 void stream_put(struct relay_stream *stream)
323 {
324 DBG("stream put for stream id %" PRIu64, stream->stream_handle);
325 /*
326 * Ensure existence of stream->reflock for stream unlock.
327 */
328 rcu_read_lock();
329 /*
330 * Stream reflock ensures that concurrent test and update of
331 * stream ref is atomic.
332 */
333 pthread_mutex_lock(&stream->reflock);
334 assert(stream->ref.refcount != 0);
335 /*
336 * Wait until we have processed all the stream packets before
337 * actually putting our last stream reference.
338 */
339 DBG("stream put stream id %" PRIu64 " refcount %d",
340 stream->stream_handle,
341 (int) stream->ref.refcount);
342 urcu_ref_put(&stream->ref, stream_release);
343 pthread_mutex_unlock(&stream->reflock);
344 rcu_read_unlock();
345 }
346
347 void try_stream_close(struct relay_stream *stream)
348 {
349 DBG("Trying to close stream %" PRIu64, stream->stream_handle);
350 pthread_mutex_lock(&stream->lock);
351 /*
352 * Can be called concurently by connection close and reception of last
353 * pending data.
354 */
355 if (stream->closed) {
356 pthread_mutex_unlock(&stream->lock);
357 DBG("closing stream %" PRIu64 " aborted since it is already marked as closed", stream->stream_handle);
358 return;
359 }
360
361 stream->close_requested = true;
362
363 if (stream->last_net_seq_num == -1ULL) {
364 /*
365 * Handle connection close without explicit stream close
366 * command.
367 *
368 * We can be clever about indexes partially received in
369 * cases where we received the data socket part, but not
370 * the control socket part: since we're currently closing
371 * the stream on behalf of the control socket, we *know*
372 * there won't be any more control information for this
373 * socket. Therefore, we can destroy all indexes for
374 * which we have received only the file descriptor (from
375 * data socket). This takes care of consumerd crashes
376 * between sending the data and control information for
377 * a packet. Since those are sent in that order, we take
378 * care of consumerd crashes.
379 */
380 relay_index_close_partial_fd(stream);
381 /*
382 * Use the highest net_seq_num we currently have pending
383 * As end of stream indicator. Leave last_net_seq_num
384 * at -1ULL if we cannot find any index.
385 */
386 stream->last_net_seq_num = relay_index_find_last(stream);
387 /* Fall-through into the next check. */
388 }
389
390 if (stream->last_net_seq_num != -1ULL &&
391 ((int64_t) (stream->prev_seq - stream->last_net_seq_num)) < 0) {
392 /*
393 * Don't close since we still have data pending. This
394 * handles cases where an explicit close command has
395 * been received for this stream, and cases where the
396 * connection has been closed, and we are awaiting for
397 * index information from the data socket. It is
398 * therefore expected that all the index fd information
399 * we need has already been received on the control
400 * socket. Matching index information from data socket
401 * should be Expected Soon(TM).
402 *
403 * TODO: We should implement a timer to garbage collect
404 * streams after a timeout to be resilient against a
405 * consumerd implementation that would not match this
406 * expected behavior.
407 */
408 pthread_mutex_unlock(&stream->lock);
409 DBG("closing stream %" PRIu64 " aborted since it still has data pending", stream->stream_handle);
410 return;
411 }
412 /*
413 * We received all the indexes we can expect.
414 */
415 stream_unpublish(stream);
416 stream->closed = true;
417 /* Relay indexes are only used by the "consumer/sessiond" end. */
418 relay_index_close_all(stream);
419 pthread_mutex_unlock(&stream->lock);
420 DBG("Succeeded in closing stream %" PRIu64, stream->stream_handle);
421 stream_put(stream);
422 }
423
424 static void print_stream_indexes(struct relay_stream *stream)
425 {
426 struct lttng_ht_iter iter;
427 struct relay_index *index;
428
429 rcu_read_lock();
430 cds_lfht_for_each_entry(stream->indexes_ht->ht, &iter.iter, index,
431 index_n.node) {
432 DBG("index %p net_seq_num %" PRIu64 " refcount %ld"
433 " stream %" PRIu64 " trace %" PRIu64
434 " session %" PRIu64,
435 index,
436 index->index_n.key,
437 stream->ref.refcount,
438 index->stream->stream_handle,
439 index->stream->trace->id,
440 index->stream->trace->session->id);
441 }
442 rcu_read_unlock();
443 }
444
445 void print_relay_streams(void)
446 {
447 struct lttng_ht_iter iter;
448 struct relay_stream *stream;
449
450 rcu_read_lock();
451 cds_lfht_for_each_entry(relay_streams_ht->ht, &iter.iter, stream,
452 node.node) {
453 if (!stream_get(stream)) {
454 continue;
455 }
456 DBG("stream %p refcount %ld stream %" PRIu64 " trace %" PRIu64
457 " session %" PRIu64,
458 stream,
459 stream->ref.refcount,
460 stream->stream_handle,
461 stream->trace->id,
462 stream->trace->session->id);
463 print_stream_indexes(stream);
464 stream_put(stream);
465 }
466 rcu_read_unlock();
467 }
This page took 0.061139 seconds and 5 git commands to generate.