X-Git-Url: http://git.efficios.com/?a=blobdiff_plain;f=src%2Fcommon%2Fkernel-consumer%2Fkernel-consumer.c;h=2281f5d8e60dca456c648e268f2b5a9d9e74d1e5;hb=29d1a7ae5d5b46002e5883366a7b06f1d9dcc115;hp=cbd1b5aac223962275b7be3472c7f498eb761077;hpb=f261ad0ad8d93634b4a6d7581628e236aa696049;p=lttng-tools.git diff --git a/src/common/kernel-consumer/kernel-consumer.c b/src/common/kernel-consumer/kernel-consumer.c index cbd1b5aac..2281f5d8e 100644 --- a/src/common/kernel-consumer/kernel-consumer.c +++ b/src/common/kernel-consumer/kernel-consumer.c @@ -16,7 +16,7 @@ * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA. */ -#define _GNU_SOURCE +#define _LGPL_SOURCE #include #include #include @@ -39,9 +39,13 @@ #include #include #include -#include +#include #include -#include +#include +#include +#include +#include +#include #include "kernel-consumer.h" @@ -60,9 +64,12 @@ int lttng_kconsumer_take_snapshot(struct lttng_consumer_stream *stream) int infd = stream->wait_fd; ret = kernctl_snapshot(infd); - if (ret != 0) { + /* + * -EAGAIN is not an error, it just means that there is no data to + * be read. + */ + if (ret != 0 && ret != -EAGAIN) { PERROR("Getting sub-buffer snapshot."); - ret = -errno; } return ret; @@ -82,7 +89,6 @@ int lttng_kconsumer_get_produced_snapshot(struct lttng_consumer_stream *stream, ret = kernctl_snapshot_get_produced(infd, pos); if (ret != 0) { PERROR("kernctl_snapshot_get_produced"); - ret = -errno; } return ret; @@ -102,23 +108,40 @@ int lttng_kconsumer_get_consumed_snapshot(struct lttng_consumer_stream *stream, ret = kernctl_snapshot_get_consumed(infd, pos); if (ret != 0) { PERROR("kernctl_snapshot_get_consumed"); - ret = -errno; } return ret; } +static +int get_current_subbuf_addr(struct lttng_consumer_stream *stream, + const char **addr) +{ + int ret; + unsigned long mmap_offset; + const char *mmap_base = stream->mmap_base; + + ret = kernctl_get_mmap_read_offset(stream->wait_fd, &mmap_offset); + if (ret < 0) { + PERROR("Failed to get mmap read offset"); + goto error; + } + + *addr = mmap_base + mmap_offset; +error: + return ret; +} + /* * Take a snapshot of all the stream of a channel * * Returns 0 on success, < 0 on error */ int lttng_kconsumer_snapshot_channel(uint64_t key, char *path, - uint64_t relayd_id, uint64_t max_stream_size, + uint64_t relayd_id, uint64_t nb_packets_per_stream, struct lttng_consumer_local_data *ctx) { int ret; - unsigned long consumed_pos, produced_pos; struct lttng_consumer_channel *channel; struct lttng_consumer_stream *stream; @@ -141,6 +164,7 @@ int lttng_kconsumer_snapshot_channel(uint64_t key, char *path, } cds_list_for_each_entry(stream, &channel->streams.head, send_node) { + unsigned long consumed_pos, produced_pos; health_code_update(); @@ -153,7 +177,7 @@ int lttng_kconsumer_snapshot_channel(uint64_t key, char *path, * Assign the received relayd ID so we can use it for streaming. The streams * are not visible to anyone so this is OK to change it. */ - stream->net_seq_idx = relayd_id; + stream->relayd_id = relayd_id; channel->relayd_id = relayd_id; if (relayd_id != (uint64_t) -1ULL) { ret = consumer_send_relayd_stream(stream, path); @@ -183,12 +207,23 @@ int lttng_kconsumer_snapshot_channel(uint64_t key, char *path, ERR("sending streams sent to relayd"); goto end_unlock; } + channel->streams_sent_to_relayd = true; } - ret = kernctl_buffer_flush(stream->wait_fd); + ret = kernctl_buffer_flush_empty(stream->wait_fd); if (ret < 0) { - ERR("Failed to flush kernel stream"); - ret = -errno; + /* + * Doing a buffer flush which does not take into + * account empty packets. This is not perfect + * for stream intersection, 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_unlock; + } goto end_unlock; } @@ -215,56 +250,57 @@ int lttng_kconsumer_snapshot_channel(uint64_t key, char *path, &stream->max_sb_size); if (ret < 0) { ERR("Getting kernel max_sb_size"); - ret = -errno; goto end_unlock; } } - /* - * The original value is sent back if max stream size is larger than - * the possible size of the snapshot. Also, we asume that the session - * daemon should never send a maximum stream size that is lower than - * subbuffer size. - */ - consumed_pos = consumer_get_consumed_maxsize(consumed_pos, - produced_pos, max_stream_size); + consumed_pos = consumer_get_consume_start_pos(consumed_pos, + produced_pos, nb_packets_per_stream, + stream->max_sb_size); while (consumed_pos < produced_pos) { ssize_t read_len; unsigned long len, padded_len; + const char *subbuf_addr; + struct lttng_buffer_view subbuf_view; health_code_update(); - DBG("Kernel consumer taking snapshot at pos %lu", consumed_pos); ret = kernctl_get_subbuf(stream->wait_fd, &consumed_pos); if (ret < 0) { - if (errno != EAGAIN) { + if (ret != -EAGAIN) { PERROR("kernctl_get_subbuf snapshot"); - ret = -errno; goto end_unlock; } DBG("Kernel consumer get subbuf failed. Skipping it."); consumed_pos += stream->max_sb_size; + stream->chan->lost_packets++; continue; } ret = kernctl_get_subbuf_size(stream->wait_fd, &len); if (ret < 0) { ERR("Snapshot kernctl_get_subbuf_size"); - ret = -errno; goto error_put_subbuf; } ret = kernctl_get_padded_subbuf_size(stream->wait_fd, &padded_len); if (ret < 0) { ERR("Snapshot kernctl_get_padded_subbuf_size"); - ret = -errno; goto error_put_subbuf; } - read_len = lttng_consumer_on_read_subbuffer_mmap(ctx, stream, len, - padded_len - len, NULL); + ret = get_current_subbuf_addr(stream, &subbuf_addr); + if (ret) { + goto error_put_subbuf; + } + + subbuf_view = lttng_buffer_view_init( + subbuf_addr, 0, padded_len); + read_len = lttng_consumer_on_read_subbuffer_mmap(ctx, + stream, &subbuf_view, + padded_len - len); /* * We write the padded len in local tracefiles but the data len * when using a relay. Display the error but continue processing @@ -285,7 +321,6 @@ int lttng_kconsumer_snapshot_channel(uint64_t key, char *path, ret = kernctl_put_subbuf(stream->wait_fd); if (ret < 0) { ERR("Snapshot kernctl_put_subbuf"); - ret = -errno; goto end_unlock; } consumed_pos += stream->max_sb_size; @@ -302,7 +337,7 @@ int lttng_kconsumer_snapshot_channel(uint64_t key, char *path, } } else { close_relayd_stream(stream); - stream->net_seq_idx = (uint64_t) -1ULL; + stream->relayd_id = (uint64_t) -1ULL; } pthread_mutex_unlock(&stream->lock); } @@ -314,7 +349,6 @@ int lttng_kconsumer_snapshot_channel(uint64_t key, char *path, error_put_subbuf: ret = kernctl_put_subbuf(stream->wait_fd); if (ret < 0) { - ret = -errno; ERR("Snapshot kernctl_put_subbuf error path"); } end_unlock: @@ -329,7 +363,7 @@ end: * * Returns 0 on success, < 0 on error */ -int lttng_kconsumer_snapshot_metadata(uint64_t key, char *path, +static int lttng_kconsumer_snapshot_metadata(uint64_t key, char *path, uint64_t relayd_id, struct lttng_consumer_local_data *ctx) { int ret, use_relayd = 0; @@ -348,11 +382,12 @@ int lttng_kconsumer_snapshot_metadata(uint64_t key, char *path, if (!metadata_channel) { ERR("Kernel snapshot metadata not found for key %" PRIu64, key); ret = -1; - goto error; + goto error_no_channel; } metadata_stream = metadata_channel->metadata_stream; assert(metadata_stream); + pthread_mutex_lock(&metadata_stream->lock); /* Flag once that we have a valid relayd for the stream. */ if (relayd_id != (uint64_t) -1ULL) { @@ -362,7 +397,7 @@ int lttng_kconsumer_snapshot_metadata(uint64_t key, char *path, if (use_relayd) { ret = consumer_send_relayd_stream(metadata_stream, path); if (ret < 0) { - goto error; + goto error_snapshot; } } else { ret = utils_create_stream_file(path, metadata_stream->name, @@ -370,7 +405,7 @@ int lttng_kconsumer_snapshot_metadata(uint64_t key, char *path, metadata_stream->tracefile_count_current, metadata_stream->uid, metadata_stream->gid, NULL); if (ret < 0) { - goto error; + goto error_snapshot; } metadata_stream->out_fd = ret; } @@ -378,12 +413,13 @@ int lttng_kconsumer_snapshot_metadata(uint64_t key, char *path, do { health_code_update(); - ret_read = lttng_kconsumer_read_subbuffer(metadata_stream, ctx); + ret_read = lttng_consumer_read_subbuffer(metadata_stream, ctx, true); if (ret_read < 0) { if (ret_read != -EAGAIN) { ERR("Kernel snapshot reading metadata subbuffer (ret: %zd)", ret_read); - goto error; + ret = ret_read; + goto error_snapshot; } /* ret_read is negative at this point so we will exit the loop. */ continue; @@ -392,7 +428,7 @@ int lttng_kconsumer_snapshot_metadata(uint64_t key, char *path, if (use_relayd) { close_relayd_stream(metadata_stream); - metadata_stream->net_seq_idx = (uint64_t) -1ULL; + metadata_stream->relayd_id = (uint64_t) -1ULL; } else { if (metadata_stream->out_fd >= 0) { ret = close(metadata_stream->out_fd); @@ -408,11 +444,12 @@ int lttng_kconsumer_snapshot_metadata(uint64_t key, char *path, } ret = 0; - +error_snapshot: + pthread_mutex_unlock(&metadata_stream->lock); cds_list_del(&metadata_stream->send_node); consumer_stream_destroy(metadata_stream, NULL); metadata_channel->metadata_stream = NULL; -error: +error_no_channel: rcu_read_unlock(); return ret; } @@ -454,10 +491,10 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, case LTTNG_CONSUMER_ADD_RELAYD_SOCKET: { /* Session daemon status message are handled in the following call. */ - ret = consumer_add_relayd_socket(msg.u.relayd_sock.net_index, + consumer_add_relayd_socket(msg.u.relayd_sock.net_index, msg.u.relayd_sock.type, ctx, sock, consumer_sockpoll, &msg.u.relayd_sock.sock, msg.u.relayd_sock.session_id, - msg.u.relayd_sock.relayd_session_id); + msg.u.relayd_sock.relayd_session_id); goto end_nosignal; } case LTTNG_CONSUMER_ADD_CHANNEL: @@ -484,7 +521,9 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, msg.u.channel.tracefile_size, msg.u.channel.tracefile_count, 0, msg.u.channel.monitor, - msg.u.channel.live_timer_interval); + msg.u.channel.live_timer_interval, + msg.u.channel.is_live, + NULL, NULL); if (new_channel == NULL) { lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_OUTFD_ERROR); goto end_nosignal; @@ -614,7 +653,10 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, health_code_update(); - new_stream = consumer_allocate_stream(channel->key, + pthread_mutex_lock(&channel->lock); + new_stream = consumer_stream_create( + channel, + channel->key, fd, LTTNG_CONSUMER_ACTIVE_STREAM, channel->name, @@ -634,20 +676,16 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_OUTFD_ERROR); break; } + pthread_mutex_unlock(&channel->lock); goto end_nosignal; } - new_stream->chan = channel; new_stream->wait_fd = fd; - switch (channel->output) { - case CONSUMER_CHANNEL_SPLICE: - new_stream->output = LTTNG_EVENT_SPLICE; - break; - case CONSUMER_CHANNEL_MMAP: - new_stream->output = LTTNG_EVENT_MMAP; - break; - default: - ERR("Stream output unknown %d", channel->output); + ret = kernctl_get_max_subbuf_size(new_stream->wait_fd, + &new_stream->max_sb_size); + if (ret < 0) { + pthread_mutex_unlock(&channel->lock); + ERR("Failed to get kernel maximal subbuffer size"); goto end_nosignal; } @@ -691,20 +729,37 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, if (!channel->monitor) { DBG("Kernel consumer add stream %s in no monitor mode with " "relayd id %" PRIu64, new_stream->name, - new_stream->net_seq_idx); + new_stream->relayd_id); cds_list_add(&new_stream->send_node, &channel->streams.head); + pthread_mutex_unlock(&channel->lock); break; } /* Send stream to relayd if the stream has an ID. */ - if (new_stream->net_seq_idx != (uint64_t) -1ULL) { + if (new_stream->relayd_id != (uint64_t) -1ULL) { ret = consumer_send_relayd_stream(new_stream, new_stream->chan->pathname); if (ret < 0) { + pthread_mutex_unlock(&channel->lock); consumer_stream_free(new_stream); goto end_nosignal; } + + /* + * If adding an extra stream to an already + * existing channel (e.g. cpu hotplug), we need + * to send the "streams_sent" command to relayd. + */ + if (channel->streams_sent_to_relayd) { + ret = consumer_send_relayd_streams_sent( + new_stream->relayd_id); + if (ret < 0) { + pthread_mutex_unlock(&channel->lock); + goto end_nosignal; + } + } } + pthread_mutex_unlock(&channel->lock); /* Get the right pipe where the stream will be sent. */ if (new_stream->metadata_flag) { @@ -797,6 +852,7 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, if (ret < 0) { goto end_nosignal; } + channel->streams_sent_to_relayd = true; } break; } @@ -881,7 +937,7 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, ret = lttng_kconsumer_snapshot_channel(msg.u.snapshot_channel.key, msg.u.snapshot_channel.pathname, msg.u.snapshot_channel.relayd_id, - msg.u.snapshot_channel.max_stream_size, + msg.u.snapshot_channel.nb_packets_per_stream, ctx); if (ret < 0) { ERR("Snapshot channel failed"); @@ -940,6 +996,66 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, goto end_nosignal; } + case LTTNG_CONSUMER_DISCARDED_EVENTS: + { + uint64_t ret; + struct lttng_consumer_channel *channel; + uint64_t id = msg.u.discarded_events.session_id; + uint64_t key = msg.u.discarded_events.channel_key; + + DBG("Kernel consumer discarded events command for session id %" + PRIu64 ", channel key %" PRIu64, id, key); + + channel = consumer_find_channel(key); + if (!channel) { + ERR("Kernel consumer discarded events channel %" + PRIu64 " not found", key); + ret = 0; + } else { + ret = channel->discarded_events; + } + + health_code_update(); + + /* Send back returned value to session daemon */ + ret = lttcomm_send_unix_sock(sock, &ret, sizeof(ret)); + if (ret < 0) { + PERROR("send discarded events"); + goto error_fatal; + } + + break; + } + case LTTNG_CONSUMER_LOST_PACKETS: + { + uint64_t ret; + struct lttng_consumer_channel *channel; + uint64_t id = msg.u.lost_packets.session_id; + uint64_t key = msg.u.lost_packets.channel_key; + + DBG("Kernel consumer lost packets command for session id %" + PRIu64 ", channel key %" PRIu64, id, key); + + channel = consumer_find_channel(key); + if (!channel) { + ERR("Kernel consumer lost packets channel %" + PRIu64 " not found", key); + ret = 0; + } else { + ret = channel->lost_packets; + } + + health_code_update(); + + /* Send back returned value to session daemon */ + ret = lttcomm_send_unix_sock(sock, &ret, sizeof(ret)); + if (ret < 0) { + PERROR("send lost packets"); + goto error_fatal; + } + + break; + } default: goto end_nosignal; } @@ -960,60 +1076,6 @@ error_fatal: return -1; } -/* - * Populate index values of a kernel stream. Values are set in big endian order. - * - * Return 0 on success or else a negative value. - */ -static int get_index_values(struct ctf_packet_index *index, int infd) -{ - int ret; - - ret = kernctl_get_timestamp_begin(infd, &index->timestamp_begin); - if (ret < 0) { - PERROR("kernctl_get_timestamp_begin"); - goto error; - } - index->timestamp_begin = htobe64(index->timestamp_begin); - - ret = kernctl_get_timestamp_end(infd, &index->timestamp_end); - if (ret < 0) { - PERROR("kernctl_get_timestamp_end"); - goto error; - } - index->timestamp_end = htobe64(index->timestamp_end); - - ret = kernctl_get_events_discarded(infd, &index->events_discarded); - if (ret < 0) { - PERROR("kernctl_get_events_discarded"); - goto error; - } - index->events_discarded = htobe64(index->events_discarded); - - ret = kernctl_get_content_size(infd, &index->content_size); - if (ret < 0) { - PERROR("kernctl_get_content_size"); - goto error; - } - index->content_size = htobe64(index->content_size); - - ret = kernctl_get_packet_size(infd, &index->packet_size); - if (ret < 0) { - PERROR("kernctl_get_packet_size"); - goto error; - } - index->packet_size = htobe64(index->packet_size); - - ret = kernctl_get_stream_id(infd, &index->stream_id); - if (ret < 0) { - PERROR("kernctl_get_stream_id"); - goto error; - } - index->stream_id = htobe64(index->stream_id); - -error: - return ret; -} /* * Sync metadata meaning request them to the session daemon and snapshot to the * metadata thread can consumer them. @@ -1037,7 +1099,7 @@ int lttng_kconsumer_sync_metadata(struct lttng_consumer_stream *metadata) ret = kernctl_snapshot(metadata->wait_fd); if (ret < 0) { - if (errno != EAGAIN) { + if (ret != -EAGAIN) { ERR("Sync metadata, taking kernel snapshot failed."); goto end; } @@ -1051,186 +1113,227 @@ end: return ret; } -/* - * Consume data on a file descriptor and write it on a trace file. - */ -ssize_t lttng_kconsumer_read_subbuffer(struct lttng_consumer_stream *stream, - struct lttng_consumer_local_data *ctx) +static +int extract_common_subbuffer_info(struct lttng_consumer_stream *stream, + struct stream_subbuffer *subbuf) { - unsigned long len, subbuf_size, padding; - int err, write_index = 1; - ssize_t ret = 0; - int infd = stream->wait_fd; - struct ctf_packet_index index; + int ret; + + ret = kernctl_get_subbuf_size( + stream->wait_fd, &subbuf->info.data.subbuf_size); + if (ret) { + goto end; + } - DBG("In read_subbuffer (infd : %d)", infd); + ret = kernctl_get_padded_subbuf_size( + stream->wait_fd, &subbuf->info.data.padded_subbuf_size); + if (ret) { + goto end; + } - /* Get the next subbuffer */ - err = kernctl_get_next_subbuf(infd); - if (err != 0) { - /* - * This is a debug message even for single-threaded consumer, - * because poll() have more relaxed criterions than get subbuf, - * so get_subbuf may fail for short race windows where poll() - * would issue wakeups. - */ - DBG("Reserving sub buffer failed (everything is normal, " - "it is due to concurrency)"); - ret = -errno; +end: + return ret; +} + +static +int extract_metadata_subbuffer_info(struct lttng_consumer_stream *stream, + struct stream_subbuffer *subbuf) +{ + int ret; + + ret = extract_common_subbuffer_info(stream, subbuf); + if (ret) { goto end; } - /* Get the full subbuffer size including padding */ - err = kernctl_get_padded_subbuf_size(infd, &len); - if (err != 0) { - PERROR("Getting sub-buffer len failed."); - err = kernctl_put_subbuf(infd); - if (err != 0) { - if (errno == EFAULT) { - PERROR("Error in unreserving sub buffer\n"); - } else if (errno == EIO) { - /* Should never happen with newer LTTng versions */ - PERROR("Reader has been pushed by the writer, last sub-buffer corrupted."); - } - ret = -errno; - goto end; - } - ret = -errno; + ret = kernctl_get_metadata_version( + stream->wait_fd, &subbuf->info.metadata.version); + if (ret) { goto end; } - if (!stream->metadata_flag) { - ret = get_index_values(&index, infd); - if (ret < 0) { - err = kernctl_put_subbuf(infd); - if (err != 0) { - if (errno == EFAULT) { - PERROR("Error in unreserving sub buffer\n"); - } else if (errno == EIO) { - /* Should never happen with newer LTTng versions */ - PERROR("Reader has been pushed by the writer, last sub-buffer corrupted."); - } - ret = -errno; - goto end; - } +end: + return ret; +} + +static +int extract_data_subbuffer_info(struct lttng_consumer_stream *stream, + struct stream_subbuffer *subbuf) +{ + int ret; + + ret = extract_common_subbuffer_info(stream, subbuf); + if (ret) { + goto end; + } + + ret = kernctl_get_packet_size( + stream->wait_fd, &subbuf->info.data.packet_size); + if (ret < 0) { + PERROR("Failed to get sub-buffer packet size"); + goto end; + } + + ret = kernctl_get_content_size( + stream->wait_fd, &subbuf->info.data.content_size); + if (ret < 0) { + PERROR("Failed to get sub-buffer content size"); + goto end; + } + + ret = kernctl_get_timestamp_begin( + stream->wait_fd, &subbuf->info.data.timestamp_begin); + if (ret < 0) { + PERROR("Failed to get sub-buffer begin timestamp"); + goto end; + } + + ret = kernctl_get_timestamp_end( + stream->wait_fd, &subbuf->info.data.timestamp_end); + if (ret < 0) { + PERROR("Failed to get sub-buffer end timestamp"); + goto end; + } + + ret = kernctl_get_events_discarded( + stream->wait_fd, &subbuf->info.data.events_discarded); + if (ret) { + PERROR("Failed to get sub-buffer events discarded count"); + goto end; + } + + ret = kernctl_get_sequence_number(stream->wait_fd, + &subbuf->info.data.sequence_number.value); + if (ret) { + /* May not be supported by older LTTng-modules. */ + if (ret != -ENOTTY) { + PERROR("Failed to get sub-buffer sequence number"); goto end; } } else { - write_index = 0; + subbuf->info.data.sequence_number.is_set = true; } - switch (stream->chan->output) { - case CONSUMER_CHANNEL_SPLICE: - /* - * XXX: The lttng-modules splice "actor" does not handle copying - * partial pages hence only using the subbuffer size without the - * padding makes the splice fail. - */ - subbuf_size = len; - padding = 0; + ret = kernctl_get_stream_id( + stream->wait_fd, &subbuf->info.data.stream_id); + if (ret < 0) { + PERROR("Failed to get stream id"); + goto end; + } - /* splice the subbuffer to the tracefile */ - ret = lttng_consumer_on_read_subbuffer_splice(ctx, stream, subbuf_size, - padding, &index); - /* - * XXX: Splice does not support network streaming so the return value - * is simply checked against subbuf_size and not like the mmap() op. - */ - if (ret != subbuf_size) { - /* - * display the error but continue processing to try - * to release the subbuffer - */ - ERR("Error splicing to tracefile (ret: %zd != len: %lu)", - ret, subbuf_size); - write_index = 0; - } - break; - case CONSUMER_CHANNEL_MMAP: - /* Get subbuffer size without padding */ - err = kernctl_get_subbuf_size(infd, &subbuf_size); - if (err != 0) { - PERROR("Getting sub-buffer len failed."); - err = kernctl_put_subbuf(infd); - if (err != 0) { - if (errno == EFAULT) { - PERROR("Error in unreserving sub buffer\n"); - } else if (errno == EIO) { - /* Should never happen with newer LTTng versions */ - PERROR("Reader has been pushed by the writer, last sub-buffer corrupted."); - } - ret = -errno; - goto end; - } - ret = -errno; + ret = kernctl_get_instance_id(stream->wait_fd, + &subbuf->info.data.stream_instance_id.value); + if (ret) { + /* May not be supported by older LTTng-modules. */ + if (ret != -ENOTTY) { + PERROR("Failed to get stream instance id"); goto end; } + } else { + subbuf->info.data.stream_instance_id.is_set = true; + } +end: + return ret; +} - /* Make sure the tracer is not gone mad on us! */ - assert(len >= subbuf_size); +static +int get_subbuffer_common(struct lttng_consumer_stream *stream, + struct stream_subbuffer *subbuffer) +{ + int ret; - padding = len - subbuf_size; + ret = kernctl_get_next_subbuf(stream->wait_fd); + if (ret) { + goto end; + } - /* write the subbuffer to the tracefile */ - ret = lttng_consumer_on_read_subbuffer_mmap(ctx, stream, subbuf_size, - padding, &index); - /* - * The mmap operation should write subbuf_size amount of data when - * network streaming or the full padding (len) size when we are _not_ - * streaming. - */ - if ((ret != subbuf_size && stream->net_seq_idx != (uint64_t) -1ULL) || - (ret != len && 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 is possible to - * happen without being a critical error. - */ - DBG("Error writing to tracefile " - "(ret: %zd != len: %lu != subbuf_size: %lu)", - ret, len, subbuf_size); - write_index = 0; - } - break; - default: - ERR("Unknown output method"); - ret = -EPERM; + ret = stream->read_subbuffer_ops.extract_subbuffer_info( + stream, subbuffer); +end: + return ret; +} + +static +int get_next_subbuffer_splice(struct lttng_consumer_stream *stream, + struct stream_subbuffer *subbuffer) +{ + int ret; + + ret = get_subbuffer_common(stream, subbuffer); + if (ret) { + goto end; } - err = kernctl_put_next_subbuf(infd); - if (err != 0) { - if (errno == EFAULT) { - PERROR("Error in unreserving sub buffer\n"); - } else if (errno == EIO) { - /* Should never happen with newer LTTng versions */ - PERROR("Reader has been pushed by the writer, last sub-buffer corrupted."); - } - ret = -errno; + subbuffer->buffer.fd = stream->wait_fd; +end: + return ret; +} + +static +int get_next_subbuffer_mmap(struct lttng_consumer_stream *stream, + struct stream_subbuffer *subbuffer) +{ + int ret; + const char *addr; + + ret = get_subbuffer_common(stream, subbuffer); + if (ret) { goto end; } - /* Write index if needed. */ - if (!write_index) { + ret = get_current_subbuf_addr(stream, &addr); + if (ret) { goto end; } - if (stream->chan->live_timer_interval && !stream->metadata_flag) { - /* - * In live, block until all the metadata is sent. - */ - err = consumer_stream_sync_metadata(ctx, stream->session_id); - if (err < 0) { - goto end; + subbuffer->buffer.buffer = lttng_buffer_view_init( + addr, 0, subbuffer->info.data.padded_subbuf_size); +end: + return ret; +} + +static +int put_next_subbuffer(struct lttng_consumer_stream *stream, + struct stream_subbuffer *subbuffer) +{ + const int ret = kernctl_put_next_subbuf(stream->wait_fd); + + if (ret) { + if (ret == -EFAULT) { + PERROR("Error in unreserving sub buffer"); + } else if (ret == -EIO) { + /* Should never happen with newer LTTng versions */ + PERROR("Reader has been pushed by the writer, last sub-buffer corrupted"); } } - err = consumer_stream_write_index(stream, &index); - if (err < 0) { - goto end; + return ret; +} + +static void lttng_kconsumer_set_stream_ops( + struct lttng_consumer_stream *stream) +{ + if (stream->chan->output == CONSUMER_CHANNEL_MMAP) { + stream->read_subbuffer_ops.get_next_subbuffer = + get_next_subbuffer_mmap; + } else { + stream->read_subbuffer_ops.get_next_subbuffer = + get_next_subbuffer_splice; } -end: - return ret; + if (stream->metadata_flag) { + stream->read_subbuffer_ops.extract_subbuffer_info = + extract_metadata_subbuffer_info; + } else { + stream->read_subbuffer_ops.extract_subbuffer_info = + extract_data_subbuffer_info; + if (stream->chan->is_live) { + stream->read_subbuffer_ops.send_live_beacon = + consumer_flush_kernel_index; + } + } + + stream->read_subbuffer_ops.put_next_subbuffer = put_next_subbuffer; } int lttng_kconsumer_on_recv_stream(struct lttng_consumer_stream *stream) @@ -1243,7 +1346,7 @@ int lttng_kconsumer_on_recv_stream(struct lttng_consumer_stream *stream) * Don't create anything if this is set for streaming or should not be * monitored. */ - if (stream->net_seq_idx == (uint64_t) -1ULL && stream->chan->monitor) { + if (stream->relayd_id == (uint64_t) -1ULL && stream->chan->monitor) { ret = utils_create_stream_file(stream->chan->pathname, stream->name, stream->chan->tracefile_size, stream->tracefile_count_current, stream->uid, stream->gid, NULL); @@ -1254,14 +1357,17 @@ int lttng_kconsumer_on_recv_stream(struct lttng_consumer_stream *stream) stream->tracefile_size_current = 0; if (!stream->metadata_flag) { - ret = index_create_file(stream->chan->pathname, + struct lttng_index_file *index_file; + + 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 (!index_file) { goto error; } - stream->index_fd = ret; + stream->index_file = index_file; } } @@ -1272,7 +1378,6 @@ int lttng_kconsumer_on_recv_stream(struct lttng_consumer_stream *stream) ret = kernctl_get_mmap_len(stream->wait_fd, &mmap_len); if (ret != 0) { PERROR("kernctl_get_mmap_len"); - ret = -errno; goto error_close_fd; } stream->mmap_len = (size_t) mmap_len; @@ -1286,6 +1391,8 @@ int lttng_kconsumer_on_recv_stream(struct lttng_consumer_stream *stream) } } + lttng_kconsumer_set_stream_ops(stream); + /* we return 0 to let the library handle the FD internally */ return 0;