X-Git-Url: https://git.lttng.org/?p=lttng-tools.git;a=blobdiff_plain;f=src%2Fcommon%2Fkernel-consumer%2Fkernel-consumer.c;h=196da6633353a44ea6f73821494642e0cd7539a3;hp=edf8acb8286cedc3c8649b6c64313ccb0ff385fc;hb=a9d36096fd1bfbeb7c55a5cfc97523512237f0d4;hpb=e5add6d004793894ef4c7e047bc0f8885763b205

diff --git a/src/common/kernel-consumer/kernel-consumer.c b/src/common/kernel-consumer/kernel-consumer.c
index edf8acb82..196da6633 100644
--- a/src/common/kernel-consumer/kernel-consumer.c
+++ b/src/common/kernel-consumer/kernel-consumer.c
@@ -231,15 +231,6 @@ static int lttng_kconsumer_snapshot_channel(
 			goto end_unlock;
 		}
 
-		if (stream->max_sb_size == 0) {
-			ret = kernctl_get_max_subbuf_size(stream->wait_fd,
-					&stream->max_sb_size);
-			if (ret < 0) {
-				ERR("Getting kernel max_sb_size");
-				goto end_unlock;
-			}
-		}
-
 		consumed_pos = consumer_get_consume_start_pos(consumed_pos,
 				produced_pos, nb_packets_per_stream,
 				stream->max_sb_size);
@@ -600,14 +591,14 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
 		ret = consumer_send_status_msg(sock, ret_code);
 		if (ret < 0) {
 			/* Somehow, the session daemon is not responding anymore. */
-			goto error_fatal;
+			goto error_add_stream_fatal;
 		}
 
 		health_code_update();
 
 		if (ret_code != LTTCOMM_CONSUMERD_SUCCESS) {
 			/* Channel was not found. */
-			goto end_nosignal;
+			goto error_add_stream_nosignal;
 		}
 
 		/* Blocking call */
@@ -615,7 +606,7 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
 		ret = lttng_consumer_poll_socket(consumer_sockpoll);
 		health_poll_exit();
 		if (ret) {
-			goto error_fatal;
+			goto error_add_stream_fatal;
 		}
 
 		health_code_update();
@@ -624,8 +615,7 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
 		ret = lttcomm_recv_fds_unix_sock(sock, &fd, 1);
 		if (ret != sizeof(fd)) {
 			lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_ERROR_RECV_FD);
-			rcu_read_unlock();
-			return ret;
+			goto end;
 		}
 
 		health_code_update();
@@ -638,7 +628,7 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
 		ret = consumer_send_status_msg(sock, ret_code);
 		if (ret < 0) {
 			/* Somehow, the session daemon is not responding anymore. */
-			goto end_nosignal;
+			goto error_add_stream_nosignal;
 		}
 
 		health_code_update();
@@ -663,11 +653,19 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
 				break;
 			}
 			pthread_mutex_unlock(&channel->lock);
-			goto end_nosignal;
+			goto error_add_stream_nosignal;
 		}
 
 		new_stream->chan = channel;
 		new_stream->wait_fd = fd;
+		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 error_add_stream_nosignal;
+		}
+
 		consumer_stream_update_channel_attributes(new_stream,
 				channel);
 		switch (channel->output) {
@@ -676,7 +674,7 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
 			ret = utils_create_pipe(new_stream->splice_pipe);
 			if (ret < 0) {
 				pthread_mutex_unlock(&channel->lock);
-				goto end_nosignal;
+				goto error_add_stream_nosignal;
 			}
 			break;
 		case CONSUMER_CHANNEL_MMAP:
@@ -685,7 +683,7 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
 		default:
 			ERR("Stream output unknown %d", channel->output);
 			pthread_mutex_unlock(&channel->lock);
-			goto end_nosignal;
+			goto error_add_stream_nosignal;
 		}
 
 		/*
@@ -717,7 +715,7 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
 				pthread_mutex_unlock(&new_stream->lock);
 				pthread_mutex_unlock(&channel->lock);
 				consumer_stream_free(new_stream);
-				goto end_nosignal;
+				goto error_add_stream_nosignal;
 			}
 		}
 		health_code_update();
@@ -734,7 +732,7 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
 			cds_list_add(&new_stream->send_node, &channel->streams.head);
 			pthread_mutex_unlock(&new_stream->lock);
 			pthread_mutex_unlock(&channel->lock);
-			break;
+			goto end_add_stream;
 		}
 
 		/* Send stream to relayd if the stream has an ID. */
@@ -745,7 +743,7 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
 				pthread_mutex_unlock(&new_stream->lock);
 				pthread_mutex_unlock(&channel->lock);
 				consumer_stream_free(new_stream);
-				goto end_nosignal;
+				goto error_add_stream_nosignal;
 			}
 
 			/*
@@ -759,7 +757,7 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
 				if (ret < 0) {
 					pthread_mutex_unlock(&new_stream->lock);
 					pthread_mutex_unlock(&channel->lock);
-					goto end_nosignal;
+					goto error_add_stream_nosignal;
 				}
 			}
 		}
@@ -790,12 +788,17 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
 			} else {
 				consumer_del_stream_for_data(new_stream);
 			}
-			goto end_nosignal;
+			goto error_add_stream_nosignal;
 		}
 
 		DBG("Kernel consumer ADD_STREAM %s (fd: %d) %s with relayd id %" PRIu64,
 				new_stream->name, fd, new_stream->chan->pathname, new_stream->relayd_stream_id);
+end_add_stream:
 		break;
+error_add_stream_nosignal:
+		goto end_nosignal;
+error_add_stream_fatal:
+		goto error_fatal;
 	}
 	case LTTNG_CONSUMER_STREAMS_SENT:
 	{
@@ -824,7 +827,7 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
 		ret = consumer_send_status_msg(sock, ret_code);
 		if (ret < 0 || ret_code != LTTCOMM_CONSUMERD_SUCCESS) {
 			/* Somehow, the session daemon is not responding anymore. */
-			goto end_nosignal;
+			goto error_streams_sent_nosignal;
 		}
 
 		health_code_update();
@@ -834,7 +837,7 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
 		 * streams in this channel.
 		 */
 		if (!channel->monitor) {
-			break;
+			goto end_error_streams_sent;
 		}
 
 		health_code_update();
@@ -843,11 +846,14 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
 			ret = consumer_send_relayd_streams_sent(
 					msg.u.sent_streams.net_seq_idx);
 			if (ret < 0) {
-				goto end_nosignal;
+				goto error_streams_sent_nosignal;
 			}
 			channel->streams_sent_to_relayd = true;
 		}
+end_error_streams_sent:
 		break;
+error_streams_sent_nosignal:
+		goto end_nosignal;
 	}
 	case LTTNG_CONSUMER_UPDATE_STREAM:
 	{
@@ -973,14 +979,14 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
 		ret = consumer_send_status_msg(sock, ret_code);
 		if (ret < 0) {
 			/* Somehow, the session daemon is not responding anymore. */
-			goto end_nosignal;
+			goto end_destroy_channel;
 		}
 
 		health_code_update();
 
 		/* Stop right now if no channel was found. */
 		if (!channel) {
-			goto end_nosignal;
+			goto end_destroy_channel;
 		}
 
 		/*
@@ -996,7 +1002,7 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
 		assert(!uatomic_sub_return(&channel->refcount, 1));
 
 		consumer_del_channel(channel);
-
+end_destroy_channel:
 		goto end_nosignal;
 	}
 	case LTTNG_CONSUMER_DISCARDED_EVENTS:
@@ -1239,15 +1245,21 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
 	}
 	case LTTNG_CONSUMER_CLOSE_TRACE_CHUNK:
 	{
+		enum lttng_trace_chunk_command_type close_command =
+				msg.u.close_trace_chunk.close_command.value;
 		const uint64_t relayd_id =
 				msg.u.close_trace_chunk.relayd_id.value;
 
 		ret_code = lttng_consumer_close_trace_chunk(
 				msg.u.close_trace_chunk.relayd_id.is_set ?
-						&relayd_id : NULL,
+						&relayd_id :
+						NULL,
 				msg.u.close_trace_chunk.session_id,
 				msg.u.close_trace_chunk.chunk_id,
-				(time_t) msg.u.close_trace_chunk.close_timestamp);
+				(time_t) msg.u.close_trace_chunk.close_timestamp,
+				msg.u.close_trace_chunk.close_command.is_set ?
+						&close_command :
+						NULL);
 		goto end_msg_sessiond;
 	}
 	case LTTNG_CONSUMER_TRACE_CHUNK_EXISTS:
@@ -1267,15 +1279,16 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
 	}
 
 end_nosignal:
-	rcu_read_unlock();
-
 	/*
 	 * Return 1 to indicate success since the 0 value can be a socket
 	 * shutdown during the recv() or send() call.
 	 */
-	health_code_update();
-	return 1;
-
+	ret = 1;
+	goto end;
+error_fatal:
+	/* This will issue a consumer stop. */
+	ret = -1;
+	goto end;
 end_msg_sessiond:
 	/*
 	 * The returned value here is not useful since either way we'll return 1 to
@@ -1286,16 +1299,11 @@ end_msg_sessiond:
 	if (ret < 0) {
 		goto error_fatal;
 	}
-	rcu_read_unlock();
-
+	ret = 1;
+end:
 	health_code_update();
-
-	return 1;
-
-error_fatal:
 	rcu_read_unlock();
-	/* This will issue a consumer stop. */
-	return -1;
+	return ret;
 }
 
 /*