Merge pull request containerd#89 from Tim-Zhang/server-timeout

Add server timeout, remove client timeout

Author: liubin

example/async-client.rs

@@ -45,7 +45,8 @@ async fn main() {
         println!(
             "Green Thread 1 - {} -> {:?} ended: {:?}",
             "health.check()",
-            thc.check(default_ctx(), &req).await,
+            thc.check(context::with_timeout(20 * 1000 * 1000), &req)
+                .await,
             now.elapsed(),
         );
     });

src/asynchronous/client.rs

@@ -8,7 +8,6 @@ use protobuf::{CodedInputStream, CodedOutputStream, Message};
 use std::collections::HashMap;
 use std::os::unix::io::RawFd;
 use std::sync::{Arc, Mutex};
-use std::time::Duration;
 
 use crate::common::MESSAGE_TYPE_RESPONSE;
 use crate::error::{Error, Result};
@@ -21,7 +20,6 @@ use tokio::{
     io::{split, AsyncWriteExt},
     sync::mpsc::{channel, Receiver, Sender},
     sync::Notify,
-    time::timeout,
 };
 
 type RequestSender = Sender<(Vec<u8>, Sender<Result<Vec<u8>>>)>;
@@ -158,21 +156,10 @@ impl Client {
             .await
             .map_err(|e| Error::Others(format!("Send packet to sender error {:?}", e)))?;
 
-        let result: Result<Vec<u8>> = if req.timeout_nano == 0 {
-            rx.recv()
-                .await
-                .ok_or_else(|| Error::Others("Recive packet from recver error".to_string()))?
-        } else {
-            match timeout(Duration::from_nanos(req.timeout_nano as u64), rx.recv()).await {
-                Ok(result) => result
-                    .ok_or_else(|| Error::Others("Recive packet from recver error".to_string()))?,
-                Err(_) => {
-                    return Err(Error::Others(
-                        "Recive packet from recver error: timeout".to_string(),
-                    ))
-                }
-            }
-        };
+        let result = rx
+            .recv()
+            .await
+            .ok_or_else(|| Error::Others("Recive packet from recver error".to_string()))?;
 
         let buf = result?;
         let mut s = CodedInputStream::from_bytes(&buf);

src/asynchronous/server.rs

@@ -9,6 +9,7 @@ use std::collections::HashMap;
 use std::os::unix::io::RawFd;
 use std::result::Result as StdResult;
 use std::sync::Arc;
+use std::time::Duration;
 
 use crate::asynchronous::stream::{receive, respond, respond_with_status};
 use crate::asynchronous::unix_incoming::UnixIncoming;
@@ -30,6 +31,7 @@ use tokio::{
     select, spawn,
     sync::mpsc::{channel, Receiver, Sender},
     sync::watch,
+    time::timeout,
 };
 use tokio_vsock::VsockListener;
 
@@ -322,10 +324,32 @@ async fn do_handle_request(
         metadata: context::from_pb(&req.metadata),
     };
 
-    method.handler(ctx, req).await.map_err(|e| {
+    let get_unknown_status_and_log_err = |e| {
         error!("method handle {} got error {:?}", path, &e);
         get_status(Code::UNKNOWN, e)
-    })
+    };
+
+    if req.timeout_nano == 0 {
+        method
+            .handler(ctx, req)
+            .await
+            .map_err(get_unknown_status_and_log_err)
+    } else {
+        timeout(
+            Duration::from_nanos(req.timeout_nano as u64),
+            method.handler(ctx, req),
+        )
+        .await
+        .map_err(|_| {
+            // Timed out
+            error!("method handle {} got error timed out", path);
+            get_status(Code::DEADLINE_EXCEEDED, "timeout")
+        })
+        .and_then(|r| {
+            // Handler finished
+            r.map_err(get_unknown_status_and_log_err)
+        })
+    }
 }
 
 async fn handle_request(