stream->last_net_seq_num = be64toh(stream_info.last_net_seq_num);
pthread_mutex_unlock(&stream->lock);
- stream_close(stream);
-
+ /*
+ * This is one of the conditions which may trigger a stream close
+ * with the others being:
+ * 1) A close command is received for a stream
+ * 2) The control connection owning the stream is closed
+ * 3) We have received all of the stream's data _after_ a close
+ * request.
+ */
+ try_stream_close(stream);
if (stream->is_metadata) {
struct relay_viewer_stream *vstream;
static int relay_recv_metadata(struct lttcomm_relayd_hdr *recv_hdr,
struct relay_connection *conn)
{
- int ret = htobe32(LTTNG_OK);
+ int ret = 0;
ssize_t size_ret;
struct relay_session *session = conn->session;
struct lttcomm_relayd_metadata_payload *metadata_struct;
}
memset(data_buffer, 0, data_size);
DBG2("Relay receiving metadata, waiting for %" PRIu64 " bytes", data_size);
- ret = conn->sock->ops->recvmsg(conn->sock, data_buffer, data_size, 0);
- if (ret < 0 || ret != data_size) {
- if (ret == 0) {
+ size_ret = conn->sock->ops->recvmsg(conn->sock, data_buffer, data_size, 0);
+ if (size_ret < 0 || size_ret != data_size) {
+ if (size_ret == 0) {
/* Orderly shutdown. Not necessary to print an error. */
DBG("Socket %d did an orderly shutdown", conn->sock->fd);
} else {
goto end_put;
}
- ret = write_padding_to_file(metadata_stream->stream_fd->fd,
+ size_ret = write_padding_to_file(metadata_stream->stream_fd->fd,
be32toh(metadata_struct->padding_size));
- if (ret < 0) {
+ if (size_ret < 0) {
goto end_put;
}
end_put:
pthread_mutex_unlock(&metadata_stream->lock);
stream_put(metadata_stream);
-
end:
return ret;
}
uint64_t net_seq_num;
uint32_t data_size;
struct relay_session *session;
- bool new_stream = false;
+ bool new_stream = false, close_requested = false;
ret = conn->sock->ops->recvmsg(conn->sock, &data_hdr,
sizeof(struct lttcomm_relayd_data_hdr), 0);
stream->prev_seq = net_seq_num;
end_stream_unlock:
+ close_requested = stream->close_requested;
pthread_mutex_unlock(&stream->lock);
+ if (close_requested) {
+ try_stream_close(stream);
+ }
+
if (new_stream) {
pthread_mutex_lock(&session->lock);
uatomic_set(&session->new_streams, 1);