projects
/
lttng-tools.git
/ blobdiff
commit
grep
author
committer
pickaxe
?
search:
re
summary
|
shortlog
|
log
|
commit
|
commitdiff
|
tree
raw
|
inline
| side by side
Fix: perform relayd socket pair cleanup on control socket error
[lttng-tools.git]
/
src
/
common
/
consumer
/
consumer.c
diff --git
a/src/common/consumer/consumer.c
b/src/common/consumer/consumer.c
index 5625e59ddd5cb8a2092d21d1145149df63ed66d6..567babadf177ade2912666c2a14ec0ee368ea7d2 100644
(file)
--- a/
src/common/consumer/consumer.c
+++ b/
src/common/consumer/consumer.c
@@
-47,6
+47,7
@@
#include <common/consumer/consumer-stream.h>
#include <common/consumer/consumer-testpoint.h>
#include <common/align.h>
#include <common/consumer/consumer-stream.h>
#include <common/consumer/consumer-testpoint.h>
#include <common/align.h>
+#include <common/consumer/consumer-metadata-cache.h>
struct lttng_consumer_global_data consumer_data = {
.stream_count = 0,
struct lttng_consumer_global_data consumer_data = {
.stream_count = 0,
@@
-66,13
+67,16
@@
struct consumer_channel_msg {
uint64_t key; /* del */
};
uint64_t key; /* del */
};
+/* Flag used to temporarily pause data consumption from testpoints. */
+int data_consumption_paused;
+
/*
* Flag to inform the polling thread to quit when all fd hung up. Updated by
* the consumer_thread_receive_fds when it notices that all fds has hung up.
* Also updated by the signal handler (consumer_should_exit()). Read by the
* polling threads.
*/
/*
* Flag to inform the polling thread to quit when all fd hung up. Updated by
* the consumer_thread_receive_fds when it notices that all fds has hung up.
* Also updated by the signal handler (consumer_should_exit()). Read by the
* polling threads.
*/
-
volatile
int consumer_quit;
+int consumer_quit;
/*
* Global hash table containing respectively metadata and data streams. The
/*
* Global hash table containing respectively metadata and data streams. The
@@
-319,6
+323,7
@@
static void free_relayd_rcu(struct rcu_head *head)
(void) relayd_close(&relayd->control_sock);
(void) relayd_close(&relayd->data_sock);
(void) relayd_close(&relayd->control_sock);
(void) relayd_close(&relayd->data_sock);
+ pthread_mutex_destroy(&relayd->ctrl_sock_mutex);
free(relayd);
}
free(relayd);
}
@@
-367,6
+372,9
@@
void consumer_del_channel(struct lttng_consumer_channel *channel)
if (channel->live_timer_enabled == 1) {
consumer_timer_live_stop(channel);
}
if (channel->live_timer_enabled == 1) {
consumer_timer_live_stop(channel);
}
+ if (channel->monitor_timer_enabled == 1) {
+ consumer_timer_monitor_stop(channel);
+ }
switch (consumer_data.type) {
case LTTNG_CONSUMER_KERNEL:
switch (consumer_data.type) {
case LTTNG_CONSUMER_KERNEL:
@@
-457,14
+465,13
@@
static void update_endpoint_status_by_netidx(uint64_t net_seq_idx,
* If a local data context is available, notify the threads that the streams'
* state have changed.
*/
* If a local data context is available, notify the threads that the streams'
* state have changed.
*/
-static void cleanup_relayd(struct consumer_relayd_sock_pair *relayd,
- struct lttng_consumer_local_data *ctx)
+void lttng_consumer_cleanup_relayd(struct consumer_relayd_sock_pair *relayd)
{
uint64_t netidx;
assert(relayd);
{
uint64_t netidx;
assert(relayd);
- DBG("Cleaning up relayd
sockets"
);
+ DBG("Cleaning up relayd
object ID %"PRIu64, relayd->net_seq_idx
);
/* Save the net sequence index before destroying the object */
netidx = relayd->net_seq_idx;
/* Save the net sequence index before destroying the object */
netidx = relayd->net_seq_idx;
@@
-484,10
+491,8
@@
static void cleanup_relayd(struct consumer_relayd_sock_pair *relayd,
* memory barrier ordering the updates of the end point status from the
* read of this status which happens AFTER receiving this notify.
*/
* memory barrier ordering the updates of the end point status from the
* read of this status which happens AFTER receiving this notify.
*/
- if (ctx) {
- notify_thread_lttng_pipe(ctx->consumer_data_pipe);
- notify_thread_lttng_pipe(ctx->consumer_metadata_pipe);
- }
+ notify_thread_lttng_pipe(relayd->ctx->consumer_data_pipe);
+ notify_thread_lttng_pipe(relayd->ctx->consumer_metadata_pipe);
}
/*
}
/*
@@
-570,7
+575,8
@@
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_fd = -1;
+ stream->index_file = NULL;
+ stream->last_sequence_number = -1ULL;
pthread_mutex_init(&stream->lock, NULL);
pthread_mutex_init(&stream->metadata_timer_lock, NULL);
pthread_mutex_init(&stream->lock, NULL);
pthread_mutex_init(&stream->metadata_timer_lock, NULL);
@@
-794,6
+800,8
@@
int consumer_send_relayd_stream(struct lttng_consumer_stream *stream,
stream->chan->tracefile_size, stream->chan->tracefile_count);
pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
if (ret < 0) {
stream->chan->tracefile_size, stream->chan->tracefile_count);
pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
if (ret < 0) {
+ ERR("Relayd add stream failed. Cleaning up relayd %" PRIu64".", relayd->net_seq_idx);
+ lttng_consumer_cleanup_relayd(relayd);
goto end;
}
goto end;
}
@@
-835,6
+843,8
@@
int consumer_send_relayd_streams_sent(uint64_t net_seq_idx)
ret = relayd_streams_sent(&relayd->control_sock);
pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
if (ret < 0) {
ret = relayd_streams_sent(&relayd->control_sock);
pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
if (ret < 0) {
+ ERR("Relayd streams sent failed. Cleaning up relayd %" PRIu64".", relayd->net_seq_idx);
+ lttng_consumer_cleanup_relayd(relayd);
goto end;
}
} else {
goto end;
}
} else {
@@
-1020,7
+1030,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;
@@
-1069,7
+1079,7
@@
int consumer_add_channel(struct lttng_consumer_channel *channel,
*/
static int update_poll_array(struct lttng_consumer_local_data *ctx,
struct pollfd **pollfd, struct lttng_consumer_stream **local_stream,
*/
static int update_poll_array(struct lttng_consumer_local_data *ctx,
struct pollfd **pollfd, struct lttng_consumer_stream **local_stream,
- struct lttng_ht *ht)
+ struct lttng_ht *ht
, int *nb_inactive_fd
)
{
int i = 0;
struct lttng_ht_iter iter;
{
int i = 0;
struct lttng_ht_iter iter;
@@
-1081,6
+1091,7
@@
static int update_poll_array(struct lttng_consumer_local_data *ctx,
assert(local_stream);
DBG("Updating poll fd array");
assert(local_stream);
DBG("Updating poll fd array");
+ *nb_inactive_fd = 0;
rcu_read_lock();
cds_lfht_for_each_entry(ht->ht, &iter.iter, stream, node.node) {
/*
rcu_read_lock();
cds_lfht_for_each_entry(ht->ht, &iter.iter, stream, node.node) {
/*
@@
-1091,9
+1102,14
@@
static int update_poll_array(struct lttng_consumer_local_data *ctx,
* just after the check. However, this is OK since the stream(s) will
* be deleted once the thread is notified that the end point state has
* changed where this function will be called back again.
* just after the check. However, this is OK since the stream(s) will
* be deleted once the thread is notified that the end point state has
* changed where this function will be called back again.
+ *
+ * We track the number of inactive FDs because they still need to be
+ * closed by the polling thread after a wakeup on the data_pipe or
+ * metadata_pipe.
*/
if (stream->state != LTTNG_CONSUMER_ACTIVE_STREAM ||
stream->endpoint_status == CONSUMER_ENDPOINT_INACTIVE) {
*/
if (stream->state != LTTNG_CONSUMER_ACTIVE_STREAM ||
stream->endpoint_status == CONSUMER_ENDPOINT_INACTIVE) {
+ (*nb_inactive_fd)++;
continue;
}
/*
continue;
}
/*
@@
-1219,7
+1235,7
@@
void lttng_consumer_should_exit(struct lttng_consumer_local_data *ctx)
{
ssize_t ret;
{
ssize_t ret;
-
consumer_quit = 1
;
+
CMM_STORE_SHARED(consumer_quit, 1)
;
ret = lttng_write(ctx->consumer_should_quit[1], "4", 1);
if (ret < 1) {
PERROR("write consumer quit");
ret = lttng_write(ctx->consumer_should_quit[1], "4", 1);
if (ret < 1) {
PERROR("write consumer quit");
@@
-1228,9
+1244,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;
/*
@@
-1261,8
+1283,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);
+ }
}
/*
}
/*
@@
-1336,6
+1362,8
@@
struct lttng_consumer_local_data *lttng_consumer_create(
goto error_metadata_pipe;
}
goto error_metadata_pipe;
}
+ ctx->channel_monitor_pipe = -1;
+
return ctx;
error_metadata_pipe:
return ctx;
error_metadata_pipe:
@@
-1519,7
+1547,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;
}
@@
-1555,6
+1582,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);
}
@@
-1578,6
+1615,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.
*/
@@
-1595,21
+1641,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. */
@@
-1662,8
+1703,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:
/*
@@
-1671,7
+1712,8
@@
write_error:
* cleanup the relayd object and all associated streams.
*/
if (relayd && relayd_hang_up) {
* cleanup the relayd object and all associated streams.
*/
if (relayd && relayd_hang_up) {
- cleanup_relayd(relayd, ctx);
+ ERR("Relayd hangup. Cleaning up relayd %" PRIu64".", relayd->net_seq_idx);
+ lttng_consumer_cleanup_relayd(relayd);
}
end:
}
end:
@@
-1743,6
+1785,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) {
@@
-1766,6
+1818,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.
*/
@@
-1784,22
+1844,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. */
@@
-1874,7
+1928,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:
@@
-1883,7
+1939,8
@@
write_error:
* cleanup the relayd object and all associated streams.
*/
if (relayd && relayd_hang_up) {
* cleanup the relayd object and all associated streams.
*/
if (relayd && relayd_hang_up) {
- cleanup_relayd(relayd, ctx);
+ ERR("Relayd hangup. Cleaning up relayd %" PRIu64".", relayd->net_seq_idx);
+ lttng_consumer_cleanup_relayd(relayd);
/* Skip splice error so the consumer does not fail */
goto end;
}
/* Skip splice error so the consumer does not fail */
goto end;
}
@@
-2014,6
+2071,10
@@
void consumer_del_metadata_stream(struct lttng_consumer_stream *stream,
pthread_mutex_lock(&consumer_data.lock);
pthread_mutex_lock(&stream->chan->lock);
pthread_mutex_lock(&stream->lock);
pthread_mutex_lock(&consumer_data.lock);
pthread_mutex_lock(&stream->chan->lock);
pthread_mutex_lock(&stream->lock);
+ if (stream->chan->metadata_cache) {
+ /* Only applicable to userspace consumers. */
+ pthread_mutex_lock(&stream->chan->metadata_cache->lock);
+ }
/* Remove any reference to that stream. */
consumer_stream_delete(stream, ht);
/* Remove any reference to that stream. */
consumer_stream_delete(stream, ht);
@@
-2037,6
+2098,9
@@
void consumer_del_metadata_stream(struct lttng_consumer_stream *stream,
*/
stream->chan->metadata_stream = NULL;
*/
stream->chan->metadata_stream = NULL;
+ if (stream->chan->metadata_cache) {
+ pthread_mutex_unlock(&stream->chan->metadata_cache->lock);
+ }
pthread_mutex_unlock(&stream->lock);
pthread_mutex_unlock(&stream->chan->lock);
pthread_mutex_unlock(&consumer_data.lock);
pthread_mutex_unlock(&stream->lock);
pthread_mutex_unlock(&stream->chan->lock);
pthread_mutex_unlock(&consumer_data.lock);
@@
-2099,7
+2163,7
@@
int consumer_add_metadata_stream(struct lttng_consumer_stream *stream)
lttng_ht_add_unique_u64(ht, &stream->node);
lttng_ht_add_unique_u64(ht, &stream->node);
- lttng_ht_add_u
nique_u
64(consumer_data.stream_per_chan_id_ht,
+ lttng_ht_add_u64(consumer_data.stream_per_chan_id_ht,
&stream->node_channel_id);
/*
&stream->node_channel_id);
/*
@@
-2223,10
+2287,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) {
@@
-2324,7
+2388,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.
*/
@@
-2351,7
+2415,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.
*/
@@
-2403,6
+2467,8
@@
void *consumer_thread_data_poll(void *data)
struct lttng_consumer_stream **local_stream = NULL, *new_stream = NULL;
/* local view of consumer_data.fds_count */
int nb_fd = 0;
struct lttng_consumer_stream **local_stream = NULL, *new_stream = NULL;
/* local view of consumer_data.fds_count */
int nb_fd = 0;
+ /* Number of FDs with CONSUMER_ENDPOINT_INACTIVE but still open. */
+ int nb_inactive_fd = 0;
struct lttng_consumer_local_data *ctx = data;
ssize_t len;
struct lttng_consumer_local_data *ctx = data;
ssize_t len;
@@
-2459,7
+2525,7
@@
void *consumer_thread_data_poll(void *data)
goto end;
}
ret = update_poll_array(ctx, &pollfd, local_stream,
goto end;
}
ret = update_poll_array(ctx, &pollfd, local_stream,
- data_ht);
+ data_ht
, &nb_inactive_fd
);
if (ret < 0) {
ERR("Error in allocating pollfd or local_outfds");
lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_POLL_ERROR);
if (ret < 0) {
ERR("Error in allocating pollfd or local_outfds");
lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_POLL_ERROR);
@@
-2472,13
+2538,17
@@
void *consumer_thread_data_poll(void *data)
pthread_mutex_unlock(&consumer_data.lock);
/* No FDs and consumer_quit, consumer_cleanup the thread */
pthread_mutex_unlock(&consumer_data.lock);
/* No FDs and consumer_quit, consumer_cleanup the thread */
- if (nb_fd == 0 && consumer_quit == 1) {
+ if (nb_fd == 0 && nb_inactive_fd == 0 &&
+ CMM_LOAD_SHARED(consumer_quit) == 1) {
err = 0; /* All is OK */
goto end;
}
/* poll on the array of fds */
restart:
DBG("polling on %d fd", nb_fd + 2);
err = 0; /* All is OK */
goto end;
}
/* poll on the array of fds */
restart:
DBG("polling on %d fd", nb_fd + 2);
+ if (testpoint(consumerd_thread_data_poll)) {
+ goto end;
+ }
health_poll_entry();
num_rdy = poll(pollfd, nb_fd + 2, -1);
health_poll_exit();
health_poll_entry();
num_rdy = poll(pollfd, nb_fd + 2, -1);
health_poll_exit();
@@
-2498,6
+2568,12
@@
void *consumer_thread_data_poll(void *data)
goto end;
}
goto end;
}
+ if (caa_unlikely(data_consumption_paused)) {
+ DBG("Data consumption paused, sleeping...");
+ sleep(1);
+ goto restart;
+ }
+
/*
* If the consumer_data_pipe triggered poll go directly to the
* beginning of the loop to update the array. We want to prioritize
/*
* If the consumer_data_pipe triggered poll go directly to the
* beginning of the loop to update the array. We want to prioritize
@@
-2806,10
+2882,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) {
@@
-3140,7
+3216,7
@@
void *consumer_thread_sessiond_poll(void *data)
err = 0;
goto end;
}
err = 0;
goto end;
}
- if (
consumer_quit
) {
+ if (
CMM_LOAD_SHARED(consumer_quit)
) {
DBG("consumer_thread_receive_fds received quit from signal");
err = 0; /* All is OK */
goto end;
DBG("consumer_thread_receive_fds received quit from signal");
err = 0; /* All is OK */
goto end;
@@
-3165,7
+3241,7
@@
end:
* when all fds have hung up, the polling thread
* can exit cleanly
*/
* when all fds have hung up, the polling thread
* can exit cleanly
*/
-
consumer_quit = 1
;
+
CMM_STORE_SHARED(consumer_quit, 1)
;
/*
* Notify the data poll thread to poll back again and test the
/*
* Notify the data poll thread to poll back again and test the
@@
-3297,7
+3373,7
@@
error:
* This will create a relayd socket pair and add it to the relayd hash table.
* The caller MUST acquire a RCU read side lock before calling it.
*/
* This will create a relayd socket pair and add it to the relayd hash table.
* The caller MUST acquire a RCU read side lock before calling it.
*/
-
int
consumer_add_relayd_socket(uint64_t net_seq_idx, int sock_type,
+
void
consumer_add_relayd_socket(uint64_t net_seq_idx, int sock_type,
struct lttng_consumer_local_data *ctx, int sock,
struct pollfd *consumer_sockpoll,
struct lttcomm_relayd_sock *relayd_sock, uint64_t sessiond_id,
struct lttng_consumer_local_data *ctx, int sock,
struct pollfd *consumer_sockpoll,
struct lttcomm_relayd_sock *relayd_sock, uint64_t sessiond_id,
@@
-3319,7
+3395,6
@@
int consumer_add_relayd_socket(uint64_t net_seq_idx, int sock_type,
/* Not found. Allocate one. */
relayd = consumer_allocate_relayd_sock_pair(net_seq_idx);
if (relayd == NULL) {
/* Not found. Allocate one. */
relayd = consumer_allocate_relayd_sock_pair(net_seq_idx);
if (relayd == NULL) {
- ret = -ENOMEM;
ret_code = LTTCOMM_CONSUMERD_ENOMEM;
goto error;
} else {
ret_code = LTTCOMM_CONSUMERD_ENOMEM;
goto error;
} else {
@@
-3352,14
+3427,12
@@
int consumer_add_relayd_socket(uint64_t net_seq_idx, int sock_type,
if (ret) {
/* Needing to exit in the middle of a command: error. */
lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_POLL_ERROR);
if (ret) {
/* Needing to exit in the middle of a command: error. */
lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_POLL_ERROR);
- ret = -EINTR;
goto error_nosignal;
}
/* Get relayd socket from session daemon */
ret = lttcomm_recv_fds_unix_sock(sock, &fd, 1);
if (ret != sizeof(fd)) {
goto error_nosignal;
}
/* Get relayd socket from session daemon */
ret = lttcomm_recv_fds_unix_sock(sock, &fd, 1);
if (ret != sizeof(fd)) {
- ret = -1;
fd = -1; /* Just in case it gets set with an invalid value. */
/*
fd = -1; /* Just in case it gets set with an invalid value. */
/*
@@
-3433,7
+3506,6
@@
int consumer_add_relayd_socket(uint64_t net_seq_idx, int sock_type,
break;
default:
ERR("Unknown relayd socket type (%d)", sock_type);
break;
default:
ERR("Unknown relayd socket type (%d)", sock_type);
- ret = -1;
ret_code = LTTCOMM_CONSUMERD_FATAL;
goto error;
}
ret_code = LTTCOMM_CONSUMERD_FATAL;
goto error;
}
@@
-3454,10
+3526,11
@@
int consumer_add_relayd_socket(uint64_t net_seq_idx, int sock_type,
* Add relayd socket pair to consumer data hashtable. If object already
* exists or on error, the function gracefully returns.
*/
* Add relayd socket pair to consumer data hashtable. If object already
* exists or on error, the function gracefully returns.
*/
+ relayd->ctx = ctx;
add_relayd(relayd);
/* All good! */
add_relayd(relayd);
/* All good! */
- return
0
;
+ return;
error:
if (consumer_send_status_msg(sock, ret_code) < 0) {
error:
if (consumer_send_status_msg(sock, ret_code) < 0) {
@@
-3475,8
+3548,6
@@
error_nosignal:
if (relayd_created) {
free(relayd);
}
if (relayd_created) {
free(relayd);
}
-
- return ret;
}
/*
}
/*
@@
-3582,6
+3653,8
@@
int consumer_data_pending(uint64_t id)
pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
if (ret < 0) {
/* Communication error thus the relayd so no data pending. */
pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
if (ret < 0) {
/* Communication error thus the relayd so no data pending. */
+ ERR("Relayd begin data pending failed. Cleaning up relayd %" PRIu64".", relayd->net_seq_idx);
+ lttng_consumer_cleanup_relayd(relayd);
goto data_not_pending;
}
}
goto data_not_pending;
}
}
@@
-3623,6
+3696,13
@@
int consumer_data_pending(uint64_t id)
stream->relayd_stream_id,
stream->next_net_seq_num - 1);
}
stream->relayd_stream_id,
stream->next_net_seq_num - 1);
}
+ if (ret < 0) {
+ ERR("Relayd data pending failed. Cleaning up relayd %" PRIu64".", relayd->net_seq_idx);
+ lttng_consumer_cleanup_relayd(relayd);
+ pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
+ pthread_mutex_unlock(&stream->lock);
+ goto data_not_pending;
+ }
pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
if (ret == 1) {
pthread_mutex_unlock(&stream->lock);
pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
if (ret == 1) {
pthread_mutex_unlock(&stream->lock);
@@
-3641,6
+3721,8
@@
int consumer_data_pending(uint64_t id)
relayd->relayd_session_id, &is_data_inflight);
pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
if (ret < 0) {
relayd->relayd_session_id, &is_data_inflight);
pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
if (ret < 0) {
+ ERR("Relayd end data pending failed. Cleaning up relayd %" PRIu64".", relayd->net_seq_idx);
+ lttng_consumer_cleanup_relayd(relayd);
goto data_not_pending;
}
if (is_data_inflight) {
goto data_not_pending;
}
if (is_data_inflight) {
This page took
0.033685 seconds
and
4
git commands to generate.