X-Git-Url: https://git.lttng.org/?p=lttng-tools.git;a=blobdiff_plain;f=src%2Fcommon%2Fkernel-consumer%2Fkernel-consumer.c;h=4f3e2871daeaa1c66faadaf64a754dccf0711b94;hp=63ca35d62a26345a7c884faf041d3b489549b0e2;hb=873b9e9a5a22f81225e3e5aaacb88b26ce739c68;hpb=cf53a8a6b321018162c7aac015a74cecab14c9e9

diff --git a/src/common/kernel-consumer/kernel-consumer.c b/src/common/kernel-consumer/kernel-consumer.c
index 63ca35d62..4f3e2871d 100644
--- a/src/common/kernel-consumer/kernel-consumer.c
+++ b/src/common/kernel-consumer/kernel-consumer.c
@@ -110,7 +110,8 @@ int lttng_kconsumer_get_consumed_snapshot(struct lttng_consumer_stream *stream,
  * Returns 0 on success, < 0 on error
  */
 int lttng_kconsumer_snapshot_channel(uint64_t key, char *path,
-		uint64_t relayd_id, struct lttng_consumer_local_data *ctx)
+		uint64_t relayd_id, uint64_t max_stream_size,
+		struct lttng_consumer_local_data *ctx)
 {
 	int ret;
 	unsigned long consumed_pos, produced_pos;
@@ -203,6 +204,15 @@ int lttng_kconsumer_snapshot_channel(uint64_t key, char *path,
 			}
 		}
 
+		/*
+		 * 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);
+
 		while (consumed_pos < produced_pos) {
 			ssize_t read_len;
 			unsigned long len, padded_len;
@@ -260,12 +270,14 @@ int lttng_kconsumer_snapshot_channel(uint64_t key, char *path,
 		}
 
 		if (relayd_id == (uint64_t) -1ULL) {
-			ret = close(stream->out_fd);
-			if (ret < 0) {
-				PERROR("Kernel consumer snapshot close out_fd");
-				goto end_unlock;
+			if (stream->out_fd >= 0) {
+				ret = close(stream->out_fd);
+				if (ret < 0) {
+					PERROR("Kernel consumer snapshot close out_fd");
+					goto end_unlock;
+				}
+				stream->out_fd = -1;
 			}
-			stream->out_fd = -1;
 		} else {
 			close_relayd_stream(stream);
 			stream->net_seq_idx = (uint64_t) -1ULL;
@@ -357,15 +369,17 @@ int lttng_kconsumer_snapshot_metadata(uint64_t key, char *path,
 		close_relayd_stream(metadata_stream);
 		metadata_stream->net_seq_idx = (uint64_t) -1ULL;
 	} else {
-		ret = close(metadata_stream->out_fd);
-		if (ret < 0) {
-			PERROR("Kernel consumer snapshot metadata close out_fd");
-			/*
-			 * Don't go on error here since the snapshot was successful at this
-			 * point but somehow the close failed.
-			 */
+		if (metadata_stream->out_fd >= 0) {
+			ret = close(metadata_stream->out_fd);
+			if (ret < 0) {
+				PERROR("Kernel consumer snapshot metadata close out_fd");
+				/*
+				 * Don't go on error here since the snapshot was successful at this
+				 * point but somehow the close failed.
+				 */
+			}
+			metadata_stream->out_fd = -1;
 		}
-		metadata_stream->out_fd = -1;
 	}
 
 	ret = 0;
@@ -392,8 +406,8 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
 
 	ret = lttcomm_recv_unix_sock(sock, &msg, sizeof(msg));
 	if (ret != sizeof(msg)) {
-		lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_ERROR_RECV_CMD);
 		if (ret > 0) {
+			lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_ERROR_RECV_CMD);
 			ret = -1;
 		}
 		return ret;
@@ -559,7 +573,8 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
 				channel->session_id,
 				msg.u.stream.cpu,
 				&alloc_ret,
-				channel->type);
+				channel->type,
+				channel->monitor);
 		if (new_stream == NULL) {
 			switch (alloc_ret) {
 			case -ENOMEM:
@@ -626,10 +641,14 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
 			break;
 		}
 
-		ret = consumer_send_relayd_stream(new_stream, NULL);
-		if (ret < 0) {
-			consumer_stream_free(new_stream);
-			goto end_nosignal;
+		/* Send stream to relayd if the stream has an ID. */
+		if (new_stream->net_seq_idx != (uint64_t) -1ULL) {
+			ret = consumer_send_relayd_stream(new_stream,
+					new_stream->chan->pathname);
+			if (ret < 0) {
+				consumer_stream_free(new_stream);
+				goto end_nosignal;
+			}
 		}
 
 		/* Get the right pipe where the stream will be sent. */
@@ -647,6 +666,8 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
 			consumer_stream_free(new_stream);
 			goto end_nosignal;
 		}
+		/* Successfully sent to the right thread. */
+		new_stream->globally_visible = 1;
 
 		DBG("Kernel consumer ADD_STREAM %s (fd: %d) with relayd id %" PRIu64,
 				new_stream->name, fd, new_stream->relayd_stream_id);
@@ -728,7 +749,9 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
 		} else {
 			ret = lttng_kconsumer_snapshot_channel(msg.u.snapshot_channel.key,
 					msg.u.snapshot_channel.pathname,
-					msg.u.snapshot_channel.relayd_id, ctx);
+					msg.u.snapshot_channel.relayd_id,
+					msg.u.snapshot_channel.max_stream_size,
+					ctx);
 			if (ret < 0) {
 				ERR("Snapshot channel failed");
 				ret_code = LTTNG_ERR_KERN_CHAN_FAIL;
@@ -984,6 +1007,11 @@ int lttng_kconsumer_data_pending(struct lttng_consumer_stream *stream)
 
 	assert(stream);
 
+	if (stream->endpoint_status != CONSUMER_ENDPOINT_ACTIVE) {
+		ret = 0;
+		goto end;
+	}
+
 	ret = kernctl_get_next_subbuf(stream->wait_fd);
 	if (ret == 0) {
 		/* There is still data so let's put back this subbuffer. */