perf: add concurrent keyspace mode (--concurrent flag) (#50)

* perf: add concurrent keyspace mode (--concurrent flag)

adds an experimental mode that uses DashMap for lock-free concurrent
access, bypassing the sharded channel architecture for GET/SET commands.

benchmark results on 8-core VM:
- SET: +13% throughput (1,010k → 1,142k ops/sec)
- GET: +11% throughput (1,115k → 1,241k ops/sec)

the improvement comes from eliminating the per-request channel overhead
(mpsc send + oneshot reply) that exists in the sharded architecture.

new files:
- concurrent.rs: DashMap-backed ConcurrentKeyspace
- concurrent_handler.rs: connection handler for concurrent mode

usage: ember-server --concurrent

* chore: fmt and clippy fixes

Author: kacy

Cargo.lock

@@ -21,6 +21,8 @@ tokio = { workspace = true }
 tracing = { workspace = true }
 rand = { workspace = true }
 ordered-float = { workspace = true }
+dashmap = "6"
+parking_lot = "0.12"
 
 [dev-dependencies]
 tempfile = "3"
@@ -34,3 +36,7 @@ harness = false
 [[bench]]
 name = "engine"
 harness = false
+
+[[bench]]
+name = "concurrent"
+harness = false

crates/ember-core/Cargo.toml

@@ -0,0 +1,108 @@
+//! Benchmark comparing sharded engine vs concurrent keyspace.
+
+use bytes::Bytes;
+use criterion::{black_box, criterion_group, criterion_main, Criterion, Throughput};
+use ember_core::concurrent::ConcurrentKeyspace;
+use ember_core::keyspace::EvictionPolicy;
+use std::sync::Arc;
+
+fn bench_concurrent_set(c: &mut Criterion) {
+    let ks = Arc::new(ConcurrentKeyspace::new(None, EvictionPolicy::NoEviction));
+
+    let mut group = c.benchmark_group("concurrent");
+    group.throughput(Throughput::Elements(1));
+
+    group.bench_function("set", |b| {
+        let mut i = 0u64;
+        b.iter(|| {
+            let key = format!("key:{}", i);
+            i = i.wrapping_add(1);
+            ks.set(key, Bytes::from_static(b"value"), None);
+            black_box(())
+        })
+    });
+
+    // Pre-populate for get benchmark
+    for i in 0..10000 {
+        ks.set(format!("key:{}", i), Bytes::from_static(b"value"), None);
+    }
+
+    group.bench_function("get_existing", |b| {
+        let mut i = 0u64;
+        b.iter(|| {
+            let key = format!("key:{}", i % 10000);
+            i = i.wrapping_add(1);
+            black_box(ks.get(&key))
+        })
+    });
+
+    group.bench_function("get_missing", |b| {
+        let mut i = 0u64;
+        b.iter(|| {
+            let key = format!("missing:{}", i);
+            i = i.wrapping_add(1);
+            black_box(ks.get(&key))
+        })
+    });
+
+    group.finish();
+}
+
+fn bench_concurrent_multithread(c: &mut Criterion) {
+    use std::thread;
+
+    let ks = Arc::new(ConcurrentKeyspace::new(None, EvictionPolicy::NoEviction));
+
+    // Pre-populate
+    for i in 0..100000 {
+        ks.set(format!("key:{}", i), Bytes::from_static(b"value"), None);
+    }
+
+    let mut group = c.benchmark_group("concurrent_mt");
+    group.throughput(Throughput::Elements(8)); // 8 threads
+
+    group.bench_function("get_8_threads", |b| {
+        b.iter(|| {
+            let handles: Vec<_> = (0..8)
+                .map(|t| {
+                    let ks = Arc::clone(&ks);
+                    thread::spawn(move || {
+                        for i in 0..1000 {
+                            let key = format!("key:{}", (t * 1000 + i) % 100000);
+                            black_box(ks.get(&key));
+                        }
+                    })
+                })
+                .collect();
+
+            for h in handles {
+                h.join().unwrap();
+            }
+        })
+    });
+
+    group.bench_function("set_8_threads", |b| {
+        b.iter(|| {
+            let handles: Vec<_> = (0..8)
+                .map(|t| {
+                    let ks = Arc::clone(&ks);
+                    thread::spawn(move || {
+                        for i in 0..1000 {
+                            let key = format!("thread:{}:key:{}", t, i);
+                            ks.set(key, Bytes::from_static(b"value"), None);
+                        }
+                    })
+                })
+                .collect();
+
+            for h in handles {
+                h.join().unwrap();
+            }
+        })
+    });
+
+    group.finish();
+}
+
+criterion_group!(benches, bench_concurrent_set, bench_concurrent_multithread);
+criterion_main!(benches);

crates/ember-core/benches/concurrent.rs

@@ -93,7 +93,7 @@ fn bench_mixed(c: &mut Criterion) {
         let mut i = 0u64;
         b.iter(|| {
             let key = format!("key:{}", i % KEY_COUNT as u64);
-            if i % 2 == 0 {
+            if i.is_multiple_of(2) {
                 let _ = black_box(ks.get(&key));
             } else {
                 black_box(ks.set(key, value.clone(), None));