Remove error count property of lttng_condition_on_event
[lttng-tools.git] / src / bin / lttng-relayd / stream.c
CommitLineData
2a174661 1/*
ab5be9fa
MJ
2 * Copyright (C) 2013 Julien Desfossez <jdesfossez@efficios.com>
3 * Copyright (C) 2013 David Goulet <dgoulet@efficios.com>
4 * Copyright (C) 2015 Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
5 * Copyright (C) 2019 Jérémie Galarneau <jeremie.galarneau@efficios.com>
2a174661 6 *
ab5be9fa 7 * SPDX-License-Identifier: GPL-2.0-only
2a174661 8 *
2a174661
DG
9 */
10
6c1c0768 11#define _LGPL_SOURCE
2a174661 12#include <common/common.h>
7591bab1 13#include <common/defaults.h>
8bb66c3c 14#include <common/fs-handle.h>
c35f9726 15#include <common/sessiond-comm/relayd.h>
8bb66c3c 16#include <common/utils.h>
7591bab1 17#include <sys/stat.h>
8bb66c3c 18#include <urcu/rculist.h>
2a174661 19
7591bab1 20#include "lttng-relayd.h"
2a174661
DG
21#include "index.h"
22#include "stream.h"
23#include "viewer-stream.h"
24
348a81dc
JG
25#include <sys/types.h>
26#include <fcntl.h>
27
c35f9726
JG
28#define FILE_IO_STACK_BUFFER_SIZE 65536
29
7591bab1
MD
30/* Should be called with RCU read-side lock held. */
31bool stream_get(struct relay_stream *stream)
32{
ce4d4083 33 return urcu_ref_get_unless_zero(&stream->ref);
7591bab1
MD
34}
35
2a174661 36/*
7591bab1
MD
37 * Get stream from stream id from the streams hash table. Return stream
38 * if found else NULL. A stream reference is taken when a stream is
39 * returned. stream_put() must be called on that stream.
2a174661 40 */
7591bab1 41struct relay_stream *stream_get_by_id(uint64_t stream_id)
2a174661
DG
42{
43 struct lttng_ht_node_u64 *node;
44 struct lttng_ht_iter iter;
45 struct relay_stream *stream = NULL;
46
7591bab1
MD
47 rcu_read_lock();
48 lttng_ht_lookup(relay_streams_ht, &stream_id, &iter);
2a174661 49 node = lttng_ht_iter_get_node_u64(&iter);
7591bab1 50 if (!node) {
2a174661
DG
51 DBG("Relay stream %" PRIu64 " not found", stream_id);
52 goto end;
53 }
54 stream = caa_container_of(node, struct relay_stream, node);
7591bab1
MD
55 if (!stream_get(stream)) {
56 stream = NULL;
57 }
2a174661 58end:
7591bab1 59 rcu_read_unlock();
2a174661
DG
60 return stream;
61}
62
c35f9726
JG
63static void stream_complete_rotation(struct relay_stream *stream)
64{
65 DBG("Rotation completed for stream %" PRIu64, stream->stream_handle);
f607fc46
MD
66 if (stream->ongoing_rotation.value.next_trace_chunk) {
67 tracefile_array_reset(stream->tfa);
68 tracefile_array_commit_seq(stream->tfa,
69 stream->index_received_seqcount);
70 }
c35f9726
JG
71 lttng_trace_chunk_put(stream->trace_chunk);
72 stream->trace_chunk = stream->ongoing_rotation.value.next_trace_chunk;
73 stream->ongoing_rotation = (typeof(stream->ongoing_rotation)) {};
80516611 74 stream->completed_rotation_count++;
c35f9726
JG
75}
76
c35f9726
JG
77static int stream_create_data_output_file_from_trace_chunk(
78 struct relay_stream *stream,
79 struct lttng_trace_chunk *trace_chunk,
80 bool force_unlink,
8bb66c3c 81 struct fs_handle **out_file)
348a81dc 82{
8bb66c3c 83 int ret;
c35f9726 84 char stream_path[LTTNG_PATH_MAX];
348a81dc
JG
85 enum lttng_trace_chunk_status status;
86 const int flags = O_RDWR | O_CREAT | O_TRUNC;
87 const mode_t mode = S_IRUSR | S_IWUSR | S_IRGRP | S_IWGRP;
348a81dc
JG
88
89 ASSERT_LOCKED(stream->lock);
348a81dc 90
348a81dc 91 ret = utils_stream_file_path(stream->path_name, stream->channel_name,
c35f9726
JG
92 stream->tracefile_size, stream->tracefile_current_index,
93 NULL, stream_path, sizeof(stream_path));
348a81dc
JG
94 if (ret < 0) {
95 goto end;
96 }
97
c35f9726
JG
98 if (stream->tracefile_wrapped_around || force_unlink) {
99 /*
100 * The on-disk ring-buffer has wrapped around.
101 * Newly created stream files will replace existing files. Since
102 * live clients may be consuming existing files, the file about
103 * to be replaced is unlinked in order to not overwrite its
104 * content.
105 */
106 status = lttng_trace_chunk_unlink_file(trace_chunk,
107 stream_path);
108 if (status != LTTNG_TRACE_CHUNK_STATUS_OK) {
109 PERROR("Failed to unlink stream file \"%s\" during trace file rotation",
110 stream_path);
111 /*
112 * Don't abort if the file doesn't exist, it is
113 * unexpected, but should not be a fatal error.
114 */
115 if (errno != ENOENT) {
116 ret = -1;
117 goto end;
118 }
119 }
120 }
121
8bb66c3c
JG
122 status = lttng_trace_chunk_open_fs_handle(trace_chunk, stream_path,
123 flags, mode, out_file, false);
348a81dc
JG
124 if (status != LTTNG_TRACE_CHUNK_STATUS_OK) {
125 ERR("Failed to open stream file \"%s\"", stream->channel_name);
126 ret = -1;
127 goto end;
128 }
348a81dc
JG
129end:
130 return ret;
131}
132
c35f9726
JG
133static int stream_rotate_data_file(struct relay_stream *stream)
134{
135 int ret = 0;
136
68c40154
MD
137 DBG("Rotating stream %" PRIu64 " data file with size %" PRIu64,
138 stream->stream_handle, stream->tracefile_size_current);
c35f9726 139
8bb66c3c
JG
140 if (stream->file) {
141 fs_handle_close(stream->file);
142 stream->file = NULL;
c35f9726
JG
143 }
144
145 stream->tracefile_wrapped_around = false;
146 stream->tracefile_current_index = 0;
147
148 if (stream->ongoing_rotation.value.next_trace_chunk) {
c35f9726
JG
149 enum lttng_trace_chunk_status chunk_status;
150
151 chunk_status = lttng_trace_chunk_create_subdirectory(
152 stream->ongoing_rotation.value.next_trace_chunk,
153 stream->path_name);
154 if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
155 ret = -1;
156 goto end;
157 }
158
159 /* Rotate the data file. */
160 ret = stream_create_data_output_file_from_trace_chunk(stream,
161 stream->ongoing_rotation.value.next_trace_chunk,
8bb66c3c 162 false, &stream->file);
c35f9726
JG
163 if (ret < 0) {
164 ERR("Failed to rotate stream data file");
165 goto end;
166 }
167 }
68c40154
MD
168 DBG("%s: reset tracefile_size_current for stream %" PRIu64 " was %" PRIu64,
169 __func__, stream->stream_handle, stream->tracefile_size_current);
c35f9726
JG
170 stream->tracefile_size_current = 0;
171 stream->pos_after_last_complete_data_index = 0;
172 stream->ongoing_rotation.value.data_rotated = true;
173
174 if (stream->ongoing_rotation.value.index_rotated) {
175 /* Rotation completed; reset its state. */
176 stream_complete_rotation(stream);
177 }
178end:
179 return ret;
180}
181
f2aea36d
JG
182/*
183 * If too much data has been written in a tracefile before we received the
184 * rotation command, we have to move the excess data to the new tracefile and
185 * perform the rotation. This can happen because the control and data
186 * connections are separate, the indexes as well as the commands arrive from
187 * the control connection and we have no control over the order so we could be
188 * in a situation where too much data has been received on the data connection
189 * before the rotation command on the control connection arrives.
190 */
191static int rotate_truncate_stream(struct relay_stream *stream)
192{
193 int ret;
194 off_t lseek_ret, previous_stream_copy_origin;
195 uint64_t copy_bytes_left, misplaced_data_size;
196 bool acquired_reference;
8bb66c3c 197 struct fs_handle *previous_stream_file = NULL;
f2aea36d
JG
198 struct lttng_trace_chunk *previous_chunk = NULL;
199
f3fe2a92 200 if (!LTTNG_OPTIONAL_GET(stream->ongoing_rotation).next_trace_chunk) {
4147107d
JG
201 ERR("Protocol error encoutered in %s(): stream rotation "
202 "sequence number is before the current sequence number "
203 "and the next trace chunk is unset. Honoring this "
204 "rotation command would result in data loss",
205 __FUNCTION__);
206 ret = -1;
207 goto end;
208 }
209
f2aea36d
JG
210 ASSERT_LOCKED(stream->lock);
211 /*
212 * Acquire a reference to the current trace chunk to ensure
213 * it is not reclaimed when `stream_rotate_data_file` is called.
214 * Failing to do so would violate the contract of the trace
215 * chunk API as an active file descriptor would outlive the
216 * trace chunk.
217 */
218 acquired_reference = lttng_trace_chunk_get(stream->trace_chunk);
219 assert(acquired_reference);
220 previous_chunk = stream->trace_chunk;
221
222 /*
223 * Steal the stream's reference to its stream_fd. A new
224 * stream_fd will be created when the rotation completes and
225 * the orinal stream_fd will be used to copy the "extra" data
226 * to the new file.
227 */
8bb66c3c
JG
228 assert(stream->file);
229 previous_stream_file = stream->file;
230 stream->file = NULL;
f2aea36d
JG
231
232 assert(!stream->is_metadata);
233 assert(stream->tracefile_size_current >
234 stream->pos_after_last_complete_data_index);
235 misplaced_data_size = stream->tracefile_size_current -
236 stream->pos_after_last_complete_data_index;
237 copy_bytes_left = misplaced_data_size;
238 previous_stream_copy_origin = stream->pos_after_last_complete_data_index;
239
240 ret = stream_rotate_data_file(stream);
241 if (ret) {
242 goto end;
243 }
244
8bb66c3c 245 assert(stream->file);
f2aea36d
JG
246 /*
247 * Seek the current tracefile to the position at which the rotation
248 * should have occurred.
249 */
8bb66c3c 250 lseek_ret = fs_handle_seek(previous_stream_file, previous_stream_copy_origin, SEEK_SET);
f2aea36d
JG
251 if (lseek_ret < 0) {
252 PERROR("Failed to seek to offset %" PRIu64
253 " while copying extra data received before a stream rotation",
254 (uint64_t) previous_stream_copy_origin);
255 ret = -1;
256 goto end;
257 }
258
259 /* Move data from the old file to the new file. */
260 while (copy_bytes_left) {
261 ssize_t io_ret;
262 char copy_buffer[FILE_IO_STACK_BUFFER_SIZE];
263 const off_t copy_size_this_pass = min_t(
264 off_t, copy_bytes_left, sizeof(copy_buffer));
265
8bb66c3c 266 io_ret = fs_handle_read(previous_stream_file, copy_buffer,
f2aea36d
JG
267 copy_size_this_pass);
268 if (io_ret < (ssize_t) copy_size_this_pass) {
269 if (io_ret == -1) {
270 PERROR("Failed to read %" PRIu64
8bb66c3c 271 " bytes from previous stream file in %s(), returned %zi: stream id = %" PRIu64,
f2aea36d 272 copy_size_this_pass,
8bb66c3c
JG
273 __FUNCTION__, io_ret,
274 stream->stream_handle);
f2aea36d
JG
275 } else {
276 ERR("Failed to read %" PRIu64
8bb66c3c 277 " bytes from previous stream file in %s(), returned %zi: stream id = %" PRIu64,
f2aea36d 278 copy_size_this_pass,
8bb66c3c
JG
279 __FUNCTION__, io_ret,
280 stream->stream_handle);
f2aea36d
JG
281 }
282 ret = -1;
283 goto end;
284 }
285
8bb66c3c
JG
286 io_ret = fs_handle_write(
287 stream->file, copy_buffer, copy_size_this_pass);
f2aea36d
JG
288 if (io_ret < (ssize_t) copy_size_this_pass) {
289 if (io_ret == -1) {
290 PERROR("Failed to write %" PRIu64
8bb66c3c 291 " bytes from previous stream file in %s(), returned %zi: stream id = %" PRIu64,
f2aea36d 292 copy_size_this_pass,
8bb66c3c
JG
293 __FUNCTION__, io_ret,
294 stream->stream_handle);
f2aea36d
JG
295 } else {
296 ERR("Failed to write %" PRIu64
8bb66c3c 297 " bytes from previous stream file in %s(), returned %zi: stream id = %" PRIu64,
f2aea36d 298 copy_size_this_pass,
8bb66c3c
JG
299 __FUNCTION__, io_ret,
300 stream->stream_handle);
f2aea36d
JG
301 }
302 ret = -1;
303 goto end;
304 }
305 copy_bytes_left -= copy_size_this_pass;
306 }
307
308 /* Truncate the file to get rid of the excess data. */
8bb66c3c
JG
309 ret = fs_handle_truncate(
310 previous_stream_file, previous_stream_copy_origin);
f2aea36d
JG
311 if (ret) {
312 PERROR("Failed to truncate current stream file to offset %" PRIu64,
313 previous_stream_copy_origin);
314 goto end;
315 }
316
317 /*
318 * Update the offset and FD of all the eventual indexes created by the
319 * data connection before the rotation command arrived.
320 */
321 ret = relay_index_switch_all_files(stream);
322 if (ret < 0) {
323 ERR("Failed to rotate index file");
324 goto end;
325 }
326
327 stream->tracefile_size_current = misplaced_data_size;
328 /* Index and data contents are back in sync. */
329 stream->pos_after_last_complete_data_index = 0;
330 ret = 0;
331end:
332 lttng_trace_chunk_put(previous_chunk);
f2aea36d
JG
333 return ret;
334}
335
c35f9726
JG
336/*
337 * Check if a stream's data file (as opposed to index) should be rotated
338 * (for session rotation).
339 * Must be called with the stream lock held.
340 *
341 * Return 0 on success, a negative value on error.
342 */
343static int try_rotate_stream_data(struct relay_stream *stream)
344{
345 int ret = 0;
346
347 if (caa_likely(!stream->ongoing_rotation.is_set)) {
348 /* No rotation expected. */
349 goto end;
350 }
351
352 if (stream->ongoing_rotation.value.data_rotated) {
353 /* Rotation of the data file has already occurred. */
354 goto end;
355 }
356
68c40154
MD
357 DBG("%s: Stream %" PRIu64
358 " (rotate_at_index_packet_seq_num = %" PRIu64
359 ", rotate_at_prev_data_net_seq = %" PRIu64
360 ", prev_data_seq = %" PRIu64 ")",
361 __func__, stream->stream_handle,
362 stream->ongoing_rotation.value.packet_seq_num,
363 stream->ongoing_rotation.value.prev_data_net_seq,
364 stream->prev_data_seq);
365
c35f9726 366 if (stream->prev_data_seq == -1ULL ||
0f83d1cc
MD
367 stream->ongoing_rotation.value.prev_data_net_seq == -1ULL ||
368 stream->prev_data_seq <
369 stream->ongoing_rotation.value.prev_data_net_seq) {
c35f9726
JG
370 /*
371 * The next packet that will be written is not part of the next
372 * chunk yet.
373 */
0f83d1cc
MD
374 DBG("Stream %" PRIu64 " data not yet ready for rotation "
375 "(rotate_at_index_packet_seq_num = %" PRIu64
376 ", rotate_at_prev_data_net_seq = %" PRIu64
c35f9726
JG
377 ", prev_data_seq = %" PRIu64 ")",
378 stream->stream_handle,
0f83d1cc
MD
379 stream->ongoing_rotation.value.packet_seq_num,
380 stream->ongoing_rotation.value.prev_data_net_seq,
c35f9726
JG
381 stream->prev_data_seq);
382 goto end;
0f83d1cc 383 } else if (stream->prev_data_seq > stream->ongoing_rotation.value.prev_data_net_seq) {
c35f9726
JG
384 /*
385 * prev_data_seq is checked here since indexes and rotation
386 * commands are serialized with respect to each other.
387 */
388 DBG("Rotation after too much data has been written in tracefile "
389 "for stream %" PRIu64 ", need to truncate before "
390 "rotating", stream->stream_handle);
391 ret = rotate_truncate_stream(stream);
392 if (ret) {
393 ERR("Failed to truncate stream");
394 goto end;
395 }
396 } else {
397 ret = stream_rotate_data_file(stream);
398 }
399
400end:
401 return ret;
402}
403
404/*
405 * Close the current index file if it is open, and create a new one.
406 *
407 * Return 0 on success, -1 on error.
408 */
409static int create_index_file(struct relay_stream *stream,
410 struct lttng_trace_chunk *chunk)
411{
412 int ret;
413 uint32_t major, minor;
414 char *index_subpath = NULL;
3a735fa7 415 enum lttng_trace_chunk_status status;
c35f9726
JG
416
417 ASSERT_LOCKED(stream->lock);
418
419 /* Put ref on previous index_file. */
420 if (stream->index_file) {
421 lttng_index_file_put(stream->index_file);
422 stream->index_file = NULL;
423 }
424 major = stream->trace->session->major;
425 minor = stream->trace->session->minor;
426
427 if (!chunk) {
428 ret = 0;
429 goto end;
430 }
431 ret = asprintf(&index_subpath, "%s/%s", stream->path_name,
432 DEFAULT_INDEX_DIR);
433 if (ret < 0) {
434 goto end;
435 }
436
3a735fa7 437 status = lttng_trace_chunk_create_subdirectory(chunk,
c35f9726
JG
438 index_subpath);
439 free(index_subpath);
3a735fa7
MD
440 if (status != LTTNG_TRACE_CHUNK_STATUS_OK) {
441 ret = -1;
c35f9726
JG
442 goto end;
443 }
3ff5c5db 444 status = lttng_index_file_create_from_trace_chunk(
c35f9726
JG
445 chunk, stream->path_name,
446 stream->channel_name, stream->tracefile_size,
447 stream->tracefile_current_index,
448 lttng_to_index_major(major, minor),
3ff5c5db
MD
449 lttng_to_index_minor(major, minor), true,
450 &stream->index_file);
451 if (status != LTTNG_TRACE_CHUNK_STATUS_OK) {
c35f9726
JG
452 ret = -1;
453 goto end;
454 }
455
456 ret = 0;
457
458end:
459 return ret;
460}
461
462/*
463 * Check if a stream's index file should be rotated (for session rotation).
464 * Must be called with the stream lock held.
465 *
466 * Return 0 on success, a negative value on error.
467 */
468static int try_rotate_stream_index(struct relay_stream *stream)
469{
470 int ret = 0;
471
472 if (!stream->ongoing_rotation.is_set) {
473 /* No rotation expected. */
474 goto end;
475 }
476
477 if (stream->ongoing_rotation.value.index_rotated) {
478 /* Rotation of the index has already occurred. */
479 goto end;
480 }
481
68c40154
MD
482 DBG("%s: Stream %" PRIu64
483 " (rotate_at_packet_seq_num = %" PRIu64
484 ", received_packet_seq_num = "
485 "(value = %" PRIu64 ", is_set = %" PRIu8 "))",
486 __func__, stream->stream_handle,
487 stream->ongoing_rotation.value.packet_seq_num,
488 stream->received_packet_seq_num.value,
489 stream->received_packet_seq_num.is_set);
490
0f83d1cc
MD
491 if (!stream->received_packet_seq_num.is_set ||
492 LTTNG_OPTIONAL_GET(stream->received_packet_seq_num) + 1 <
493 stream->ongoing_rotation.value.packet_seq_num) {
494 DBG("Stream %" PRIu64 " index not yet ready for rotation "
495 "(rotate_at_packet_seq_num = %" PRIu64
496 ", received_packet_seq_num = "
497 "(value = %" PRIu64 ", is_set = %" PRIu8 "))",
c35f9726 498 stream->stream_handle,
0f83d1cc
MD
499 stream->ongoing_rotation.value.packet_seq_num,
500 stream->received_packet_seq_num.value,
501 stream->received_packet_seq_num.is_set);
c35f9726
JG
502 goto end;
503 } else {
0f83d1cc
MD
504 /*
505 * The next index belongs to the new trace chunk; rotate.
506 * In overwrite mode, the packet seq num may jump over the
507 * rotation position.
508 */
509 assert(LTTNG_OPTIONAL_GET(stream->received_packet_seq_num) + 1 >=
510 stream->ongoing_rotation.value.packet_seq_num);
c35f9726
JG
511 DBG("Rotating stream %" PRIu64 " index file",
512 stream->stream_handle);
f607fc46
MD
513 if (stream->index_file) {
514 lttng_index_file_put(stream->index_file);
515 stream->index_file = NULL;
516 }
c35f9726
JG
517 stream->ongoing_rotation.value.index_rotated = true;
518
0f83d1cc
MD
519 /*
520 * Set the rotation pivot position for the data, now that we have the
521 * net_seq_num matching the packet_seq_num index pivot position.
522 */
523 stream->ongoing_rotation.value.prev_data_net_seq =
524 stream->prev_index_seq;
c35f9726
JG
525 if (stream->ongoing_rotation.value.data_rotated &&
526 stream->ongoing_rotation.value.index_rotated) {
527 /* Rotation completed; reset its state. */
528 DBG("Rotation completed for stream %" PRIu64,
529 stream->stream_handle);
530 stream_complete_rotation(stream);
531 }
532 }
533
534end:
535 return ret;
536}
537
348a81dc
JG
538static int stream_set_trace_chunk(struct relay_stream *stream,
539 struct lttng_trace_chunk *chunk)
540{
541 int ret = 0;
542 enum lttng_trace_chunk_status status;
543 bool acquired_reference;
544
348a81dc
JG
545 status = lttng_trace_chunk_create_subdirectory(chunk,
546 stream->path_name);
547 if (status != LTTNG_TRACE_CHUNK_STATUS_OK) {
548 ret = -1;
549 goto end;
550 }
551
552 lttng_trace_chunk_put(stream->trace_chunk);
553 acquired_reference = lttng_trace_chunk_get(chunk);
554 assert(acquired_reference);
555 stream->trace_chunk = chunk;
c35f9726 556
8bb66c3c
JG
557 if (stream->file) {
558 fs_handle_close(stream->file);
559 stream->file = NULL;
c35f9726
JG
560 }
561 ret = stream_create_data_output_file_from_trace_chunk(stream, chunk,
8bb66c3c 562 false, &stream->file);
348a81dc 563end:
348a81dc
JG
564 return ret;
565}
566
2a174661 567/*
7591bab1 568 * We keep ownership of path_name and channel_name.
2a174661 569 */
7591bab1
MD
570struct relay_stream *stream_create(struct ctf_trace *trace,
571 uint64_t stream_handle, char *path_name,
572 char *channel_name, uint64_t tracefile_size,
348a81dc 573 uint64_t tracefile_count)
2a174661 574{
7591bab1
MD
575 int ret;
576 struct relay_stream *stream = NULL;
577 struct relay_session *session = trace->session;
348a81dc
JG
578 bool acquired_reference = false;
579 struct lttng_trace_chunk *current_trace_chunk;
2a174661 580
7591bab1
MD
581 stream = zmalloc(sizeof(struct relay_stream));
582 if (stream == NULL) {
583 PERROR("relay stream zmalloc");
7591bab1
MD
584 goto error_no_alloc;
585 }
2a174661 586
7591bab1 587 stream->stream_handle = stream_handle;
a8f9f353 588 stream->prev_data_seq = -1ULL;
7a45c7e6 589 stream->prev_index_seq = -1ULL;
bda7c7b9 590 stream->last_net_seq_num = -1ULL;
7591bab1
MD
591 stream->ctf_stream_id = -1ULL;
592 stream->tracefile_size = tracefile_size;
593 stream->tracefile_count = tracefile_count;
594 stream->path_name = path_name;
595 stream->channel_name = channel_name;
2f9c3030 596 stream->beacon_ts_end = -1ULL;
7591bab1
MD
597 lttng_ht_node_init_u64(&stream->node, stream->stream_handle);
598 pthread_mutex_init(&stream->lock, NULL);
7591bab1
MD
599 urcu_ref_init(&stream->ref);
600 ctf_trace_get(trace);
601 stream->trace = trace;
2a174661 602
348a81dc
JG
603 pthread_mutex_lock(&trace->session->lock);
604 current_trace_chunk = trace->session->current_trace_chunk;
605 if (current_trace_chunk) {
606 acquired_reference = lttng_trace_chunk_get(current_trace_chunk);
607 }
608 pthread_mutex_unlock(&trace->session->lock);
609 if (!acquired_reference) {
610 ERR("Cannot create stream for channel \"%s\" as a reference to the session's current trace chunk could not be acquired",
611 channel_name);
7591bab1
MD
612 ret = -1;
613 goto end;
2a174661
DG
614 }
615
348a81dc
JG
616 stream->indexes_ht = lttng_ht_new(0, LTTNG_HT_TYPE_U64);
617 if (!stream->indexes_ht) {
618 ERR("Cannot created indexes_ht");
619 ret = -1;
7591bab1
MD
620 goto end;
621 }
2a174661 622
c35f9726 623 pthread_mutex_lock(&stream->lock);
348a81dc 624 ret = stream_set_trace_chunk(stream, current_trace_chunk);
c35f9726 625 pthread_mutex_unlock(&stream->lock);
348a81dc
JG
626 if (ret) {
627 ERR("Failed to set the current trace chunk of session \"%s\" on newly created stream of channel \"%s\"",
628 trace->session->session_name,
629 stream->channel_name);
7591bab1
MD
630 ret = -1;
631 goto end;
2a174661 632 }
a44ca2ca
MD
633 stream->tfa = tracefile_array_create(stream->tracefile_count);
634 if (!stream->tfa) {
635 ret = -1;
636 goto end;
637 }
7591bab1 638
348a81dc
JG
639 stream->is_metadata = !strcmp(stream->channel_name,
640 DEFAULT_METADATA_NAME);
7591bab1
MD
641 stream->in_recv_list = true;
642
643 /*
644 * Add the stream in the recv list of the session. Once the end stream
645 * message is received, all session streams are published.
646 */
647 pthread_mutex_lock(&session->recv_list_lock);
648 cds_list_add_rcu(&stream->recv_node, &session->recv_list);
649 session->stream_count++;
650 pthread_mutex_unlock(&session->recv_list_lock);
651
652 /*
653 * Both in the ctf_trace object and the global stream ht since the data
654 * side of the relayd does not have the concept of session.
655 */
656 lttng_ht_add_unique_u64(relay_streams_ht, &stream->node);
77f7bd85 657 stream->in_stream_ht = true;
2a174661 658
7591bab1
MD
659 DBG("Relay new stream added %s with ID %" PRIu64, stream->channel_name,
660 stream->stream_handle);
661 ret = 0;
662
663end:
664 if (ret) {
8bb66c3c
JG
665 if (stream->file) {
666 fs_handle_close(stream->file);
667 stream->file = NULL;
2a174661 668 }
7591bab1
MD
669 stream_put(stream);
670 stream = NULL;
2a174661 671 }
317eadef
JG
672 if (acquired_reference) {
673 lttng_trace_chunk_put(current_trace_chunk);
674 }
7591bab1 675 return stream;
2a174661 676
7591bab1
MD
677error_no_alloc:
678 /*
679 * path_name and channel_name need to be freed explicitly here
680 * because we cannot rely on stream_put().
681 */
682 free(path_name);
683 free(channel_name);
684 return NULL;
685}
686
687/*
688 * Called with the session lock held.
689 */
690void stream_publish(struct relay_stream *stream)
691{
692 struct relay_session *session;
693
694 pthread_mutex_lock(&stream->lock);
695 if (stream->published) {
696 goto unlock;
2a174661
DG
697 }
698
7591bab1 699 session = stream->trace->session;
2a174661 700
7591bab1
MD
701 pthread_mutex_lock(&session->recv_list_lock);
702 if (stream->in_recv_list) {
703 cds_list_del_rcu(&stream->recv_node);
704 stream->in_recv_list = false;
705 }
706 pthread_mutex_unlock(&session->recv_list_lock);
2a174661 707
7591bab1
MD
708 pthread_mutex_lock(&stream->trace->stream_list_lock);
709 cds_list_add_rcu(&stream->stream_node, &stream->trace->stream_list);
710 pthread_mutex_unlock(&stream->trace->stream_list_lock);
2a174661 711
7591bab1
MD
712 stream->published = true;
713unlock:
2a174661 714 pthread_mutex_unlock(&stream->lock);
2a174661
DG
715}
716
7591bab1 717/*
77f7bd85 718 * Stream must be protected by holding the stream lock or by virtue of being
ce4d4083 719 * called from stream_destroy.
7591bab1
MD
720 */
721static void stream_unpublish(struct relay_stream *stream)
2a174661 722{
77f7bd85
MD
723 if (stream->in_stream_ht) {
724 struct lttng_ht_iter iter;
725 int ret;
726
727 iter.iter.node = &stream->node.node;
728 ret = lttng_ht_del(relay_streams_ht, &iter);
729 assert(!ret);
730 stream->in_stream_ht = false;
731 }
732 if (stream->published) {
733 pthread_mutex_lock(&stream->trace->stream_list_lock);
734 cds_list_del_rcu(&stream->stream_node);
735 pthread_mutex_unlock(&stream->trace->stream_list_lock);
736 stream->published = false;
7591bab1 737 }
7591bab1
MD
738}
739
740static void stream_destroy(struct relay_stream *stream)
741{
742 if (stream->indexes_ht) {
49e614cb
MD
743 /*
744 * Calling lttng_ht_destroy in call_rcu worker thread so
745 * we don't hold the RCU read-side lock while calling
746 * it.
747 */
7591bab1
MD
748 lttng_ht_destroy(stream->indexes_ht);
749 }
a44ca2ca
MD
750 if (stream->tfa) {
751 tracefile_array_destroy(stream->tfa);
752 }
7591bab1
MD
753 free(stream->path_name);
754 free(stream->channel_name);
755 free(stream);
756}
757
758static void stream_destroy_rcu(struct rcu_head *rcu_head)
759{
760 struct relay_stream *stream =
761 caa_container_of(rcu_head, struct relay_stream, rcu_node);
762
763 stream_destroy(stream);
764}
765
766/*
767 * No need to take stream->lock since this is only called on the final
768 * stream_put which ensures that a single thread may act on the stream.
7591bab1
MD
769 */
770static void stream_release(struct urcu_ref *ref)
771{
772 struct relay_stream *stream =
773 caa_container_of(ref, struct relay_stream, ref);
774 struct relay_session *session;
2a174661 775
7591bab1
MD
776 session = stream->trace->session;
777
778 DBG("Releasing stream id %" PRIu64, stream->stream_handle);
779
780 pthread_mutex_lock(&session->recv_list_lock);
781 session->stream_count--;
782 if (stream->in_recv_list) {
783 cds_list_del_rcu(&stream->recv_node);
784 stream->in_recv_list = false;
785 }
786 pthread_mutex_unlock(&session->recv_list_lock);
2a174661 787
7591bab1
MD
788 stream_unpublish(stream);
789
8bb66c3c
JG
790 if (stream->file) {
791 fs_handle_close(stream->file);
792 stream->file = NULL;
7591bab1 793 }
f8f3885c
MD
794 if (stream->index_file) {
795 lttng_index_file_put(stream->index_file);
796 stream->index_file = NULL;
7591bab1
MD
797 }
798 if (stream->trace) {
799 ctf_trace_put(stream->trace);
800 stream->trace = NULL;
801 }
c35f9726 802 stream_complete_rotation(stream);
348a81dc
JG
803 lttng_trace_chunk_put(stream->trace_chunk);
804 stream->trace_chunk = NULL;
7591bab1
MD
805
806 call_rcu(&stream->rcu_node, stream_destroy_rcu);
2a174661
DG
807}
808
7591bab1 809void stream_put(struct relay_stream *stream)
2a174661 810{
7591bab1 811 rcu_read_lock();
7591bab1
MD
812 assert(stream->ref.refcount != 0);
813 /*
814 * Wait until we have processed all the stream packets before
815 * actually putting our last stream reference.
816 */
7591bab1 817 urcu_ref_put(&stream->ref, stream_release);
7591bab1
MD
818 rcu_read_unlock();
819}
820
c35f9726
JG
821int stream_set_pending_rotation(struct relay_stream *stream,
822 struct lttng_trace_chunk *next_trace_chunk,
823 uint64_t rotation_sequence_number)
824{
825 int ret = 0;
826 const struct relay_stream_rotation rotation = {
0f83d1cc
MD
827 .data_rotated = false,
828 .index_rotated = false,
829 .packet_seq_num = rotation_sequence_number,
830 .prev_data_net_seq = -1ULL,
c35f9726
JG
831 .next_trace_chunk = next_trace_chunk,
832 };
833
834 if (stream->ongoing_rotation.is_set) {
835 ERR("Attempted to set a pending rotation on a stream already being rotated (protocol error)");
836 ret = -1;
837 goto end;
838 }
839
840 if (next_trace_chunk) {
841 const bool reference_acquired =
842 lttng_trace_chunk_get(next_trace_chunk);
843
844 assert(reference_acquired);
845 }
846 LTTNG_OPTIONAL_SET(&stream->ongoing_rotation, rotation);
847
0f83d1cc
MD
848 DBG("Setting pending rotation: stream_id = %" PRIu64
849 ", rotate_at_packet_seq_num = %" PRIu64,
c35f9726
JG
850 stream->stream_handle, rotation_sequence_number);
851 if (stream->is_metadata) {
852 /*
853 * A metadata stream has no index; consider it already rotated.
854 */
855 stream->ongoing_rotation.value.index_rotated = true;
f607fc46
MD
856 if (next_trace_chunk) {
857 /*
858 * The metadata will be received again in the new chunk.
859 */
860 stream->metadata_received = 0;
861 }
c35f9726
JG
862 ret = stream_rotate_data_file(stream);
863 } else {
0f83d1cc 864 ret = try_rotate_stream_index(stream);
c35f9726
JG
865 if (ret < 0) {
866 goto end;
867 }
868
0f83d1cc 869 ret = try_rotate_stream_data(stream);
c35f9726
JG
870 if (ret < 0) {
871 goto end;
872 }
873 }
874end:
875 return ret;
876}
877
bda7c7b9 878void try_stream_close(struct relay_stream *stream)
7591bab1 879{
98ba050e
JR
880 bool session_aborted;
881 struct relay_session *session = stream->trace->session;
882
bda7c7b9 883 DBG("Trying to close stream %" PRIu64, stream->stream_handle);
98ba050e
JR
884
885 pthread_mutex_lock(&session->lock);
886 session_aborted = session->aborted;
887 pthread_mutex_unlock(&session->lock);
888
7591bab1 889 pthread_mutex_lock(&stream->lock);
bda7c7b9
JG
890 /*
891 * Can be called concurently by connection close and reception of last
892 * pending data.
893 */
894 if (stream->closed) {
895 pthread_mutex_unlock(&stream->lock);
896 DBG("closing stream %" PRIu64 " aborted since it is already marked as closed", stream->stream_handle);
897 return;
898 }
899
900 stream->close_requested = true;
3d07a857
MD
901
902 if (stream->last_net_seq_num == -1ULL) {
903 /*
904 * Handle connection close without explicit stream close
905 * command.
906 *
907 * We can be clever about indexes partially received in
908 * cases where we received the data socket part, but not
909 * the control socket part: since we're currently closing
910 * the stream on behalf of the control socket, we *know*
911 * there won't be any more control information for this
912 * socket. Therefore, we can destroy all indexes for
913 * which we have received only the file descriptor (from
914 * data socket). This takes care of consumerd crashes
915 * between sending the data and control information for
916 * a packet. Since those are sent in that order, we take
917 * care of consumerd crashes.
918 */
5312a3ed 919 DBG("relay_index_close_partial_fd");
3d07a857
MD
920 relay_index_close_partial_fd(stream);
921 /*
922 * Use the highest net_seq_num we currently have pending
923 * As end of stream indicator. Leave last_net_seq_num
924 * at -1ULL if we cannot find any index.
925 */
926 stream->last_net_seq_num = relay_index_find_last(stream);
5312a3ed 927 DBG("Updating stream->last_net_seq_num to %" PRIu64, stream->last_net_seq_num);
3d07a857
MD
928 /* Fall-through into the next check. */
929 }
930
bda7c7b9 931 if (stream->last_net_seq_num != -1ULL &&
a8f9f353 932 ((int64_t) (stream->prev_data_seq - stream->last_net_seq_num)) < 0
98ba050e 933 && !session_aborted) {
3d07a857
MD
934 /*
935 * Don't close since we still have data pending. This
936 * handles cases where an explicit close command has
937 * been received for this stream, and cases where the
938 * connection has been closed, and we are awaiting for
939 * index information from the data socket. It is
940 * therefore expected that all the index fd information
941 * we need has already been received on the control
942 * socket. Matching index information from data socket
943 * should be Expected Soon(TM).
944 *
945 * TODO: We should implement a timer to garbage collect
946 * streams after a timeout to be resilient against a
947 * consumerd implementation that would not match this
948 * expected behavior.
949 */
bda7c7b9
JG
950 pthread_mutex_unlock(&stream->lock);
951 DBG("closing stream %" PRIu64 " aborted since it still has data pending", stream->stream_handle);
952 return;
953 }
3d07a857
MD
954 /*
955 * We received all the indexes we can expect.
956 */
77f7bd85 957 stream_unpublish(stream);
2229a09c 958 stream->closed = true;
bda7c7b9 959 /* Relay indexes are only used by the "consumer/sessiond" end. */
7591bab1 960 relay_index_close_all(stream);
d91bc249
MD
961
962 /*
963 * If we are closed by an application exiting (per-pid buffers),
964 * we need to put our reference on the stream trace chunk right
965 * away, because otherwise still holding the reference on the
966 * trace chunk could allow a viewer stream (which holds a reference
967 * to the stream) to postpone destroy waiting for the chunk to cease
968 * to exist endlessly until the viewer is detached.
969 */
970
971 /* Put stream fd before put chunk. */
8bb66c3c
JG
972 if (stream->file) {
973 fs_handle_close(stream->file);
974 stream->file = NULL;
d91bc249
MD
975 }
976 if (stream->index_file) {
977 lttng_index_file_put(stream->index_file);
978 stream->index_file = NULL;
979 }
980 lttng_trace_chunk_put(stream->trace_chunk);
981 stream->trace_chunk = NULL;
7591bab1 982 pthread_mutex_unlock(&stream->lock);
bda7c7b9 983 DBG("Succeeded in closing stream %" PRIu64, stream->stream_handle);
7591bab1
MD
984 stream_put(stream);
985}
986
c35f9726
JG
987int stream_init_packet(struct relay_stream *stream, size_t packet_size,
988 bool *file_rotated)
989{
990 int ret = 0;
991
992 ASSERT_LOCKED(stream->lock);
1ad1103b 993
8bb66c3c 994 if (!stream->file || !stream->trace_chunk) {
1ad1103b
JG
995 ERR("Protocol error: received a packet for a stream that doesn't have a current trace chunk: stream_id = %" PRIu64 ", channel_name = %s",
996 stream->stream_handle, stream->channel_name);
997 ret = -1;
998 goto end;
999 }
1000
c35f9726
JG
1001 if (caa_likely(stream->tracefile_size == 0)) {
1002 /* No size limit set; nothing to check. */
1003 goto end;
1004 }
1005
1006 /*
1007 * Check if writing the new packet would exceed the maximal file size.
1008 */
1009 if (caa_unlikely((stream->tracefile_size_current + packet_size) >
1010 stream->tracefile_size)) {
1011 const uint64_t new_file_index =
1012 (stream->tracefile_current_index + 1) %
1013 stream->tracefile_count;
1014
1015 if (new_file_index < stream->tracefile_current_index) {
1016 stream->tracefile_wrapped_around = true;
1017 }
1018 DBG("New stream packet causes stream file rotation: stream_id = %" PRIu64
1019 ", current_file_size = %" PRIu64
76ee9245 1020 ", packet_size = %zu, current_file_index = %" PRIu64
c35f9726
JG
1021 " new_file_index = %" PRIu64,
1022 stream->stream_handle,
1023 stream->tracefile_size_current, packet_size,
1024 stream->tracefile_current_index, new_file_index);
78118e3b 1025 tracefile_array_file_rotate(stream->tfa, TRACEFILE_ROTATE_WRITE);
c35f9726
JG
1026 stream->tracefile_current_index = new_file_index;
1027
8bb66c3c
JG
1028 if (stream->file) {
1029 fs_handle_close(stream->file);
1030 stream->file = NULL;
c35f9726
JG
1031 }
1032 ret = stream_create_data_output_file_from_trace_chunk(stream,
8bb66c3c 1033 stream->trace_chunk, false, &stream->file);
c35f9726
JG
1034 if (ret) {
1035 ERR("Failed to perform trace file rotation of stream %" PRIu64,
1036 stream->stream_handle);
1037 goto end;
1038 }
1039
1040 /*
1041 * Reset current size because we just performed a stream
1042 * rotation.
1043 */
68c40154
MD
1044 DBG("%s: reset tracefile_size_current for stream %" PRIu64 " was %" PRIu64,
1045 __func__, stream->stream_handle, stream->tracefile_size_current);
c35f9726
JG
1046 stream->tracefile_size_current = 0;
1047 *file_rotated = true;
1048 } else {
1049 *file_rotated = false;
1050 }
1051end:
1052 return ret;
1053}
1054
1055/* Note that the packet is not necessarily complete. */
1056int stream_write(struct relay_stream *stream,
1057 const struct lttng_buffer_view *packet, size_t padding_len)
1058{
1059 int ret = 0;
1060 ssize_t write_ret;
1061 size_t padding_to_write = padding_len;
1062 char padding_buffer[FILE_IO_STACK_BUFFER_SIZE];
1063
1064 ASSERT_LOCKED(stream->lock);
1065 memset(padding_buffer, 0,
1066 min(sizeof(padding_buffer), padding_to_write));
1067
8bb66c3c 1068 if (!stream->file || !stream->trace_chunk) {
1ad1103b
JG
1069 ERR("Protocol error: received a packet for a stream that doesn't have a current trace chunk: stream_id = %" PRIu64 ", channel_name = %s",
1070 stream->stream_handle, stream->channel_name);
1071 ret = -1;
1072 goto end;
1073 }
c35f9726 1074 if (packet) {
8bb66c3c
JG
1075 write_ret = fs_handle_write(
1076 stream->file, packet->data, packet->size);
c35f9726
JG
1077 if (write_ret != packet->size) {
1078 PERROR("Failed to write to stream file of %sstream %" PRIu64,
1079 stream->is_metadata ? "metadata " : "",
1080 stream->stream_handle);
1081 ret = -1;
1082 goto end;
1083 }
1084 }
1085
1086 while (padding_to_write > 0) {
1087 const size_t padding_to_write_this_pass =
1088 min(padding_to_write, sizeof(padding_buffer));
1089
8bb66c3c
JG
1090 write_ret = fs_handle_write(stream->file, padding_buffer,
1091 padding_to_write_this_pass);
c35f9726
JG
1092 if (write_ret != padding_to_write_this_pass) {
1093 PERROR("Failed to write padding to file of %sstream %" PRIu64,
1094 stream->is_metadata ? "metadata " : "",
1095 stream->stream_handle);
1096 ret = -1;
1097 goto end;
1098 }
1099 padding_to_write -= padding_to_write_this_pass;
1100 }
1101
1102 if (stream->is_metadata) {
94f73d08
MD
1103 size_t recv_len;
1104
1105 recv_len = packet ? packet->size : 0;
1106 recv_len += padding_len;
1107 stream->metadata_received += recv_len;
1108 if (recv_len) {
1109 stream->no_new_metadata_notified = false;
1110 }
c35f9726
JG
1111 }
1112
8c865d87 1113 DBG("Wrote to %sstream %" PRIu64 ": data_length = %zu, padding_length = %zu",
c35f9726
JG
1114 stream->is_metadata ? "metadata " : "",
1115 stream->stream_handle,
8c865d87 1116 packet ? packet->size : (size_t) 0, padding_len);
c35f9726
JG
1117end:
1118 return ret;
1119}
1120
1121/*
1122 * Update index after receiving a packet for a data stream.
1123 *
1124 * Called with the stream lock held.
1125 *
1126 * Return 0 on success else a negative value.
1127 */
1128int stream_update_index(struct relay_stream *stream, uint64_t net_seq_num,
1129 bool rotate_index, bool *flushed, uint64_t total_size)
1130{
1131 int ret = 0;
1132 uint64_t data_offset;
1133 struct relay_index *index;
1134
92761772 1135 assert(stream->trace_chunk);
c35f9726
JG
1136 ASSERT_LOCKED(stream->lock);
1137 /* Get data offset because we are about to update the index. */
1138 data_offset = htobe64(stream->tracefile_size_current);
1139
1140 DBG("handle_index_data: stream %" PRIu64 " net_seq_num %" PRIu64 " data offset %" PRIu64,
1141 stream->stream_handle, net_seq_num, stream->tracefile_size_current);
1142
1143 /*
1144 * Lookup for an existing index for that stream id/sequence
1145 * number. If it exists, the control thread has already received the
1146 * data for it, thus we need to write it to disk.
1147 */
1148 index = relay_index_get_by_id_or_create(stream, net_seq_num);
1149 if (!index) {
1150 ret = -1;
1151 goto end;
1152 }
1153
1154 if (rotate_index || !stream->index_file) {
1155 ret = create_index_file(stream, stream->trace_chunk);
1156 if (ret) {
1157 ERR("Failed to create index file for stream %" PRIu64,
1158 stream->stream_handle);
1159 /* Put self-ref for this index due to error. */
1160 relay_index_put(index);
1161 index = NULL;
1162 goto end;
1163 }
1164 }
1165
1166 if (relay_index_set_file(index, stream->index_file, data_offset)) {
1167 ret = -1;
1168 /* Put self-ref for this index due to error. */
1169 relay_index_put(index);
1170 index = NULL;
1171 goto end;
1172 }
1173
1174 ret = relay_index_try_flush(index);
1175 if (ret == 0) {
78118e3b 1176 tracefile_array_file_rotate(stream->tfa, TRACEFILE_ROTATE_READ);
62f6e9ef 1177 tracefile_array_commit_seq(stream->tfa, stream->index_received_seqcount);
c35f9726 1178 stream->index_received_seqcount++;
0f83d1cc
MD
1179 LTTNG_OPTIONAL_SET(&stream->received_packet_seq_num,
1180 be64toh(index->index_data.packet_seq_num));
c35f9726
JG
1181 *flushed = true;
1182 } else if (ret > 0) {
1183 index->total_size = total_size;
1184 /* No flush. */
1185 ret = 0;
1186 } else {
1187 /*
1188 * ret < 0
1189 *
1190 * relay_index_try_flush is responsible for the self-reference
1191 * put of the index object on error.
1192 */
1193 ERR("relay_index_try_flush error %d", ret);
1194 ret = -1;
1195 }
1196end:
1197 return ret;
1198}
1199
1200int stream_complete_packet(struct relay_stream *stream, size_t packet_total_size,
1201 uint64_t sequence_number, bool index_flushed)
1202{
1203 int ret = 0;
1204
1205 ASSERT_LOCKED(stream->lock);
1206
1207 stream->tracefile_size_current += packet_total_size;
1208 if (index_flushed) {
1209 stream->pos_after_last_complete_data_index =
1210 stream->tracefile_size_current;
1211 stream->prev_index_seq = sequence_number;
1212 ret = try_rotate_stream_index(stream);
1213 if (ret < 0) {
1214 goto end;
1215 }
1216 }
1217
1218 stream->prev_data_seq = sequence_number;
1219 ret = try_rotate_stream_data(stream);
d2ec9b88 1220
c35f9726
JG
1221end:
1222 return ret;
1223}
1224
1225int stream_add_index(struct relay_stream *stream,
1226 const struct lttcomm_relayd_index *index_info)
1227{
1228 int ret = 0;
1229 struct relay_index *index;
1230
1231 ASSERT_LOCKED(stream->lock);
1232
68c40154
MD
1233 DBG("stream_add_index for stream %" PRIu64, stream->stream_handle);
1234
c35f9726
JG
1235 /* Live beacon handling */
1236 if (index_info->packet_size == 0) {
1237 DBG("Received live beacon for stream %" PRIu64,
1238 stream->stream_handle);
1239
1240 /*
1241 * Only flag a stream inactive when it has already
1242 * received data and no indexes are in flight.
1243 */
1244 if (stream->index_received_seqcount > 0
1245 && stream->indexes_in_flight == 0) {
1246 stream->beacon_ts_end = index_info->timestamp_end;
1247 }
1248 ret = 0;
1249 goto end;
1250 } else {
1251 stream->beacon_ts_end = -1ULL;
1252 }
1253
1254 if (stream->ctf_stream_id == -1ULL) {
1255 stream->ctf_stream_id = index_info->stream_id;
1256 }
1257
1258 index = relay_index_get_by_id_or_create(stream, index_info->net_seq_num);
1259 if (!index) {
1260 ret = -1;
1261 ERR("Failed to get or create index %" PRIu64,
1262 index_info->net_seq_num);
1263 goto end;
1264 }
1265 if (relay_index_set_control_data(index, index_info,
1266 stream->trace->session->minor)) {
1267 ERR("set_index_control_data error");
1268 relay_index_put(index);
1269 ret = -1;
1270 goto end;
1271 }
1272 ret = relay_index_try_flush(index);
1273 if (ret == 0) {
78118e3b 1274 tracefile_array_file_rotate(stream->tfa, TRACEFILE_ROTATE_READ);
62f6e9ef 1275 tracefile_array_commit_seq(stream->tfa, stream->index_received_seqcount);
c35f9726
JG
1276 stream->index_received_seqcount++;
1277 stream->pos_after_last_complete_data_index += index->total_size;
1278 stream->prev_index_seq = index_info->net_seq_num;
0f83d1cc
MD
1279 LTTNG_OPTIONAL_SET(&stream->received_packet_seq_num,
1280 index_info->packet_seq_num);
c35f9726
JG
1281
1282 ret = try_rotate_stream_index(stream);
1283 if (ret < 0) {
1284 goto end;
1285 }
0f83d1cc
MD
1286 ret = try_rotate_stream_data(stream);
1287 if (ret < 0) {
1288 goto end;
1289 }
c35f9726
JG
1290 } else if (ret > 0) {
1291 /* no flush. */
1292 ret = 0;
1293 } else {
1294 /*
1295 * ret < 0
1296 *
1297 * relay_index_try_flush is responsible for the self-reference
1298 * put of the index object on error.
1299 */
1300 ERR("relay_index_try_flush error %d", ret);
1301 ret = -1;
1302 }
1303end:
1304 return ret;
1305}
1306
da412cde
MD
1307static void print_stream_indexes(struct relay_stream *stream)
1308{
1309 struct lttng_ht_iter iter;
1310 struct relay_index *index;
1311
1312 rcu_read_lock();
1313 cds_lfht_for_each_entry(stream->indexes_ht->ht, &iter.iter, index,
1314 index_n.node) {
1315 DBG("index %p net_seq_num %" PRIu64 " refcount %ld"
1316 " stream %" PRIu64 " trace %" PRIu64
1317 " session %" PRIu64,
1318 index,
1319 index->index_n.key,
1320 stream->ref.refcount,
1321 index->stream->stream_handle,
1322 index->stream->trace->id,
1323 index->stream->trace->session->id);
1324 }
1325 rcu_read_unlock();
1326}
1327
c35f9726
JG
1328int stream_reset_file(struct relay_stream *stream)
1329{
1330 ASSERT_LOCKED(stream->lock);
1331
8bb66c3c
JG
1332 if (stream->file) {
1333 int ret;
1334
1335 ret = fs_handle_close(stream->file);
1336 if (ret) {
1337 ERR("Failed to close stream file handle: channel name = \"%s\", id = %" PRIu64,
1338 stream->channel_name,
1339 stream->stream_handle);
1340 }
1341 stream->file = NULL;
c35f9726
JG
1342 }
1343
68c40154
MD
1344 DBG("%s: reset tracefile_size_current for stream %" PRIu64 " was %" PRIu64,
1345 __func__, stream->stream_handle, stream->tracefile_size_current);
c35f9726
JG
1346 stream->tracefile_size_current = 0;
1347 stream->prev_data_seq = 0;
1348 stream->prev_index_seq = 0;
1349 /* Note that this does not reset the tracefile array. */
1350 stream->tracefile_current_index = 0;
1351 stream->pos_after_last_complete_data_index = 0;
1352
1353 return stream_create_data_output_file_from_trace_chunk(stream,
8bb66c3c 1354 stream->trace_chunk, true, &stream->file);
c35f9726
JG
1355}
1356
7591bab1
MD
1357void print_relay_streams(void)
1358{
1359 struct lttng_ht_iter iter;
1360 struct relay_stream *stream;
1361
ce3f3ba3
JG
1362 if (!relay_streams_ht) {
1363 return;
1364 }
1365
7591bab1
MD
1366 rcu_read_lock();
1367 cds_lfht_for_each_entry(relay_streams_ht->ht, &iter.iter, stream,
1368 node.node) {
1369 if (!stream_get(stream)) {
1370 continue;
1371 }
1372 DBG("stream %p refcount %ld stream %" PRIu64 " trace %" PRIu64
1373 " session %" PRIu64,
1374 stream,
1375 stream->ref.refcount,
1376 stream->stream_handle,
1377 stream->trace->id,
1378 stream->trace->session->id);
da412cde 1379 print_stream_indexes(stream);
7591bab1
MD
1380 stream_put(stream);
1381 }
1382 rcu_read_unlock();
2a174661 1383}
This page took 0.111062 seconds and 4 git commands to generate.