nexar

Not supported yet

• Python bindings.
• GPU adapters other than CudaAdapter (NVIDIA, in nexar-rdma). The repository has no ROCm or Intel adapter.
• GPU collectives without NCCL. nexar-nccl requires NCCL and CUDA.

Quick Start

Add the crates to your project:

cargo add nexar
cargo add tokio --features full

Start a 4-rank local cluster and run an in-place allreduce:

use nexar::{NexarClient, CpuAdapter, DataType, ReduceOp};
use std::sync::Arc;

#[tokio::main]
async fn main() -> nexar::Result<()> {
    let adapter = Arc::new(CpuAdapter::new());
    let clients = NexarClient::bootstrap_local(4, adapter).await?;

    // Each client holds rank 0..3 with connections to all peers.
    let client = &clients[0];
    assert_eq!(client.rank(), 0);
    assert_eq!(client.world_size(), 4);

    // Run allreduce in-place across all ranks.
    let mut data = vec![1.0f32; 1024];
    unsafe {
        client.all_reduce(
            data.as_mut_ptr() as u64,
            1024,
            DataType::F32,
            ReduceOp::Sum,
        ).await?;
    }

    Ok(())
}

The raw-pointer collective methods are unsafe. The caller must keep the buffer valid and sized for count elements of dtype until the call returns.

Multi-host deployment

bootstrap_local runs every rank in one process. For one process per rank, start rank 0 with bootstrap_seed and every other rank with bootstrap_join. Each host sets advertise_addr to its own reachable IP:

use nexar::{ClusterConfig, CpuAdapter, NexarClient};
use std::sync::Arc;

async fn join(is_seed: bool) -> nexar::Result<NexarClient> {
    let seed_addr = "10.0.0.1:7000".parse().unwrap();
    let cfg = ClusterConfig::new(4)
        .with_bind_addr("0.0.0.0:7001".parse().unwrap())
        .with_advertise_addr("10.0.0.2:7001".parse().unwrap())
        .with_cluster_token(std::env::var("NEXAR_CLUSTER_TOKEN").unwrap());
    let adapter = Arc::new(CpuAdapter::new());
    if is_seed {
        NexarClient::bootstrap_seed(seed_addr, &cfg, adapter).await
    } else {
        NexarClient::bootstrap_join(seed_addr, &cfg, adapter).await
    }
}

Formation runs in four steps:

1. Each worker sends Hello to the seed. The seed assigns ranks and answers with Welcome: the rank and an mTLS certificate from an ephemeral cluster CA.
2. Each rank binds its mesh listener and reports the address with MeshListen.
3. The seed broadcasts the PeerTable once every rank reports. Each rank dials higher ranks and accepts lower ranks.
4. Each rank enters the seed's ready barrier. The bootstrap returns once every rank is ready.

ClusterConfig fields:

Set these on every host:

• Seed address: every process uses the address the seed binds. Rank 0 binds it.
• Token: the bootstrap connection does not verify the seed's certificate. The token is the only defense against a stranger who joins or poses as the seed. A bind, seed, or advertise address that is not loopback requires a token, on every SeedNode constructor too. Use the same secret on every process.
• Firewall: open the seed port for UDP. Open each rank's mesh port for UDP (QUIC) and TCP (bulk sidecar). With bind_addr port 0, the ports are ephemeral. Fix the port to write a firewall rule.
• NAT and containers: set advertise_addr to the address peers can reach. With a published container port, map the same port for UDP and TCP.
• Sidecar: set the same enable_tcp_bulk_sidecar and encrypt_bulk_transport on every rank. A mismatch fails formation. Set max_frame_bytes (default 4 GiB) to the largest message you send; a longer frame closes the sidecar with FrameTooLarge.

Shutdown

Call client.shutdown().await on every rank after its last send and recv, before the process exits. Closing a QUIC connection discards data the peer has not acknowledged, so a rank that exits right after its last collective can lose that collective's final message.

shutdown flushes and closes the TCP sidecars, exchanges a DONE byte with every direct peer, and closes each connection once the peer is done too. A peer that dropped its client counts as done. After the barrier timeout, or the limit passed to shutdown_with_timeout, every connection closes anyway and the call returns ShutdownIncomplete with the peers that did not finish. A peer that answers with another byte, finishes its stream early, or resets it lands in ShutdownIncomplete::bad_signal.

shutdown takes the client by value. A client shared as Arc<NexarClient> needs every other holder dropped first, then Arc::into_inner.

Dropping the last client that shares a mesh closes its connections at once. Unlike shutdown, a drop can lose unacknowledged data.

A failure on one rank aborts formation on every rank with a typed error: RankConflict, ClusterTokenMismatch, ProtocolMismatch, ClusterFormationTimeout, JoinTimeout, or ConnectionFailed { rank }.

Collectives

NexarClient exposes 11 collectives. Each one has an _nb variant that returns a CollectiveHandle. Call wait on the handle to finish the operation.

The client also exposes all_reduce_host, all_reduce_device, and the other _host and _device variants. The _device variants take memory managed by a DeviceAdapter.

Features

• Point-to-point: tagged send and recv between any two ranks. A receive takes only its own tag, in send order. See Receive semantics.
• Elastic membership: rebuild_adding and rebuild_excluding change the communicator at run time.
• Fault recovery: RecoveryPolicy::Automatic, Manual, or Abort. Healthy ranks agree on the failed set and rebuild.
• Sparse topologies: TopologyStrategy::FullMesh, KRegular { degree }, and Hypercube. Traffic to non-neighbors relays through peers.
• Compressed allreduce: TopKCompressor and random-K gradient compression.
• RPC: register_rpc registers a handler by function ID. rpc and rpc_with_timeout call it on a rank. Responses match requests by request ID.
• Device adapters: the DeviceAdapter trait stages device memory for network I/O. CpuAdapter ships with nexar. CudaAdapter ships with nexar-rdma behind the gpudirect feature.

Examples

Runnable examples live in nexar/examples/:

Run an example from the repository root:

cargo run -p nexar --example send_recv
cargo run -p nexar --example allreduce

Architecture

seed node (discovery only, no data routing)
    │
    ├── worker 0 ──── worker 1
    │       \            /
    │        \          /
    │         worker 2 ── worker 3
    │              ...
    └── direct peer-to-peer mesh

Workers connect to the seed to get a rank, a certificate, and the peer table. They then open direct mTLS QUIC connections to their peers. The seed is not on the data path.

Each peer connection runs a router. The router accepts incoming QUIC streams and sends each message to the receiver of its key:

Receive semantics

• A receive for (src, key) waits for a message with that key from src. It never takes, or fails on, a message with another key. That message stays queued for its own receiver.
• Ordered: the messages of one byte lane from one sender, in the order their sends started on the connection. Each communicator (root, split, rebuilt) has its own tag space.
• Not ordered: different keys, different senders, control messages within one key, and traffic that relays through other ranks in a sparse topology.
• Each key queues NexarConfig::recv_queue_depth messages per peer (default 256). A message past that waits unread in its QUIC stream, and QUIC flow control slows its sender. Other keys from the same peer go on while the peer's open streams stay under the QUIC limit of 1024.
• A send that fails or is cancelled does not stall the lane. The sender reports its sequence number in a skip notice, and later messages pass it.
• When a peer disconnects, receivers first get the complete messages already queued, then NexarError::PeerDisconnected.
• A message longer than NexarConfig::max_frame_bytes is refused before any allocation: the router stops its stream, logs FrameTooLarge, and the sender's write fails. The router of each peer reads at most NexarConfig::recv_inflight_bytes of payload at once (default 4 GiB). A message past that waits unread in its QUIC stream. A message longer than the budget reads alone.
• The receiver refuses a sequence number 65536 or more past the lowest unknown one. A hostile peer cannot grow the reorder state.
• Receives are cancel-safe: a receive dropped by a timeout consumes nothing.
• The TCP bulk sidecar that collectives use is one ordered TCP stream per peer. A lagging tag on it delays the tags behind it.

Stream protocol

Each QUIC unidirectional stream starts with a 1-byte tag:

Integers are little-endian. One counter per connection numbers every sequenced message.

Messages use rkyv serialization.

Crates

Build each crate from the repository root:

cargo build -p nexar --release
cargo build -p nexar-rdma --release
cargo build -p nexar-rdma --release --features gpudirect
cargo build -p nexar-nccl --release

Requires Rust 1.89 or later. See CONTRIBUTING.md for test and lint commands.

When to Use nexar

Use nexar when:

• You need collectives such as allreduce or broadcast across machines.
• You want async, non-blocking communication in Rust.
• You want to avoid MPI installation, mpirun, and C FFI.
• You are building distributed ML training or inference.

Use NCCL directly when:

• All GPUs sit in one machine. NCCL uses the local interconnect without a network hop.

Contributing

Contributions are welcome. See CONTRIBUTING.md for conventions and checks.

Questions and design discussion happen on Discord.

License

Licensed under Apache-2.0. See LICENSE.