Increase length delimited codec to maximum of 2GiB

Otherwise the server just stalls indefinitely

Author: johnramsden

example.server.toml

@@ -1,9 +1,9 @@
 [server]
 socket = "/tmp/oxcache.sock"
 disk = "/dev/nvme3n1"
-writer_threads = 14
-reader_threads = 14
-chunk_size = 8192
+writer_threads = 1
+reader_threads = 1
+chunk_size = 268435456
 
 [remote]
 remote_type = "emulated" # emulated | S3

oxcache/src/bin/simpleevaluationclient.rs

@@ -30,10 +30,12 @@ struct Cli {
     num_clients: usize,
 }
 
+const MAX_FRAME_LENGTH: usize = 2 * 1024 * 1024 * 1024; // 2 GB
+
 #[tokio::main]
 async fn main() -> Result<(), Box<dyn std::error::Error>> {
     let nr_queries = 10000;
-    let nr_uuids = 1000;
+    let nr_uuids = 100;
     let mut queries: Arc<Mutex<Vec<GetRequest>>> = Arc::new(Mutex::new(Vec::new()));
     let args = Cli::parse();
     let counter = Arc::new(AtomicUsize::new(0));
@@ -55,7 +57,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
 
             queries.push(GetRequest {
                 key: uuid.to_string(),
-                size: 8192,
+                size: 268435456,
                 offset: 0,
             });
         }
@@ -77,9 +79,14 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
             let stream = UnixStream::connect(&sock).await?;
             println!("[t.{}] Client connected to {}", c, sock);
 
-            let (read_half, write_half) = split(stream);
-            let mut reader = FramedRead::new(read_half, LengthDelimitedCodec::new());
-            let mut writer = FramedWrite::new(write_half, LengthDelimitedCodec::new());
+            let (read_half, write_half) = tokio::io::split(stream);
+
+            let codec = LengthDelimitedCodec::builder()
+                .max_frame_length(MAX_FRAME_LENGTH)
+                .new_codec();
+
+            let mut reader = FramedRead::new(read_half, codec.clone());
+            let mut writer = FramedWrite::new(write_half, codec);
 
             loop {
                 let query_num = counter.fetch_add(1, Ordering::Relaxed) + 1;

oxcache/src/server.rs

@@ -148,6 +148,8 @@ impl<T: RemoteBackend + Send + Sync + 'static> Server<T> {
     }
 }
 
+const MAX_FRAME_LENGTH: usize = 2 * 1024 * 1024 * 1024; // 2 GB
+
 async fn handle_connection<T: RemoteBackend + Send + Sync + 'static>(
     stream: UnixStream,
     writer_pool: Arc<WriterPool>,
@@ -156,9 +158,14 @@ async fn handle_connection<T: RemoteBackend + Send + Sync + 'static>(
     cache: Arc<Cache>,
     chunk_size: usize,
 ) -> tokio::io::Result<()> {
-    let (read_half, write_half) = split(stream);
-    let mut reader = FramedRead::new(read_half, LengthDelimitedCodec::new());
-    let writer = Arc::new(Mutex::new(FramedWrite::new(write_half, LengthDelimitedCodec::new())));
+    let (read_half, write_half) = tokio::io::split(stream);
+
+    let codec = LengthDelimitedCodec::builder()
+        .max_frame_length(MAX_FRAME_LENGTH)
+        .new_codec();
+
+    let mut reader = FramedRead::new(read_half, codec.clone());
+    let writer = Arc::new(Mutex::new(FramedWrite::new(write_half, codec)));
 
     while let Some(frame) = reader.next().await {
         let f = frame.map_err(|e| std::io::Error::new(std::io::ErrorKind::Other, format!("frame read failed: {}", e)))?;
@@ -169,7 +176,7 @@ async fn handle_connection<T: RemoteBackend + Send + Sync + 'static>(
 
         match msg {
             Ok((request, _)) => {
-                // println!("Received: {:?}", request);
+                println!("Received req");
                 match request {
                     request::Request::Get(req) => {
                         if let Err(e) = req.validate(chunk_size) {
@@ -259,11 +266,11 @@ async fn handle_connection<T: RemoteBackend + Send + Sync + 'static>(
                                             return Err(std::io::Error::new(std::io::ErrorKind::Other, format!("remote.get failed: {}", e)));
                                         }
                                     };
-                                    
+
                                     let encoded = bincode::serde::encode_to_vec(
                                         request::GetResponse::Response(resp.clone()),
                                         bincode::config::standard()
-                                    ).unwrap();
+                                    ).map_err(|e| std::io::Error::new(std::io::ErrorKind::Other, format!("serialization failed: {}", e)))?;
                                     {
                                         let mut w = writer.lock().await;
                                         w.send(Bytes::from(encoded)).await.map_err(|e| {