X-Git-Url: http://git.efficios.com/?a=blobdiff_plain;f=src%2Fcommon%2Fkernel-consumer%2Fkernel-consumer.c;h=7039b3b80cfc833ff830eb84c8ef70cc03b5c0bb;hb=e1f421592683038467f8f12177975e19502ffc56;hp=bdd5ec86d03d926c45b3bebb54e3a1701e4147c8;hpb=2947da290dc03bd6b5919241aaa8fcf96ba11525;p=lttng-tools.git diff --git a/src/common/kernel-consumer/kernel-consumer.c b/src/common/kernel-consumer/kernel-consumer.c index bdd5ec86d..7039b3b80 100644 --- a/src/common/kernel-consumer/kernel-consumer.c +++ b/src/common/kernel-consumer/kernel-consumer.c @@ -60,7 +60,11 @@ 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."); } @@ -187,14 +191,6 @@ int lttng_kconsumer_snapshot_channel(uint64_t key, char *path, DBG("Kernel consumer snapshot stream %s/%s (%" PRIu64 ")", path, stream->name, stream->key); } - if (relayd_id != -1ULL) { - ret = consumer_send_relayd_streams_sent(relayd_id); - if (ret < 0) { - ERR("sending streams sent to relayd"); - goto end_unlock; - } - channel->streams_sent_to_relayd = true; - } ret = kernctl_buffer_flush_empty(stream->wait_fd); if (ret < 0) { @@ -340,7 +336,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; @@ -359,11 +355,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) { @@ -373,7 +370,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, @@ -381,7 +378,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; } @@ -394,7 +391,8 @@ int lttng_kconsumer_snapshot_metadata(uint64_t key, char *path, 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; @@ -419,11 +417,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; } @@ -465,10 +464,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: @@ -983,7 +982,8 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, } case LTTNG_CONSUMER_DISCARDED_EVENTS: { - uint64_t ret; + ssize_t ret; + uint64_t count; struct lttng_consumer_channel *channel; uint64_t id = msg.u.discarded_events.session_id; uint64_t key = msg.u.discarded_events.channel_key; @@ -995,15 +995,15 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, if (!channel) { ERR("Kernel consumer discarded events channel %" PRIu64 " not found", key); - ret = 0; + count = 0; } else { - ret = channel->discarded_events; + count = channel->discarded_events; } health_code_update(); /* Send back returned value to session daemon */ - ret = lttcomm_send_unix_sock(sock, &ret, sizeof(ret)); + ret = lttcomm_send_unix_sock(sock, &count, sizeof(count)); if (ret < 0) { PERROR("send discarded events"); goto error_fatal; @@ -1013,7 +1013,8 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, } case LTTNG_CONSUMER_LOST_PACKETS: { - uint64_t ret; + ssize_t ret; + uint64_t count; struct lttng_consumer_channel *channel; uint64_t id = msg.u.lost_packets.session_id; uint64_t key = msg.u.lost_packets.channel_key; @@ -1025,15 +1026,15 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, if (!channel) { ERR("Kernel consumer lost packets channel %" PRIu64 " not found", key); - ret = 0; + count = 0; } else { - ret = channel->lost_packets; + count = channel->lost_packets; } health_code_update(); /* Send back returned value to session daemon */ - ret = lttcomm_send_unix_sock(sock, &ret, sizeof(ret)); + ret = lttcomm_send_unix_sock(sock, &count, sizeof(count)); if (ret < 0) { PERROR("send lost packets"); goto error_fatal; @@ -1090,6 +1091,31 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, } break; } + case LTTNG_CONSUMER_CLEAR_CHANNEL: + { + struct lttng_consumer_channel *channel; + uint64_t key = msg.u.clear_channel.key; + + channel = consumer_find_channel(key); + if (!channel) { + DBG("Channel %" PRIu64 " not found", key); + ret_code = LTTCOMM_CONSUMERD_CHAN_NOT_FOUND; + } else { + ret = lttng_consumer_clear_channel(channel); + if (ret) { + ERR("Clear channel failed"); + ret_code = ret; + } + + health_code_update(); + } + ret = consumer_send_status_msg(sock, ret_code); + if (ret < 0) { + /* Somehow, the session daemon is not responding anymore. */ + goto end_nosignal; + } + break; + } default: goto end_nosignal; } @@ -1166,7 +1192,6 @@ static int get_index_values(struct ctf_packet_index *index, int infd) if (ret == -ENOTTY) { /* Command not implemented by lttng-modules. */ index->stream_instance_id = -1ULL; - ret = 0; } else { PERROR("kernctl_get_instance_id"); goto error; @@ -1238,7 +1263,6 @@ int update_stream_stats(struct lttng_consumer_stream *stream) if (ret == -ENOTTY) { /* Command not implemented by lttng-modules. */ seq = -1ULL; - ret = 0; } else { PERROR("kernctl_get_sequence_number"); goto end;