X-Git-Url: http://git.lttng.org/?a=blobdiff_plain;f=src%2Fcommon%2Fust-consumer%2Fust-consumer.c;h=3d3a30dcd9e2e41d47b7c8d2bc7a92edbce57b35;hb=ac5b946733aa5c81e5b3098a0b243deaf9134ce6;hp=147fe8aafe899a092a91ee8be58357786d7f074f;hpb=e9404c27e7cc9d841785e6c4292c1add19fbc1cc;p=lttng-tools.git diff --git a/src/common/ust-consumer/ust-consumer.c b/src/common/ust-consumer/ust-consumer.c index 147fe8aaf..3d3a30dcd 100644 --- a/src/common/ust-consumer/ust-consumer.c +++ b/src/common/ust-consumer/ust-consumer.c @@ -51,7 +51,6 @@ extern struct lttng_consumer_global_data consumer_data; extern int consumer_poll_timeout; -extern volatile int consumer_quit; /* * Free channel object and all streams associated with it. This MUST be used @@ -832,6 +831,7 @@ static int close_metadata(uint64_t chan_key) { int ret = 0; struct lttng_consumer_channel *channel; + unsigned int channel_monitor; DBG("UST consumer close metadata key %" PRIu64, chan_key); @@ -850,13 +850,48 @@ static int close_metadata(uint64_t chan_key) pthread_mutex_lock(&consumer_data.lock); pthread_mutex_lock(&channel->lock); - + channel_monitor = channel->monitor; if (cds_lfht_is_node_deleted(&channel->node.node)) { goto error_unlock; } lttng_ustconsumer_close_metadata(channel); + pthread_mutex_unlock(&channel->lock); + pthread_mutex_unlock(&consumer_data.lock); + /* + * The ownership of a metadata channel depends on the type of + * session to which it belongs. In effect, the monitor flag is checked + * to determine if this metadata channel is in "snapshot" mode or not. + * + * In the non-snapshot case, the metadata channel is created along with + * a single stream which will remain present until the metadata channel + * is destroyed (on the destruction of its session). In this case, the + * metadata stream in "monitored" by the metadata poll thread and holds + * the ownership of its channel. + * + * Closing the metadata will cause the metadata stream's "metadata poll + * pipe" to be closed. Closing this pipe will wake-up the metadata poll + * thread which will teardown the metadata stream which, in return, + * deletes the metadata channel. + * + * In the snapshot case, the metadata stream is created and destroyed + * on every snapshot record. Since the channel doesn't have an owner + * other than the session daemon, it is safe to destroy it immediately + * on reception of the CLOSE_METADATA command. + */ + if (!channel_monitor) { + /* + * The channel and consumer_data locks must be + * released before this call since consumer_del_channel + * re-acquires the channel and consumer_data locks to teardown + * the channel and queue its reclamation by the "call_rcu" + * worker thread. + */ + consumer_del_channel(channel); + } + + return ret; error_unlock: pthread_mutex_unlock(&channel->lock); pthread_mutex_unlock(&consumer_data.lock); @@ -1077,9 +1112,6 @@ static int snapshot_channel(uint64_t key, char *path, uint64_t relayd_id, DBG("UST consumer snapshot channel %" PRIu64, key); cds_list_for_each_entry(stream, &channel->streams.head, send_node) { - /* Are we at a position _before_ the first available packet ? */ - bool before_first_packet = true; - health_code_update(); /* Lock stream because we are about to change its state. */ @@ -1105,12 +1137,6 @@ static int snapshot_channel(uint64_t key, char *path, uint64_t relayd_id, DBG("UST 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) { - goto error_unlock; - } - } /* * If tracing is active, we want to perform a "full" buffer flush. @@ -1151,7 +1177,6 @@ static int snapshot_channel(uint64_t key, char *path, uint64_t relayd_id, while (consumed_pos < produced_pos) { ssize_t read_len; unsigned long len, padded_len; - int lost_packet = 0; health_code_update(); @@ -1165,15 +1190,7 @@ static int snapshot_channel(uint64_t key, char *path, uint64_t relayd_id, } DBG("UST consumer get subbuf failed. Skipping it."); consumed_pos += stream->max_sb_size; - - /* - * Start accounting lost packets only when we - * already have extracted packets (to match the - * content of the final snapshot). - */ - if (!before_first_packet) { - lost_packet = 1; - } + stream->chan->lost_packets++; continue; } @@ -1209,16 +1226,6 @@ static int snapshot_channel(uint64_t key, char *path, uint64_t relayd_id, goto error_close_stream; } consumed_pos += stream->max_sb_size; - - /* - * Only account lost packets located between - * succesfully extracted packets (do not account before - * and after since they are not visible in the - * resulting snapshot). - */ - stream->chan->lost_packets += lost_packet; - lost_packet = 0; - before_first_packet = false; } /* Simply close the stream so we can use it on the next snapshot. */ @@ -1355,7 +1362,7 @@ int lttng_ustconsumer_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); @@ -1456,6 +1463,7 @@ int lttng_ustconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, attr.read_timer_interval = msg.u.ask_channel.read_timer_interval; attr.chan_id = msg.u.ask_channel.chan_id; memcpy(attr.uuid, msg.u.ask_channel.uuid, sizeof(attr.uuid)); + attr.blocking_timeout= msg.u.ask_channel.blocking_timeout; /* Match channel buffer type to the UST abi. */ switch (msg.u.ask_channel.output) { @@ -2914,7 +2922,7 @@ int lttng_ustconsumer_request_metadata(struct lttng_consumer_local_data *ctx, request.key = channel->key; DBG("Sending metadata request to sessiond, session id %" PRIu64 - ", per-pid %" PRIu64 ", app UID %u and channek key %" PRIu64, + ", per-pid %" PRIu64 ", app UID %u and channel key %" PRIu64, request.session_id, request.session_id_per_pid, request.uid, request.key);