vmm: detect dead connections during live migration

During a live migration, connections may get interrupted silently,
without being reset or closed. Interrupted idle connections may stay
alive indefinitely, for example if a connection dies during prefaulting
on the receiver side.

To detect dead idle connections, we enable
`SO_KEEPALIVE`. With `SO_KEEPALIVE`, the kernel will send keepalive
probes if a connection is idle and close the connection if the probes
remain unacknowledged.

To detect dead connections when actively sending data in a timely
manner, we enable `TCP_USER_TIMEOUT`, to reduce the timeout for closing
a connection where the peer doesn't acknowledge sent data.

On-behalf-of: SAP julian.schindel@sap.com
Signed-off-by: Julian Schindel <julian.schindel@cyberus-technology.de>

Cargo.lock

@@ -2310,6 +2310,16 @@ version = "1.15.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "8ed6a63f02c8539c91a8685a86f4099661ba3da017932f6ebbea6de3f0fa7c90"
 
+[[package]]
+name = "socket2"
+version = "0.6.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "c3d1e2c7f27f8d4cb10542a02c49005dbd6e93095799d6f3be745fae9f8fedd4"
+dependencies = [
+ "libc",
+ "windows-sys 0.61.2",
+]
+
 [[package]]
 name = "spin"
 version = "0.9.9"
@@ -2852,6 +2862,7 @@ dependencies = [
  "serial_buffer",
  "sha2",
  "signal-hook",
+ "socket2",
  "tempfile",
  "thiserror",
  "tracer",

vmm/Cargo.toml

@@ -84,6 +84,7 @@ serde_with = { workspace = true, features = ["macros"] }
 serial_buffer = { path = "../serial_buffer" }
 sha2 = { workspace = true }
 signal-hook = { workspace = true }
+socket2 = { version = "0.6.5", features = ["all"] }
 thiserror = { workspace = true }
 tracer = { path = "../tracer" }
 uuid = { workspace = true }

vmm/src/migration/transport.rs

@@ -20,6 +20,7 @@ use anyhow::{Context, anyhow};
 use log::{debug, error, info, warn};
 use seccompiler::{BpfProgram, SeccompAction, apply_filter};
 use serde_json;
+use socket2::{SockRef, TcpKeepalive};
 use thiserror::Error;
 use vm_memory::bitmap::BitmapSlice;
 use vm_memory::{
@@ -51,11 +52,18 @@ impl ReceiveListener {
     /// Block until a connection is accepted.
     pub(crate) fn accept(&mut self) -> Result<SocketStream, MigratableError> {
         match self {
-            ReceiveListener::Tcp(listener) => listener
-                .accept()
-                .map(|(socket, _)| SocketStream::Tcp(socket))
-                .context("Failed to accept TCP migration connection")
-                .map_err(MigratableError::MigrateReceive),
+            ReceiveListener::Tcp(listener) => {
+                let (socket, _) = listener
+                    .accept()
+                    .context("Failed to accept TCP migration connection")
+                    .map_err(MigratableError::MigrateReceive)?;
+
+                set_tcp_keepalive_and_user_timeout(&socket)
+                    .context("Failed to set socket options")
+                    .map_err(MigratableError::MigrateReceive)?;
+
+                Ok(SocketStream::Tcp(socket))
+            }
             ReceiveListener::Unix(listener) => listener
                 .accept()
                 .map(|(socket, _)| SocketStream::Unix(socket))
@@ -67,6 +75,10 @@ impl ReceiveListener {
                     .context("Failed to accept TCP connection")
                     .map_err(MigratableError::MigrateReceive)?;
 
+                set_tcp_keepalive_and_user_timeout(&socket)
+                    .context("Failed to set socket options")
+                    .map_err(MigratableError::MigrateReceive)?;
+
                 TlsStream::new_server(socket, config)
                     .map(Box::new)
                     .map(SocketStream::Tls)
@@ -1015,6 +1027,33 @@ pub fn tcp_address_to_server_name(address: &str) -> Result<&str, TcpAddressParse
     Ok(host)
 }
 
+/// Enables `SO_KEEPALIVE` and `TCP_USER_TIMEOUT` for the `tcp_stream`'s socket.
+///
+/// The set options target a failure detection time of two to three minutes.
+fn set_tcp_keepalive_and_user_timeout(tcp_stream: &TcpStream) -> io::Result<()> {
+    /// [`TcpKeepalive`] config for migration TCP sockets.
+    ///
+    /// After 60 seconds, the kernel starts sending keepalive probes.
+    /// Every 15 seconds another keepalive probe is sent.
+    /// If 4 probes are sent without response, the connection is dropped.
+    const MIGRATION_TCP_KEEPALIVE: TcpKeepalive = TcpKeepalive::new()
+        .with_retries(4)
+        .with_time(Duration::from_secs(60))
+        .with_interval(Duration::from_secs(15));
+
+    let socket_ref = SockRef::from(&tcp_stream);
+
+    // `TCP_USER_TIMEOUT` overrides `SO_KEEPALIVE` partially as per documentation:
+    // "[...] TCP_USER_TIMEOUT will override keepalive to determine when
+    // to close a connection due to keepalive failure."
+    // https://man7.org/linux/man-pages/man7/tcp.7.html
+    //TODO: Look into setting the keepalive via `std` once
+    // https://github.com/rust-lang/rust/issues/155889 is stabilized.
+    socket_ref.set_tcp_keepalive(&MIGRATION_TCP_KEEPALIVE)?;
+
+    socket_ref.set_tcp_user_timeout(Some(Duration::from_secs(120)))
+}
+
 /// Connect to a migration endpoint and return the established stream.
 pub(crate) fn send_migration_socket(
     destination_url: &str,
@@ -1027,6 +1066,10 @@ pub(crate) fn send_migration_socket(
             .context("Error connecting to TCP socket")
             .map_err(MigratableError::MigrateSend)?;
 
+        set_tcp_keepalive_and_user_timeout(&socket)
+            .context("Failed to set socket options")
+            .map_err(MigratableError::MigrateSend)?;
+
         if let Some(tls_dir) = tls_dir {
             // The address should have been validated by the API using this exact function.
             // Any error here has to be treated as a programming error.