How can I make tokio run multiple sqlx executions at the same time?

[fritshoogland-yugabyte]

I am trying to build a tool to execute SQL against YugabyteDB, a PostgreSQL compatible database.
So this is about tokyo and sqlx.

This works great for sequentially executed SQL.
But now I want to run one or more different SQL statements at the same time, preferably with a controlled limit on the concurrent amount. I cannot get this to work. This is what I have got so far:

use sqlx::postgres::PgPoolOptions;

#[tokio::main]
async fn main() -> Result<(), sqlx::Error> {
    dbg!("start");
    let pool = PgPoolOptions::new()
        .min_connections(5)
        .max_connections(5)
        .test_before_acquire(false)
        .connect("postgres://yugabyte:[email protected]:5433/yugabyte").await?;

    dbg!("loop");
    tokio::spawn( async move {
        for _ in 0..100000000 {
            let row: (i64, ) = sqlx::query_as("SELECT $1").bind(150_i64).fetch_one(&pool).await?;
            print!(".");
        };
    });

    dbg!("end");
    Ok(())
}

this returns the following error:

error[E0277]: the trait bound `(): FromResidual<Result<Infallible, sqlx::Error>>` is not satisfied
  --> src/main.rs:23:30
   |
23 |       tokio::spawn( async move {
   |  ______________________________-
24 | |         for _ in 0..100000000 {
25 | |             //let row: (i64, ) = sqlx::query_as("SELECT $1").bind(150_i64).fetch_one(&pool).await?;
26 | |             let row: (i64, ) = sqlx::query_as("SELECT $1").bind(150_i64).fetch_one(&pool).await?;
...  |
29 | |         };
30 | |     });
   | |     ^
   | |     |
   | |_____the trait `FromResidual<Result<Infallible, sqlx::Error>>` is not implemented for `()`
   |       required by a bound introduced by this call

How can I make rust execute multiple sqlx queries at the same time?

[Darksonn]

The async block that you are spawning doesn't return a Result, so you can't use the question mark operator. Either stop using the question mark operator or have it return a Result. It has nothing to do with sqlx.

[fritshoogland-yugabyte]

thank you for your answer. So you are saying that the question mark creates a result, and the block that is executed cannot consume the result, and therefore throws this error; and thus for the spawned thread to be able to run, the code block should not return anything?

[PoorlyDefinedBehaviour]

Let me try to help 😄

The ? operator is a nice way to propagate errors to the caller of the current function.

Since you are using the ? operator an error may be returned which means the function returns a Result<T, E>

tokio::spawn(async move {
  for _ in 0..100000000 {
    // Err(E) is returned if sqlx::query_as(...) fails because of `?`
    let row: (i64, ) = sqlx::query_as("SELECT $1").bind(150_i64).fetch_one(&pool).await?;
  }
});

The problem is that if nothing goes wrong and no errors are returned by the sqlx query, you are not returning something of type Result<T, E> or that can be converted to Result<T, E> at the end of your function.

This code does not compile for the same reason:

fn foo() -> Result<i32, i32> {
    Ok(1)
}

#[tokio::main]
async fn main() {
    tokio::spawn(async move {
        let x = foo()?;
    });
}

but this one does:

fn foo() -> Result<i32, i32> {
    Ok(1)
}

#[tokio::main]
async fn main() {
    tokio::spawn(async move {
        let x = foo()?;

        Result::<i32, i32>::Ok(1)
    });
}