X-Git-Url: https://git.lttng.org/?p=lttng-tools.git;a=blobdiff_plain;f=src%2Fcommon%2Fconsumer%2Fconsumer.c;h=be440e69492b020b4281f83cc94c12fe29c9c38f;hp=d70a5d1d406b758e1b02cc61f22b440ee006a7b1;hb=04ed9e10dfa0b3c88d4a7abe9fa59b8e03b7e49a;hpb=fd424d99bfddf3581bc55fdf14799a50661beddd diff --git a/src/common/consumer/consumer.c b/src/common/consumer/consumer.c index d70a5d1d4..be440e694 100644 --- a/src/common/consumer/consumer.c +++ b/src/common/consumer/consumer.c @@ -7,6 +7,7 @@ * */ +#include "common/index/ctf-index.h" #define _LGPL_SOURCE #include #include @@ -62,6 +63,12 @@ struct consumer_channel_msg { uint64_t key; /* del */ }; +enum open_packet_status { + OPEN_PACKET_STATUS_OPENED, + OPEN_PACKET_STATUS_NO_SPACE, + OPEN_PACKET_STATUS_ERROR, +}; + /* Flag used to temporarily pause data consumption from testpoints. */ int data_consumption_paused; @@ -568,95 +575,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. */ @@ -1063,6 +981,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) { @@ -1094,6 +1013,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); @@ -1462,7 +1382,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)) @@ -1669,11 +1589,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; @@ -1765,10 +1683,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; } @@ -1777,7 +1691,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 @@ -1845,8 +1759,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; @@ -1950,7 +1863,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 +2429,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 +2456,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 +2672,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 +2703,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. */ @@ -3404,40 +3316,296 @@ error_testpoint: return NULL; } -ssize_t lttng_consumer_read_subbuffer(struct lttng_consumer_stream *stream, - struct lttng_consumer_local_data *ctx) +static +int consumer_flush_buffer(struct lttng_consumer_stream *stream, + int producer_active) { - ssize_t ret; - - pthread_mutex_lock(&stream->chan->lock); - pthread_mutex_lock(&stream->lock); - if (stream->metadata_flag) { - pthread_mutex_lock(&stream->metadata_rdv_lock); - } + int ret = 0; switch (consumer_data.type) { case LTTNG_CONSUMER_KERNEL: - ret = lttng_kconsumer_read_subbuffer(stream, ctx); + if (producer_active) { + ret = kernctl_buffer_flush(stream->wait_fd); + if (ret < 0) { + ERR("Failed to flush kernel stream"); + goto end; + } + } else { + ret = kernctl_buffer_flush_empty(stream->wait_fd); + if (ret < 0) { + /* + * Doing a buffer flush which does not take into + * account empty packets. This is not perfect, + * but required as a fall-back when + * "flush_empty" is not implemented by + * lttng-modules. + */ + ret = kernctl_buffer_flush(stream->wait_fd); + if (ret < 0) { + ERR("Failed to flush kernel stream"); + goto end; + } + } + } break; case LTTNG_CONSUMER32_UST: case LTTNG_CONSUMER64_UST: - ret = lttng_ustconsumer_read_subbuffer(stream, ctx); + lttng_ustconsumer_flush_buffer(stream, producer_active); break; default: ERR("Unknown consumer_data type"); - assert(0); - ret = -ENOSYS; - break; + abort(); } - if (stream->metadata_flag) { - pthread_cond_broadcast(&stream->metadata_rdv); - pthread_mutex_unlock(&stream->metadata_rdv_lock); +end: + return ret; +} + +static enum open_packet_status open_packet(struct lttng_consumer_stream *stream) +{ + int ret; + enum open_packet_status status; + unsigned long produced_pos_before, produced_pos_after; + + ret = lttng_consumer_sample_snapshot_positions(stream); + if (ret < 0) { + ERR("Failed to snapshot positions before post-rotation empty packet flush: stream id = %" PRIu64 + ", channel name = %s, session id = %" PRIu64, + stream->key, stream->chan->name, + stream->chan->session_id); + status = OPEN_PACKET_STATUS_ERROR; + goto end; + } + + ret = lttng_consumer_get_produced_snapshot( + stream, &produced_pos_before); + if (ret < 0) { + ERR("Failed to read produced position before post-rotation empty packet flush: stream id = %" PRIu64 + ", channel name = %s, session id = %" PRIu64, + stream->key, stream->chan->name, + stream->chan->session_id); + status = OPEN_PACKET_STATUS_ERROR; + goto end; + } + + ret = consumer_flush_buffer(stream, 0); + if (ret) { + ERR("Failed to flush an empty packet at rotation point: stream id = %" PRIu64 + ", channel name = %s, session id = %" PRIu64, + stream->key, stream->chan->name, + stream->chan->session_id); + status = OPEN_PACKET_STATUS_ERROR; + goto end; + } + + ret = lttng_consumer_sample_snapshot_positions(stream); + if (ret < 0) { + ERR("Failed to snapshot positions after post-rotation empty packet flush: stream id = %" PRIu64 + ", channel name = %s, session id = %" PRIu64, + stream->key, stream->chan->name, + stream->chan->session_id); + status = OPEN_PACKET_STATUS_ERROR; + goto end; + } + + ret = lttng_consumer_get_produced_snapshot(stream, &produced_pos_after); + if (ret < 0) { + ERR("Failed to read produced position after post-rotation empty packet flush: stream id = %" PRIu64 + ", channel name = %s, session id = %" PRIu64, + stream->key, stream->chan->name, + stream->chan->session_id); + status = OPEN_PACKET_STATUS_ERROR; + goto end; + } + + /* + * Determine if the flush had an effect by comparing the produced + * positons before and after the flush. + */ + status = produced_pos_before != produced_pos_after ? + OPEN_PACKET_STATUS_OPENED : + OPEN_PACKET_STATUS_NO_SPACE; + if (status == OPEN_PACKET_STATUS_OPENED) { + stream->opened_packet_in_current_trace_chunk = true; + } +end: + return status; +} + +static bool stream_is_rotating_to_null_chunk( + const struct lttng_consumer_stream *stream) +{ + bool rotating_to_null_chunk = false; + + if (stream->rotate_position == -1ULL) { + /* No rotation ongoing. */ + goto end; + } + + if (stream->trace_chunk == stream->chan->trace_chunk || + !stream->chan->trace_chunk) { + rotating_to_null_chunk = true; + } +end: + return rotating_to_null_chunk; +} + +ssize_t lttng_consumer_read_subbuffer(struct lttng_consumer_stream *stream, + struct lttng_consumer_local_data *ctx, + bool locked_by_caller) +{ + ssize_t ret, written_bytes = 0; + int rotation_ret; + struct stream_subbuffer subbuffer = {}; + + if (!locked_by_caller) { + stream->read_subbuffer_ops.lock(stream); + } + + if (stream->read_subbuffer_ops.on_wake_up) { + ret = stream->read_subbuffer_ops.on_wake_up(stream); + if (ret) { + goto end; + } + } + + /* + * 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); + /* + * Should write subbuf_size amount of data when network streaming or + * the full padded size when we are not streaming. + */ + if ((written_bytes != subbuffer.info.data.subbuf_size && + stream->net_seq_idx != (uint64_t) -1ULL) || + (written_bytes != subbuffer.info.data.padded_subbuf_size && + stream->net_seq_idx == + (uint64_t) -1ULL)) { + /* + * Display the error but continue processing to try to + * release the subbuffer. This is a DBG statement + * since this can happen without being a critical + * error. + */ + DBG("Failed to write to tracefile (written_bytes: %zd != padded subbuffer size: %lu, subbuffer size: %lu)", + written_bytes, subbuffer.info.data.padded_subbuf_size, + subbuffer.info.data.subbuf_size); + } + + 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; + } + + /* + * TODO roll into a post_consume op as this doesn't apply to metadata + * streams. + */ + if (!stream->opened_packet_in_current_trace_chunk && + stream->trace_chunk && !stream->metadata_flag && + !stream_is_rotating_to_null_chunk(stream)) { + const enum open_packet_status status = open_packet(stream); + + switch (status) { + case OPEN_PACKET_STATUS_OPENED: + DBG("Opened a packet after consuming a packet rotation: stream id = %" PRIu64 + ", channel name = %s, session id = %" PRIu64, + stream->key, stream->chan->name, + stream->chan->session_id); + break; + case OPEN_PACKET_STATUS_NO_SPACE: + /* + * Can't open a packet as there is no space left. + * This means that new events were produced, resulting + * in a packet being opened, which is what we wanted + * anyhow. + */ + DBG("No space left to open a packet after consuming a packet: stream id = %" PRIu64 + ", channel name = %s, session id = %" PRIu64, + stream->key, stream->chan->name, + stream->chan->session_id); + stream->opened_packet_in_current_trace_chunk = true; + break; + case OPEN_PACKET_STATUS_ERROR: + /* Logged by callee. */ + ret = -1; + goto end; + default: + abort(); + } + } + +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) @@ -3911,50 +4079,6 @@ unsigned long consumer_get_consume_start_pos(unsigned long consumed_pos, return start_pos; } -static -int consumer_flush_buffer(struct lttng_consumer_stream *stream, int producer_active) -{ - int ret = 0; - - switch (consumer_data.type) { - case LTTNG_CONSUMER_KERNEL: - if (producer_active) { - ret = kernctl_buffer_flush(stream->wait_fd); - if (ret < 0) { - ERR("Failed to flush kernel stream"); - goto end; - } - } else { - ret = kernctl_buffer_flush_empty(stream->wait_fd); - if (ret < 0) { - /* - * Doing a buffer flush which does not take into - * account empty packets. This is not perfect, - * but required as a fall-back when - * "flush_empty" is not implemented by - * lttng-modules. - */ - ret = kernctl_buffer_flush(stream->wait_fd); - if (ret < 0) { - ERR("Failed to flush kernel stream"); - goto end; - } - } - } - break; - case LTTNG_CONSUMER32_UST: - case LTTNG_CONSUMER64_UST: - lttng_ustconsumer_flush_buffer(stream, producer_active); - break; - default: - ERR("Unknown consumer_data type"); - abort(); - } - -end: - return ret; -} - /* * Sample the rotate position for all the streams of a channel. If a stream * is already at the rotate position (produced == consumed), we flag it as @@ -4112,6 +4236,84 @@ int lttng_consumer_rotate_channel(struct lttng_consumer_channel *channel, } stream_count++; } + + stream->opened_packet_in_current_trace_chunk = false; + + if (rotating_to_new_chunk && !stream->metadata_flag) { + /* + * Attempt to flush an empty packet as close to the + * rotation point as possible. In the event where a + * stream remains inactive after the rotation point, + * this ensures that the new trace chunk has a + * beginning timestamp set at the begining of the + * trace chunk instead of only creating an empty + * packet when the trace chunk is stopped. + * + * This indicates to the viewers that the stream + * was being recorded, but more importantly it + * allows viewers to determine a useable trace + * intersection. + * + * This presents a problem in the case where the + * ring-buffer is completely full. + * + * Consider the following scenario: + * - The consumption of data is slow (slow network, + * for instance), + * - The ring buffer is full, + * - A rotation is initiated, + * - The flush below does nothing (no space left to + * open a new packet), + * - The other streams rotate very soon, and new + * data is produced in the new chunk, + * - This stream completes its rotation long after the + * rotation was initiated + * - The session is stopped before any event can be + * produced in this stream's buffers. + * + * The resulting trace chunk will have a single packet + * temporaly at the end of the trace chunk for this + * stream making the stream intersection more narrow + * than it should be. + * + * To work-around this, an empty flush is performed + * after the first consumption of a packet during a + * rotation if open_packet fails. The idea is that + * consuming a packet frees enough space to switch + * packets in this scenario and allows the tracer to + * "stamp" the beginning of the new trace chunk at the + * earliest possible point. + */ + const enum open_packet_status status = + open_packet(stream); + + switch (status) { + case OPEN_PACKET_STATUS_OPENED: + DBG("Opened a packet after a rotation: stream id = %" PRIu64 + ", channel name = %s, session id = %" PRIu64, + stream->key, stream->chan->name, + stream->chan->session_id); + break; + case OPEN_PACKET_STATUS_NO_SPACE: + /* + * Can't open a packet as there is no space left + * in the buffer. A new packet will be opened + * once one has been consumed. + */ + DBG("No space left to open a packet after a rotation: stream id = %" PRIu64 + ", channel name = %s, session id = %" PRIu64, + stream->key, stream->chan->name, + stream->chan->session_id); + break; + case OPEN_PACKET_STATUS_ERROR: + /* Logged by callee. */ + ret = -1; + goto end_unlock_stream; + default: + abort(); + } + } + pthread_mutex_unlock(&stream->lock); } stream = NULL; @@ -4967,3 +5169,70 @@ int lttng_consumer_clear_channel(struct lttng_consumer_channel *channel) end: return ret; } + +enum lttcomm_return_code lttng_consumer_open_channel_packets( + struct lttng_consumer_channel *channel) +{ + struct lttng_consumer_stream *stream; + enum lttcomm_return_code ret = LTTCOMM_CONSUMERD_SUCCESS; + + if (channel->metadata_stream) { + ERR("Open channel packets command attempted on a metadata channel"); + ret = LTTCOMM_CONSUMERD_INVALID_PARAMETERS; + goto end; + } + + rcu_read_lock(); + cds_list_for_each_entry(stream, &channel->streams.head, send_node) { + enum open_packet_status status; + + pthread_mutex_lock(&stream->lock); + if (cds_lfht_is_node_deleted(&stream->node.node)) { + goto next; + } + + status = open_packet(stream); + switch (status) { + case OPEN_PACKET_STATUS_OPENED: + DBG("Opened a packet in \"open channel packets\" command: stream id = %" PRIu64 + ", channel name = %s, session id = %" PRIu64, + stream->key, stream->chan->name, + stream->chan->session_id); + stream->opened_packet_in_current_trace_chunk = true; + break; + case OPEN_PACKET_STATUS_NO_SPACE: + DBG("No space left to open a packet in \"open channel packets\" command: stream id = %" PRIu64 + ", channel name = %s, session id = %" PRIu64, + stream->key, stream->chan->name, + stream->chan->session_id); + break; + case OPEN_PACKET_STATUS_ERROR: + /* + * Only unexpected internal errors can lead to this + * failing. Report an unknown error. + */ + ERR("Failed to flush empty buffer in \"open channel packets\" command: stream id = %" PRIu64 + ", channel id = %" PRIu64 + ", channel name = %s" + ", session id = %" PRIu64, + stream->key, channel->key, + channel->name, channel->session_id); + ret = LTTCOMM_CONSUMERD_UNKNOWN_ERROR; + goto error_unlock; + default: + abort(); + } + + next: + pthread_mutex_unlock(&stream->lock); + } + +end_rcu_unlock: + rcu_read_unlock(); +end: + return ret; + +error_unlock: + pthread_mutex_unlock(&stream->lock); + goto end_rcu_unlock; +}