Rust with async

A Rust program can await TinyStore instead of blocking a thread on it. The async feature adds tinystore::aio, whose futures run on tokio, async-std, smol or any other executor, because TinyStore does not bring a runtime of its own. Every engine is there: each call that reads, writes or waits returns a future, and you run a transaction and close the store with .await too.

#Turn it on

[dependencies]
tinystore = { version = "0.1", features = ["async"] }
tokio = { version = "1", features = ["rt-multi-thread", "macros"] }

use tinystore::Options;
use tinystore::aio::Store;

#[tokio::main]
async fn main() -> tinystore::Result<()> {
    let store = Store::open("data", Options::default())?; // a plain call, made once at start
    let stock = store.bucket::<i64>("stock").open()?; // so is opening a handle

    stock.set("sku-1", &10).await?;
    assert_eq!(stock.get("sku-1").await?, Some(10));

    let taking = stock.clone(); // the closure owns its handle
    let left = store
        .tx(move |tx| -> tinystore::Result<i64> {
            let left = tx.with(&taking).get("sku-1")?.unwrap_or(10);
            tx.with(&taking).set("sku-1", &(left - 1))?;
            Ok(left - 1)
        })
        .await?; // 9: the transaction ran on the store's own threads
    println!("{left} left");

    store.close().await // closes on a thread of its own
}

aio::Store opens as tinystore::Store does. Opening is a plain call that you make once at start, because a first open may migrate a file. Every engine's handle opens from aio::Store with the same builder as in the blocking API, and you await its calls. A bucket, a queue or a table's search index kept in a database opens from the async database the same way. Each engine's guide lists its awaited calls, and the table at the end links them.

Every future is Send + 'static, so you can pass it to tokio::spawn, except those that borrow what you give them. run of once is 'static only when the key, the function and the function's future are, and a lock's or a semaphore's hold and try_hold only when their function owns what it uses. A job run's step borrows the run, and an upload's write and a file's read_at borrow their bytes: you await those in place. A future does nothing until it is first polled.

#Where the work runs

No call commits, syncs the disk or waits on the thread that polls its future (contract STORE-10). So your executor's threads never wait for the disk. A point read that would wait in the poll, for a reader or for a write, runs again on the store's threads instead.

On a node that copies a file from its home, a write passes on to the home, and a once run claims its key there. Each of them waits for the home's answer on a thread that TinyStore starts for the call, not on the store's threads. So a slow home, or a run under way at the home, never holds the threads that your other calls need.

Once you poll a write, it commits even if you drop its future (contract DUR-11). A write whose future you never poll writes nothing. A read whose future you drop stops where it is and gives its reader back (contract STORE-09), so tokio::time::timeout around a read stops it.

The store's threads are a pool of at most sixteen, and each running worker adds a thread of its own beside them for its claims. TinyStore starts them as calls need them, ends each one after a minute without work, and stops all of them when the store closes. If all sixteen are busy when a commit needs a thread, the commit gets a thread of its own. This matters because the busy threads may be waiting for the memory that the commit gives back.

#What dropping a future does

A future that is never polled reads and writes nothing. After its first poll:

• A read stops. Its SQLite statement is interrupted and its reader goes back at once, as if a CancellationToken was cancelled (contract STORE-09). This is what a web handler wants when its client goes away. tokio::time::timeout(duration, call) stops a read this way when the time runs out.
• A write commits. Writes share a commit, and a commit cannot stop half way, so a write that was polled commits whether or not anyone still awaits it (contract DUR-11). Only its answer is lost. To know that a write happened, await it.
• A transaction runs to its end, as a write does.

A CancellationToken still works around a call. Make the future inside token.run(...), and its read stops when you call token.cancel(): it fails with a cancelled error (ErrorKind::Cancelled).

#Transactions

let taking = stock.clone();
let placed = store
    .tx(move |tx| -> tinystore::Result<bool> {
        let left = tx.with(&taking).get("sku-1")?.unwrap_or(0);
        if left < 1 {
            return Ok(false); // sold out: nothing is written
        }
        tx.with(&taking).set("sku-1", &(left - 1))?;
        Ok(true)
    })
    .await?;

This block uses the store and stock of the first one. The closure of store.tx is blocking, and it runs on one of the store's threads, never on your executor's thread. A transaction holds its file's writer, so it must not wait for your program's other tasks. Inside the closure, the calls go through tx.with(&handle), exactly as in the blocking API. If the closure returns an error, the transaction rolls back and the future returns that error. If the closure panics, the transaction rolls back and the panic continues where you await the future.

Once its future is polled, a transaction runs to its end, even if you drop the future. To know whether it committed, await it.

Don't call block_on on an awaited call inside the closure. That call would wait for the writer the transaction holds, so it fails with an invalid error (ErrorKind::Invalid) instead.

#Streams

use futures::StreamExt; // next() on a Stream

let events = store.channel::<String>("chat-events").open()?;
let mut each = events.under("chat-7").each()?; // listens from here
while let Some(message) = each.next().await {
    println!("{}", message?.value);
}

This block uses the store of the first one. A loop over a channel's messages, a watch of a config or a queue, a walk over every key, job, file or record and a tail of records are each a Stream of the futures-core crate, and so is a file that Blobs' get returns, piece by piece. So next() from futures::StreamExt or tokio_stream::StreamExt reads them, and every item is a Result. A stream of messages or changes blocks no thread while it waits, because a publish or a change wakes it, and a walk reads one page at a time on the store's threads. A channel has no handler form in the async API: to handle its messages in the background, spawn a task that reads the stream.

#Blocking and async in one program

let store = Store::open("data", Options::default())?;
let tool = store.blocking(); // the same store, for blocking calls
let again = Store::from(tool.clone()); // the same store, its calls awaited

Store::from and store.blocking() cross between the two APIs. Both are cheap copies of one store, its engines and its files, never a second store. A command-line tool can use the blocking API while the server beside it uses the async one.

#Errors and closing

An awaited call fails with the same tinystore::Error and ErrorKind as its blocking twin. A call made after its store closed fails with a closed error (ErrorKind::Closed), and so does a call that was still waiting for the store's threads when the store closed.

There is one way to time out a call: your runtime's timeout, such as tokio::time::timeout. When it drops the future, a read stops and a write still commits.

store.close().await closes the store on a thread that the call starts. It waits for background work, for the handlers that run and for each engine's last writes, so your executor's thread waits for nothing. If you drop the last handle instead, the store closes on the thread that drops it, which may be your executor's thread. Await close instead.

#Each engine's calls

Each engine's guide shows its awaited calls in a section named Async Rust:

#Why this design?

Why no runtime of its own. An async library that depends on tokio forces tokio on every program that uses it, and one that supports several runtimes needs a feature for each. TinyStore's work is SQLite, which is synchronous, so a runtime would only move it to a thread. TinyStore already has its own threads for that, so its futures wake through the standard Waker and any executor runs them.

Why a transaction's closure is blocking. A transaction holds its file's writer until it commits. If its closure could await other tasks, every write to that file would wait for them, and one slow request would stall all the others. A blocking closure on the store's threads keeps the writer for the time the transaction's own reads and writes take.

Why a polled write commits. Writes that arrive together share one commit and one sync of the disk. Once a write is queued, other writes may be committing with it, and taking it out would mean failing them too. So a dropped future only stops waiting for the answer. A read is different: it changes nothing, so stopping it is always safe.

#See also

• One directory, many engines: the store, its files and the ways to reach it.
• Store API for Rust: aio::Store and its calls.