# Concurrency

Multiple threads or asynchronous tasks can be used to speed up queries to a SurrealDB database

While the Rust SDK for SurrealDB uses the tokio async runtime, the operation of the database itself will only take place concurrently if the code itself uses concurrency. The following example shows how to do this and a comparison of the performance between synchronous and asynchronous usage.

## Getting started

Start a running database using the following command:

```bash
surreal start --user root --pass secret 
```

Then use the `cargo add` command to add the `surrealdb` and `tokio` crates.

## The code

The SurrealDB client by nature has no maximum capacity for the number of channels that can be made to perform queries. To change this setting, the [`.with_capacity()`](https://docs.rs/surrealdb/latest/surrealdb/struct.Connect.html#method.with_capacity) method can be used. As tokio's documentation notes, this method is useful if your client is running so many queries concurrently that the machine used to execute them is running out of memory.

> This is used to set bounds of the channels used internally as well set the capacity of the HashMap used for routing responses in case of the WebSocket client. Setting this capacity to 0 (the default) means that unbounded channels will be used. If your queries per second are so high that the client is running out of memory, it might be helpful to set this to a number that works best for you.

To experiment with this, we will create three clients: one with no maximum capacity, a second with a maximum capacity of 1, and a third with a maximum capacity of 1000. To avoid any code duplication, we'll use an enum to set these clients up.

```rust
enum DbType {
    Standard,
    With1,
    With1000,
}

impl DbType {
    async fn generate(self) -> Result<Surreal<Client>, Error> {
        let db = match self {
            DbType::Standard => Surreal::new::<Ws>("localhost:8000").await,
            DbType::With1 => Surreal::new::<Ws>("localhost:8000").with_capacity(1).await,
            DbType::With1000 => {
                Surreal::new::<Ws>("localhost:8000")
                    .with_capacity(1000)
                    .await
            }
        }?;
        db.use_ns("main").use_db("main").await?;
        db.signin(Root {
            username: "root",
            password: "secret",
        })
        .await?;
        Ok(db)
    }
}
```

Each of these clients will be put into a `for` loop 50000 times, in which they will simply return the index number of the current iteration of the loop. Note that the [`.take()`](https://docs.rs/surrealdb/latest/surrealdb/struct.IndexedResults.html#method.take) method returns a query result as anything that can be deserialised, including primitive types like `usize`. It will also take a `test_num` field so that we can see which test is running as the index numbers fly by on the terminal.

```rust
async fn select_index(db: &Surreal<Client>, idx: usize, test_num: &'static str) {
    let mut result = db
        .query("SELECT * FROM $idx")
        .bind(("idx", idx))
        .await
        .unwrap();

    let db_idx: Option<usize> = result.take(0).unwrap();
    if let Some(db_idx) = db_idx {
        println!("{test_num} - {idx}: {db_idx}");
    }
}
```

The first test inside main will be done synchronously, returning the `Duration` taken by the test once it is done.

```rust
let start = std::time::Instant::now();
for idx in 0..=MAX {
	select_index(&db, idx).await;
}
let res_1 = format!("Regular DB: {:?}", Instant::now() - start);
```

The other three tests will all be done asynchronously, so we'll put them into their own function. Each one will spawn a tokio task which will execute and return the result whenever it is finished. Each task will be put into a `Vec` of `JoinHandle`s that will be awaited on at the end to ensure that we don't exit the function before they have finished.

```rust
async fn async_test(db: Arc<Surreal<Client>>, test_num: &'static str) -> std::time::Duration {
    let start = std::time::Instant::now();
    let mut handles = vec![];
    for idx in 0..=MAX {
        let db = db.clone();
        handles.push(tokio::spawn(async move {
            select_index(&db, idx, test_num).await;
        }));
    }
    for h in handles {
        h.await.unwrap();
    }
    Instant::now() - start
}
```

We will then make a `String` variable from the output of each test, and print them out at the end when everything has run.

The entire code is as follows:

```rust
use std::sync::Arc;
use std::time::Instant;
use surrealdb::engine::remote::ws::{Client, Ws};
use surrealdb::opt::auth::Root;
use surrealdb::{Error, Surreal};

const MAX: usize = 50_000;

enum DbType {
    Standard,
    With1,
    With1000,
}

impl DbType {
    async fn generate(self) -> Result<Surreal<Client>, Error> {
        let db = match self {
            DbType::Standard => Surreal::new::<Ws>("localhost:8000").await,
            DbType::With1 => Surreal::new::<Ws>("localhost:8000").with_capacity(1).await,
            DbType::With1000 => {
                Surreal::new::<Ws>("localhost:8000")
                    .with_capacity(1000)
                    .await
            }
        }?;
        db.use_ns("main").use_db("main").await?;
        db.signin(Root {
            username: "root",
            password: "secret",
        })
        .await?;
        Ok(db)
    }
}

async fn select_index(db: &Surreal<Client>, idx: usize, test_num: &'static str) {
    let mut result = db
        .query("SELECT * FROM $idx")
        .bind(("idx", idx))
        .await
        .unwrap();

    let db_idx: Option<usize> = result.take(0).unwrap();
    if let Some(db_idx) = db_idx {
        println!("{test_num} - {idx}: {db_idx}");
    }
}

async fn async_test(db: Arc<Surreal<Client>>, test_num: &'static str) -> std::time::Duration {
    let start = std::time::Instant::now();
    let mut handles = vec![];
    for idx in 0..=MAX {
        let db = db.clone();
        handles.push(tokio::spawn(async move {
            select_index(&db, idx, test_num).await;
        }));
    }
    for h in handles {
        h.await.unwrap();
    }
    Instant::now() - start
}

#[tokio::main]
async fn main() -> Result<(), Error> {

    let db_standard = Arc::new(DbType::Standard.generate().await?);
    let db_with_1 = Arc::new(DbType::With1.generate().await?);
    let db_with_1000 = Arc::new(DbType::With1000.generate().await?);

    let start = std::time::Instant::now();
    for idx in 0..=MAX {
        select_index(&db_standard, idx, "Test1").await;
    }

    let res_1 = format!("Regular DB: {:?}", Instant::now() - start);
    let res_2 = format!("Async with capacity 1: {:?}", async_test(db_with_1, "Test2").await);
    let res_3 = format!(
        "Async with capacity 1000: {:?}",
        async_test(db_with_1000, "Test3").await
    );
    let res_4 = format!(
        "Async with unbounded capacity: {:?}",
        async_test(db_standard, "Test4").await
    );

    println!("{res_1}\n{res_2}\n{res_3}\n{res_4}");

    Ok(())
}
```

Running the code, you should see the following:

1) The first test runs one index and one query at a time, taking by far the longest time.
2) The second test will run much faster. Despite only having a capacity of 1, it still runs concurrently and does not need to wait for the output of the previous query to send in its own.
3) The third test with a capacity of 1000 will run even faster,
4) The last test using the first unbounded database should run fastest of all. If running the test on a particularly slow computer, however, you may see a slowdown compared to the other two async tests if the computer's memory capacity is reached during the test.

A sample of the output at the end:

```text
Regular DB: 5.244320833s
Async with capacity 1: 1.523926416s
Async with capacity 1000: 1.441300833s
Async with unbounded capacity: 1.183820584s
```

## Using a channel instead of a JoinHandle

Besides the classic method of using a `JoinHandle` for each thread or task to wait until all have completed their operation, a [channel](https://docs.rs/tokio/latest/tokio/sync/mpsc/fn.channel.html) can also be used. A channel can be used in the following way:

* Create a channel with a buffer of 1 (the minimum size), as the channel will not be used to actually send data,
* Clone the sender and send it into each iteration of the `for` loop,
* Drop the cloned sender once the database query is done,
* Drop the original sender after the `for` loop,
* Call [`.recv()`](https://docs.rs/tokio/latest/tokio/sync/mpsc/struct.Receiver.html#method.recv) on the receiver at the very end.

As tokio's documentation shows, the `.recv()` method will cause the receiver to sleep as it waits for each task to complete, and at the end it will close once the last sender has been dropped.

```text
This method returns None if the channel has been closed and there are no remaining messages in the channel’s buffer. This indicates that no further values can ever be received from this Receiver. The channel is closed when all senders have been dropped, or when close is called.

If there are no messages in the channel’s buffer, but the channel has not yet been closed, this method will sleep until a message is sent or the channel is closed.
```

Here is what the `async_test()` function looks like using this method.

```rust
async fn async_test(db: Arc<Surreal<Client>>, test_num: &'static str) -> std::time::Duration {
    let (tx, mut rx) = mpsc::channel::<()>(1);
    let start = std::time::Instant::now();

    for idx in 0..=MAX {
        let sender = tx.clone();
        let db = db.clone();
        tokio::spawn(async move {
            select_index(&db, idx, test_num).await;
            drop(sender);
        });
    }
    drop(tx);

	rx.recv().await;

    Instant::now() - start
}
```