Still a work in progress.

This is a very direct port of the existing codebase. I'm avoiding rearchitecting as much as I can, and instead doing a mostly mechanical transformation. Seems to be going okay so far, but I'm not quite done yet (need https://github.com/tokio-rs/tokio/pull/1537 for example). There are definitely a bunch of places the codebase could be improved (in particular, a bunch of Boxes that could be removed), but at least this will get us off the ground.

Fixes #63.

In my project, I need a listener for MySQL's binlog events. There are 2 crates available that can parse the Binlog event data, but there is no network-based listener. I think this could be a nice addition to this crate.

Over the weekend, I built a prototype that adds commands for registering to the binlog events. I am happy to work on this further in a PR, but first wanted to get your feedback on this. See: https://github.com/blackbeam/mysql_async/compare/master...normanrz:expose-binlog

The API is as follows: The client code creates a Conn, registers that conn as a slave and then requests binlog events. After that, the client can repeatedly call read_packet to receive the next binlog event, which needs to be parsed and used. The conn can't be used for regular queries anymore.

Here is a runnable test: https://github.com/normanrz/mysql_async/blob/expose-binlog/src/binlog/mod.rs Please note that mysqld needs to be configured for binlogs: https://github.com/normanrz/mysql_async/blob/expose-binlog/.circleci/config.yml#L17 In my example code, I use the mysql_binlog crate for parsing. There is a PR for my changes in that crate.

Would you rather implement the command code for COM_REGISTER_SLAVE and COM_BINLOG_DUMP in mysql_common? Also, I couldn't figure out how to expose read_packet for Conn as a public function.

My implementation is built upon @dannysu's fork: https://github.com/blackbeam/rust-mysql-simple/compare/master...dannysu:binlogsync

Try running the following test:

#[tokio::test(threaded_scheduler)]
async fn can_handle_the_pressure() {
    use futures_util::stream::StreamExt;
    {
        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel::<()>();
        for _ in 0..500 {
            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);
            });
        }
        drop(tx);
        // see that all the tx's eventually complete
        assert_eq!(rx.next().await, None);
    }
    eprintln!("MID");
    {
        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel::<()>();
        for _ in 0..500 {
            let tx = tx.clone();
            tokio::spawn(async move {
                eprintln!("GETCONN");
                let c = Conn::new(get_opts()).await.unwrap();
                eprintln!("GOTCONN");
                let c = c.drop_query("SELECT 1").await.unwrap();
                c.disconnect().await.unwrap();
                drop(tx);
            });
        }
        drop(tx);
        // see that all the tx's eventually complete
        assert_eq!(rx.next().await, None);
    }
    eprintln!("DONE");
}

Against a local MySQL with max_connections = 1000. It passes just fine. Now, run some CPU-heavy program that takes up most of the cores on the machine, and run the test again. It will frequently hang between printing MID and printing DONE. Digging a big deeper, you will see that GETCONN gets printed 500 times, but GOTCONN only N < 500 times (the exact value of N varies). Some further digging reveals that the issue lies in handle_handshake. Specifically, if you put a print statement above and below the call to read_packet().await, you'll see that the print before gets printed 2*500-N times, and the print after 2*500-2*N times. The explanation for this is that every connection goes through Conn::new twice: once as a "plain" connection, and once as an "upgraded" socket connection. But some (500-N) plain connections get "stuck" on that .await. They therefore do not get upgraded, so they fail to hit the second print twice, and the first print once (as a socket).

In any case, the issue for the stuck connections is that read_packet().await never returns for plain connections. Reading through the code, it appears that there are only three ways this can happen: tokio::net::TcpStream is broken (unlikely), tokio_util::codec::Framed is broken (possible), or mysql_common::proto::codec::PacketCodec either blocks (unlikely) or returns None when it should return Some (a likely culprit). How any of these are related to load, I have no idea. Why it is necessary to first spin up and drop 500 connections, I also have no idea. But that's what I'm seeing. Which sort of points at a tokio bug, but that seems weird too?

Well, I figured I should write up my findings no matter what, so that hopefully we can get to the bottom of it. I ran into this when running code on EC2 which is infamous for having a lot of scheduling noise, similar to that of a loaded machine, so it's probably worth trying to fix.

While connecting to the same server referenced for this mysql simple issue https://github.com/blackbeam/rust-mysql-simple/issues/160 I get an error about unexpected packet order using mysql_async. My first thought is perhaps the feature that was added in mysql simple just hasn't been implement in mysql_async. I can get any extra details you need. This is using SSL without a server public certificate and with domain verification off.

Did a cargo update, now my pool get_conn throws these errors:

Io(Custom { kind: UnexpectedEof, error: "can't parse: buf doesn't have enough data" })

Reverted to the previous Cargo.toml, works fine. Not sure if it's mysql_async or a dependent package, but I can't debug it more fine-grained at the moment.

I have a wrapper around a Pool object (using the new async update) which has a disconnect method:

pub async fn disconnect(&self) -> Result<()> {
    self.pool.clone().disconnect().await?;
    Ok(())
}

When I want to exit my program, I run this on my tokio runtime with ::block_on. However, it never returns.

cc @jonhoo

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:

#[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).

This should be a first step towards fixing #68.

We currently don't maintain pool invariants if a connection errors or is dropped (exist to be specific). This patch should fix that.

Also renames should_hold_bounds to should_hold_bounds1 to make it easier to run that test by itself.

Hi, I am trying mysql_async with warp something like this:

The pool:

let mut opts = mysql_async::OptsBuilder::new();
    opts.user(dsn.username);
    opts.pass(dsn.password.clone());
    opts.ip_or_hostname(dsn.host.unwrap());
    opts.socket(dsn.socket);
    opts.db_name(dsn.database);
    if let Some(port) = dsn.port {
        opts.tcp_port(port);
    }
    opts.pool_options(mysql_async::PoolOptions::with_constraints(
        mysql_async::PoolConstraints::new(10, 100).unwrap(),
    ));

    // mysql ssl options
    let mut ssl_opts = mysql_async::SslOpts::default();
    if let Some(tls) = dsn.params.get("tls") {
        if *tls == "skip-verify" {
            ssl_opts.set_danger_accept_invalid_certs(true);
        }
    }
    opts.ssl_opts(ssl_opts);

    mysql_async::Pool::new(opts)

Then with warp:

let pool = options::new();
    pool.get_conn().await.unwrap_or_else(|e| {
        eprintln!("Could not connect to MySQL: {}", e);
        process::exit(1);
    });
    println!("{:#?}", pool);

 let db = warp::any().map(move || pool.clone());

    let register = warp::post()
        .and(warp::path("register"))
        .and(warp::body::json())
        .and(db.clone())
        .and_then(register);

    warp::serve(register).run(([0, 0, 0, 0], 8000)).await;


async fn register(
    user: User,
    pool: mysql_async::Pool,
) -> Result<impl warp::Reply, warp::Rejection> {
 match queries::find_user(pool.clone(), &user.username, &user.email).await {
        Ok(rs) => {
             println!("{:?}", rs);
             Ok(StatusCode::CREATED)
       }
        err => {
            eprintln!("{:?}", err);
           Ok(StatusCode::INTERNAL_SERVER_ERROR)
        }
    }
}

and queries is something like:

pub async fn find_user(
    pool: mysql_async::Pool,
    username: &str,
    email: &str,
) -> Result<usize, Box<dyn Error>> {
    let conn = pool.get_conn().await?;
    let (_, row) = conn
        .first_exec::<_,_,usize>(
            "SELECT IF((SELECT count(*) FROM users WHERE email=?), 2, (IF((SELECT count(*) FROM users WHERE name=?), 1, 0)))",
            (&email, &username),
        ).await?;
    Ok(row.ok_or("expecting a row")?)
}

It works but in some cases, I am getting an Err(Driver(ConnectionClosed)), also the pool is always one connection instead of 10 as defined, therefore wondering what could be wrong or if I should close the connection after every use.

Any ideas?

Hi! I meet a problem while executing transaction exec_drop,some times I got error result like: Io(Io(Custom { kind: UnexpectedEof, error: "failed to fill whole buffer" })) But when i tried again,it's just ok.

On version 0.25.0 my code like this: trans.exec_drop( "INSERT INTO stock_order_sku (order_id,sku_id,amount) VALUES(?,?,?)", (order_id, sku, amount), ).await.map_err(|e| error!("fail to insert sku info {:?}",e)) .is_ok();

In light of stable async/await, a lot of these collect/try_collect/map/etc. methods seem redundant if we could instead stream the async results.

e.g. this pattern is replicated with minor changes in several places:

    pub async fn for_each<F>(self, mut fun: F) -> Result<Self>
    where
        F: FnMut(Row),
    {
        if self.is_empty() {
            Ok(self)
        } else {
            let mut qr = self;
            loop {
                let (qr_, row) = qr.get_row().await?;
                qr = qr_;
                if let Some(row) = row {
                    fun(row);
                } else {
                    break Ok(qr);
                }
            }
        }
    }

Why not simply swap out the fun(row); and instead make that the poll on a Stream? The big thing this would open up for me, is I could impl From<my::Row> (which would call the FromRow), and then let the From and standard collect() calls build my result sets.

As it's written, I'm stuck forced to build full Vecs of my entire QueryResult, when a stream would let me map, filter, zip, etc., and let type inference handle my result from collecting (could even make a HashMap or whatever). Would also make it a lot easier to return an Err on the very first TryFrom<my::Row> that fails, instead of building the full result with all errors like try_collect does now.