Async all the way down: streaming results and connection multiplexing

This is the post I'd been looking forward to writing, because it's where the Rust bet from Part 1 really pays rent. Two things landed: result sets became async streams, and the async client learned to multiplex many concurrent queries over a few shared sockets. Then I benchmarked it, because claims without numbers are just vibes.

Results are streams

On the async client, a result set's data is a Stream<Item = FalkorResult<Row>>. Same fallible, header-aware row from Part 3 — now it arrives lazily and composes with the entire futures toolbox.

sequenceDiagram
    autonumber
    participant App
    participant S as Row stream
    participant DB as FalkorDB
    App->>DB: GRAPH.RO_QUERY
    DB-->>S: header + result set
    loop for each row, lazily
        App->>S: .next().await
        S-->>App: Ok(Row) — or Err you can `?`
    end
    Note over App,S: compose with the futures toolbox:<br/>map / try_collect / buffer_unordered(8)

Here's the real examples/async_stream.rs:

/*
 * Copyright FalkorDB Ltd. 2023 - present
 * Licensed under the MIT License.
 */

#![recursion_limit = "256"]

//! Async streaming results with `futures::StreamExt` / `TryStreamExt`.
//!
//! Run with: `cargo run --example async_stream --features tokio`
//!
//! Requires a running FalkorDB instance (defaults to `127.0.0.1:6379`).

use falkordb::{FalkorClientBuilder, FalkorResult};
use futures::{StreamExt, TryStreamExt};

#[tokio::main(flavor = "multi_thread")]
async fn main() -> FalkorResult<()> {
    let client = FalkorClientBuilder::new_async()
        .with_connection_info("falkor://127.0.0.1:6379".try_into()?)
        .build()
        .await?;
    let mut graph = client.select_graph("async_stream_example");
    let _ = graph.delete().await;

    graph
        .query("UNWIND range(1, 5) AS i CREATE (:Movie {year: 1990 + i})")
        .execute()
        .await?;

    // ── 1. Collect a whole typed stream in one line ─────────────────────────
    // `data` is a `Stream`; map each row to a typed value and `try_collect`.
    let years: Vec<i64> = graph
        .query("MATCH (m:Movie) RETURN m.year AS year ORDER BY year")
        .execute()
        .await?
        .data
        .map(|row| row?.try_get::<i64>("year"))
        .try_collect()
        .await?;
    println!("years: {years:?}");

    // ── 2. Fan out a follow-up query per row, with bounded concurrency ──────
    // The graph handle is `Send + Clone` (it shares one schema cache), so each unit of work just
    // clones it. `buffer_unordered` keeps at most 8 follow-up queries in flight at once.
    let mut next_years: Vec<i64> = graph
        .query("MATCH (m:Movie) RETURN m.year AS year")
        .execute()
        .await?
        .data
        .map(|row| {
            let mut g = graph.clone();
            async move {
                let year: i64 = row?.try_get("year")?;
                let mut r = g
                    .query(format!("RETURN {year} + 1 AS next"))
                    .execute()
                    .await?;
                r.data
                    .try_next()
                    .await?
                    .expect("a row")
                    .try_get::<i64>("next")
            }
        })
        .buffer_unordered(8)
        .try_collect()
        .await?;
    next_years.sort_unstable();
    println!("next years: {next_years:?}");

    graph.delete().await?;
    Ok(())
}

Source: examples/async_stream.rs — compiled in CI.

The bit that makes me happy is the second half: a stream of rows, each fanning out into a follow-up query, with buffer_unordered(8) keeping at most eight in flight. That's backpressure-aware concurrency in about five lines, and it falls out of treating results as a Stream instead of inventing a bespoke cursor API. Reuse beats reinvention.

Many queries, few sockets: multiplexing

The other half is how those concurrent queries hit the wire. The async client supports two connection strategies:

• Pooled — a fixed set of connections; each query borrows one exclusively for its duration.
• Multiplexed — a few auto-reconnecting sockets, each carrying several in-flight commands at once. Queries are pipelined and their replies correlated back to the right caller.

sequenceDiagram
    autonumber
    participant T1 as task #1
    participant T2 as task #2
    participant T3 as task #64
    participant M as multiplexer (4 sockets)
    participant DB as FalkorDB
    T1->>M: ro_query("RETURN 1")
    T2->>M: ro_query("RETURN 1")
    T3->>M: ro_query("RETURN 1")
    Note over M,DB: requests are pipelined over the shared sockets,<br/>not blocked on an exclusive connection each
    M->>DB: GRAPH.RO_QUERY (pipelined)
    DB-->>M: replies, correlated back to each task
    M-->>T1: typed Row
    M-->>T2: typed Row
    M-->>T3: typed Row

examples/multiplexed_async.rs fires 64 independent queries across 4 shared sockets:

/*
 * Copyright FalkorDB Ltd. 2023 - present
 * Licensed under the MIT License.
 */

#![recursion_limit = "256"]

//! Demonstrates the multiplexed async connection strategy: many concurrent queries
//! share a small number of auto-reconnecting sockets, each carrying several in-flight
//! commands at once.
//!
//! Run a FalkorDB instance first (e.g. `docker run -p 6379:6379 falkordb/falkordb`), then
//! `cargo run --example multiplexed_async --features tokio`.

use falkordb::{ConnectionStrategy, FalkorClientBuilder, FalkorResult};
use futures::StreamExt;
use std::num::NonZeroU8;
use std::sync::Arc;
use tokio::task::JoinSet;

// Usage of the asynchronous client REQUIRES the multi-threaded rt
#[tokio::main(flavor = "multi_thread")]
async fn main() -> FalkorResult<()> {
    // Explicitly select the multiplexed strategy with 4 shared sockets. This is also the
    // default for `new_async()` (with 8 sockets), so `with_connection_strategy` here is
    // purely to make the choice visible.
    let client = Arc::new(
        FalkorClientBuilder::new_async()
            .with_connection_info("falkor://127.0.0.1:6379".try_into()?)
            .with_connection_strategy(ConnectionStrategy::Multiplexed {
                connections: NonZeroU8::new(4).unwrap(),
            })
            .build()
            .await?,
    );

    println!(
        "effective strategy: {:?} ({} connections)",
        client.connection_strategy(),
        client.connection_pool_size()
    );

    // Fire many independent queries concurrently. With multiplexing, these are pipelined
    // over the shared sockets rather than each waiting for an exclusive connection.
    let mut join_set = JoinSet::new();
    for i in 0..64 {
        let client = client.clone();
        join_set.spawn(async move {
            let mut graph = client.select_graph("multiplexed_demo");
            let mut res = graph.ro_query("RETURN 1").execute().await?;
            let value = res
                .data
                .next()
                .await
                .and_then(|row| row.ok().and_then(|r| r.try_get_at::<i64>(0).ok()));
            FalkorResult::Ok((i, value))
        });
    }

    let mut completed = 0;
    while let Some(joined) = join_set.join_next().await {
        let (_i, _value) = joined.expect("task should not panic")?;
        completed += 1;
    }
    println!("completed {completed} concurrent queries over 4 multiplexed sockets");

    client.select_graph("multiplexed_demo").delete().await.ok();

    Ok(())
}

Source: examples/multiplexed_async.rs — compiled in CI.

Now, the numbers

Here's where I have to be honest instead of triumphant. Multiplexing is not free, and it is not always faster — it's faster under contention. The benchmark is the real benches/async_strategies.rs harness; it runs the same batch of concurrent RETURN 1 reads against both strategies at a fixed 8 connections, across rising concurrency. (RETURN 1 is deliberate: it isolates the client's connection behavior from query-planning cost.)

Read that table honestly:

• At concurrency 1, pooling wins. One query at a time has nothing to multiplex, and the multiplexer's bookkeeping is pure overhead. The speedup is below 1.0 — multiplexing is slightly slower here, and pretending otherwise would be lying to you.
• As concurrency climbs, multiplexing pulls ahead. By 64 concurrent reads it's ~1.5× faster, and by 256 it's ~1.7×, because the pooled strategy is serializing waves of queries through 8 exclusive connections while the multiplexer keeps many commands in flight per socket.

So the guidance writes itself: a request/response app doing one query at a time? Pooled is simple and great. A server fanning out hundreds of concurrent graph reads? Multiplex. The default for the async client is multiplexed with 8 sockets, which is the right call for the server workloads people actually point this at.

The harness, by the way, skips itself cleanly when no server is reachable, so it stays runnable in CI without flaking — a small thing I care about a lot, and the subject of more than one bug fix.

You can reproduce the table with just bench against your own server; your numbers will differ, which is exactly why they're labeled with the machine and date. Next up, the feature that means you don't even need a server running to try any of this: the embedded FalkorDB.