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);
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 */
ret = lttng_consumer_poll_socket(consumer_sockpoll);
health_poll_exit();
if (ret) {
- goto error_fatal;
+ goto error_add_stream_fatal;
}
health_code_update();
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();
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();
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) {
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:
default:
ERR("Stream output unknown %d", channel->output);
pthread_mutex_unlock(&channel->lock);
- goto end_nosignal;
+ goto error_add_stream_nosignal;
}
/*
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();
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. */
pthread_mutex_unlock(&new_stream->lock);
pthread_mutex_unlock(&channel->lock);
consumer_stream_free(new_stream);
- goto end_nosignal;
+ goto error_add_stream_nosignal;
}
/*
if (ret < 0) {
pthread_mutex_unlock(&new_stream->lock);
pthread_mutex_unlock(&channel->lock);
- goto end_nosignal;
+ goto error_add_stream_nosignal;
}
}
}
} 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:
{
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();
* streams in this channel.
*/
if (!channel->monitor) {
- break;
+ goto end_error_streams_sent;
}
health_code_update();
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:
{
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;
}
/*
assert(!uatomic_sub_return(&channel->refcount, 1));
consumer_del_channel(channel);
-
+end_destroy_channel:
goto end_nosignal;
}
case LTTNG_CONSUMER_DISCARDED_EVENTS:
}
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:
}
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
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;
}
/*
projects / lttng-tools.git / blobdiff