From bccca99cf6c9e39b3c365885db2917bfb044fe36 Mon Sep 17 00:00:00 2001 From: jokemanfire Date: Mon, 10 Feb 2025 17:09:19 +0800 Subject: [PATCH] Use pipe in shim , not use fifo directly Do io copy like goshim Signed-off-by: jokemanfire --- crates/runc-shim/src/common.rs | 35 ++++++++++++++++++++----------- crates/runc-shim/src/processes.rs | 4 +++- crates/runc-shim/src/runc.rs | 33 ++++++++++++++++++++++++----- 3 files changed, 54 insertions(+), 18 deletions(-) diff --git a/crates/runc-shim/src/common.rs b/crates/runc-shim/src/common.rs index 6421ce6e..3291b106 100644 --- a/crates/runc-shim/src/common.rs +++ b/crates/runc-shim/src/common.rs @@ -30,9 +30,7 @@ use std::{ use containerd_shim::{ api::{ExecProcessRequest, Options}, - io_error, other, other_error, - util::IntoOption, - Error, + io_error, other, other_error, Error, }; use log::{debug, warn}; use nix::{ @@ -44,7 +42,7 @@ use nix::{ }; use oci_spec::runtime::{LinuxNamespaceType, Spec}; use runc::{ - io::{Io, NullIo, FIFO}, + io::{IOOption, Io, NullIo, PipedIo}, options::GlobalOpts, Runc, Spawner, }; @@ -77,8 +75,8 @@ pub struct ProcessIO { pub fn create_io( id: &str, - _io_uid: u32, - _io_gid: u32, + io_uid: u32, + io_gid: u32, stdio: &Stdio, ) -> containerd_shim::Result { let mut pio = ProcessIO::default(); @@ -101,19 +99,32 @@ pub fn create_io( if scheme == FIFO_SCHEME { debug!( - "create named pipe io for container {}, stdin: {}, stdout: {}, stderr: {}", + "create pipe io for container {}, stdin: {}, stdout: {}, stderr: {}", id, stdio.stdin.as_str(), stdio.stdout.as_str(), stdio.stderr.as_str() ); - let io = FIFO { - stdin: stdio.stdin.to_string().none_if(|x| x.is_empty()), - stdout: stdio.stdout.to_string().none_if(|x| x.is_empty()), - stderr: stdio.stderr.to_string().none_if(|x| x.is_empty()), + + // let io = FIFO { + // stdin: stdio.stdin.to_string().none_if(|x| x.is_empty()), + // stdout: stdio.stdout.to_string().none_if(|x| x.is_empty()), + // stderr: stdio.stderr.to_string().none_if(|x| x.is_empty()), + // }; + // pio.copy = false; + + if stdio.stdin.is_empty() { + debug!("stdin is empty"); + } + let opts = IOOption { + open_stdin: !stdio.stdin.is_empty(), + open_stdout: !stdio.stdout.is_empty(), + open_stderr: !stdio.stderr.is_empty(), }; + let io = PipedIo::new(io_uid, io_gid, &opts).unwrap(); + pio.copy = true; + pio.io = Some(Arc::new(io)); - pio.copy = false; } Ok(pio) } diff --git a/crates/runc-shim/src/processes.rs b/crates/runc-shim/src/processes.rs index 16c1295c..74090753 100644 --- a/crates/runc-shim/src/processes.rs +++ b/crates/runc-shim/src/processes.rs @@ -37,7 +37,7 @@ use tokio::{ sync::oneshot::{channel, Receiver, Sender}, }; -use crate::io::Stdio; +use crate::{common::ProcessIO, io::Stdio}; #[allow(dead_code)] #[async_trait] @@ -77,6 +77,7 @@ pub struct ProcessTemplate { pub state: Status, pub id: String, pub stdio: Stdio, + pub io: Option>, pub pid: i32, pub exit_code: i32, pub exited_at: Option, @@ -92,6 +93,7 @@ impl ProcessTemplate { state: Status::CREATED, id: id.to_string(), stdio, + io: None, pid: 0, exit_code: 0, exited_at: None, diff --git a/crates/runc-shim/src/runc.rs b/crates/runc-shim/src/runc.rs index e2eb0660..3b894a29 100644 --- a/crates/runc-shim/src/runc.rs +++ b/crates/runc-shim/src/runc.rs @@ -163,8 +163,10 @@ impl RuncFactory { (Some(s), None) } else { let pio = create_io(&id, opts.io_uid, opts.io_gid, stdio)?; - create_opts.io = pio.io.as_ref().cloned(); - (None, Some(pio)) + let ref_pio = Arc::new(pio); + create_opts.io = ref_pio.io.clone(); + init.io = Some(ref_pio.clone()); + (None, Some(ref_pio)) }; let resp = init @@ -178,6 +180,22 @@ impl RuncFactory { } return Err(runtime_error(bundle, e, "OCI runtime create failed").await); } + if !init.stdio.stdin.is_empty() { + let stdin_clone = init.stdio.stdin.clone(); + let stdin_w = init.stdin.clone(); + // Open the write side in advance to make sure read side will not block, + // open it in another thread otherwise it will block too. + tokio::spawn(async move { + if let Ok(stdin_w_file) = OpenOptions::new() + .write(true) + .open(stdin_clone.as_str()) + .await + { + let mut lock_guard = stdin_w.lock().unwrap(); + *lock_guard = Some(stdin_w_file); + } + }); + } copy_io_or_console(init, socket, pio, init.lifecycle.exit_signal.clone()).await?; let pid = read_file_to_str(pid_path).await?.parse::()?; init.pid = pid; @@ -232,6 +250,7 @@ impl ProcessFactory for RuncExecFactory { stderr: req.stderr.to_string(), terminal: req.terminal, }, + io: None, pid: 0, exit_code: 0, exited_at: None, @@ -299,6 +318,7 @@ impl ProcessLifecycle for RuncInitLifecycle { ); } } + self.exit_signal.signal(); Ok(()) } @@ -434,8 +454,10 @@ impl ProcessLifecycle for RuncExecLifecycle { (Some(s), None) } else { let pio = create_io(&p.id, self.io_uid, self.io_gid, &p.stdio)?; - exec_opts.io = pio.io.as_ref().cloned(); - (None, Some(pio)) + let ref_pio = Arc::new(pio); + exec_opts.io = ref_pio.io.clone(); + p.io = Some(ref_pio.clone()); + (None, Some(ref_pio)) }; //TODO checkpoint support let exec_result = self @@ -698,7 +720,7 @@ where async fn copy_io_or_console

( p: &mut ProcessTemplate

, socket: Option, - pio: Option, + pio: Option>, exit_signal: Arc, ) -> Result<()> { if p.stdio.terminal { @@ -736,6 +758,7 @@ impl Spawner for ShimExecutor { } }; let pid = child.id().unwrap(); + let (stdout, stderr, exit_code) = tokio::join!( read_std(child.stdout), read_std(child.stderr),