[PATCH] iproto: init coio watcher before join/subscribe

[Alexander Turenko <alexander.turenko@tarantool.org>]

box_process_join() and box_process_subscribe() use coio_write_xrow(),
which calls coio_writev_timeout() under hood. If a socket will block at
write() the function calls ev_io_start() to wake the fiber up when the
socket will be ready to write. This code assumes that the watcher
(struct ev_io) is initialized as coio watcher, i.e. coio_create() has
been called.

The reason why the code works before is that coio_write_xrow() in
box_process_{join,subscribe}() writes a small piece of data and so the
situation when a socket write buffer has less free space then needed is
rare.

Fixes #4110.
---

How to reproduce the issue:

 | echo 'use_unix_sockets_iproto = True' >> test/vinyl/suite.ini
 | (cd test && ./test-run.py vinyl/replica_rejoin.test.lua)

It is hard to write a test w/o a network emulation layer: we should
trigger a situation when a socket write buffer is almost full right
after initial join data sending (but before vclock sending).

I don't sure I use terminology in the right way in the commit message.
Please, let me know if the wording feels bad.

https://github.com/tarantool/tarantool/issues/4110
https://github.com/tarantool/tarantool/tree/Totktonada/gh-4110-fix-segfault-in-iproto

 src/box/iproto.cc | 6 ++++--
 1 file changed, 4 insertions(+), 2 deletions(-)

diff --git a/src/box/iproto.cc b/src/box/iproto.cc
index 02b558ede..8f899fed8 100644
--- a/src/box/iproto.cc
+++ b/src/box/iproto.cc
@@ -1693,6 +1693,8 @@ tx_process_join_subscribe(struct cmsg *m)
 {
 	struct iproto_msg *msg = tx_accept_msg(m);
 	struct iproto_connection *con = msg->connection;
+	struct ev_io io;
+	coio_create(&io, con->input.fd);
 	try {
 		switch (msg->header.type) {
 		case IPROTO_JOIN:
@@ -1701,7 +1703,7 @@ tx_process_join_subscribe(struct cmsg *m)
 			 * the lambda in the beginning of the block
 			 * will re-activate the watchers for us.
 			 */
-			box_process_join(&con->input, &msg->header);
+			box_process_join(&io, &msg->header);
 			break;
 		case IPROTO_SUBSCRIBE:
 			/*
@@ -1710,7 +1712,7 @@ tx_process_join_subscribe(struct cmsg *m)
 			 * the write watcher will be re-activated
 			 * the same way as for JOIN.
 			 */
-			box_process_subscribe(&con->input, &msg->header);
+			box_process_subscribe(&io, &msg->header);
 			break;
 		default:
 			unreachable();
-- 
2.21.0