sandbox: bugfix infinite loop in new_ttrpc_client()

Signed-off-by: Zhang Tianyang <[email protected]>

Author: Burning1020

vmm/sandbox/src/client.rs

@@ -31,7 +31,7 @@ use nix::{
     unistd::close,
 };
 use tokio::{
-    io::{AsyncReadExt, AsyncWriteExt},
+    io::{AsyncBufReadExt, AsyncWriteExt, BufReader},
     net::UnixStream,
     time::timeout,
 };
@@ -41,17 +41,17 @@ use vmm_common::api::{sandbox::*, sandbox_ttrpc::SandboxServiceClient};
 use crate::network::{NetworkInterface, Route};
 
 const HVSOCK_RETRY_TIMEOUT_IN_MS: u64 = 10;
+// TODO: reduce to 10s
+const NEW_TTRPC_CLIENT_TIMEOUT: u64 = 45;
 const TIME_SYNC_PERIOD: u64 = 60;
 const TIME_DIFF_TOLERANCE_IN_MS: u64 = 10;
 
 pub(crate) async fn new_sandbox_client(address: &str) -> Result<SandboxServiceClient> {
-    let client = new_ttrpc_client(address).await?;
+    let client = new_ttrpc_client_with_timeout(address, NEW_TTRPC_CLIENT_TIMEOUT).await?;
     Ok(SandboxServiceClient::new(client))
 }
 
-async fn new_ttrpc_client(address: &str) -> Result<Client> {
-    let ctx_timeout = 10;
-
+async fn new_ttrpc_client_with_timeout(address: &str, t: u64) -> Result<Client> {
     let mut last_err = Error::Other(anyhow!(""));
 
     let fut = async {
@@ -63,12 +63,17 @@ async fn new_ttrpc_client(address: &str) -> Result<Client> {
                 }
                 Err(e) => last_err = e,
             }
+            // In case that the address doesn't exist, the executed function in this loop are all
+            // sync, making the first time of future poll in timeout hang forever. As a result, the
+            // timeout will hang too. To solve this, add a async function in this loop or call
+            // `tokio::task::yield_now()` to give up current cpu time slice.
+            tokio::time::sleep(Duration::from_millis(10)).await;
         }
     };
 
-    let client = timeout(Duration::from_secs(ctx_timeout), fut)
+    let client = timeout(Duration::from_secs(t), fut)
         .await
-        .map_err(|_| anyhow!("{}s timeout connecting socket: {}", ctx_timeout, last_err))?;
+        .map_err(|_| anyhow!("{}s timeout connecting socket: {}", t, last_err))?;
     Ok(client)
 }
 
@@ -167,31 +172,21 @@ async fn connect_to_hvsocket(address: &str) -> Result<RawFd> {
 
     let fut = async {
         let mut stream = UnixStream::connect(addr).await?;
-
-        match stream.write(format!("CONNECT {}\n", port).as_bytes()).await {
-            Ok(_) => {
-                let mut buf = [0; 4096];
-                match stream.read(&mut buf).await {
-                    Ok(0) => Err(anyhow!("stream closed")),
-                    Ok(n) => {
-                        if String::from_utf8(buf[..n].to_vec())
-                            .unwrap_or_default()
-                            .contains("OK")
-                        {
-                            return Ok(stream.into_std()?.into_raw_fd());
-                        }
-                        Err(anyhow!("failed to connect"))
-                    }
-                    Err(ref e) if e.kind() == std::io::ErrorKind::WouldBlock => {
-                        Err(anyhow!("{}", e))
-                    }
-                    Err(e) => Err(anyhow!("failed to read from hvsock: {}", e)),
-                }
-            }
-            Err(ref e) if e.kind() == std::io::ErrorKind::WouldBlock => Err(anyhow!("{}", e)),
-            Err(e) => Err(anyhow!("failed to write CONNECT to hvsock: {}", e)),
+        stream
+            .write_all(format!("CONNECT {}\n", port).as_bytes())
+            .await
+            .map_err(|e| anyhow!("hvsock connected but failed to write CONNECT: {}", e))?;
+
+        let mut response = String::new();
+        BufReader::new(&mut stream)
+            .read_line(&mut response)
+            .await
+            .map_err(|e| anyhow!("CONNECT sent but failed to get response: {}", e))?;
+        if response.starts_with("OK") {
+            Ok(stream.into_std()?.into_raw_fd())
+        } else {
+            Err(anyhow!("CONNECT sent but response is not OK: {}", response).into())
         }
-        .map_err(Error::Other)
     };
 
     timeout(Duration::from_millis(HVSOCK_RETRY_TIMEOUT_IN_MS), fut)
@@ -312,3 +307,16 @@ pub(crate) async fn client_sync_clock(client: &SandboxServiceClient, id: &str) {
         }
     });
 }
+
+#[cfg(test)]
+mod tests {
+    use crate::client::new_ttrpc_client_with_timeout;
+
+    #[tokio::test]
+    async fn test_new_ttrpc_client_timeout() {
+        // Expect new_ttrpc_client would return timeout error, instead of blocking.
+        assert!(new_ttrpc_client_with_timeout("hvsock://fake.sock:1024", 1)
+            .await
+            .is_err());
+    }
+}

vmm/sandbox/src/cloud_hypervisor/client.rs

@@ -16,15 +16,15 @@ limitations under the License.
 
 use std::{
     os::unix::net::UnixStream,
-    path::Path,
     thread::sleep,
     time::{Duration, SystemTime},
 };
 
 use anyhow::anyhow;
 use api_client::{simple_api_command, simple_api_full_command_with_fds_and_response};
 use containerd_sandbox::error::Result;
-use log::{error, trace};
+use log::{debug, error, trace};
+use tokio::task::spawn_blocking;
 
 use crate::{
     cloud_hypervisor::devices::{block::DiskConfig, AddDeviceResponse, RemoveDeviceRequest},
@@ -38,25 +38,32 @@ pub struct ChClient {
 }
 
 impl ChClient {
-    pub fn new<P: AsRef<Path>>(socket_path: P) -> Result<Self> {
+    pub async fn new(socket_path: String) -> Result<Self> {
+        let s = socket_path.to_string();
         let start_time = SystemTime::now();
-        loop {
-            match UnixStream::connect(&socket_path) {
-                Ok(socket) => {
-                    return Ok(Self { socket });
-                }
-                Err(e) => {
-                    trace!("failed to create client: {:?}", e);
-                    if start_time.elapsed().unwrap().as_secs()
-                        > CLOUD_HYPERVISOR_START_TIMEOUT_IN_SEC
-                    {
-                        error!("failed to create client: {:?}", e);
-                        return Err(anyhow!("timeout connect client, {}", e).into());
+        let socket = spawn_blocking(move || -> Result<UnixStream> {
+            loop {
+                match UnixStream::connect(&socket_path) {
+                    Ok(socket) => {
+                        return Ok(socket);
+                    }
+                    Err(e) => {
+                        trace!("failed to create client: {:?}", e);
+                        if start_time.elapsed().unwrap().as_secs()
+                            > CLOUD_HYPERVISOR_START_TIMEOUT_IN_SEC
+                        {
+                            error!("failed to create client: {:?}", e);
+                            return Err(anyhow!("timeout connect client, {}", e).into());
+                        }
+                        sleep(Duration::from_millis(10));
                     }
-                    sleep(Duration::from_millis(10));
                 }
             }
-        }
+        })
+        .await
+        .map_err(|e| anyhow!("failed to join thread {}", e))??;
+        debug!("connected to api server {}", s);
+        Ok(Self { socket })
     }
 
     pub fn hot_attach(&mut self, device_info: DeviceInfo) -> Result<String> {

vmm/sandbox/src/cloud_hypervisor/mod.rs

@@ -113,7 +113,7 @@ impl CloudHypervisorVM {
     }
 
     async fn create_client(&self) -> Result<ChClient> {
-        ChClient::new(&self.config.api_socket)
+        ChClient::new(self.config.api_socket.to_string()).await
     }
 
     fn get_client(&mut self) -> Result<&mut ChClient> {