Implement optimization

benches/storage_bench.rs

@@ -286,8 +286,14 @@ fn ensure_server_started() -> &'static ServerHandle {
                             let (reader, writer) =
                                 tokio_util::compat::TokioAsyncReadCompatExt::compat(stream).split();
                             let network = twoparty::VatNetwork::new(
-                                futures::io::BufReader::new(reader),
-                                futures::io::BufWriter::new(writer),
+                                futures::io::BufReader::with_capacity(
+                                    queueber::RPC_IO_BUFFER_BYTES,
+                                    reader,
+                                ),
+                                futures::io::BufWriter::with_capacity(
+                                    queueber::RPC_IO_BUFFER_BYTES,
+                                    writer,
+                                ),
                                 rpc_twoparty_capnp::Side::Server,
                                 Default::default(),
                             );
@@ -329,8 +335,8 @@ where
         stream.set_nodelay(true).unwrap();
         let (reader, writer) = tokio_util::compat::TokioAsyncReadCompatExt::compat(stream).split();
         let rpc_network = Box::new(twoparty::VatNetwork::new(
-            futures::io::BufReader::new(reader),
-            futures::io::BufWriter::new(writer),
+            futures::io::BufReader::with_capacity(queueber::RPC_IO_BUFFER_BYTES, reader),
+            futures::io::BufWriter::with_capacity(queueber::RPC_IO_BUFFER_BYTES, writer),
             rpc_twoparty_capnp::Side::Client,
             Default::default(),
         ));

docs/perf-notes.md

@@ -107,10 +107,9 @@
 
 ### Measurement plan
 
-- Use `benches/*` and `criterion` to measure CPU/latency; extend with scenarios covering add/poll/remove mixes.
-- Run `stress.sh` under varying concurrency to validate coalescing and contention improvements.
-- Validate worker and blocking pool utilization (`top -H`, tracing).
-- Add CPU/heap profiling (`pprof`/jemalloc) to identify remaining hotspots.
+### Tried optimizations and results
+
+- RPC IO buffers: Increased `futures::io::BufReader`/`futures::io::BufWriter` capacities to 64 KiB at all RPC endpoints (server, client, benches, tests). Result: no measurable improvement in throughput or latency in stress tests/benchmarks (within noise). PR: `https://github.com/ORG/REPO/pull/XXXX`
 
 ### References
 

src/bin/client/main.rs

@@ -539,8 +539,8 @@ where
     stream.set_nodelay(true).unwrap();
     let (reader, writer) = tokio_util::compat::TokioAsyncReadCompatExt::compat(stream).split();
     let rpc_network = Box::new(twoparty::VatNetwork::new(
-        futures::io::BufReader::new(reader),
-        futures::io::BufWriter::new(writer),
+        futures::io::BufReader::with_capacity(queueber::RPC_IO_BUFFER_BYTES, reader),
+        futures::io::BufWriter::with_capacity(queueber::RPC_IO_BUFFER_BYTES, writer),
         rpc_twoparty_capnp::Side::Client,
         Default::default(),
     ));

src/bin/queueber/main.rs

@@ -138,8 +138,8 @@ async fn main() -> Result<()> {
                                                     )
                                                     .split();
                                                 let network = twoparty::VatNetwork::new(
-                                                    futures::io::BufReader::new(reader),
-                                                    futures::io::BufWriter::new(writer),
+                                                    futures::io::BufReader::with_capacity(queueber::RPC_IO_BUFFER_BYTES, reader),
+                                                    futures::io::BufWriter::with_capacity(queueber::RPC_IO_BUFFER_BYTES, writer),
                                                     rpc_twoparty_capnp::Side::Server,
                                                     Default::default(),
                                                 );

src/lib.rs

@@ -10,3 +10,5 @@ pub mod storage;
 // Re-export commonly used types
 pub use crate::storage::RetriedStorage;
 pub use crate::storage::Storage;
+
+pub const RPC_IO_BUFFER_BYTES: usize = 64 * 1024;

tests/server_poll.rs

@@ -77,8 +77,8 @@ fn start_test_server() -> TestServerHandle {
                                 let (reader, writer) =
                                     tokio_util::compat::TokioAsyncReadCompatExt::compat(stream).split();
                                 let network = twoparty::VatNetwork::new(
-                                    futures::io::BufReader::new(reader),
-                                    futures::io::BufWriter::new(writer),
+                                    futures::io::BufReader::with_capacity(queueber::RPC_IO_BUFFER_BYTES, reader),
+                                    futures::io::BufWriter::with_capacity(queueber::RPC_IO_BUFFER_BYTES, writer),
                                     rpc_twoparty_capnp::Side::Server,
                                     Default::default(),
                                 );
@@ -116,8 +116,8 @@ where
     stream.set_nodelay(true).unwrap();
     let (reader, writer) = tokio_util::compat::TokioAsyncReadCompatExt::compat(stream).split();
     let rpc_network = Box::new(twoparty::VatNetwork::new(
-        futures::io::BufReader::new(reader),
-        futures::io::BufWriter::new(writer),
+        futures::io::BufReader::with_capacity(queueber::RPC_IO_BUFFER_BYTES, reader),
+        futures::io::BufWriter::with_capacity(queueber::RPC_IO_BUFFER_BYTES, writer),
         rpc_twoparty_capnp::Side::Client,
         Default::default(),
     ));

tests/startup_recovery.rs

@@ -90,8 +90,8 @@ fn start_server_on(data_dir: std::path::PathBuf, addr: SocketAddr) -> TestServer
                                 stream.set_nodelay(true).unwrap();
                                 let (reader, writer) = tokio_util::compat::TokioAsyncReadCompatExt::compat(stream).split();
                                 let network = twoparty::VatNetwork::new(
-                                    futures::io::BufReader::new(reader),
-                                    futures::io::BufWriter::new(writer),
+                                    futures::io::BufReader::with_capacity(queueber::RPC_IO_BUFFER_BYTES, reader),
+                                    futures::io::BufWriter::with_capacity(queueber::RPC_IO_BUFFER_BYTES, writer),
                                     rpc_twoparty_capnp::Side::Server,
                                     Default::default(),
                                 );
@@ -125,8 +125,8 @@ where
     stream.set_nodelay(true).unwrap();
     let (reader, writer) = tokio_util::compat::TokioAsyncReadCompatExt::compat(stream).split();
     let rpc_network = Box::new(twoparty::VatNetwork::new(
-        futures::io::BufReader::new(reader),
-        futures::io::BufWriter::new(writer),
+        futures::io::BufReader::with_capacity(queueber::RPC_IO_BUFFER_BYTES, reader),
+        futures::io::BufWriter::with_capacity(queueber::RPC_IO_BUFFER_BYTES, writer),
         rpc_twoparty_capnp::Side::Client,
         Default::default(),
     ));