X-Git-Url: https://git.liburcu.org/?a=blobdiff_plain;f=src%2Fcommon%2Fconsumer%2Fconsumer.c;h=9709705b733b17fe2c53970d60482e964a2027df;hb=c124970bc6f688ba8e13cf21cc9933e291cbb76a;hp=6525a7203130e7bbf3ce3a97cba489735029a61b;hpb=b770aa7ff007d18a53a3293a396880ad7ce42c8e;p=lttng-tools.git diff --git a/src/common/consumer/consumer.c b/src/common/consumer/consumer.c index 6525a7203..9709705b7 100644 --- a/src/common/consumer/consumer.c +++ b/src/common/consumer/consumer.c @@ -17,6 +17,7 @@ * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA. */ +#include "common/index/ctf-index.h" #define _LGPL_SOURCE #include #include @@ -564,95 +565,6 @@ void consumer_stream_update_channel_attributes( channel->tracefile_size; } -struct lttng_consumer_stream *consumer_allocate_stream(uint64_t channel_key, - uint64_t stream_key, - const char *channel_name, - uint64_t relayd_id, - uint64_t session_id, - struct lttng_trace_chunk *trace_chunk, - int cpu, - int *alloc_ret, - enum consumer_channel_type type, - unsigned int monitor) -{ - int ret; - struct lttng_consumer_stream *stream; - - stream = zmalloc(sizeof(*stream)); - if (stream == NULL) { - PERROR("malloc struct lttng_consumer_stream"); - ret = -ENOMEM; - goto end; - } - - if (trace_chunk && !lttng_trace_chunk_get(trace_chunk)) { - ERR("Failed to acquire trace chunk reference during the creation of a stream"); - ret = -1; - goto error; - } - - rcu_read_lock(); - stream->key = stream_key; - stream->trace_chunk = trace_chunk; - stream->out_fd = -1; - stream->out_fd_offset = 0; - stream->output_written = 0; - stream->net_seq_idx = relayd_id; - stream->session_id = session_id; - stream->monitor = monitor; - stream->endpoint_status = CONSUMER_ENDPOINT_ACTIVE; - stream->index_file = NULL; - stream->last_sequence_number = -1ULL; - stream->rotate_position = -1ULL; - pthread_mutex_init(&stream->lock, NULL); - pthread_mutex_init(&stream->metadata_timer_lock, NULL); - - /* If channel is the metadata, flag this stream as metadata. */ - if (type == CONSUMER_CHANNEL_TYPE_METADATA) { - stream->metadata_flag = 1; - /* Metadata is flat out. */ - strncpy(stream->name, DEFAULT_METADATA_NAME, sizeof(stream->name)); - /* Live rendez-vous point. */ - pthread_cond_init(&stream->metadata_rdv, NULL); - pthread_mutex_init(&stream->metadata_rdv_lock, NULL); - } else { - /* Format stream name to _ */ - ret = snprintf(stream->name, sizeof(stream->name), "%s_%d", - channel_name, cpu); - if (ret < 0) { - PERROR("snprintf stream name"); - goto error; - } - } - - /* Key is always the wait_fd for streams. */ - lttng_ht_node_init_u64(&stream->node, stream->key); - - /* Init node per channel id key */ - lttng_ht_node_init_u64(&stream->node_channel_id, channel_key); - - /* Init session id node with the stream session id */ - lttng_ht_node_init_u64(&stream->node_session_id, stream->session_id); - - DBG3("Allocated stream %s (key %" PRIu64 ", chan_key %" PRIu64 - " relayd_id %" PRIu64 ", session_id %" PRIu64, - stream->name, stream->key, channel_key, - stream->net_seq_idx, stream->session_id); - - rcu_read_unlock(); - return stream; - -error: - rcu_read_unlock(); - lttng_trace_chunk_put(stream->trace_chunk); - free(stream); -end: - if (alloc_ret) { - *alloc_ret = ret; - } - return NULL; -} - /* * Add a stream to the global list protected by a mutex. */ @@ -966,6 +878,43 @@ error: return outfd; } +/* + * Write a character on the metadata poll pipe to wake the metadata thread. + * Returns 0 on success, -1 on error. + */ +int consumer_metadata_wakeup_pipe(const struct lttng_consumer_channel *channel) +{ + int ret = 0; + + DBG("Waking up metadata poll thread (writing to pipe): channel name = '%s'", + channel->name); + if (channel->monitor && channel->metadata_stream) { + const char dummy = 'c'; + const ssize_t write_ret = lttng_write( + channel->metadata_stream->ust_metadata_poll_pipe[1], + &dummy, 1); + + if (write_ret < 1) { + if (errno == EWOULDBLOCK) { + /* + * This is fine, the metadata poll thread + * is having a hard time keeping-up, but + * it will eventually wake-up and consume + * the available data. + */ + ret = 0; + } else { + PERROR("Failed to write to UST metadata pipe while attempting to wake-up the metadata poll thread"); + ret = -1; + goto end; + } + } + } + +end: + return ret; +} + /* * Trigger a dump of the metadata content. Following/during the succesful * completion of this call, the metadata poll thread will start receiving @@ -1064,6 +1013,7 @@ struct lttng_consumer_channel *consumer_allocate_channel(uint64_t key, uint64_t session_id_per_pid, unsigned int monitor, unsigned int live_timer_interval, + bool is_in_live_session, const char *root_shm_path, const char *shm_path) { @@ -1095,6 +1045,7 @@ struct lttng_consumer_channel *consumer_allocate_channel(uint64_t key, channel->tracefile_count = tracefile_count; channel->monitor = monitor; channel->live_timer_interval = live_timer_interval; + channel->is_live = is_in_live_session; pthread_mutex_init(&channel->lock, NULL); pthread_mutex_init(&channel->timer_lock, NULL); @@ -1463,7 +1414,7 @@ void lttng_consumer_sync_trace_file(struct lttng_consumer_stream *stream, struct lttng_consumer_local_data *lttng_consumer_create( enum lttng_consumer_type type, ssize_t (*buffer_ready)(struct lttng_consumer_stream *stream, - struct lttng_consumer_local_data *ctx), + struct lttng_consumer_local_data *ctx, bool locked_by_caller), int (*recv_channel)(struct lttng_consumer_channel *channel), int (*recv_stream)(struct lttng_consumer_stream *stream), int (*update_stream)(uint64_t stream_key, uint32_t state)) @@ -1670,11 +1621,9 @@ end: * Returns the number of bytes written */ ssize_t lttng_consumer_on_read_subbuffer_mmap( - struct lttng_consumer_local_data *ctx, struct lttng_consumer_stream *stream, const struct lttng_buffer_view *buffer, - unsigned long padding, - struct ctf_packet_index *index) + unsigned long padding) { ssize_t ret = 0; off_t orig_offset = stream->out_fd_offset; @@ -1766,10 +1715,6 @@ ssize_t lttng_consumer_on_read_subbuffer_mmap( orig_offset = 0; } stream->tracefile_size_current += buffer->size; - if (index) { - index->offset = htobe64(stream->out_fd_offset); - } - write_len = buffer->size; } @@ -1778,7 +1723,7 @@ ssize_t lttng_consumer_on_read_subbuffer_mmap( * receive a ret value that is bigger than len. */ ret = lttng_write(outfd, buffer->data, write_len); - DBG("Consumer mmap write() ret %zd (len %lu)", ret, write_len); + DBG("Consumer mmap write() ret %zd (len %zu)", ret, write_len); if (ret < 0 || ((size_t) ret != write_len)) { /* * Report error to caller if nothing was written else at least send the @@ -1846,8 +1791,7 @@ end: ssize_t lttng_consumer_on_read_subbuffer_splice( struct lttng_consumer_local_data *ctx, struct lttng_consumer_stream *stream, unsigned long len, - unsigned long padding, - struct ctf_packet_index *index) + unsigned long padding) { ssize_t ret = 0, written = 0, ret_splice = 0; loff_t offset = 0; @@ -1951,7 +1895,6 @@ ssize_t lttng_consumer_on_read_subbuffer_splice( orig_offset = 0; } stream->tracefile_size_current += len; - index->offset = htobe64(stream->out_fd_offset); } while (len > 0) { @@ -2517,7 +2460,7 @@ restart: do { health_code_update(); - len = ctx->on_buffer_ready(stream, ctx); + len = ctx->on_buffer_ready(stream, ctx, false); /* * We don't check the return value here since if we get * a negative len, it means an error occurred thus we @@ -2544,7 +2487,7 @@ restart: do { health_code_update(); - len = ctx->on_buffer_ready(stream, ctx); + len = ctx->on_buffer_ready(stream, ctx, false); /* * We don't check the return value here since if we get * a negative len, it means an error occurred thus we @@ -2760,7 +2703,7 @@ void *consumer_thread_data_poll(void *data) if (pollfd[i].revents & POLLPRI) { DBG("Urgent read on fd %d", pollfd[i].fd); high_prio = 1; - len = ctx->on_buffer_ready(local_stream[i], ctx); + len = ctx->on_buffer_ready(local_stream[i], ctx, false); /* it's ok to have an unavailable sub-buffer */ if (len < 0 && len != -EAGAIN && len != -ENODATA) { /* Clean the stream and free it. */ @@ -2791,7 +2734,7 @@ void *consumer_thread_data_poll(void *data) local_stream[i]->hangup_flush_done || local_stream[i]->has_data) { DBG("Normal read on fd %d", pollfd[i].fd); - len = ctx->on_buffer_ready(local_stream[i], ctx); + len = ctx->on_buffer_ready(local_stream[i], ctx, false); /* it's ok to have an unavailable sub-buffer */ if (len < 0 && len != -EAGAIN && len != -ENODATA) { /* Clean the stream and free it. */ @@ -3347,7 +3290,7 @@ void *consumer_thread_sessiond_poll(void *data) err = 0; /* All is OK */ goto end; } - DBG("received command on sock"); + DBG("Received command on sock"); } /* All is OK */ err = 0; @@ -3405,39 +3348,109 @@ error_testpoint: } ssize_t lttng_consumer_read_subbuffer(struct lttng_consumer_stream *stream, - struct lttng_consumer_local_data *ctx) + struct lttng_consumer_local_data *ctx, + bool locked_by_caller) { - ssize_t ret; + ssize_t ret, written_bytes = 0; + int rotation_ret; + struct stream_subbuffer subbuffer = {}; - pthread_mutex_lock(&stream->chan->lock); - pthread_mutex_lock(&stream->lock); - if (stream->metadata_flag) { - pthread_mutex_lock(&stream->metadata_rdv_lock); + if (!locked_by_caller) { + stream->read_subbuffer_ops.lock(stream); } - switch (consumer_data.type) { - case LTTNG_CONSUMER_KERNEL: - ret = lttng_kconsumer_read_subbuffer(stream, ctx); - break; - case LTTNG_CONSUMER32_UST: - case LTTNG_CONSUMER64_UST: - ret = lttng_ustconsumer_read_subbuffer(stream, ctx); - break; - default: - ERR("Unknown consumer_data type"); - assert(0); - ret = -ENOSYS; - break; + if (stream->read_subbuffer_ops.on_wake_up) { + ret = stream->read_subbuffer_ops.on_wake_up(stream); + if (ret) { + goto end; + } } - if (stream->metadata_flag) { - pthread_cond_broadcast(&stream->metadata_rdv); - pthread_mutex_unlock(&stream->metadata_rdv_lock); + /* + * If the stream was flagged to be ready for rotation before we extract + * the next packet, rotate it now. + */ + if (stream->rotate_ready) { + DBG("Rotate stream before consuming data"); + ret = lttng_consumer_rotate_stream(ctx, stream); + if (ret < 0) { + ERR("Stream rotation error before consuming data"); + goto end; + } + } + + ret = stream->read_subbuffer_ops.get_next_subbuffer(stream, &subbuffer); + if (ret) { + if (ret == -ENODATA) { + /* Not an error. */ + ret = 0; + goto sleep_stream; + } + goto end; + } + + ret = stream->read_subbuffer_ops.pre_consume_subbuffer( + stream, &subbuffer); + if (ret) { + goto error_put_subbuf; + } + + written_bytes = stream->read_subbuffer_ops.consume_subbuffer( + ctx, stream, &subbuffer); + if (written_bytes <= 0) { + ERR("Error consuming subbuffer: (%zd)", written_bytes); + ret = (int) written_bytes; + goto error_put_subbuf; + } + + ret = stream->read_subbuffer_ops.put_next_subbuffer(stream, &subbuffer); + if (ret) { + goto end; + } + + if (stream->read_subbuffer_ops.post_consume) { + ret = stream->read_subbuffer_ops.post_consume(stream, &subbuffer, ctx); + if (ret) { + goto end; + } + } + + /* + * After extracting the packet, we check if the stream is now ready to + * be rotated and perform the action immediately. + * + * Don't overwrite `ret` as callers expect the number of bytes + * consumed to be returned on success. + */ + rotation_ret = lttng_consumer_stream_is_rotate_ready(stream); + if (rotation_ret == 1) { + rotation_ret = lttng_consumer_rotate_stream(ctx, stream); + if (rotation_ret < 0) { + ret = rotation_ret; + ERR("Stream rotation error after consuming data"); + goto end; + } + } else if (rotation_ret < 0) { + ret = rotation_ret; + ERR("Failed to check if stream was ready to rotate after consuming data"); + goto end; + } + +sleep_stream: + if (stream->read_subbuffer_ops.on_sleep) { + stream->read_subbuffer_ops.on_sleep(stream, ctx); + } + + ret = written_bytes; +end: + if (!locked_by_caller) { + stream->read_subbuffer_ops.unlock(stream); } - pthread_mutex_unlock(&stream->lock); - pthread_mutex_unlock(&stream->chan->lock); return ret; +error_put_subbuf: + (void) stream->read_subbuffer_ops.put_next_subbuffer(stream, &subbuffer); + goto end; } int lttng_consumer_on_recv_stream(struct lttng_consumer_stream *stream)