[tarantool-patches] [PATCH v2 10/10] session: introduce binary box.session.push
Konstantin Osipov
kostja at tarantool.org
Thu May 10 22:50:38 MSK 2018
* Vladislav Shpilevoy <v.shpilevoy at 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
More information about the Tarantool-patches
mailing list