Pool::acquire times out

[skuzins]

I'm using a connection pool with a larger number of tokio tasks, see the example below. The tasks run a small query, do some work without holding on to a connection, then run another query. In the example, the second call to acquire quickly starts failing with PoolTimeout or it takes too long, roughly connect_timeout seconds.

Any ideas would be appreciated.

use sqlx::mysql::*;
async fn run_queries(i: u32, pool: &sqlx::Pool<MySql>) {
    {
        let mut conn = pool.acquire().await.unwrap();

        let _ = sqlx::query("SELECT 1").fetch_one(&mut conn).await.unwrap();

        // conn gets dropped here and should be returned to the pool
    }

    // (do some other work here without holding on to a connection)

    {
        let start = std::time::Instant::now();

        // this often fails (PoolTimedOut) or takes connect_timeout seconds
        let mut _conn = pool.acquire().await.unwrap();

        println!("{} pool.acquire() took {:?}", i, std::time::Instant::now() - start);

        // (run another query...)

    }
}

#[tokio::main]
async fn main() {
    let pool = MySqlPoolOptions::new()
        .max_connections(1) // also fails with higher counts, e.g. 5
        .connect(&std::env::var("DATABASE_URL").unwrap())
        .await
        .unwrap();

    let mut handles = vec![];
    for i in 0..100 {
        let pool = pool.clone();

        handles.push(tokio::spawn(async move {
            run_queries(i, &pool).await;
        }));
    }

    futures::future::join_all(handles).await;
}

[dependencies]
sqlx = { version = "0.4.0-beta.1", default-features = false, features = [ "runtime-tokio", "mysql" ] }
tokio = "0.2.22"
futures = "0.3.5"

Output snippets:

thread 'tokio-runtime-worker' panicked at 'called `Result::unwrap()` on an `Err` value: PoolTimedOut'
pool.acquire() took 30.001958452s

MySQL Engine version 8.0.17. Tested with stable-x86_64-unknown-linux-gnu and stable-x86_64-pc-windows-msvc.

[cetra3]

I've ran into the same issue. I am not seeing connections recycled when they are dropped, which is causing issues when multiple queries happen on one web request/future.

I have worked around it by using deadpool for connection pooling, which appears to be more reliable in my limited tests.

Here's my (very simple) implementation. (also note: I needed to use DerefMut to coax the compiler, as I am not sure what combination of * and & is needed here.)

use std::ops::DerefMut;
use sqlx::{Connect, Connection, Error as SqlxError, Executor, PgConnection};

use async_trait::async_trait;
use deadpool::managed::{Manager, RecycleResult};
use log::*;
type Pool = deadpool::managed::Pool<PgConnection, SqlxError>;

struct DbPool {
    url: String,
}

impl DbPool {
    fn new(url: String, size: usize) -> Pool {
        Pool::new(DbPool { url }, size)
    }
}

#[async_trait]
impl Manager<PgConnection, SqlxError> for DbPool {
    async fn create(&self) -> Result<PgConnection, SqlxError> {
        PgConnection::connect(&self.url).await
    }
    async fn recycle(&self, obj: &mut PgConnection) -> RecycleResult<SqlxError> {
        Ok(obj.ping().await?)
    }
}



async fn main() {
    ...
    let pool = DbPool::new(url, 16);
    ...
    sqlx::query_as("select * from users where email = $1")
      .bind(&email)
      .fetch_one(pool.get().await?.deref_mut())
      .await?;
     ... 
}

[skuzins]

Thanks for sharing your workaround! deadpool looks great.

[abonander]

Currently the pool doesn't behave well if the acquire() future is cancelled; it fails to remove the waiter from the list, which holds up the queue. This could have a knock-on effect with other acquire() calls timing out due to the same problem. Each time a connection is returned to the pool it only attempts to wake the task at the head of the queue. If there's more dead tasks than connections in the queue then the pool will stall.

@skuzins in your example, could you please try debug-printing the pool after the second .acquire() call times out, instead of panicking:

if let Err(e) = pool.acquire().await {
    println!("acquire error: {}, pool state: {:?}", e, pool);
}

[skuzins]

I'm not sure any future would/should normally be cancelled in this example - I believe they should all just run to completion.

Here's the debug message, please let me know if you need more information.

acquire error: pool timed out while waiting for an open connection, pool state: 
Pool { size: 1, num_idle: 0, is_closed: false, options: PoolOptions { 
max_connections: 1, min_connections: 0, connect_timeout: 30s, 
max_lifetime: Some(1800s), idle_timeout: Some(600s), test_before_acquire: true } }

[abonander]

The timeout itself cancels the internal future that's actually waiting in the queue.

However, I'd expect to see num_idle: 1 if that was the case. How about with .max_connections(5)?

[cetra3]

I am not seeing cancellations in my code and the example posted doesn't have them either. It feels more like an issue with waking up after the pool has stopped using a connection.

[skuzins]

@abonander The thing is, there shouldn't be any timeouts. The task acquires the connection, performs a short query, drops the connection and does the same thing once more. The whole program should finish within milliseconds without any timeouts.

Here's the output with .max_connections(5):

acquire error: pool timed out while waiting for an open connection, pool state: Pool 
{ size: 5, num_idle: 0, is_closed: false, options: PoolOptions { 
max_connections: 5, min_connections: 0, connect_timeout: 30s, 
max_lifetime: Some(1800s), idle_timeout: Some(600s), test_before_acquire: true } }

[jgrund]

FWIW, I think I'm hitting this (or something like it) when using try_acquire with Postgres.

I have a loop which waits 5 seconds and calls try_acquire to see if any connections are available. The loop also does some work with the pool in a spawned task.

Once the 5 seconds elapse the pool shows all connections idle, but try_acquire returns None.