Struct ClusterClient 

pub struct ClusterClient { /* private fields */ }

One open connection per distinct shard node, with a slot → shard index so a single-key command goes straight to its owner.

Implementations

impl ClusterClient

pub fn connect(host: &str, port: u16) -> Result<Self>

Connect via a seed node, discover the topology (CLUSTER SLOTS), and open one connection per shard.

Examples found in repository

examples/cluster.rs (line 25)

fn main() -> std::io::Result<()> {
    let seed: u16 = std::env::args()
        .nth(1)
        .and_then(|s| s.parse().ok())
        .unwrap_or(6005);

    // Connect to a seed shard; the full topology is discovered from it.
    let mut cc = ClusterClient::connect("127.0.0.1", seed)?;
    println!("connected — {} shard(s)", cc.shard_count());

    // Each of these keys may live on a different shard; the client routes each
    // to its owner with no -MOVED and no forwarding hop.
    cc.set(b"user:1", b"alice")?;
    cc.set(b"user:2", b"bob")?;
    cc.set(b"user:3", b"carol")?;

    println!("user:1 = {:?}", cc.get(b"user:1")?.map(String::from_utf8));

    // INCR routes by key like any single-key command.
    let n = cc.incr(b"counter")?;
    println!("counter = {n}");

    // DEL/EXISTS take keys that may span shards — routed per key and summed.
    let exists = cc.exists(&[b"user:1", b"user:2", b"missing"])?;
    println!("exists(user:1, user:2, missing) = {exists}"); // 2

    let removed = cc.del(&[b"user:1", b"user:2", b"user:3"])?;
    println!("deleted {removed} keys"); // 3

    // DBSIZE / FLUSHALL are whole-cluster: kevy fans them out server-side, so
    // one call covers every shard.
    println!("dbsize (whole cluster) = {}", cc.dbsize()?);

    Ok(())
}

examples/cluster_bench.rs (line 25)

fn main() {
    let a: Vec<String> = std::env::args().collect();
    let seed: u16 = a.get(1).and_then(|s| s.parse().ok()).unwrap_or(7001);
    let iters: usize = a.get(2).and_then(|s| s.parse().ok()).unwrap_or(20_000);
    let keys: usize = a.get(3).and_then(|s| s.parse().ok()).unwrap_or(1000);
    let conc: usize = a.get(4).and_then(|s| s.parse().ok()).unwrap_or(8);

    // Populate.
    {
        let mut cc = ClusterClient::connect("127.0.0.1", seed).unwrap();
        for k in 0..keys {
            cc.set(format!("k{k}").as_bytes(), b"v").unwrap();
        }
    }

    let t0 = Instant::now();
    let handles: Vec<_> = (0..conc)
        .map(|c| {
            std::thread::spawn(move || {
                let mut cc = ClusterClient::connect("127.0.0.1", seed).unwrap();
                let mut lat = Vec::with_capacity(iters);
                for i in 0..iters {
                    let key = format!("k{}", (i * 7 + c * 131) % keys);
                    let t = Instant::now();
                    let _ = cc.get(key.as_bytes()).unwrap();
                    lat.push(t.elapsed().as_nanos() as u64);
                }
                lat
            })
        })
        .collect();
    let mut all = Vec::new();
    for h in handles {
        all.extend(h.join().unwrap());
    }
    let wall = t0.elapsed().as_secs_f64();
    all.sort_unstable();
    let p = |q: f64| all[((all.len() as f64 * q) as usize).min(all.len() - 1)] as f64 / 1000.0;
    println!(
        "ClusterClient conc={conc} n={} ops/s={:.0} p50={:.1}us p99={:.1}us p999={:.1}us",
        all.len(),
        all.len() as f64 / wall,
        p(0.50),
        p(0.99),
        p(0.999),
    );
}

pub fn request_keyed(&mut self, key: &[u8], args: &[Vec<u8>]) -> Result<Reply>

Route a single-key command (args) to the shard owning key.

pub fn request_unkeyed(&mut self, args: &[Vec<u8>]) -> Result<Reply>

A keyless command (PING / etc.) — answered identically by any shard, so send it to the first.

pub fn shard_count(&self) -> usize

Number of distinct shard nodes.

Examples found in repository

examples/cluster.rs (line 26)

fn main() -> std::io::Result<()> {
    let seed: u16 = std::env::args()
        .nth(1)
        .and_then(|s| s.parse().ok())
        .unwrap_or(6005);

    // Connect to a seed shard; the full topology is discovered from it.
    let mut cc = ClusterClient::connect("127.0.0.1", seed)?;
    println!("connected — {} shard(s)", cc.shard_count());

    // Each of these keys may live on a different shard; the client routes each
    // to its owner with no -MOVED and no forwarding hop.
    cc.set(b"user:1", b"alice")?;
    cc.set(b"user:2", b"bob")?;
    cc.set(b"user:3", b"carol")?;

    println!("user:1 = {:?}", cc.get(b"user:1")?.map(String::from_utf8));

    // INCR routes by key like any single-key command.
    let n = cc.incr(b"counter")?;
    println!("counter = {n}");

    // DEL/EXISTS take keys that may span shards — routed per key and summed.
    let exists = cc.exists(&[b"user:1", b"user:2", b"missing"])?;
    println!("exists(user:1, user:2, missing) = {exists}"); // 2

    let removed = cc.del(&[b"user:1", b"user:2", b"user:3"])?;
    println!("deleted {removed} keys"); // 3

    // DBSIZE / FLUSHALL are whole-cluster: kevy fans them out server-side, so
    // one call covers every shard.
    println!("dbsize (whole cluster) = {}", cc.dbsize()?);

    Ok(())
}

impl ClusterClient

pub fn ping(&mut self) -> Result<()>

PING — answered by any shard.

pub fn publish(&mut self, channel: &[u8], message: &[u8]) -> Result<usize>

PUBLISH channel message — returns the number of subscribers reached. kevy’s pub/sub is process-global (delivered to subscribers on every core), so a cluster PUBLISH goes to any one shard.

pub fn set(&mut self, key: &[u8], value: &[u8]) -> Result<()>

SET key value.

Examples found in repository

examples/cluster.rs (line 30)

fn main() -> std::io::Result<()> {
    let seed: u16 = std::env::args()
        .nth(1)
        .and_then(|s| s.parse().ok())
        .unwrap_or(6005);

    // Connect to a seed shard; the full topology is discovered from it.
    let mut cc = ClusterClient::connect("127.0.0.1", seed)?;
    println!("connected — {} shard(s)", cc.shard_count());

    // Each of these keys may live on a different shard; the client routes each
    // to its owner with no -MOVED and no forwarding hop.
    cc.set(b"user:1", b"alice")?;
    cc.set(b"user:2", b"bob")?;
    cc.set(b"user:3", b"carol")?;

    println!("user:1 = {:?}", cc.get(b"user:1")?.map(String::from_utf8));

    // INCR routes by key like any single-key command.
    let n = cc.incr(b"counter")?;
    println!("counter = {n}");

    // DEL/EXISTS take keys that may span shards — routed per key and summed.
    let exists = cc.exists(&[b"user:1", b"user:2", b"missing"])?;
    println!("exists(user:1, user:2, missing) = {exists}"); // 2

    let removed = cc.del(&[b"user:1", b"user:2", b"user:3"])?;
    println!("deleted {removed} keys"); // 3

    // DBSIZE / FLUSHALL are whole-cluster: kevy fans them out server-side, so
    // one call covers every shard.
    println!("dbsize (whole cluster) = {}", cc.dbsize()?);

    Ok(())
}

examples/cluster_bench.rs (line 27)

fn main() {
    let a: Vec<String> = std::env::args().collect();
    let seed: u16 = a.get(1).and_then(|s| s.parse().ok()).unwrap_or(7001);
    let iters: usize = a.get(2).and_then(|s| s.parse().ok()).unwrap_or(20_000);
    let keys: usize = a.get(3).and_then(|s| s.parse().ok()).unwrap_or(1000);
    let conc: usize = a.get(4).and_then(|s| s.parse().ok()).unwrap_or(8);

    // Populate.
    {
        let mut cc = ClusterClient::connect("127.0.0.1", seed).unwrap();
        for k in 0..keys {
            cc.set(format!("k{k}").as_bytes(), b"v").unwrap();
        }
    }

    let t0 = Instant::now();
    let handles: Vec<_> = (0..conc)
        .map(|c| {
            std::thread::spawn(move || {
                let mut cc = ClusterClient::connect("127.0.0.1", seed).unwrap();
                let mut lat = Vec::with_capacity(iters);
                for i in 0..iters {
                    let key = format!("k{}", (i * 7 + c * 131) % keys);
                    let t = Instant::now();
                    let _ = cc.get(key.as_bytes()).unwrap();
                    lat.push(t.elapsed().as_nanos() as u64);
                }
                lat
            })
        })
        .collect();
    let mut all = Vec::new();
    for h in handles {
        all.extend(h.join().unwrap());
    }
    let wall = t0.elapsed().as_secs_f64();
    all.sort_unstable();
    let p = |q: f64| all[((all.len() as f64 * q) as usize).min(all.len() - 1)] as f64 / 1000.0;
    println!(
        "ClusterClient conc={conc} n={} ops/s={:.0} p50={:.1}us p99={:.1}us p999={:.1}us",
        all.len(),
        all.len() as f64 / wall,
        p(0.50),
        p(0.99),
        p(0.999),
    );
}

pub fn set_with_ttl( &mut self, key: &[u8], value: &[u8], ttl: Duration, ) -> Result<()>

SET key value PX ttl_ms — value with an expiry.

pub fn get(&mut self, key: &[u8]) -> Result<Option<Vec<u8>>>

GET key. None if absent or expired.

Examples found in repository

examples/cluster.rs (line 34)

fn main() -> std::io::Result<()> {
    let seed: u16 = std::env::args()
        .nth(1)
        .and_then(|s| s.parse().ok())
        .unwrap_or(6005);

    // Connect to a seed shard; the full topology is discovered from it.
    let mut cc = ClusterClient::connect("127.0.0.1", seed)?;
    println!("connected — {} shard(s)", cc.shard_count());

    // Each of these keys may live on a different shard; the client routes each
    // to its owner with no -MOVED and no forwarding hop.
    cc.set(b"user:1", b"alice")?;
    cc.set(b"user:2", b"bob")?;
    cc.set(b"user:3", b"carol")?;

    println!("user:1 = {:?}", cc.get(b"user:1")?.map(String::from_utf8));

    // INCR routes by key like any single-key command.
    let n = cc.incr(b"counter")?;
    println!("counter = {n}");

    // DEL/EXISTS take keys that may span shards — routed per key and summed.
    let exists = cc.exists(&[b"user:1", b"user:2", b"missing"])?;
    println!("exists(user:1, user:2, missing) = {exists}"); // 2

    let removed = cc.del(&[b"user:1", b"user:2", b"user:3"])?;
    println!("deleted {removed} keys"); // 3

    // DBSIZE / FLUSHALL are whole-cluster: kevy fans them out server-side, so
    // one call covers every shard.
    println!("dbsize (whole cluster) = {}", cc.dbsize()?);

    Ok(())
}

examples/cluster_bench.rs (line 40)

fn main() {
    let a: Vec<String> = std::env::args().collect();
    let seed: u16 = a.get(1).and_then(|s| s.parse().ok()).unwrap_or(7001);
    let iters: usize = a.get(2).and_then(|s| s.parse().ok()).unwrap_or(20_000);
    let keys: usize = a.get(3).and_then(|s| s.parse().ok()).unwrap_or(1000);
    let conc: usize = a.get(4).and_then(|s| s.parse().ok()).unwrap_or(8);

    // Populate.
    {
        let mut cc = ClusterClient::connect("127.0.0.1", seed).unwrap();
        for k in 0..keys {
            cc.set(format!("k{k}").as_bytes(), b"v").unwrap();
        }
    }

    let t0 = Instant::now();
    let handles: Vec<_> = (0..conc)
        .map(|c| {
            std::thread::spawn(move || {
                let mut cc = ClusterClient::connect("127.0.0.1", seed).unwrap();
                let mut lat = Vec::with_capacity(iters);
                for i in 0..iters {
                    let key = format!("k{}", (i * 7 + c * 131) % keys);
                    let t = Instant::now();
                    let _ = cc.get(key.as_bytes()).unwrap();
                    lat.push(t.elapsed().as_nanos() as u64);
                }
                lat
            })
        })
        .collect();
    let mut all = Vec::new();
    for h in handles {
        all.extend(h.join().unwrap());
    }
    let wall = t0.elapsed().as_secs_f64();
    all.sort_unstable();
    let p = |q: f64| all[((all.len() as f64 * q) as usize).min(all.len() - 1)] as f64 / 1000.0;
    println!(
        "ClusterClient conc={conc} n={} ops/s={:.0} p50={:.1}us p99={:.1}us p999={:.1}us",
        all.len(),
        all.len() as f64 / wall,
        p(0.50),
        p(0.99),
        p(0.999),
    );
}

pub fn incr(&mut self, key: &[u8]) -> Result<i64>

INCR key. Returns the post-increment value.

Examples found in repository

examples/cluster.rs (line 37)

fn main() -> std::io::Result<()> {
    let seed: u16 = std::env::args()
        .nth(1)
        .and_then(|s| s.parse().ok())
        .unwrap_or(6005);

    // Connect to a seed shard; the full topology is discovered from it.
    let mut cc = ClusterClient::connect("127.0.0.1", seed)?;
    println!("connected — {} shard(s)", cc.shard_count());

    // Each of these keys may live on a different shard; the client routes each
    // to its owner with no -MOVED and no forwarding hop.
    cc.set(b"user:1", b"alice")?;
    cc.set(b"user:2", b"bob")?;
    cc.set(b"user:3", b"carol")?;

    println!("user:1 = {:?}", cc.get(b"user:1")?.map(String::from_utf8));

    // INCR routes by key like any single-key command.
    let n = cc.incr(b"counter")?;
    println!("counter = {n}");

    // DEL/EXISTS take keys that may span shards — routed per key and summed.
    let exists = cc.exists(&[b"user:1", b"user:2", b"missing"])?;
    println!("exists(user:1, user:2, missing) = {exists}"); // 2

    let removed = cc.del(&[b"user:1", b"user:2", b"user:3"])?;
    println!("deleted {removed} keys"); // 3

    // DBSIZE / FLUSHALL are whole-cluster: kevy fans them out server-side, so
    // one call covers every shard.
    println!("dbsize (whole cluster) = {}", cc.dbsize()?);

    Ok(())
}

pub fn incr_by(&mut self, key: &[u8], delta: i64) -> Result<i64>

INCRBY key delta.

pub fn expire(&mut self, key: &[u8], ttl: Duration) -> Result<bool>

PEXPIRE key ttl_ms. Whether the key existed and got a TTL.

pub fn persist(&mut self, key: &[u8]) -> Result<bool>

PERSIST key. Whether a TTL was removed.

pub fn ttl_ms(&mut self, key: &[u8]) -> Result<i64>

PTTL key. ms remaining, -2 no key, -1 no TTL.

pub fn del(&mut self, keys: &[&[u8]]) -> Result<usize>

DEL key [key ...] — routed per key (each to its owner) and summed, so keys spanning shards work without a same-slot constraint.

Examples found in repository

examples/cluster.rs (line 44)

fn main() -> std::io::Result<()> {
    let seed: u16 = std::env::args()
        .nth(1)
        .and_then(|s| s.parse().ok())
        .unwrap_or(6005);

    // Connect to a seed shard; the full topology is discovered from it.
    let mut cc = ClusterClient::connect("127.0.0.1", seed)?;
    println!("connected — {} shard(s)", cc.shard_count());

    // Each of these keys may live on a different shard; the client routes each
    // to its owner with no -MOVED and no forwarding hop.
    cc.set(b"user:1", b"alice")?;
    cc.set(b"user:2", b"bob")?;
    cc.set(b"user:3", b"carol")?;

    println!("user:1 = {:?}", cc.get(b"user:1")?.map(String::from_utf8));

    // INCR routes by key like any single-key command.
    let n = cc.incr(b"counter")?;
    println!("counter = {n}");

    // DEL/EXISTS take keys that may span shards — routed per key and summed.
    let exists = cc.exists(&[b"user:1", b"user:2", b"missing"])?;
    println!("exists(user:1, user:2, missing) = {exists}"); // 2

    let removed = cc.del(&[b"user:1", b"user:2", b"user:3"])?;
    println!("deleted {removed} keys"); // 3

    // DBSIZE / FLUSHALL are whole-cluster: kevy fans them out server-side, so
    // one call covers every shard.
    println!("dbsize (whole cluster) = {}", cc.dbsize()?);

    Ok(())
}

pub fn exists(&mut self, keys: &[&[u8]]) -> Result<usize>

EXISTS key [key ...] — routed per key and summed (a repeated key counts each time, matching Redis).

Examples found in repository

examples/cluster.rs (line 41)

fn main() -> std::io::Result<()> {
    let seed: u16 = std::env::args()
        .nth(1)
        .and_then(|s| s.parse().ok())
        .unwrap_or(6005);

    // Connect to a seed shard; the full topology is discovered from it.
    let mut cc = ClusterClient::connect("127.0.0.1", seed)?;
    println!("connected — {} shard(s)", cc.shard_count());

    // Each of these keys may live on a different shard; the client routes each
    // to its owner with no -MOVED and no forwarding hop.
    cc.set(b"user:1", b"alice")?;
    cc.set(b"user:2", b"bob")?;
    cc.set(b"user:3", b"carol")?;

    println!("user:1 = {:?}", cc.get(b"user:1")?.map(String::from_utf8));

    // INCR routes by key like any single-key command.
    let n = cc.incr(b"counter")?;
    println!("counter = {n}");

    // DEL/EXISTS take keys that may span shards — routed per key and summed.
    let exists = cc.exists(&[b"user:1", b"user:2", b"missing"])?;
    println!("exists(user:1, user:2, missing) = {exists}"); // 2

    let removed = cc.del(&[b"user:1", b"user:2", b"user:3"])?;
    println!("deleted {removed} keys"); // 3

    // DBSIZE / FLUSHALL are whole-cluster: kevy fans them out server-side, so
    // one call covers every shard.
    println!("dbsize (whole cluster) = {}", cc.dbsize()?);

    Ok(())
}

pub fn dbsize(&mut self) -> Result<usize>

DBSIZE — the cluster-wide total. kevy answers DBSIZE by fanning out across shards internally (Route::Dbsize), so a single shard already reports the whole-cluster count; no client-side summing.

Examples found in repository

examples/cluster.rs (line 49)

fn main() -> std::io::Result<()> {
    let seed: u16 = std::env::args()
        .nth(1)
        .and_then(|s| s.parse().ok())
        .unwrap_or(6005);

    // Connect to a seed shard; the full topology is discovered from it.
    let mut cc = ClusterClient::connect("127.0.0.1", seed)?;
    println!("connected — {} shard(s)", cc.shard_count());

    // Each of these keys may live on a different shard; the client routes each
    // to its owner with no -MOVED and no forwarding hop.
    cc.set(b"user:1", b"alice")?;
    cc.set(b"user:2", b"bob")?;
    cc.set(b"user:3", b"carol")?;

    println!("user:1 = {:?}", cc.get(b"user:1")?.map(String::from_utf8));

    // INCR routes by key like any single-key command.
    let n = cc.incr(b"counter")?;
    println!("counter = {n}");

    // DEL/EXISTS take keys that may span shards — routed per key and summed.
    let exists = cc.exists(&[b"user:1", b"user:2", b"missing"])?;
    println!("exists(user:1, user:2, missing) = {exists}"); // 2

    let removed = cc.del(&[b"user:1", b"user:2", b"user:3"])?;
    println!("deleted {removed} keys"); // 3

    // DBSIZE / FLUSHALL are whole-cluster: kevy fans them out server-side, so
    // one call covers every shard.
    println!("dbsize (whole cluster) = {}", cc.dbsize()?);

    Ok(())
}

pub fn flushall(&mut self) -> Result<()>

FLUSHALL — clears every shard. kevy fans FLUSHALL out internally (Route::Flush), so one call wipes the whole cluster.

impl ClusterClient

pub fn hset(&mut self, key: &[u8], pairs: &[(&[u8], &[u8])]) -> Result<usize>

HSET key field value [field value ...] — count of newly added fields.

pub fn hget(&mut self, key: &[u8], field: &[u8]) -> Result<Option<Vec<u8>>>

HGET key field. None when key or field absent.

pub fn hdel(&mut self, key: &[u8], fields: &[&[u8]]) -> Result<usize>

HDEL key field [field ...] — count of fields removed.

pub fn hlen(&mut self, key: &[u8]) -> Result<usize>

HLEN key.

pub fn hgetall(&mut self, key: &[u8]) -> Result<Vec<Vec<u8>>>

HGETALL key — flat [f0, v0, f1, v1, …].

pub fn hkeys(&mut self, key: &[u8]) -> Result<Vec<Vec<u8>>>

HKEYS key.

pub fn hvals(&mut self, key: &[u8]) -> Result<Vec<Vec<u8>>>

HVALS key.

pub fn lpush(&mut self, key: &[u8], values: &[&[u8]]) -> Result<usize>

LPUSH key value [value ...] — new list length.

pub fn rpush(&mut self, key: &[u8], values: &[&[u8]]) -> Result<usize>

RPUSH key value [value ...] — new list length.

pub fn lpop(&mut self, key: &[u8], n: usize) -> Result<Vec<Vec<u8>>>

LPOP key count.

pub fn rpop(&mut self, key: &[u8], n: usize) -> Result<Vec<Vec<u8>>>

RPOP key count.

pub fn llen(&mut self, key: &[u8]) -> Result<usize>

LLEN key.

pub fn lrange( &mut self, key: &[u8], start: i64, stop: i64, ) -> Result<Vec<Vec<u8>>>

LRANGE key start stop (Redis-style negative indices).

pub fn sadd(&mut self, key: &[u8], members: &[&[u8]]) -> Result<usize>

SADD key member [member ...] — count newly added.

pub fn srem(&mut self, key: &[u8], members: &[&[u8]]) -> Result<usize>

SREM key member [member ...] — count removed.

pub fn smembers(&mut self, key: &[u8]) -> Result<Vec<Vec<u8>>>

SMEMBERS key.

pub fn scard(&mut self, key: &[u8]) -> Result<usize>

SCARD key.

pub fn sismember(&mut self, key: &[u8], member: &[u8]) -> Result<bool>

SISMEMBER key member.

pub fn sinter(&mut self, keys: &[&[u8]]) -> Result<Vec<Vec<u8>>>

SINTER key [key ...] — all keys must share a slot (route by the first).

pub fn sunion(&mut self, keys: &[&[u8]]) -> Result<Vec<Vec<u8>>>

SUNION key [key ...] — same-slot.

pub fn sdiff(&mut self, keys: &[&[u8]]) -> Result<Vec<Vec<u8>>>

SDIFF key [key ...] — same-slot.

pub fn zadd(&mut self, key: &[u8], pairs: &[(f64, &[u8])]) -> Result<usize>

ZADD key score member [score member ...] — count newly added.

pub fn zrem(&mut self, key: &[u8], members: &[&[u8]]) -> Result<usize>

ZREM key member [member ...] — count removed.

pub fn zscore(&mut self, key: &[u8], member: &[u8]) -> Result<Option<f64>>

ZSCORE key member. None if absent.

pub fn zcard(&mut self, key: &[u8]) -> Result<usize>

ZCARD key.

pub fn zrange( &mut self, key: &[u8], start: i64, stop: i64, ) -> Result<Vec<Vec<u8>>>

ZRANGE key start stop (ascending score; negative indices from tail).