projects
/
lttng-tools.git
/ blobdiff
commit
grep
author
committer
pickaxe
?
search:
re
summary
|
shortlog
|
log
|
commit
|
commitdiff
|
tree
raw
|
inline
| side by side
Fix: relayd vs consumerd compatibility
[lttng-tools.git]
/
src
/
common
/
consumer
/
consumer.c
diff --git
a/src/common/consumer/consumer.c
b/src/common/consumer/consumer.c
index 7911a5b4e8d4d0701d40e88d3ebdd6841c4061af..929be6553a7ea76dc54694cb149d988b448efa3b 100644
(file)
--- a/
src/common/consumer/consumer.c
+++ b/
src/common/consumer/consumer.c
@@
-570,7
+570,7
@@
struct lttng_consumer_stream *consumer_allocate_stream(uint64_t channel_key,
stream->session_id = session_id;
stream->monitor = monitor;
stream->endpoint_status = CONSUMER_ENDPOINT_ACTIVE;
stream->session_id = session_id;
stream->monitor = monitor;
stream->endpoint_status = CONSUMER_ENDPOINT_ACTIVE;
- stream->index_f
d = -1
;
+ stream->index_f
ile = NULL
;
stream->last_sequence_number = -1ULL;
pthread_mutex_init(&stream->lock, NULL);
pthread_mutex_init(&stream->metadata_timer_lock, NULL);
stream->last_sequence_number = -1ULL;
pthread_mutex_init(&stream->lock, NULL);
pthread_mutex_init(&stream->metadata_timer_lock, NULL);
@@
-1021,7
+1021,7
@@
struct lttng_consumer_channel *consumer_allocate_channel(uint64_t key,
CDS_INIT_LIST_HEAD(&channel->streams.head);
CDS_INIT_LIST_HEAD(&channel->streams.head);
- DBG("Allocated channel (key %" PRIu64 ")", channel->key)
+ DBG("Allocated channel (key %" PRIu64 ")", channel->key)
;
end:
return channel;
end:
return channel;
@@
-1229,9
+1229,15
@@
void lttng_consumer_should_exit(struct lttng_consumer_local_data *ctx)
DBG("Consumer flag that it should quit");
}
DBG("Consumer flag that it should quit");
}
+
+/*
+ * Flush pending writes to trace output disk file.
+ */
+static
void lttng_consumer_sync_trace_file(struct lttng_consumer_stream *stream,
off_t orig_offset)
{
void lttng_consumer_sync_trace_file(struct lttng_consumer_stream *stream,
off_t orig_offset)
{
+ int ret;
int outfd = stream->out_fd;
/*
int outfd = stream->out_fd;
/*
@@
-1262,8
+1268,12
@@
void lttng_consumer_sync_trace_file(struct lttng_consumer_stream *stream,
* defined. So it can be expected to lead to lower throughput in
* streaming.
*/
* defined. So it can be expected to lead to lower throughput in
* streaming.
*/
- posix_fadvise(outfd, orig_offset - stream->max_sb_size,
+
ret =
posix_fadvise(outfd, orig_offset - stream->max_sb_size,
stream->max_sb_size, POSIX_FADV_DONTNEED);
stream->max_sb_size, POSIX_FADV_DONTNEED);
+ if (ret && ret != -ENOSYS) {
+ errno = ret;
+ PERROR("posix_fadvise on fd %i", outfd);
+ }
}
/*
}
/*
@@
-1520,7
+1530,6
@@
ssize_t lttng_consumer_on_read_subbuffer_mmap(
mmap_base = stream->mmap_base;
ret = kernctl_get_mmap_read_offset(stream->wait_fd, &mmap_offset);
if (ret < 0) {
mmap_base = stream->mmap_base;
ret = kernctl_get_mmap_read_offset(stream->wait_fd, &mmap_offset);
if (ret < 0) {
- ret = -errno;
PERROR("tracer ctl get_mmap_read_offset");
goto end;
}
PERROR("tracer ctl get_mmap_read_offset");
goto end;
}
@@
-1556,6
+1565,16
@@
ssize_t lttng_consumer_on_read_subbuffer_mmap(
if (stream->metadata_flag) {
/* Metadata requires the control socket. */
pthread_mutex_lock(&relayd->ctrl_sock_mutex);
if (stream->metadata_flag) {
/* Metadata requires the control socket. */
pthread_mutex_lock(&relayd->ctrl_sock_mutex);
+ if (stream->reset_metadata_flag) {
+ ret = relayd_reset_metadata(&relayd->control_sock,
+ stream->relayd_stream_id,
+ stream->metadata_version);
+ if (ret < 0) {
+ relayd_hang_up = 1;
+ goto write_error;
+ }
+ stream->reset_metadata_flag = 0;
+ }
netlen += sizeof(struct lttcomm_relayd_metadata_payload);
}
netlen += sizeof(struct lttcomm_relayd_metadata_payload);
}
@@
-1579,6
+1598,15
@@
ssize_t lttng_consumer_on_read_subbuffer_mmap(
/* No streaming, we have to set the len with the full padding */
len += padding;
/* No streaming, we have to set the len with the full padding */
len += padding;
+ if (stream->metadata_flag && stream->reset_metadata_flag) {
+ ret = utils_truncate_stream_file(stream->out_fd, 0);
+ if (ret < 0) {
+ ERR("Reset metadata file");
+ goto end;
+ }
+ stream->reset_metadata_flag = 0;
+ }
+
/*
* Check if we need to change the tracefile before writing the packet.
*/
/*
* Check if we need to change the tracefile before writing the packet.
*/
@@
-1596,21
+1624,16
@@
ssize_t lttng_consumer_on_read_subbuffer_mmap(
}
outfd = stream->out_fd;
}
outfd = stream->out_fd;
- if (stream->index_fd >= 0) {
- ret = close(stream->index_fd);
- if (ret < 0) {
- PERROR("Closing index");
- goto end;
- }
- stream->index_fd = -1;
- ret = index_create_file(stream->chan->pathname,
+ if (stream->index_file) {
+ lttng_index_file_put(stream->index_file);
+ stream->index_file = lttng_index_file_create(stream->chan->pathname,
stream->name, stream->uid, stream->gid,
stream->chan->tracefile_size,
stream->name, stream->uid, stream->gid,
stream->chan->tracefile_size,
- stream->tracefile_count_current);
- if (ret < 0) {
+ stream->tracefile_count_current,
+ CTF_INDEX_MAJOR, CTF_INDEX_MINOR);
+ if (!stream->index_file) {
goto end;
}
goto end;
}
- stream->index_fd = ret;
}
/* Reset current size because we just perform a rotation. */
}
/* Reset current size because we just perform a rotation. */
@@
-1663,8
+1686,8
@@
ssize_t lttng_consumer_on_read_subbuffer_mmap(
lttng_sync_file_range(outfd, stream->out_fd_offset, len,
SYNC_FILE_RANGE_WRITE);
stream->out_fd_offset += len;
lttng_sync_file_range(outfd, stream->out_fd_offset, len,
SYNC_FILE_RANGE_WRITE);
stream->out_fd_offset += len;
+ lttng_consumer_sync_trace_file(stream, orig_offset);
}
}
- lttng_consumer_sync_trace_file(stream, orig_offset);
write_error:
/*
write_error:
/*
@@
-1744,6
+1767,16
@@
ssize_t lttng_consumer_on_read_subbuffer_splice(
*/
pthread_mutex_lock(&relayd->ctrl_sock_mutex);
*/
pthread_mutex_lock(&relayd->ctrl_sock_mutex);
+ if (stream->reset_metadata_flag) {
+ ret = relayd_reset_metadata(&relayd->control_sock,
+ stream->relayd_stream_id,
+ stream->metadata_version);
+ if (ret < 0) {
+ relayd_hang_up = 1;
+ goto write_error;
+ }
+ stream->reset_metadata_flag = 0;
+ }
ret = write_relayd_metadata_id(splice_pipe[1], stream, relayd,
padding);
if (ret < 0) {
ret = write_relayd_metadata_id(splice_pipe[1], stream, relayd,
padding);
if (ret < 0) {
@@
-1767,6
+1800,14
@@
ssize_t lttng_consumer_on_read_subbuffer_splice(
/* No streaming, we have to set the len with the full padding */
len += padding;
/* No streaming, we have to set the len with the full padding */
len += padding;
+ if (stream->metadata_flag && stream->reset_metadata_flag) {
+ ret = utils_truncate_stream_file(stream->out_fd, 0);
+ if (ret < 0) {
+ ERR("Reset metadata file");
+ goto end;
+ }
+ stream->reset_metadata_flag = 0;
+ }
/*
* Check if we need to change the tracefile before writing the packet.
*/
/*
* Check if we need to change the tracefile before writing the packet.
*/
@@
-1785,22
+1826,16
@@
ssize_t lttng_consumer_on_read_subbuffer_splice(
}
outfd = stream->out_fd;
}
outfd = stream->out_fd;
- if (stream->index_fd >= 0) {
- ret = close(stream->index_fd);
- if (ret < 0) {
- PERROR("Closing index");
- goto end;
- }
- stream->index_fd = -1;
- ret = index_create_file(stream->chan->pathname,
+ if (stream->index_file) {
+ lttng_index_file_put(stream->index_file);
+ stream->index_file = lttng_index_file_create(stream->chan->pathname,
stream->name, stream->uid, stream->gid,
stream->chan->tracefile_size,
stream->name, stream->uid, stream->gid,
stream->chan->tracefile_size,
- stream->tracefile_count_current
);
- if (ret < 0) {
- written = ret;
+ stream->tracefile_count_current
,
+ CTF_INDEX_MAJOR, CTF_INDEX_MINOR);
+ if (!stream->index_file) {
goto end;
}
goto end;
}
- stream->index_fd = ret;
}
/* Reset current size because we just perform a rotation. */
}
/* Reset current size because we just perform a rotation. */
@@
-1875,7
+1910,9
@@
ssize_t lttng_consumer_on_read_subbuffer_splice(
stream->output_written += ret_splice;
written += ret_splice;
}
stream->output_written += ret_splice;
written += ret_splice;
}
- lttng_consumer_sync_trace_file(stream, orig_offset);
+ if (!relayd) {
+ lttng_consumer_sync_trace_file(stream, orig_offset);
+ }
goto end;
write_error:
goto end;
write_error:
@@
-2224,10
+2261,10
@@
restart:
DBG("Metadata poll return from wait with %d fd(s)",
LTTNG_POLL_GETNB(&events));
health_poll_exit();
DBG("Metadata poll return from wait with %d fd(s)",
LTTNG_POLL_GETNB(&events));
health_poll_exit();
- DBG("Metadata event ca
tched
in thread");
+ DBG("Metadata event ca
ught
in thread");
if (ret < 0) {
if (errno == EINTR) {
if (ret < 0) {
if (errno == EINTR) {
- ERR("Poll EINTR ca
tched
");
+ ERR("Poll EINTR ca
ught
");
goto restart;
}
if (LTTNG_POLL_GETNB(&events) == 0) {
goto restart;
}
if (LTTNG_POLL_GETNB(&events) == 0) {
@@
-2325,7
+2362,7
@@
restart:
len = ctx->on_buffer_ready(stream, ctx);
/*
* We don't check the return value here since if we get
len = ctx->on_buffer_ready(stream, ctx);
/*
* We don't check the return value here since if we get
- * a negative len, it means an error occured thus we
+ * a negative len, it means an error occur
r
ed thus we
* simply remove it from the poll set and free the
* stream.
*/
* simply remove it from the poll set and free the
* stream.
*/
@@
-2352,7
+2389,7
@@
restart:
len = ctx->on_buffer_ready(stream, ctx);
/*
* We don't check the return value here since if we get
len = ctx->on_buffer_ready(stream, ctx);
/*
* We don't check the return value here since if we get
- * a negative len, it means an error occured thus we
+ * a negative len, it means an error occur
r
ed thus we
* simply remove it from the poll set and free the
* stream.
*/
* simply remove it from the poll set and free the
* stream.
*/
@@
-2807,10
+2844,10
@@
restart:
DBG("Channel poll return from wait with %d fd(s)",
LTTNG_POLL_GETNB(&events));
health_poll_exit();
DBG("Channel poll return from wait with %d fd(s)",
LTTNG_POLL_GETNB(&events));
health_poll_exit();
- DBG("Channel event ca
tched
in thread");
+ DBG("Channel event ca
ught
in thread");
if (ret < 0) {
if (errno == EINTR) {
if (ret < 0) {
if (errno == EINTR) {
- ERR("Poll EINTR ca
tched
");
+ ERR("Poll EINTR ca
ught
");
goto restart;
}
if (LTTNG_POLL_GETNB(&events) == 0) {
goto restart;
}
if (LTTNG_POLL_GETNB(&events) == 0) {
This page took
0.029395 seconds
and
4
git commands to generate.