[Tarantool-patches] [PATCH v3 5/8] iproto: implement streams in iproto

[Vladimir Davydov]

On Wed, Aug 11, 2021 at 11:56:55AM +0300, mechanik20051988 wrote:
> diff --git a/src/box/iproto.cc b/src/box/iproto.cc
> index dcf60e1be..3b792130b 100644
> --- a/src/box/iproto.cc
> +++ b/src/box/iproto.cc
> @@ -1873,12 +2049,52 @@ tx_process_replication(struct cmsg *m)
>  	}
>  }
>  
> +static void
> +iproto_msg_finish_processing_in_stream(struct iproto_msg *msg)
> +{
> +	struct iproto_connection *con = msg->connection;
> +	struct iproto_stream *stream = msg->stream;
> +
> +	assert(stream != NULL);
> +	struct iproto_msg *tmp =
> +		stailq_shift_entry(&stream->pending_requests,
> +				   struct iproto_msg, in_stream);
> +	assert(tmp == msg);
> +	(void)tmp;
> +	errinj_stream_msg_count_add(-1);
> +
> +	if (stailq_empty(&stream->pending_requests)) {
> +		struct mh_i64ptr_node_t node = { stream->id, NULL };
> +		mh_i64ptr_remove(con->streams, &node, 0);
> +		iproto_stream_delete(stream);
> +	} else {
> +		/*
> +		 * If there are new messages for this stream
> +		 * then schedule their processing.
> +		 */
> +		struct iproto_msg *next =
> +			stailq_first_entry(&stream->pending_requests,
> +					   struct iproto_msg,
> +					   in_stream);
> +		assert(next != NULL);
> +		next->wpos = con->wpos;
> +		cpipe_push_input(&con->iproto_thread->tx_pipe, &next->base);
> +		cpipe_flush_input(&con->iproto_thread->tx_pipe);
> +	}
> +}
> +
>  static void
>  net_send_msg(struct cmsg *m)
>  {
>  	struct iproto_msg *msg = (struct iproto_msg *) m;
>  	struct iproto_connection *con = msg->connection;
> +	struct iproto_stream *stream = msg->stream;
> +
> +	if (stream == NULL)
> +		goto send_msg;

Please fold this check into iproto_msg_finish_processing_in_stream
and call the latter unconditionally, like you do in case of
iproto_msg_start_processing_in_stream.

After this LGTM.

>  
> +	iproto_msg_finish_processing_in_stream(msg);
> +send_msg:
>  	if (msg->len != 0) {
>  		/* Discard request (see iproto_enqueue_batch()). */
>  		msg->p_ibuf->rpos += msg->len;