Datum

Appearance

Pipelining And Parallelism ​

Datum has two separate tools for doing more than one thing at a time:

  • bounded future concurrency in the linear Source / Flow DSL, using Tokio-dispatched futures;
  • fused-region isolation with a bounded async-boundary handoff, using async_boundary() in the linear Source / Flow DSL or the AsyncBoundary graph stage in GraphDsl.

Use future concurrency when each element needs async work such as an RPC, disk call, or actor round-trip. Use an async boundary when a fused region should be isolated by the async-boundary runner with a bounded handoff.

Bounded future concurrency ​

map_async(parallelism, f) runs up to parallelism futures at once and preserves input order in the output. The future factory receives an owned element and returns StreamResult<Next>.

rust
use datum::{Sink, Source};
use std::sync::{
    Arc,
    atomic::{AtomicUsize, Ordering},
};
use std::time::Duration;

let active = Arc::new(AtomicUsize::new(0));
let max_active = Arc::new(AtomicUsize::new(0));

let values: Vec<u64> = Source::from_iter(1_u64..=6)
    .map_async(2, {
        let active = Arc::clone(&active);
        let max_active = Arc::clone(&max_active);
        move |item| {
            let active = Arc::clone(&active);
            let max_active = Arc::clone(&max_active);
            async move {
                let now_active = active.fetch_add(1, Ordering::SeqCst) + 1;
                max_active.fetch_max(now_active, Ordering::SeqCst);
                tokio::time::sleep(Duration::from_millis(2)).await;
                active.fetch_sub(1, Ordering::SeqCst);
                Ok(item * 10)
            }
        }
    })
    .run_with(Sink::collect())
    .unwrap()
    .wait()
    .unwrap();

The parallelism argument must be greater than zero. Pending futures are dispatched onto Tokio; futures that are immediately ready can complete without a Tokio handoff.

map_async_unordered(parallelism, f) uses the same bounded concurrency but emits elements in completion order. Use it when element order does not matter.

rust
use datum::{Sink, Source};

let mut values: Vec<u64> = Source::from_iter(1_u64..=4)
    .map_async_unordered(4, |item| async move { Ok(item * item) })
    .run_with(Sink::collect())
    .unwrap()
    .wait()
    .unwrap();

values.sort_unstable();

map_async_partitioned(parallelism, per_partition, partition_fn, f) preserves order within each partition key while allowing different keys to proceed independently.

rust
use datum::{Sink, Source};

let mut values: Vec<String> = Source::from_iter(0_u64..6)
    .map_async_partitioned(
        4, // maximum total futures
        1, // one in-flight future per partition key
        |item| item % 2,
        |item| async move { Ok(format!("key-{}:{item}", item % 2)) },
    )
    .run_with(Sink::collect())
    .unwrap()
    .wait()
    .unwrap();

values.sort();

The key returned by partition_fn must be Clone + Eq + Hash + Send + 'static. parallelism limits total in-flight futures; per_partition limits concurrent futures for one key.

Async boundaries in linear streams ​

Source::async_boundary() and Flow::async_boundary() insert a boundary between the upstream and downstream fused regions. The boundary preserves element order and values, but the upstream region runs on the Ractor-backed async-boundary worker and hands elements to the downstream region through a bounded queue.

async_boundary() is the Rust-friendly primary name. Datum also provides .r#async() as an Akka-mirroring alias for callers that prefer the original operator name despite Rust's async keyword.

Use async_boundary_with_config(AsyncBoundaryExecutionConfig { ... }) when you need an explicit handoff size. buffer_size is the bounded queue capacity.

rust
use datum::{AsyncBoundaryExecutionConfig, Sink, Source};

let values: Vec<u64> = Source::from_iter(1_u64..=3)
    .map(|item| item + 1)
    .async_boundary_with_config(AsyncBoundaryExecutionConfig {
        buffer_size: 4,
        ..AsyncBoundaryExecutionConfig::default()
    })
    .map(|item| item * 2)
    .run_with(Sink::collect())
    .unwrap()
    .wait()
    .unwrap();

Async boundaries in GraphDSL ​

AsyncBoundary<T>::new() is a GraphStage with FlowShape<T, T>. Add it inside a GraphDsl graph, not as a linear Flow::via(...) operator.

Attach an InputBuffer hint with GraphBuilder::add_with_attributes when the boundary stage should carry stage attributes. The async-boundary execution/report path takes an AsyncBoundaryExecutionConfig; its buffer_size is the bounded handoff queue size and fused contains the fused-region event limit.

rust
use datum::{
    AsyncBoundary, AsyncBoundaryExecutionConfig, Attributes, FusedExecutionConfig, GraphDsl,
    GraphFlowShape, MapStage,
};

let graph = GraphDsl::try_create(|builder| {
    let first = builder.add(MapStage::new(|item: u64| item + 1));
    let boundary = builder
        .add_with_attributes(AsyncBoundary::<u64>::new(), Attributes::input_buffer(4, 8));
    let second = builder.add(MapStage::new(|item: u64| item * 2));

    builder.connect(first.outlet(), boundary.inlet())?;
    builder.connect(boundary.outlet(), second.inlet())?;

    Ok(GraphFlowShape::new(first.inlet(), second.outlet()))
})
.unwrap();

let output = graph.run_with_input([1, 2, 3]).unwrap();
let report = graph
    .run_async_boundary_count_with_input_report(
        1_u64..=3,
        AsyncBoundaryExecutionConfig {
            fused: FusedExecutionConfig { event_limit: 1024 },
            buffer_size: 4,
        },
    )
    .unwrap();

The ordinary run_with_input path still validates the graph and produces output. The run_async_boundary_count_with_input_report path is useful when you specifically want to exercise the Ractor-backed async-boundary handoff and inspect async_boundary_crossings.

Choosing the tool ​

Use map_async when the expensive work is naturally per element and can be represented as a future. Use map_async_unordered when latency matters more than ordering. Use map_async_partitioned when per-key ordering matters.

Use async_boundary() when a linear stream should split work across fused regions with a bounded handoff. Use AsyncBoundary when you are designing graph topology and want the same boundary inside GraphDsl. Async boundaries are isolation tools, not replacements for bounded async element work.

Next steps ​