Fix a potential memory corruption on large iproto output.

Remove ev_clear_pending(), which is already done by ev_io_stop().
Update comments.

src/box/iproto.cc

@@ -425,7 +425,7 @@ iproto_connection_input_iobuf(struct iproto_connection *con)
 		 * iobuf idle, time to trim buffers.
 		 */
 		if (iobuf_is_idle(oldbuf))
-			iobuf_reset(oldbuf);
+			iobuf_reset_mt(oldbuf);
 	}
 	/*
 	 * Rotate buffers. Not strictly necessary, but
@@ -482,15 +482,14 @@ iproto_enqueue_batch(struct iproto_connection *con, struct ibuf *in)
 			   msg->header.type == IPROTO_JOIN) {
 			/**
 			 * Don't mess with the file descriptor
-			 * while join is running. Clear the
-			 * pending events as well, since their
-			 * invocation may re-start the watcher,
-			 * ruining our efforts.
+			 * while join is running. ev_io_stop()
+			 * also clears any pending events, which
+			 * is good, since their invocation may
+			 * re-start the watcher, ruining our
+			 * efforts.
 			 */
 			ev_io_stop(con->loop, &con->output);
-			ev_clear_pending(con->loop, &con->output);
 			ev_io_stop(con->loop, &con->input);
-			ev_clear_pending(con->loop, &con->input);
 			stop_input = true;
 		}
 		msg->request.header = &msg->header;
@@ -603,7 +602,7 @@ iproto_flush(struct iobuf *iobuf, struct iproto_connection *con)
 				/* Quickly recycle the buffer if it's idle. */
 				assert(end->used == obuf_size(&iobuf->out));
 				/* resets wpos and wpend to zero pos */
-				iobuf_reset(iobuf);
+				iobuf_reset_mt(iobuf);
 			} else { /* Avoid assignment reordering. */
 				/* Advance write position. */
 				*begin = *end;

src/iobuf.cc

@@ -122,6 +122,31 @@ iobuf_reset(struct iobuf *iobuf)
 	}
 }
 
+void
+iobuf_reset_mt(struct iobuf *iobuf)
+{
+	/*
+	 * If we happen to have fully processed the input,
+	 * move the pos to the start of the input buffer.
+	 */
+	if (ibuf_used(&iobuf->in) == 0) {
+		if (ibuf_capacity(&iobuf->in) < iobuf_max_size()) {
+			ibuf_reset(&iobuf->in);
+		} else {
+			struct slab_cache *slabc = iobuf->in.slabc;
+			ibuf_destroy(&iobuf->in);
+			ibuf_create(&iobuf->in, slabc, iobuf_readahead);
+		}
+	}
+	/*
+	 * We can't re-create the output buffer in iproto thread,
+	 * since obuf->slabc is from tx thread.
+	 * FIXME: send a message to tx thread to garbage-collect
+	 * the buffer when it's too big.
+	 */
+	obuf_reset(&iobuf->out);
+}
+
 void
 iobuf_init()
 {

src/iobuf.h

@@ -80,6 +80,9 @@ iobuf_delete_mt(struct iobuf *iobuf);
 void
 iobuf_reset(struct iobuf *iobuf);
 
+void
+iobuf_reset_mt(struct iobuf *iobuf);
+
 /** Return true if there is no input and no output and
  * no one has pinned the buffer - i.e. it's safe to
  * destroy it.