X-Git-Url: http://git.efficios.com/?a=blobdiff_plain;f=src%2Fcommon%2Fust-consumer%2Fust-consumer.c;h=c5c2a6548e1bddbcf9772d36fe8266f1fab365be;hb=66d583dc40af49e00d85385eb6f148c0fe3f7d50;hp=4297d60cffbaaf5a82d3d575e30eea964ff9713b;hpb=10211f5c32e51e398cb37508851956a2f8b7dbbf;p=lttng-tools.git

diff --git a/src/common/ust-consumer/ust-consumer.c b/src/common/ust-consumer/ust-consumer.c
index 4297d60cf..c5c2a6548 100644
--- a/src/common/ust-consumer/ust-consumer.c
+++ b/src/common/ust-consumer/ust-consumer.c
@@ -206,28 +206,21 @@ static int send_stream_to_thread(struct lttng_consumer_stream *stream,
 
 	/* Get the right pipe where the stream will be sent. */
 	if (stream->metadata_flag) {
-		ret = consumer_add_metadata_stream(stream);
-		if (ret) {
-			ERR("Consumer add metadata stream %" PRIu64 " failed.",
-					stream->key);
-			goto error;
-		}
+		consumer_add_metadata_stream(stream);
 		stream_pipe = ctx->consumer_metadata_pipe;
 	} else {
-		ret = consumer_add_data_stream(stream);
-		if (ret) {
-			ERR("Consumer add stream %" PRIu64 " failed.",
-					stream->key);
-			goto error;
-		}
+		consumer_add_data_stream(stream);
 		stream_pipe = ctx->consumer_data_pipe;
 	}
 
 	/*
 	 * From this point on, the stream's ownership has been moved away from
-	 * the channel and becomes globally visible.
+	 * the channel and it becomes globally visible. Hence, remove it from
+	 * the local stream list to prevent the stream from being both local and
+	 * global.
 	 */
 	stream->globally_visible = 1;
+	cds_list_del(&stream->send_node);
 
 	ret = lttng_pipe_write(stream_pipe, &stream, sizeof(stream));
 	if (ret < 0) {
@@ -239,7 +232,9 @@ static int send_stream_to_thread(struct lttng_consumer_stream *stream,
 		} else {
 			consumer_del_stream_for_data(stream);
 		}
+		goto error;
 	}
+
 error:
 	return ret;
 }
@@ -721,14 +716,8 @@ static int send_streams_to_thread(struct lttng_consumer_channel *channel,
 			 * If we are unable to send the stream to the thread, there is
 			 * a big problem so just stop everything.
 			 */
-			/* Remove node from the channel stream list. */
-			cds_list_del(&stream->send_node);
 			goto error;
 		}
-
-		/* Remove node from the channel stream list. */
-		cds_list_del(&stream->send_node);
-
 	}
 
 error:
@@ -918,6 +907,10 @@ static int setup_metadata(struct lttng_consumer_local_data *ctx, uint64_t key)
 		}
 	}
 
+	/*
+	 * Ownership of metadata stream is passed along. Freeing is handled by
+	 * the callee.
+	 */
 	ret = send_streams_to_thread(metadata, ctx);
 	if (ret < 0) {
 		/*
@@ -925,7 +918,7 @@ static int setup_metadata(struct lttng_consumer_local_data *ctx, uint64_t key)
 		 * a big problem so just stop everything.
 		 */
 		ret = LTTCOMM_CONSUMERD_FATAL;
-		goto error;
+		goto send_streams_error;
 	}
 	/* List MUST be empty after or else it could be reused. */
 	assert(cds_list_empty(&metadata->streams.head));
@@ -943,6 +936,7 @@ error:
 	consumer_stream_destroy(metadata->metadata_stream, NULL);
 	cds_list_del(&metadata->metadata_stream->send_node);
 	metadata->metadata_stream = NULL;
+send_streams_error:
 error_no_stream:
 end:
 	return ret;
@@ -1076,9 +1070,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. */
@@ -1104,12 +1095,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.
@@ -1150,7 +1135,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();
 
@@ -1164,15 +1148,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;
 			}
 
@@ -1208,16 +1184,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. */
@@ -1354,7 +1320,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);