Repository: blackbeam/mysql_async (language: rust)

Issue:
PacketOutOfOrder error on connect with too many connections
`Conn::new` returns a `PacketOutOfOrder` error if too many connections are established at the same time to a MySQL database. This doesn't seem like the right error to return? You can try this with something like:
```rust
#[tokio::test(threaded_scheduler)]
async fn ooo_error() {
    use futures_util::stream::StreamExt;

    let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel::<()>();
    for _ in 0..1000 {
        let tx = tx.clone();
        tokio::spawn(async move {
            let c = Conn::new(get_opts()).await.unwrap();
            let c = c.drop_query("SELECT 1").await.unwrap();
            c.disconnect().await.unwrap();
            drop(tx);
        });
    }

    // wait until all connections have been dropped
    drop(tx);
    assert_eq!(rx.next().await, None);
}
```
Which will panic at `Conn::new(get_opts()).await.unwrap()` (expected) with a `PacketOutOfOrder` error (unexpected).
PacketOutOfOrder error on connect with too many connections
`Conn::new` returns a `PacketOutOfOrder` error if too many connections are established at the same time to a MySQL database. This doesn't seem like the right error to return? You can try this with something like:
```rust
#[tokio::test(threaded_scheduler)]
async fn ooo_error() {
    use futures_util::stream::StreamExt;

    let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel::<()>();
    for _ in 0..1000 {
        let tx = tx.clone();
        tokio::spawn(async move {
            let c = Conn::new(get_opts()).await.unwrap();
            let c = c.drop_query("SELECT 1").await.unwrap();
            c.disconnect().await.unwrap();
            drop(tx);
        });
    }

    // wait until all connections have been dropped
    drop(tx);
    assert_eq!(rx.next().await, None);
}
```
Which will panic at `Conn::new(get_opts()).await.unwrap()` (expected) with a `PacketOutOfOrder` error (unexpected).


Discussion / hints:
Note that the test above will appear to pass. You'll need to run it with `-- --nocapture` to see the panics.
In my experience, the test above fails if the number of connections established is `>=` **half** of MySQL's `max_connections`. The reason for this is that when a `Plain` connection is upgraded to a `Socket` connection, there is a period of time during which _both_ versions of the connection are present, and so the connection counts double against the connection limit.
I have a weak suspicion that this might be related to #94. @blackbeam can you reproduce this issue? It should be relatively self-contained and easy enough to debug given that you know the protocol :) Especially because it doesn't require much concurrency at all.
I just ran into this issue again today sadly, so doesn't appear to have been fixed by the resolution to #94.
Oh, this is quite sad. I'll look into it again.
Could you please, if possible, share your server and OS version, and also a database configuration?
Please note that the biggest problem for me here was to reproduce this issue (I simply couldn't).
This was on an EC2 Ubuntu machine, running latest LTS, with the latest MariaDB version available there.
I was using the stock database configuration (though note the relationship to `max-connections` in the first post).
Btw, which version of `mysql_common` is in your `Cargo.lock`?
This was with `mysql_async = "0.23.1"`, and the latest `mysql_common` that brings in, which I think is `mysql_common = "0.22.1"`.
I've tried to reproduce it on t2.micro instance with bionic and stock MariaDB 10.1.44.
I got "to may connections", "connection reset" and plantly of other errors but no PacketOutOfOrder.
I tried with and without `--release`.

Maybe I'm missing something? Here is the code:

```rust
use mysql_async::{Conn, prelude::*};

use std::future::Future;

#[tokio::main(core_threads=8)]
async fn main() {
    use futures::stream::StreamExt;

    for _ in 0..100_000 {
        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel::<()>();
        for i in 0..160 {
            let tx = tx.clone();
            tokio::spawn(async move {
                let c = Conn::new("mysql://user:password@127.0.0.1/mysql").await.unwrap();
                let c = c.drop_query("SELECT 1").await.unwrap();
                c.disconnect().await.unwrap();
                drop(tx);
            });
        }
        drop(tx);
        assert_eq!(rx.next().await, None);
    }
}
```
That's very weird — I don't have a good answer for you unfortunately. It could be that it only happens when there's significant concurrency, which might require more cores than a t2.micro. I'm running on a 16-core instance when I see the issue.
Hi, @jonhoo. Could you please test the following revision in your environment?

```toml
[dependencies]
mysql_async = { git = "https://github.com/blackbeam/mysql_async.git", rev = "4602469d303321ca4a307c7377b8502e8b5a77cf" }
```
Ooh, interesting, #4602469d303321ca4a307c7377b8502e8b5a77cf does look like a promising fix! Unfortunately I'm pretty swamped in the coming month(s) as I'm also moving cross-country, so may not be able to test this for a while. I'll add it to my backlog though!
Note that the test above will appear to pass. You'll need to run it with `-- --nocapture` to see the panics.
In my experience, the test above fails if the number of connections established is `>=` **half** of MySQL's `max_connections`. The reason for this is that when a `Plain` connection is upgraded to a `Socket` connection, there is a period of time during which _both_ versions of the connection are present, and so the connection counts double against the connection limit.
I have a weak suspicion that this might be related to #94. @blackbeam can you reproduce this issue? It should be relatively self-contained and easy enough to debug given that you know the protocol :) Especially because it doesn't require much concurrency at all.
I just ran into this issue again today sadly, so doesn't appear to have been fixed by the resolution to #94.
Oh, this is quite sad. I'll look into it again.
Could you please, if possible, share your server and OS version, and also a database configuration?
Please note that the biggest problem for me here was to reproduce this issue (I simply couldn't).
This was on an EC2 Ubuntu machine, running latest LTS, with the latest MariaDB version available there.
I was using the stock database configuration (though note the relationship to `max-connections` in the first post).
Btw, which version of `mysql_common` is in your `Cargo.lock`?
This was with `mysql_async = "0.23.1"`, and the latest `mysql_common` that brings in, which I think is `mysql_common = "0.22.1"`.
I've tried to reproduce it on t2.micro instance with bionic and stock MariaDB 10.1.44.
I got "to may connections", "connection reset" and plantly of other errors but no PacketOutOfOrder.
I tried with and without `--release`.

Maybe I'm missing something? Here is the code:

```rust
use mysql_async::{Conn, prelude::*};

use std::future::Future;

#[tokio::main(core_threads=8)]
async fn main() {
    use futures::stream::StreamExt;

    for _ in 0..100_000 {
        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel::<()>();
        for i in 0..160 {
            let tx = tx.clone();
            tokio::spawn(async move {
                let c = Conn::new("mysql://user:password@127.0.0.1/mysql").await.unwrap();
                let c = c.drop_query("SELECT 1").await.unwrap();
                c.disconnect().await.unwrap();
                drop(tx);
            });
        }
        drop(tx);
        assert_eq!(rx.next().await, None);
    }
}
```
That's very weird — I don't have a good answer for you unfortunately. It could be that it only happens when there's significant concurrency, which might require more cores than a t2.micro. I'm running on a 16-core instance when I see the issue.
Hi, @jonhoo. Could you please test the following revision in your environment?

```toml
[dependencies]
mysql_async = { git = "https://github.com/blackbeam/mysql_async.git", rev = "4602469d303321ca4a307c7377b8502e8b5a77cf" }
```
Ooh, interesting, #4602469d303321ca4a307c7377b8502e8b5a77cf does look like a promising fix! Unfortunately I'm pretty swamped in the coming month(s) as I'm also moving cross-country, so may not be able to test this for a while. I'll add it to my backlog though!

Fix patch:
diff --git a/Cargo.toml b/Cargo.toml
index fdb1b3fd..6c968e45 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -36,6 +36,7 @@ url = "2.1"
 
 [dev-dependencies]
 tempfile = "3.1.0"
+rand = "0.8.0"
 
 [features]
 nightly = []
diff --git a/azure-pipelines.yml b/azure-pipelines.yml
index 26391878..07125d90 100644
--- a/azure-pipelines.yml
+++ b/azure-pipelines.yml
@@ -60,7 +60,8 @@ jobs:
           echo 'local_infile=1' >> /usr/local/etc/my.cnf
           echo 'socket=/tmp/mysql.sock' >> /usr/local/etc/my.cnf
           brew services start mysql
-          sleep 5
+          /usr/local/Cellar/mysql/*/bin/mysqld --verbose --help
+          sleep 20
           /usr/local/Cellar/mysql/*/bin/mysql -e "SET GLOBAL max_allowed_packet = 36700160;" -uroot
         displayName: Install MySql
       - bash: |
diff --git a/src/conn/mod.rs b/src/conn/mod.rs
index 9184a9e5..b3d4f85b 100644
--- a/src/conn/mod.rs
+++ b/src/conn/mod.rs
@@ -42,7 +42,10 @@ use crate::{
     OptsBuilder,
 };
 
+use self::routines::Routine;
+
 pub mod pool;
+pub mod routines;
 pub mod stmt_cache;
 
 /// Helper that asynchronously disconnects the givent connection on the default tokio executor.
@@ -265,6 +268,25 @@ impl Conn {
         self.inner.status
     }
 
+    pub(crate) async fn routine<'a, F, T>(&mut self, mut f: F) -> crate::Result<T>
+    where
+        F: Routine<T> + 'a,
+    {
+        self.inner.disconnected = true;
+        let result = f.call(&mut *self).await;
+        match result {
+            result @ Ok(_) | result @ Err(crate::Error::Server(_)) => {
+                // either OK or non-fatal error
+                self.inner.disconnected = false;
+                result
+            }
+            Err(err) => {
+                self.take_stream().close().await?;
+                Err(err)
+            }
+        }
+    }
+
     /// Returns server version.
     pub fn server_version(&self) -> (u16, u16, u16) {
         self.inner.version
@@ -701,9 +723,7 @@ impl Conn {
         let pool = self.inner.pool.clone();
 
         if self.inner.version > (5, 7, 2) {
-            self.write_command_data(Command::COM_RESET_CONNECTION, &[])
-                .await?;
-            self.read_packet().await?;
+            self.routine(routines::ResetRoutine).await?;
         } else {
             let opts = self.inner.opts.clone();
             let old_conn = std::mem::replace(self, Conn::new(opts).await?);
diff --git a/src/conn/pool/mod.rs b/src/conn/pool/mod.rs
index 16d772ea..5ab83d41 100644
--- a/src/conn/pool/mod.rs
+++ b/src/conn/pool/mod.rs
@@ -287,7 +287,13 @@ impl Drop for Conn {
 
 #[cfg(test)]
 mod test {
-    use futures_util::{future::try_join_all, stream::StreamExt, try_join};
+    use futures_util::{
+        future::{join_all, select, select_all, try_join_all},
+        stream::StreamExt,
+        try_join, FutureExt,
+    };
+    use mysql_common::row::Row;
+    use tokio::time::delay_for;
 
     use std::time::Duration;
 
@@ -379,7 +385,7 @@ mod test {
             // now check, that they're still in the pool..
             assert_eq!(ex_field!(pool, available).len(), NUM_CONNS);
 
-            tokio::time::delay_for(std::time::Duration::from_millis(500)).await;
+            delay_for(Duration::from_millis(500)).await;
 
             // now get new connection..
             let _conn = pool.get_conn().await?;
@@ -470,16 +476,16 @@ mod test {
         drop(conns);
 
         // wait for a bit to let the connections be reclaimed
-        tokio::time::delay_for(std::time::Duration::from_millis(100)).await;
+        delay_for(Duration::from_millis(100)).await;
 
         // check that connections are still in the pool because of inactive_connection_ttl
         assert_eq!(ex_field!(pool_clone, available).len(), POOL_MAX);
 
         // then, wait for ttl_check_interval
-        tokio::time::delay_for(TTL_CHECK_INTERVAL).await;
+        delay_for(TTL_CHECK_INTERVAL).await;
 
         // wait a bit more to let the connections be reclaimed by the ttl check
-        tokio::time::delay_for(std::time::Duration::from_millis(200)).await;
+        delay_for(Duration::from_millis(500)).await;
 
         // check that we have the expected number of connections
         assert_eq!(ex_field!(pool_clone, available).len(), POOL_MIN);
@@ -512,7 +518,7 @@ mod test {
             let _ = conns.pop();
 
             // then, wait for a bit to let the connection be reclaimed
-            tokio::time::delay_for(std::time::Duration::from_millis(50)).await;
+            delay_for(Duration::from_millis(50)).await;
 
             // now check that we have the expected # of connections
             // this may look a little funky, but think of it this way:
@@ -593,7 +599,7 @@ mod test {
         assert_eq!(wait_timeout, Some(3));
         assert_eq!(ex_field!(pool, exist), 1);
 
-        tokio::time::delay_for(std::time::Duration::from_secs(6)).await;
+        delay_for(Duration::from_secs(6)).await;
 
         let mut conn = pool.get_conn().await?;
         let id2: Option<usize> = conn.query_first("SELECT CONNECTION_ID()").await?;
@@ -727,6 +733,46 @@ mod test {
         Ok(())
     }
 
+    #[tokio::test]
+    async fn should_work_if_pooled_connection_operation_is_cancelled() -> super::Result<()> {
+        let pool = Pool::new(get_opts());
+
+        // warm up
+        join_all((0..10).map(|_| pool.get_conn())).await;
+
+        /// some operation
+        async fn op(pool: &Pool) {
+            let _: Option<Row> = pool
+                .get_conn()
+                .await
+                .unwrap()
+                .exec_first("SELECT ?, ?", (42, "foo"))
+                .await
+                .unwrap();
+        }
+
+        // Measure the delay
+        let mut max_delay = 0_u128;
+        for _ in 0..10_usize {
+            let start = std::time::Instant::now();
+            op(&pool).await;
+            max_delay = std::cmp::max(max_delay, start.elapsed().as_micros());
+        }
+
+        for _ in 0_usize..128 {
+            let fut = select_all((0_usize..5).map(|_| op(&pool).boxed()));
+
+            // we need to cancel the op in the middle
+            // this should not lead to the `packet out of order` error.
+            let delay_micros = rand::random::<u128>() % max_delay;
+            select(delay_for(Duration::from_micros(delay_micros as u64)), fut).await;
+
+            // give some time for connections to return to the pool
+            delay_for(Duration::from_millis(100)).await;
+        }
+        Ok(())
+    }
+
     #[cfg(feature = "nightly")]
     mod bench {
         use futures_util::future::{FutureExt, TryFutureExt};
diff --git a/src/conn/routines/exec.rs b/src/conn/routines/exec.rs
new file mode 100644
index 00000000..aa89b158
--- /dev/null
+++ b/src/conn/routines/exec.rs
@@ -0,0 +1,82 @@
+use std::mem;
+
+use futures_core::future::BoxFuture;
+use futures_util::FutureExt;
+use mysql_common::{packets::ComStmtExecuteRequestBuilder, params::Params};
+
+use crate::{BinaryProtocol, Conn, DriverError, Statement};
+
+use super::Routine;
+
+/// A routine that executes `COM_STMT_EXECUTE`.
+#[derive(Debug, Clone)]
+pub struct ExecRoutine<'a> {
+    stmt: &'a Statement,
+    params: Params,
+}
+
+impl<'a> ExecRoutine<'a> {
+    pub fn new(stmt: &'a Statement, params: Params) -> Self {
+        Self { stmt, params }
+    }
+}
+
+impl Routine<()> for ExecRoutine<'_> {
+    fn call<'a>(&'a mut self, conn: &'a mut Conn) -> BoxFuture<'a, crate::Result<()>> {
+        async move {
+            loop {
+                match self.params {
+                    Params::Positional(ref params) => {
+                        if self.stmt.num_params() as usize != params.len() {
+                            Err(DriverError::StmtParamsMismatch {
+                                required: self.stmt.num_params(),
+                                supplied: params.len() as u16,
+                            })?
+                        }
+
+                        let (body, as_long_data) =
+                            ComStmtExecuteRequestBuilder::new(self.stmt.id()).build(&*params);
+
+                        if as_long_data {
+                            conn.send_long_data(self.stmt.id(), params.iter()).await?;
+                        }
+
+                        conn.write_command_raw(body).await?;
+                        conn.read_result_set::<BinaryProtocol>(true).await?;
+                        break;
+                    }
+                    Params::Named(_) => {
+                        if self.stmt.named_params.is_none() {
+                            let error = DriverError::NamedParamsForPositionalQuery.into();
+                            return Err(error);
+                        }
+
+                        let named = mem::replace(&mut self.params, Params::Empty);
+                        self.params =
+                            named.into_positional(self.stmt.named_params.as_ref().unwrap())?;
+
+                        continue;
+                    }
+                    Params::Empty => {
+                        if self.stmt.num_params() > 0 {
+                            let error = DriverError::StmtParamsMismatch {
+                                required: self.stmt.num_params(),
+                                supplied: 0,
+                            }
+                            .into();
+                            return Err(error);
+                        }
+
+                        let (body, _) =
+                            ComStmtExecuteRequestBuilder::new(self.stmt.id()).build(&[]);
+                        conn.write_command_raw(body).await?;
+                        conn.read_result_set::<BinaryProtocol>(true).await?;
+                        break;
+                    }
+                }
+            }
+            Ok(())
+        }
+        .boxed()
+    }
+}
diff --git a/src/conn/routines/helpers.rs b/src/conn/routines/helpers.rs
new file mode 100644
index 00000000..2ee9b2e6
--- /dev/null
+++ b/src/conn/routines/helpers.rs
@@ -0,0 +1,126 @@
+//! Private routine helpers.
+
+use std::sync::Arc;
+
+use mysql_common::{
+    constants::MAX_PAYLOAD_LEN,
+    io::ReadMysqlExt,
+    packets::{parse_local_infile_packet, ComStmtSendLongData},
+    value::Value,
+};
+use tokio::io::AsyncReadExt;
+
+use crate::{
+    queryable::{query_result::ResultSetMeta, Protocol},
+    Conn, DriverError, Error,
+};
+
+impl Conn {
+    /// Helper, that sends all `Value::Bytes` in the given list of paramenters as long data.
+    pub(super) async fn send_long_data<'a, I>(
+        &mut self,
+        statement_id: u32,
+        params: I,
+    ) -> crate::Result<()>
+    where
+        I: Iterator<Item = &'a Value>,
+    {
+        for (i, value) in params.enumerate() {
+            if let Value::Bytes(bytes) = value {
+                let chunks = bytes.chunks(MAX_PAYLOAD_LEN - 6);
+                let chunks = chunks.chain(if bytes.is_empty() {
+                    Some(&[][..])
+                } else {
+                    None
+                });
+                for chunk in chunks {
+                    let com = ComStmtSendLongData::new(statement_id, i, chunk);
+                    self.write_command_raw(com.into()).await?;
+                }
+            }
+        }
+
+        Ok(())
+    }
+
+    /// Will read result set and write pending result into `self` (if any).
+    pub(super) async fn read_result_set<P>(
+        &mut self,
+        is_first_result_set: bool,
+    ) -> crate::Result<()>
+    where
+        P: Protocol,
+    {
+        let packet = match self.read_packet().await {
+            Ok(packet) => packet,
+            Err(err @ Error::Server(_)) if is_first_result_set => {
+                // shortcut to emit an error right to the caller of a query/execute
+                return Err(err);
+            }
+            Err(Error::Server(error)) => {
+                // error will be consumed as a part of a multi-result set
+                self.set_pending_result(Some(ResultSetMeta::Error(error)));
+                return Ok(());
+            }
+            Err(err) => {
+                // non-server errors are fatal
+                return Err(err);
+            }
+        };
+
+        match packet.get(0) {
+            Some(0x00) => {
+                self.set_pending_result(Some(P::result_set_meta(Arc::from(
+                    Vec::new().into_boxed_slice(),
+                ))));
+            }
+            Some(0xFB) => self.handle_local_infile::<P>(&*packet).await?,
+            _ => self.handle_result_set::<P>(&*packet).await?,
+        }
+
+        Ok(())
+    }
+
+    /// Will handle local infile packet.
+    pub(super) async fn handle_local_infile<P>(&mut self, packet: &[u8]) -> crate::Result<()>
+    where
+        P: Protocol,
+    {
+        let local_infile = parse_local_infile_packet(&*packet)?;
+        let (local_infile, handler) = match self.opts().local_infile_handler() {
+            Some(handler) => ((local_infile.into_owned(), handler)),
+            None => return Err(DriverError::NoLocalInfileHandler.into()),
+        };
+        let mut reader = handler.handle(local_infile.file_name_ref()).await?;
+
+        let mut buf = [0; 4096];
+        loop {
+            let read = reader.read(&mut buf[..]).await?;
+            self.write_packet(&buf[..read]).await?;
+
+            if read == 0 {
+                break;
+            }
+        }
+
+        self.read_packet().await?;
+        self.set_pending_result(Some(P::result_set_meta(Arc::from(
+            Vec::new().into_boxed_slice(),
+        ))));
+        Ok(())
+    }
+
+    /// Helper that handles result set packet.
+    ///
+    /// Requires that `packet` contains non-zero length-encoded integer.
+    pub(super) async fn handle_result_set<P>(&mut self, mut packet: &[u8]) -> crate::Result<()>
+    where
+        P: Protocol,
+    {
+        let column_count = packet.read_lenenc_int()?;
+        let columns = self.read_column_defs(column_count as usize).await?;
+        let meta = P::result_set_meta(Arc::from(columns.into_boxed_slice()));
+        self.set_pending_result(Some(meta));
+        Ok(())
+    }
+}
diff --git a/src/conn/routines/mod.rs b/src/conn/routines/mod.rs
new file mode 100644
index 00000000..928bd771
--- /dev/null
+++ b/src/conn/routines/mod.rs
@@ -0,0 +1,19 @@
+use futures_core::future::BoxFuture;
+
+use crate::Conn;
+
+pub use self::{exec::*, next_set::*, ping::*, prepare::*, query::*, reset::*};
+
+mod exec;
+mod next_set;
+mod ping;
+mod prepare;
+mod query;
+mod reset;
+
+mod helpers;
+
+/// Connection will be broken if this operation isn't finished.
+pub trait Routine<T> {
+    fn call<'a>(&'a mut self, conn: &'a mut Conn) -> BoxFuture<'a, crate::Result<T>>;
+}
diff --git a/src/conn/routines/next_set.rs b/src/conn/routines/next_set.rs
new file mode 100644
index 00000000..f29b4bf5
--- /dev/null
+++ b/src/conn/routines/next_set.rs
@@ -0,0 +1,32 @@
+use std::marker::PhantomData;
+
+use futures_core::future::BoxFuture;
+use futures_util::FutureExt;
+
+use crate::{queryable::Protocol, Conn};
+
+use super::Routine;
+
+/// A routine that handles subsequent result of a mutlti-result set.
+#[derive(Debug, Clone, Copy)]
+pub struct NextSetRoutine<P>(PhantomData<P>);
+
+impl<P> NextSetRoutine<P> {
+    pub fn new() -> Self {
+        Self(PhantomData)
+    }
+}
+
+impl<P> Routine<()> for NextSetRoutine<P>
+where
+    P: Protocol,
+{
+    fn call<'a>(&'a mut self, conn: &'a mut Conn) -> BoxFuture<'a, crate::Result<()>> {
+        conn.sync_seq_id();
+        async move {
+            conn.read_result_set::<P>(false).await?;
+            Ok(())
+        }
+        .boxed()
+    }
+}
diff --git a/src/conn/routines/ping.rs b/src/conn/routines/ping.rs
new file mode 100644
index 00000000..e5abe0b8
--- /dev/null
+++ b/src/conn/routines/ping.rs
@@ -0,0 +1,23 @@
+use futures_core::future::BoxFuture;
+use futures_util::FutureExt;
+use mysql_common::constants::Command;
+
+use crate::Conn;
+
+use super::Routine;
+
+/// A routine that executes `COM_PING`.
+#[derive(Debug, Copy, Clone)]
+pub struct PingRoutine;
+
+impl Routine<()> for PingRoutine {
+    fn call<'a>(&'a mut self, conn: &'a mut Conn) -> BoxFuture<'a, crate::Result<()>> {
+        async move {
+            conn.write_command_raw(vec![Command::COM_PING as u8])
+                .await?;
+            conn.read_packet().await?;
+            Ok(())
+        }
+        .boxed()
+    }
+}
diff --git a/src/conn/routines/prepare.rs b/src/conn/routines/prepare.rs
new file mode 100644
index 00000000..813317e7
--- /dev/null
+++ b/src/conn/routines/prepare.rs
@@ -0,0 +1,48 @@
+use std::{borrow::Cow, sync::Arc};
+
+use futures_core::future::BoxFuture;
+use futures_util::FutureExt;
+use mysql_common::constants::Command;
+
+use crate::{queryable::stmt::StmtInner, Conn};
+
+use super::Routine;
+
+/// A routine that performs `COM_STMT_PREPARE`.
+#[derive(Debug, Clone)]
+pub struct PrepareRoutine {
+    query: Arc<str>,
+}
+
+impl PrepareRoutine {
+    pub fn new(raw_query: Cow<'_, str>) -> Self {
+        Self {
+            query: raw_query.into_owned().into_boxed_str().into(),
+        }
+    }
+}
+
+impl Routine<Arc<StmtInner>> for PrepareRoutine {
+    fn call<'a>(&'a mut self, conn: &'a mut Conn) -> BoxFuture<'a, crate::Result<Arc<StmtInner>>> {
+        async move {
+            conn.write_command_data(Command::COM_STMT_PREPARE, self.query.as_bytes())
+                .await?;
+
+            let packet = conn.read_packet().await?;
+            let mut inner_stmt = StmtInner::from_payload(&*packet, conn.id(), self.query.clone())?;
+
+            if inner_stmt.num_params() > 0 {
+                let params = conn.read_column_defs(inner_stmt.num_params()).await?;
+                inner_stmt = inner_stmt.with_params(params);
+            }
+
+            if inner_stmt.num_columns() > 0 {
+                let columns = conn.read_column_defs(inner_stmt.num_columns()).await?;
+                inner_stmt = inner_stmt.with_columns(columns);
+            }
+
+            Ok(Arc::new(inner_stmt))
+        }
+        .boxed()
+    }
+}
diff --git a/src/conn/routines/query.rs b/src/conn/routines/query.rs
new file mode 100644
index 00000000..e9c59d0a
--- /dev/null
+++ b/src/conn/routines/query.rs
@@ -0,0 +1,31 @@
+use futures_core::future::BoxFuture;
+use futures_util::FutureExt;
+use mysql_common::constants::Command;
+
+use crate::{Conn, TextProtocol};
+
+use super::Routine;
+
+/// A routine that performs `COM_QUERY`.
+#[derive(Debug, Copy, Clone)]
+pub struct QueryRoutine<'a> {
+    data: &'a [u8],
+}
+
+impl<'a> QueryRoutine<'a> {
+    pub fn new(data: &'a [u8]) -> Self {
+        Self { data }
+    }
+}
+
+impl Routine<()> for QueryRoutine<'_> {
+    fn call<'a>(&'a mut self, conn: &'a mut Conn) -> BoxFuture<'a, crate::Result<()>> {
+        async move {
+            conn.write_command_data(Command::COM_QUERY, self.data)
+                .await?;
+            conn.read_result_set::<TextProtocol>(true).await?;
+            Ok(())
+        }
+        .boxed()
+    }
+}
diff --git a/src/conn/routines/reset.rs b/src/conn/routines/reset.rs
new file mode 100644
index 00000000..596e92b7
--- /dev/null
+++ b/src/conn/routines/reset.rs
@@ -0,0 +1,23 @@
+use futures_core::future::BoxFuture;
+use futures_util::FutureExt;
+use mysql_common::constants::Command;
+
+use crate::Conn;
+
+use super::Routine;
+
+/// A routine that performs `COM_RESET_CONNECTION`.
+#[derive(Debug, Copy, Clone)]
+pub struct ResetRoutine;
+
+impl Routine<()> for ResetRoutine {
+    fn call<'a>(&'a mut self, conn: &'a mut Conn) -> BoxFuture<'a, crate::Result<()>> {
+        async move {
+            conn.write_command_data(Command::COM_RESET_CONNECTION, &[])
+                .await?;
+            conn.read_packet().await?;
+            Ok(())
+        }
+        .boxed()
+    }
+}
diff --git a/src/queryable/mod.rs b/src/queryable/mod.rs
index 939db640..8713869a 100644
--- a/src/queryable/mod.rs
+++ b/src/queryable/mod.rs
@@ -21,7 +21,8 @@ use self::{
 };
 
 use crate::{
-    consts::{CapabilityFlags, Command},
+    conn::routines::{PingRoutine, QueryRoutine},
+    consts::CapabilityFlags,
     error::*,
     prelude::{FromRow, StatementLike},
     queryable::query_result::ResultSetMeta,
@@ -95,10 +96,8 @@ impl Conn {
     where
         Q: AsRef<str> + Send + Sync + 'a,
     {
-        self.write_command_data(Command::COM_QUERY, query.as_ref().as_bytes())
-            .await?;
-        self.read_result_set::<TextProtocol>(true).await?;
-        Ok(())
+        self.routine(QueryRoutine::new(query.as_ref().as_bytes()))
+            .await
     }
 }
 
@@ -256,9 +255,7 @@ pub trait Queryable: Send {
 impl Queryable for Conn {
     fn ping(&mut self) -> BoxFuture<'_, ()> {
         BoxFuture(Box::pin(async move {
-            self.write_command_raw(vec![Command::COM_PING as u8])
-                .await?;
-            self.read_packet().await?;
+            self.routine(PingRoutine).await?;
             Ok(())
         }))
     }
@@ -271,7 +268,8 @@ impl Queryable for Conn {
         Q: AsRef<str> + Send + Sync + 'a,
     {
         BoxFuture(Box::pin(async move {
-            self.raw_query(query).await?;
+            self.routine(QueryRoutine::new(query.as_ref().as_bytes()))
+                .await?;
             Ok(QueryResult::new(self))
         }))
     }
diff --git a/src/queryable/query_result.rs b/src/queryable/query_result.rs
index 40bd9d42..1571bd60 100644
--- a/src/queryable/query_result.rs
+++ b/src/queryable/query_result.rs
@@ -7,12 +7,11 @@
 // modified, or distributed except according to those terms.
 
 use mysql_common::row::convert::FromRowError;
-use mysql_common::{io::ReadMysqlExt, packets::parse_local_infile_packet};
-use tokio::prelude::*;
 
 use std::{borrow::Cow, marker::PhantomData, result::Result as StdResult, sync::Arc};
 
 use crate::{
+    conn::routines::NextSetRoutine,
     connection_like::Connection,
     error::*,
     prelude::{FromRow, Protocol},
@@ -140,8 +139,7 @@ where
                     // Consumed result set.
                     if self.conn.more_results_exists() {
                         // More data will follow.
-                        self.conn.sync_seq_id();
-                        self.conn.read_result_set::<P>(false).await?;
+                        self.conn.routine(NextSetRoutine::<P>::new()).await?;
                         return Ok(None);
                     } else {
                         // The end of a query result.
@@ -349,83 +347,3 @@ where
             .and_then(|meta| meta.columns().map(|columns| columns.clone()).ok())
     }
 }
-
-impl crate::Conn {
-    /// Will read result set and write pending result into `self` (if any).
-    pub(crate) async fn read_result_set<P>(&mut self, is_first_result_set: bool) -> Result<()>
-    where
-        P: Protocol,
-    {
-        let packet = match self.read_packet().await {
-            Ok(packet) => packet,
-            Err(err @ Error::Server(_)) if is_first_result_set => {
-                // shortcut to emit an error right to the caller of a query/execute
-                return Err(err);
-            }
-            Err(Error::Server(error)) => {
-                // error will be consumed as a part of a multi-result set
-                self.set_pending_result(Some(ResultSetMeta::Error(error)));
-                return Ok(());
-            }
-            Err(err) => {
-                // non-server errors are fatal
-                return Err(err);
-            }
-        };
-
-        match packet.get(0) {
-            Some(0x00) => {
-                self.set_pending_result(Some(P::result_set_meta(Arc::from(
-                    Vec::new().into_boxed_slice(),
-                ))));
-            }
-            Some(0xFB) => self.handle_local_infile::<P>(&*packet).await?,
-            _ => self.handle_result_set::<P>(&*packet).await?,
-        }
-
-        Ok(())
-    }
-
-    /// Will handle local infile packet.
-    pub(crate) async fn handle_local_infile<P>(&mut self, packet: &[u8]) -> Result<()>
-    where
-        P: Protocol,
-    {
-        let local_infile = parse_local_infile_packet(&*packet)?;
-        let (local_infile, handler) = match self.opts().local_infile_handler() {
-            Some(handler) => ((local_infile.into_owned(), handler)),
-            None => return Err(DriverError::NoLocalInfileHandler.into()),
-        };
-        let mut reader = handler.handle(local_infile.file_name_ref()).await?;
-
-        let mut buf = [0; 4096];
-        loop {
-            let read = reader.read(&mut buf[..]).await?;
-            self.write_packet(&buf[..read]).await?;
-
-            if read == 0 {
-                break;
-            }
-        }
-
-        self.read_packet().await?;
-        self.set_pending_result(Some(P::result_set_meta(Arc::from(
-            Vec::new().into_boxed_slice(),
-        ))));
-        Ok(())
-    }
-
-    /// Helper that handles result set packet.
-    ///
-    /// Requires that `packet` contains non-zero length-encoded integer.
-    pub(crate) async fn handle_result_set<P>(&mut self, mut packet: &[u8]) -> Result<()>
-    where
-        P: Protocol,
-    {
-        let column_count = packet.read_lenenc_int()?;
-        let columns = self.read_column_defs(column_count as usize).await?;
-        let meta = P::result_set_meta(Arc::from(columns.into_boxed_slice()));
-        self.set_pending_result(Some(meta));
-        Ok(())
-    }
-}
diff --git a/src/queryable/stmt.rs b/src/queryable/stmt.rs
index 7c6a8025..818e67ae 100644
--- a/src/queryable/stmt.rs
+++ b/src/queryable/stmt.rs
@@ -7,21 +7,17 @@
 // modified, or distributed except according to those terms.
 
 use mysql_common::{
-    constants::MAX_PAYLOAD_LEN,
     named_params::parse_named_params,
-    packets::{
-        column_from_payload, parse_stmt_packet, ComStmtClose, ComStmtExecuteRequestBuilder,
-        ComStmtSendLongData, StmtPacket,
-    },
+    packets::{column_from_payload, parse_stmt_packet, ComStmtClose, StmtPacket},
 };
 
 use std::{borrow::Cow, sync::Arc};
 
 use crate::{
-    consts::{CapabilityFlags, Command},
+    conn::routines::{ExecRoutine, PrepareRoutine},
+    consts::CapabilityFlags,
     error::*,
-    queryable::BinaryProtocol,
-    Column, Params, Value,
+    Column, Params,
 };
 
 /// Result of a `StatementLike::to_statement` call.
@@ -217,25 +213,7 @@ impl crate::Conn {
     ///
     /// `raw_query` is a query with `?` placeholders (if any).
     async fn prepare_statement(&mut self, raw_query: Cow<'_, str>) -> Result<Arc<StmtInner>> {
-        let raw_query: Arc<str> = raw_query.into_owned().into_boxed_str().into();
-
-        self.write_command_data(Command::COM_STMT_PREPARE, raw_query.as_bytes())
-            .await?;
-
-        let packet = self.read_packet().await?;
-        let mut inner_stmt = StmtInner::from_payload(&*packet, self.id(), raw_query)?;
-
-        if inner_stmt.num_params() > 0 {
-            let params = self.read_column_defs(inner_stmt.num_params()).await?;
-            inner_stmt = inner_stmt.with_params(params);
-        }
-
-        if inner_stmt.num_columns() > 0 {
-            let columns = self.read_column_defs(inner_stmt.num_columns()).await?;
-            inner_stmt = inner_stmt.with_columns(columns);
-        }
-
-        let inner_stmt = Arc::new(inner_stmt);
+        let inner_stmt = self.routine(PrepareRoutine::new(raw_query)).await?;
 
         if let Some(old_stmt) = self.cache_stmt(&inner_stmt) {
             self.close_statement(old_stmt.id()).await?;
@@ -253,84 +231,8 @@ impl crate::Conn {
     where
         P: Into<Params>,
     {
-        let mut params = params.into();
-        loop {
-            match params {
-                Params::Positional(params) => {
-                    if statement.num_params() as usize != params.len() {
-                        Err(DriverError::StmtParamsMismatch {
-                            required: statement.num_params(),
-                            supplied: params.len() as u16,
-                        })?
-                    }
-
-                    let params = params.into_iter().collect::<Vec<_>>();
-
-                    let (body, as_long_data) =
-                        ComStmtExecuteRequestBuilder::new(statement.id()).build(&*params);
-
-                    if as_long_data {
-                        self.send_long_data(statement.id(), params.iter()).await?;
-                    }
-
-                    self.write_command_raw(body).await?;
-                    self.read_result_set::<BinaryProtocol>(true).await?;
-                    break;
-                }
-                Params::Named(_) => {
-                    if statement.named_params.is_none() {
-                        let error = DriverError::NamedParamsForPositionalQuery.into();
-                        return Err(error);
-                    }
-
-                    params = match params.into_positional(statement.named_params.as_ref().unwrap())
-                    {
-                        Ok(positional_params) => positional_params,
-                        Err(error) => return Err(error.into()),
-                    };
-
-                    continue;
-                }
-                Params::Empty => {
-                    if statement.num_params() > 0 {
-                        let error = DriverError::StmtParamsMismatch {
-                            required: statement.num_params(),
-                            supplied: 0,
-                        }
-                        .into();
-                        return Err(error);
-                    }
-
-                    let (body, _) = ComStmtExecuteRequestBuilder::new(statement.id()).build(&[]);
-                    self.write_command_raw(body).await?;
-                    self.read_result_set::<BinaryProtocol>(true).await?;
-                    break;
-                }
-            }
-        }
-        Ok(())
-    }
-
-    /// Helper, that sends all `Value::Bytes` in the given list of paramenters as long data.
-    async fn send_long_data<'a, I>(&mut self, statement_id: u32, params: I) -> Result<()>
-    where
-        I: Iterator<Item = &'a Value>,
-    {
-        for (i, value) in params.enumerate() {
-            if let Value::Bytes(bytes) = value {
-                let chunks = bytes.chunks(MAX_PAYLOAD_LEN - 6);
-                let chunks = chunks.chain(if bytes.is_empty() {
-                    Some(&[][..])
-                } else {
-                    None
-                });
-                for chunk in chunks {
-                    let com = ComStmtSendLongData::new(statement_id, i, chunk);
-                    self.write_command_raw(com.into()).await?;
-                }
-            }
-        }
-
+        self.routine(ExecRoutine::new(statement, params.into()))
+            .await?;
         Ok(())
     }
 


Test patch: