Re: [tarantool-patches] [PATCH v2 10/10] session: introduce binary box.session.push

[Konstantin Osipov <kostja@tarantool.org>]

* Vladislav Shpilevoy <v.shpilevoy@tarantool.org> [18/04/20 16:25]:
> Box.session.push() allows to send a message to a client with no
> finishing a main request.
> 
> Tarantool after this patch supports pushes over binary protocol.
> 
> IProto message is encoded using a new header code - IPROTO_CHUNK.
> TX thread to notify IProto thread about new data in obuf sends
> a message 'push_msg'. IProto thread, got this message, notifies
> libev about new data, and then sends 'push_msg' back with
> updated write position. TX thread, received the message back,
> updates its version of a write position. If IProto would not send
> a write position, then TX would write to the same obuf again and
> again, because it can not know that IProto already flushed
> another obuf.
> 
> To avoid multiple 'push_msg' in fly between IProto and TX, the
> only one 'push_msg' per connection is used. To deliver pushes,
> appeared when 'push_msg' was in fly, TX thread sets a flag every
> time when sees, that 'push_msg' is sent, and there is a new push.
> When 'push_msg' returns, it checks this flag, and if it is set,
> the IProto is notified again.

I don't see any reason for this restriction.
Any connection has two independent rotating output buffers
of infinite size. If you ever want to block a push message, you
should block it because both buffers are busy.

> +/**
> + * Message to notify IProto thread about new data in an output
> + * buffer. Struct iproto_msg is not used here, because push
> + * notification can be much more compact: it does not have
> + * request, ibuf, length, flags ...
> + */
> +struct iproto_push_msg {
> +	struct cmsg base;
> +	/**
> +	 * Before sending to IProto thread, the wpos is set to a
> +	 * current position in an output buffer. Before IProto
> +	 * returns the message to TX, it sets wpos to the last
> +	 * flushed position (works like iproto_msg.wpos).
> +	 */
> +	struct iproto_wpos wpos;
> +};
> +

Looks good to me.

> +	 * Is_push_in_progress is set, when a push_msg is sent to
> +	 * IProto thread, and reset, when the message is returned
> +	 * to TX. If a new push sees, that a push_msg is already
> +	 * sent to IProto, then has_new_pushes is set. After push
> +	 * notification is returned to TX, it checks
> +	 * has_new_pushes. If it is set, then the notification is
> +	 * sent again. This ping-pong continues, until TX stopped
> +	 * pushing. It allows to
> +	 * 1) avoid multiple push_msg from one session in fly,
> +	 * 2) do not block push() until a previous push() is
> +	 *    finished.

Please make it radically simpler, every push can create a new
message which has an independent life cycle. Messages can never
run one over each other, so you have nothing to worry about.

> @@ -1038,7 +1085,7 @@ tx_process_disconnect(struct cmsg *m)
>  	struct iproto_msg *msg = (struct iproto_msg *) m;
>  	struct iproto_connection *con = msg->connection;
>  	if (con->session) {
> -		tx_fiber_init(con->session, 0);
> +		tx_fiber_init(con->session, NULL);

Why do you need to make it more complex than it is now?

Every Lua procedure which makes a push takes a long-polling 
reference to the connection already. Until this procedure
ends, you can't disconnect a connection.

> +static void
> +tx_accept_wpos(struct iproto_connection *con, const struct iproto_wpos *wpos)
>  {
> -	struct iproto_msg *msg = (struct iproto_msg *) m;
> -	struct iproto_connection *con = msg->connection;
> -
>  	struct obuf *prev = &con->obuf[con->tx.p_obuf == con->obuf];
> -	if (msg->wpos.obuf == con->tx.p_obuf) {
> +	if (wpos->obuf == con->tx.p_obuf) {
>  		/*
>  		 * We got a message advancing the buffer which
>  		 * is being appended to. The previous buffer is
> @@ -1134,6 +1182,13 @@ tx_accept_msg(struct cmsg *m)
>  		 */
>  		con->tx.p_obuf = prev;
>  	}
> +}
> +
> +static inline struct iproto_msg *
> +tx_accept_msg(struct cmsg *m)
> +{
> +	struct iproto_msg *msg = (struct iproto_msg *) m;
> +	tx_accept_wpos(msg->connection, &msg->wpos);
>  	return msg;
>  }

This somehow looks half-baked, I don't know how yet.

> +c:call('push_null', {}, {on_push = on_push})

What happens if on_push handler is not set? Can I get the entire
data set in a result when all pushes are over? 

Can I get a data set as an iterable and yield in the iterator
instead?



-- 
Konstantin Osipov, Moscow, Russia, +7 903 626 22 32
http://tarantool.io - www.twitter.com/kostja_osipov