build(deps): bump nix from 0.25.0 to 0.28.0 in /vmm/task

vmm/task/Cargo.toml

@@ -8,9 +8,9 @@ edition = "2021"
 panic = 'abort'
 
 [dependencies]
-vmm-common = {path = "../common"}
+vmm-common = { path = "../common" }
 log = "0.4"
-nix = "0.25"
+nix = { version = "0.28.0", features = ["sched", "term", "time", "hostname", "signal", "mount", "uio", "socket"] }
 libc = "0.2.95"
 time = { version = "=0.3.7", features = ["serde", "std"] }
 serde = { version = "1.0.133", features = ["derive"] }
@@ -19,21 +19,21 @@ oci-spec = "0.5.4"
 crossbeam = "0.8.1"
 env_logger = "0.9.0"
 lazy_static = "1.4.0"
-netlink-sys = { version = "0.7.0", features = ["tokio_socket"]}
+netlink-sys = { version = "0.7.0", features = ["tokio_socket"] }
 rtnetlink = "0.12"
 netlink-packet-route = "0.15"
 netlink-packet-core = "0.5.0"
 ipnetwork = "0.20"
 anyhow = { version = "1.0.66", default-features = false, features = ["std", "backtrace"] }
 
 # Async dependencies
-async-trait = { version = "0.1.51"}
-tokio = { version = "1.17.0", features = ["full"]}
-futures = { version = "0.3.21"}
-signal-hook-tokio = {version = "0.3.1", features = ["futures-v0_3"]}
+async-trait = { version = "0.1.51" }
+tokio = { version = "1.17.0", features = ["full"] }
+futures = { version = "0.3.21" }
+signal-hook-tokio = { version = "0.3.1", features = ["futures-v0_3"] }
 tokio-vsock = "0.3.1"
 pin-project-lite = "0.2.7"
 ttrpc = { version = "0.7", features = ["async"] }
 
-containerd-shim = { git="https://github.com/kuasar-io/rust-extensions.git", features=["async"] }
-runc = { git="https://github.com/kuasar-io/rust-extensions.git", features=["async"] }
+containerd-shim = { git = "https://github.com/kuasar-io/rust-extensions.git", features = ["async"] }
+runc = { git = "https://github.com/kuasar-io/rust-extensions.git", features = ["async"] }

vmm/task/src/debug.rs

@@ -14,7 +14,10 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-use std::{os::unix::prelude::FromRawFd, process::Stdio};
+use std::{
+    os::{fd::IntoRawFd, unix::prelude::FromRawFd},
+    process::Stdio,
+};
 
 use containerd_shim::{
     io_error,
@@ -47,9 +50,10 @@ pub async fn debug_console(stream: VsockStream) -> Result<()> {
     let pty = openpty(None, None)?;
     let pty_master = pty.master;
     let mut cmd = Command::new("/bin/bash");
-    cmd.stdin(unsafe { Stdio::from_raw_fd(pty.slave) });
-    cmd.stdout(unsafe { Stdio::from_raw_fd(pty.slave) });
-    cmd.stderr(unsafe { Stdio::from_raw_fd(pty.slave) });
+    let pty_fd = pty.slave.into_raw_fd();
+    cmd.stdin(unsafe { Stdio::from_raw_fd(pty_fd) });
+    cmd.stdout(unsafe { Stdio::from_raw_fd(pty_fd) });
+    cmd.stderr(unsafe { Stdio::from_raw_fd(pty_fd) });
     unsafe {
         cmd.pre_exec(move || {
             setsid()?;

vmm/task/src/io.rs

@@ -17,7 +17,10 @@
 use std::{
     io::{ErrorKind, IoSliceMut},
     ops::Deref,
-    os::unix::prelude::{AsRawFd, FromRawFd, RawFd},
+    os::{
+        fd::{IntoRawFd, OwnedFd},
+        unix::prelude::{AsRawFd, FromRawFd, RawFd},
+    },
     path::PathBuf,
     pin::Pin,
     sync::Arc,
@@ -145,8 +148,8 @@ async fn copy_console<P>(
 ) -> Result<Console> {
     debug!("copy_console: waiting for runtime to send console fd");
     let stream = console_socket.accept().await?;
-    let fd = asyncify(move || -> Result<RawFd> { receive_socket(stream.as_raw_fd()) }).await?;
-    let f = unsafe { File::from_raw_fd(fd) };
+    let fd = asyncify(move || -> Result<OwnedFd> { receive_socket(stream.as_raw_fd()) }).await?;
+    let f = unsafe { File::from_raw_fd(fd.into_raw_fd()) };
     if !stdio.stdin.is_empty() {
         debug!("copy_console: pipe stdin to console");
 
@@ -356,7 +359,7 @@ where
     }
 }
 
-pub fn receive_socket(stream_fd: RawFd) -> containerd_shim::Result<RawFd> {
+pub fn receive_socket(stream_fd: RawFd) -> Result<OwnedFd> {
     let mut buf = [0u8; 4096];
     let mut iovec = [IoSliceMut::new(&mut buf)];
     let mut space = cmsg_space!([RawFd; 2]);
@@ -386,8 +389,9 @@ pub fn receive_socket(stream_fd: RawFd) -> containerd_shim::Result<RawFd> {
         "copy_console: console socket get path: {}, fd: {}",
         path, &fds[0]
     );
-    tcgetattr(fds[0])?;
-    Ok(fds[0])
+    let fd = unsafe { OwnedFd::from_raw_fd(fds[0]) };
+    tcgetattr(&fd)?;
+    Ok(fd)
 }
 
 // TODO we still have to create pipes, otherwise the device maybe opened multiple times in container,

vmm/task/src/main.rs

@@ -140,7 +140,7 @@ lazy_static! {
     ]);
 }
 
-async fn initialize() -> anyhow::Result<()> {
+async fn start_task_server() -> anyhow::Result<()> {
     early_init_call().await?;
 
     let config = TaskConfig::new().await?;
@@ -172,26 +172,16 @@ async fn initialize() -> anyhow::Result<()> {
 
     late_init_call().await?;
 
+    start_ttrpc_server().await?.start().await?;
+
     Ok(())
 }
 
 #[tokio::main]
 async fn main() {
-    if let Err(e) = initialize().await {
-        error!("failed to do init call: {:?}", e);
-        exit(-1);
-    }
-
-    // Keep server alive in main function
-    let mut server = match create_ttrpc_server().await {
-        Ok(s) => s,
-        Err(e) => {
-            error!("failed to create ttrpc server: {:?}", e);
-            exit(-1);
-        }
-    };
-    if let Err(e) = server.start().await {
-        error!("failed to start ttrpc server: {:?}", e);
+    // start task server
+    if let Err(e) = start_task_server().await {
+        error!("failed to start task server: {:?}", e);
         exit(-1);
     }
 
@@ -362,9 +352,9 @@ async fn mount_static_mounts(mounts: Vec<StaticMount>) -> Result<()> {
     Ok(())
 }
 
-// create_ttrpc_server will create all the ttrpc service and register them to a server that
+// start_ttrpc_server will create all the ttrpc service and register them to a server that
 // bind to vsock 1024 port.
-async fn create_ttrpc_server() -> anyhow::Result<Server> {
+async fn start_ttrpc_server() -> anyhow::Result<Server> {
     let task = create_task_service().await?;
     let task_service = create_task(Arc::new(Box::new(task)));
 

vmm/task/src/netlink.rs

@@ -371,7 +371,7 @@ impl Handle {
                 }
 
                 if let Err(rtnetlink::Error::NetlinkError(message)) = request.execute().await {
-                    if Errno::from_i32(message.code.abs()) != Errno::EEXIST {
+                    if Errno::from_raw(message.code.abs()) != Errno::EEXIST {
                         return Err(other!(
                             "Failed to add IP v6 route (src: {}, dst: {}, gtw: {},Err: {})",
                             route.source,
@@ -418,7 +418,7 @@ impl Handle {
                 }
 
                 if let Err(rtnetlink::Error::NetlinkError(message)) = request.execute().await {
-                    if Errno::from_i32(message.code.abs()) != Errno::EEXIST {
+                    if Errno::from_raw(message.code.abs()) != Errno::EEXIST {
                         return Err(other!(
                             "Failed to add IP v4 route (src: {}, dst: {}, gtw: {},Err: {})",
                             route.source,