#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,
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.
*/
-volatile int consumer_quit;
+int consumer_quit;
/*
* Global hash table containing respectively metadata and data streams. The
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:
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);
CDS_INIT_LIST_HEAD(&channel->streams.head);
- DBG("Allocated channel (key %" PRIu64 ")", channel->key)
+ DBG("Allocated channel (key %" PRIu64 ")", channel->key);
end:
return channel;
{
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");
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)
{
+ int ret;
int outfd = stream->out_fd;
/*
* 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);
+ if (ret && ret != -ENOSYS) {
+ errno = ret;
+ PERROR("posix_fadvise on fd %i", outfd);
+ }
}
/*
goto error_metadata_pipe;
}
+ ctx->channel_monitor_pipe = -1;
+
return ctx;
error_metadata_pipe:
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;
}
}
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->tracefile_count_current);
- if (ret < 0) {
+ stream->tracefile_count_current,
+ CTF_INDEX_MAJOR, CTF_INDEX_MINOR);
+ if (!stream->index_file) {
goto end;
}
- stream->index_fd = ret;
}
/* Reset current size because we just perform a rotation. */
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:
/*
}
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->tracefile_count_current);
- if (ret < 0) {
- written = ret;
+ stream->tracefile_count_current,
+ CTF_INDEX_MAJOR, CTF_INDEX_MINOR);
+ if (!stream->index_file) {
goto end;
}
- stream->index_fd = ret;
}
/* Reset current size because we just perform a rotation. */
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:
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);
*/
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);
DBG("Metadata poll return from wait with %d fd(s)",
LTTNG_POLL_GETNB(&events));
health_poll_exit();
- DBG("Metadata event catched in thread");
+ DBG("Metadata event caught in thread");
if (ret < 0) {
if (errno == EINTR) {
- ERR("Poll EINTR catched");
+ ERR("Poll EINTR caught");
goto restart;
}
if (LTTNG_POLL_GETNB(&events) == 0) {
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 occurred thus we
* simply remove it from the poll set and free the
* stream.
*/
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 occurred thus we
* simply remove it from the poll set and free the
* stream.
*/
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 && 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);
+ if (testpoint(consumerd_thread_data_poll)) {
+ goto end;
+ }
health_poll_entry();
num_rdy = poll(pollfd, nb_fd + 2, -1);
health_poll_exit();
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
DBG("Channel poll return from wait with %d fd(s)",
LTTNG_POLL_GETNB(&events));
health_poll_exit();
- DBG("Channel event catched in thread");
+ DBG("Channel event caught in thread");
if (ret < 0) {
if (errno == EINTR) {
- ERR("Poll EINTR catched");
+ ERR("Poll EINTR caught");
goto restart;
}
if (LTTNG_POLL_GETNB(&events) == 0) {
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;
* 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
projects / lttng-tools.git / blobdiff