From mboxrd@z Thu Jan 1 00:00:00 1970 Return-Path: Date: Thu, 10 May 2018 22:50:38 +0300 From: Konstantin Osipov Subject: Re: [tarantool-patches] [PATCH v2 10/10] session: introduce binary box.session.push Message-ID: <20180510195038.GJ30593@atlas> References: <2e8899477ec65522992dd33d246e21c92ebbe246.1524228894.git.v.shpilevoy@tarantool.org> MIME-Version: 1.0 Content-Type: text/plain; charset=us-ascii Content-Disposition: inline In-Reply-To: <2e8899477ec65522992dd33d246e21c92ebbe246.1524228894.git.v.shpilevoy@tarantool.org> To: tarantool-patches@freelists.org Cc: vdavydov.dev@gmail.com List-ID: * Vladislav Shpilevoy [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