From 946014f04bf7015c3dc851f50968c5b0a27225ca Mon Sep 17 00:00:00 2001
From: Kacy Fortner <kacy@kacyfortner.com>
Date: Sat, 7 Feb 2026 13:42:06 -0500
Subject: [PATCH] perf: dispatch pipelined commands concurrently

instead of processing pipelined commands serially (awaiting each one
before starting the next), dispatch all commands to shards at once
and await them together using join_all.

for a pipeline of 16 commands, the old code did 16 sequential channel
round-trips. the new code dispatches all 16 concurrently, allowing
shards to process them in parallel.

this should significantly improve pipelined throughput by utilizing
all shards simultaneously rather than one at a time.
---
 Cargo.lock                            |  1 +
 crates/ember-server/Cargo.toml        |  1 +
 crates/ember-server/src/connection.rs | 27 ++++++++++++++++++++++-----
 3 files changed, 24 insertions(+), 5 deletions(-)

diff --git a/Cargo.lock b/Cargo.lock
index 19b79cf0..374d5282 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -545,6 +545,7 @@ dependencies = [
  "ember-persistence",
  "ember-protocol",
  "emberkv-core",
+ "futures",
  "metrics",
  "metrics-exporter-prometheus",
  "tikv-jemallocator",
diff --git a/crates/ember-server/Cargo.toml b/crates/ember-server/Cargo.toml
index 233eb9c3..ad96d653 100644
--- a/crates/ember-server/Cargo.toml
+++ b/crates/ember-server/Cargo.toml
@@ -25,6 +25,7 @@ tracing-subscriber = { workspace = true }
 clap = { workspace = true }
 metrics = { workspace = true }
 metrics-exporter-prometheus = { workspace = true }
+futures = "0.3"
 
 # optional: better multi-threaded allocation performance
 tikv-jemallocator = { version = "0.6", optional = true }
diff --git a/crates/ember-server/src/connection.rs b/crates/ember-server/src/connection.rs
index 6b8a90fd..1d96c8b5 100644
--- a/crates/ember-server/src/connection.rs
+++ b/crates/ember-server/src/connection.rs
@@ -2,7 +2,7 @@
 //!
 //! Reads RESP3 frames from a TCP stream, routes them through the
 //! sharded engine, and writes responses back. Supports pipelining
-//! by processing multiple frames from a single read.
+//! by dispatching multiple commands concurrently to shards.
 
 use std::sync::atomic::Ordering;
 use std::sync::Arc;
@@ -11,6 +11,7 @@ use std::time::{Duration, Instant};
 use bytes::{Bytes, BytesMut};
 use ember_core::{Engine, KeyspaceStats, ShardRequest, ShardResponse, TtlResult, Value};
 use ember_protocol::{parse_frame, Command, Frame, SetExpire};
+use futures::future::join_all;
 use tokio::io::{AsyncReadExt, AsyncWriteExt};
 use tokio::net::TcpStream;
 
@@ -67,15 +68,16 @@ pub async fn handle(
             Err(_) => return Ok(()), // idle timeout — close silently
         }
 
-        // process as many complete frames as the buffer holds (pipelining),
-        // batching all responses into a single write buffer
+        // parse all complete frames from the buffer first, then dispatch
+        // them concurrently to shards. this allows pipelined commands to
+        // be processed in parallel rather than serially.
         out.clear();
+        let mut frames = Vec::new();
         loop {
             match parse_frame(&buf) {
                 Ok(Some((frame, consumed))) => {
                     let _ = buf.split_to(consumed);
-                    let response = process(frame, &engine, ctx, slow_log).await;
-                    response.serialize(&mut out);
+                    frames.push(frame);
                 }
                 Ok(None) => break, // need more data
                 Err(e) => {
@@ -87,6 +89,21 @@ pub async fn handle(
             }
         }
 
+        // dispatch all commands concurrently — this is the key optimization.
+        // instead of await-ing each command serially (16 round-trips for a
+        // pipeline of 16), we dispatch all at once and await them together
+        // (effectively 1 round-trip worth of latency for all 16).
+        if !frames.is_empty() {
+            let futures: Vec<_> = frames
+                .into_iter()
+                .map(|frame| process(frame, &engine, ctx, slow_log))
+                .collect();
+            let responses = join_all(futures).await;
+            for response in responses {
+                response.serialize(&mut out);
+            }
+        }
+
         if !out.is_empty() {
             stream.write_all(&out).await?;
         }