shim: implement close_io

Signed-off-by: Zhang Tianyang <burning9699@gmail.com>

crates/shim/src/asynchronous/container.rs

@@ -49,6 +49,7 @@ pub trait Container {
     async fn update(&mut self, resources: &LinuxResources) -> Result<()>;
     async fn stats(&self) -> Result<Metrics>;
     async fn all_processes(&self) -> Result<Vec<ProcessInfo>>;
+    async fn close_io(&mut self, exec_id: Option<&str>) -> Result<()>;
 }
 
 #[async_trait]
@@ -182,6 +183,11 @@ where
     async fn all_processes(&self) -> Result<Vec<ProcessInfo>> {
         self.init.ps().await
     }
+
+    async fn close_io(&mut self, exec_id: Option<&str>) -> Result<()> {
+        let process = self.get_mut_process(exec_id)?;
+        process.close_io().await
+    }
 }
 
 impl<T, E, P> ContainerTemplate<T, E, P>

crates/shim/src/asynchronous/processes.rs

@@ -24,6 +24,8 @@ use containerd_shim_protos::{
 };
 use oci_spec::runtime::LinuxResources;
 use time::OffsetDateTime;
+use tokio::fs::File;
+use tokio::sync::Mutex;
 use tokio::sync::oneshot::{channel, Receiver, Sender};
 
 use crate::{io::Stdio, ioctl_set_winsz, util::asyncify, Console};
@@ -43,6 +45,7 @@ pub trait Process {
     async fn update(&mut self, resources: &LinuxResources) -> crate::Result<()>;
     async fn stats(&self) -> crate::Result<Metrics>;
     async fn ps(&self) -> crate::Result<Vec<ProcessInfo>>;
+    async fn close_io(&mut self) -> crate::Result<()>;
 }
 
 #[async_trait]
@@ -65,6 +68,7 @@ pub struct ProcessTemplate<S> {
     pub wait_chan_tx: Vec<Sender<()>>,
     pub console: Option<Console>,
     pub lifecycle: Arc<S>,
+    pub stdin: Arc<Mutex<Option<File>>>,
 }
 
 impl<S> ProcessTemplate<S> {
@@ -79,6 +83,7 @@ impl<S> ProcessTemplate<S> {
             wait_chan_tx: vec![],
             console: None,
             lifecycle: Arc::new(lifecycle),
+            stdin: Arc::new(Mutex::new(None)),
         }
     }
 }
@@ -176,4 +181,12 @@ where
     async fn ps(&self) -> crate::Result<Vec<ProcessInfo>> {
         self.lifecycle.ps(self).await
     }
+
+    async fn close_io(&mut self) -> crate::Result<()> {
+        let mut lock_guard = self.stdin.lock().await;
+        if let Some(stdin_w_file) = lock_guard.take() {
+            drop(stdin_w_file);
+        }
+        Ok(())
+    }
 }

crates/shim/src/asynchronous/task.rs

@@ -269,8 +269,9 @@ where
         Ok(Empty::new())
     }
 
-    async fn close_io(&self, _ctx: &TtrpcContext, _req: CloseIORequest) -> TtrpcResult<Empty> {
-        // TODO call close_io of container
+    async fn close_io(&self, _ctx: &TtrpcContext, req: CloseIORequest) -> TtrpcResult<Empty> {
+        let mut container = self.get_container(req.id()).await?;
+        container.close_io(req.exec_id().as_option()).await?;
         Ok(Empty::new())
     }