From 01593ad18181854ef6a5a3d5d96916855983c608 Mon Sep 17 00:00:00 2001 From: Rob Bradford Date: Wed, 10 Jun 2026 16:01:34 +0100 Subject: [PATCH] virtio-devices: vsock: Handle host half-close When the host half-closed the socket this was wrongly interpreted as a full shutdown preventing the guest from sending any more data. Instead propagate the half-close by setting just `VSOCK_FLAGS_SHUTDOWN_SEND`, leaving the connection alive so that guest-to-host writes are still forwarded. The connection is only torn down once the guest also shuts down its send side or a host write fails. Fixes: #8300 Assisted-by: Claude:claude-opus-4-8 Signed-off-by: Rob Bradford --- virtio-devices/src/vsock/csm/connection.rs | 298 +++++++++++++++++++-- virtio-devices/src/vsock/csm/mod.rs | 5 +- virtio-devices/src/vsock/unix/muxer.rs | 32 +++ 3 files changed, 313 insertions(+), 22 deletions(-) diff --git a/virtio-devices/src/vsock/csm/connection.rs b/virtio-devices/src/vsock/csm/connection.rs index e43e1affb..d03e298d0 100644 --- a/virtio-devices/src/vsock/csm/connection.rs +++ b/virtio-devices/src/vsock/csm/connection.rs @@ -140,6 +140,9 @@ pub struct VsockConnection VsockChannel for VsockConnection @@ -200,6 +203,10 @@ where // A data packet is only valid for established connections, and connections for // which our peer has initiated a graceful shutdown, but can still receive data. ConnState::Established | ConnState::PeerClosed(false, _) => (), + // A full host close (EPOLLHUP) after a host send-side half-close drives a read + // here; let it reach the full-close handler below to surface a clean shutdown + // rather than a forceful reset. + ConnState::LocalClosed(_, true) if self.host_hung_up => (), _ => { // Any other connection state is invalid at this point, and we need to kill it // with fire. @@ -209,8 +216,11 @@ where } // Oh wait, before we start bringing in the big data, can our peer handle receiving so - // much bytes goodness? - if self.need_credit_update_from_peer() { + // much bytes goodness? A hung-up host send side has no data left to deliver, so skip + // the credit dance (it could otherwise strand the teardown) and let the read return 0. + if self.need_credit_update_from_peer() + && !matches!(self.state, ConnState::LocalClosed(_, true)) + { self.last_fwd_cnt_to_peer = self.fwd_cnt; pkt.set_op(uapi::VSOCK_OP_CREDIT_REQUEST); return Ok(()); @@ -226,16 +236,39 @@ where match pkt.read_volatile_from(&mut self.stream, max_len) { Ok(read_cnt) => { if read_cnt == 0 { - // A 0-length read means the host stream was closed down. In that case, - // we'll ask our peer to shut down the connection. We can neither send nor - // receive any more data. - self.state = ConnState::LocalClosed; - self.expiry = Some( - Instant::now() + Duration::from_millis(defs::CONN_SHUTDOWN_TIMEOUT_MS), - ); - pkt.set_op(uapi::VSOCK_OP_SHUTDOWN) - .set_flag(uapi::VSOCK_FLAGS_SHUTDOWN_RCV) - .set_flag(uapi::VSOCK_FLAGS_SHUTDOWN_SEND); + if self.host_hung_up { + // A 0-length read with EPOLLHUP is a full host close: we can neither + // send nor receive any more data. + self.state = ConnState::LocalClosed(true, true); + self.expiry = Some( + Instant::now() + + Duration::from_millis(defs::CONN_SHUTDOWN_TIMEOUT_MS), + ); + pkt.set_op(uapi::VSOCK_OP_SHUTDOWN) + .set_flag(uapi::VSOCK_FLAGS_SHUTDOWN_RCV) + .set_flag(uapi::VSOCK_FLAGS_SHUTDOWN_SEND); + } else if matches!(self.state, ConnState::PeerClosed(_, true)) { + // Host send-side half-close, but the guest already closed its send + // side: nothing can flow either way. Terminate once the TX buffer + // drains, rather than reopening the connection for guest writes. + self.state = ConnState::PeerClosed(true, true); + if self.tx_buf.is_empty() { + pkt.set_op(uapi::VSOCK_OP_RST); + } else { + self.expiry = Some( + Instant::now() + + Duration::from_millis(defs::CONN_SHUTDOWN_TIMEOUT_MS), + ); + pkt.set_op(uapi::VSOCK_OP_SHUTDOWN) + .set_flag(uapi::VSOCK_FLAGS_SHUTDOWN_SEND); + } + } else { + // Signal only the host send-side shutdown and keep the connection + // alive for guest-to-host writes. + self.state = ConnState::LocalClosed(false, true); + pkt.set_op(uapi::VSOCK_OP_SHUTDOWN) + .set_flag(uapi::VSOCK_FLAGS_SHUTDOWN_SEND); + } } else { // On a successful data read, we fill in the packet with the RW op, and // length of the read data. @@ -299,7 +332,9 @@ where // Most frequent case: this is an established connection that needs to forward some // data to the host stream. Also works for a connection that has begun shutting // down, but the peer still has some data to send. - ConnState::Established | ConnState::PeerClosed(_, false) + ConnState::Established + | ConnState::PeerClosed(_, false) + | ConnState::LocalClosed(false, _) if pkt.op() == uapi::VSOCK_OP_RW => { if !pkt.has_buf() { @@ -372,6 +407,22 @@ where } } + // Host send side already closed; if the guest now closes its send side too, record the + // send-off (further RW packets are dropped) and terminate once the TX buffer drains. + ConnState::LocalClosed(_, _) if pkt.op() == uapi::VSOCK_OP_SHUTDOWN => { + let send_off = pkt.flags() & uapi::VSOCK_FLAGS_SHUTDOWN_SEND != 0; + if send_off { + self.state = ConnState::PeerClosed(true, true); + if self.tx_buf.is_empty() { + self.pending_rx.insert(PendingRx::Rst); + } else { + self.expiry = Some( + Instant::now() + Duration::from_millis(defs::CONN_SHUTDOWN_TIMEOUT_MS), + ); + } + } + } + // A credit update from our peer is valid only in a state which allows data // transfer towards the peer. ConnState::Established | ConnState::PeerInit | ConnState::PeerClosed(false, _) @@ -382,7 +433,10 @@ where // A credit request from our peer is valid only in a state which allows data // transfer from the peer. We'll respond with a credit update packet. - ConnState::Established | ConnState::PeerInit | ConnState::PeerClosed(_, false) + ConnState::Established + | ConnState::PeerInit + | ConnState::PeerClosed(_, false) + | ConnState::LocalClosed(false, _) if pkt.op() == uapi::VSOCK_OP_CREDIT_REQUEST => { self.pending_rx.insert(PendingRx::CreditUpdate); @@ -437,7 +491,16 @@ where // We're generally interested in being notified when data can be read from the host // stream, unless we're in a state which doesn't allow moving data from host to guest. match self.state { - ConnState::Killed | ConnState::LocalClosed | ConnState::PeerClosed(true, _) => (), + ConnState::Killed | ConnState::PeerClosed(true, _) => (), + // The host send side is closed but it can still receive guest-to-host data, so the + // connection stays alive. Watch for EPOLLHUP (but not EPOLLIN, which would spin on the + // read EOF) so that an eventual full host close is still observed while the guest is + // idle, rather than leaving the fd unwatched until the next guest write. + ConnState::LocalClosed(_, true) => { + if !self.host_hung_up { + evset.insert(epoll::Events::EPOLLHUP); + } + } _ if self.need_credit_update_from_peer() => (), _ => evset.insert(epoll::Events::EPOLLIN), } @@ -447,6 +510,11 @@ where /// Notify the connection about an event (or set of events) that it was interested in. /// fn notify(&mut self, evset: epoll::Events) { + if evset.contains(epoll::Events::EPOLLHUP) { + self.host_hung_up = true; + self.pending_rx.insert(PendingRx::Rw); + } + if evset.contains(epoll::Events::EPOLLIN) { // Data can be read from the host stream. Setting a Rw pending indication, so that // the muxer will know to call `recv_pkt()` later. @@ -527,6 +595,7 @@ where pending_rx: PendingRxSet::from(PendingRx::Response), expiry: None, host_write_shutdown: false, + host_hung_up: false, } } @@ -555,6 +624,7 @@ where pending_rx: PendingRxSet::from(PendingRx::Request), expiry: None, host_write_shutdown: false, + host_hung_up: false, } } @@ -956,6 +1026,12 @@ mod unit_tests { assert!(self.conn.has_pending_rx()); } + fn notify_epollin_hup(&mut self) { + self.conn + .notify(epoll::Events::EPOLLIN | epoll::Events::EPOLLHUP); + assert!(self.conn.has_pending_rx()); + } + fn notify_epollout(&mut self) { self.conn.notify(epoll::Events::EPOLLOUT); } @@ -1050,7 +1126,7 @@ mod unit_tests { } // A recv attempt in an invalid state should yield an instant reset packet. - ctx.conn.state = ConnState::LocalClosed; + ctx.conn.state = ConnState::LocalClosed(true, true); ctx.notify_epollin(); ctx.recv(); assert_eq!(ctx.pkt.op(), uapi::VSOCK_OP_RST); @@ -1062,11 +1138,11 @@ mod unit_tests { let mut stream = TestStream::new(); stream.read_state = StreamState::Closed; ctx.set_stream(stream); - ctx.notify_epollin(); + ctx.notify_epollin_hup(); ctx.recv(); - // When the host-side stream is closed, we can neither send not receive any more data. - // Therefore, the vsock shutdown packet that we'll deliver to the guest must contain both - // the no-more-send and the no-more-recv indications. + // When the host-side stream is fully closed (a read EOF accompanied by EPOLLHUP), we can + // neither send nor receive any more data. Therefore, the vsock shutdown packet that we'll + // deliver to the guest must contain both the no-more-send and the no-more-recv indications. assert_eq!(ctx.pkt.op(), uapi::VSOCK_OP_SHUTDOWN); assert_ne!(ctx.pkt.flags() & uapi::VSOCK_FLAGS_SHUTDOWN_SEND, 0); assert_ne!(ctx.pkt.flags() & uapi::VSOCK_FLAGS_SHUTDOWN_RCV, 0); @@ -1079,6 +1155,188 @@ mod unit_tests { ); } + #[test] + fn test_local_read_eof_sends_half_shutdown() { + let mut ctx = CsmTestContext::new_established(); + let mut stream = TestStream::new(); + stream.read_state = StreamState::Closed; + ctx.set_stream(stream); + + ctx.notify_epollin(); + ctx.recv(); + + assert_eq!(ctx.pkt.op(), uapi::VSOCK_OP_SHUTDOWN); + assert_ne!(ctx.pkt.flags() & uapi::VSOCK_FLAGS_SHUTDOWN_SEND, 0); + assert_eq!(ctx.pkt.flags() & uapi::VSOCK_FLAGS_SHUTDOWN_RCV, 0); + } + + #[test] + fn test_local_read_eof_polls_for_hup() { + // After a host send-side half-close the connection stays alive for guest-to-host writes, so + // it must remain registered for EPOLLHUP (but not EPOLLIN, which would spin on the read + // EOF) to observe an eventual full host close even while the guest is idle. + let mut ctx = CsmTestContext::new_established(); + let mut stream = TestStream::new(); + stream.read_state = StreamState::Closed; + ctx.set_stream(stream); + + ctx.notify_epollin(); + ctx.recv(); + assert_eq!(ctx.conn.state, ConnState::LocalClosed(false, true)); + + let evset = ctx.conn.get_polled_evset(); + assert!(evset.contains(epoll::Events::EPOLLHUP)); + assert!(!evset.contains(epoll::Events::EPOLLIN)); + } + + #[test] + fn test_local_read_eof_preserves_guest_to_host_writes() { + let mut ctx = CsmTestContext::new_established(); + let mut stream = TestStream::new(); + stream.read_state = StreamState::Closed; + ctx.set_stream(stream); + + ctx.notify_epollin(); + ctx.recv(); + assert_eq!(ctx.pkt.op(), uapi::VSOCK_OP_SHUTDOWN); + + let data = &[5, 6, 7, 8]; + ctx.init_data_pkt(data); + ctx.send(); + assert_eq!(ctx.conn.stream.write_buf, data.to_vec()); + } + + #[test] + fn test_local_half_close_then_hup_sends_full_shutdown() { + // After a host send-side half-close, an eventual full host close (EPOLLHUP) must surface a + // clean bidirectional shutdown with the kill timer, not a forceful reset. + let mut ctx = CsmTestContext::new_established(); + let mut stream = TestStream::new(); + stream.read_state = StreamState::Closed; + ctx.set_stream(stream); + + // Host send-side half-close: a read EOF without EPOLLHUP. + ctx.notify_epollin(); + ctx.recv(); + assert_eq!(ctx.conn.state, ConnState::LocalClosed(false, true)); + assert_eq!(ctx.pkt.op(), uapi::VSOCK_OP_SHUTDOWN); + + // The host then fully closes: EPOLLHUP fires while we are idle. + ctx.conn.notify(epoll::Events::EPOLLHUP); + assert!(ctx.conn.has_pending_rx()); + ctx.recv(); + assert_eq!(ctx.pkt.op(), uapi::VSOCK_OP_SHUTDOWN); + assert_ne!(ctx.pkt.flags() & uapi::VSOCK_FLAGS_SHUTDOWN_SEND, 0); + assert_ne!(ctx.pkt.flags() & uapi::VSOCK_FLAGS_SHUTDOWN_RCV, 0); + assert_eq!(ctx.conn.state, ConnState::LocalClosed(true, true)); + assert!(ctx.conn.will_expire()); + } + + #[test] + fn test_local_half_close_then_hup_terminates_with_zero_credit() { + // The full-close teardown after a host half-close must not get stuck behind flow control: + // even with no peer credit it surfaces the shutdown (and kill timer), never a stranding + // credit request. + let mut ctx = CsmTestContext::new_established(); + let mut stream = TestStream::new(); + stream.read_state = StreamState::Closed; + ctx.set_stream(stream); + + ctx.notify_epollin(); + ctx.recv(); + assert_eq!(ctx.conn.state, ConnState::LocalClosed(false, true)); + + // The guest has no receive buffer space left when the host fully closes. + ctx.set_peer_credit(0); + ctx.conn.notify(epoll::Events::EPOLLHUP); + ctx.recv(); + assert_eq!(ctx.pkt.op(), uapi::VSOCK_OP_SHUTDOWN); + assert_ne!(ctx.pkt.flags() & uapi::VSOCK_FLAGS_SHUTDOWN_SEND, 0); + assert_ne!(ctx.pkt.flags() & uapi::VSOCK_FLAGS_SHUTDOWN_RCV, 0); + assert!(ctx.conn.will_expire()); + } + + #[test] + fn test_local_read_eof_then_peer_send_shutdown_terminates() { + // After a host send-side half-close, a guest send-side shutdown should terminate the + // connection (RST). + let mut ctx = CsmTestContext::new_established(); + let mut stream = TestStream::new(); + stream.read_state = StreamState::Closed; + ctx.set_stream(stream); + + ctx.notify_epollin(); + ctx.recv(); + assert_eq!(ctx.pkt.op(), uapi::VSOCK_OP_SHUTDOWN); + assert_eq!(ctx.conn.state, ConnState::LocalClosed(false, true)); + + ctx.init_pkt(uapi::VSOCK_OP_SHUTDOWN, 0) + .set_flags(uapi::VSOCK_FLAGS_SHUTDOWN_SEND); + ctx.send(); + assert_eq!(ctx.conn.state, ConnState::PeerClosed(true, true)); + assert!(ctx.conn.has_pending_rx()); + + // The guest's send-off is now recorded, so a further RW packet is dropped, not forwarded. + let data = &[1, 2, 3, 4]; + ctx.init_data_pkt(data); + ctx.send(); + assert!(ctx.conn.stream.write_buf.is_empty()); + + ctx.recv(); + assert_eq!(ctx.pkt.op(), uapi::VSOCK_OP_RST); + } + + #[test] + fn test_local_closed_peer_send_shutdown_drains_then_terminates() { + // With guest-to-host data still buffered, a guest send-side shutdown after a host + // half-close must defer termination until the TX buffer drains, then queue an RST. + let mut ctx = CsmTestContext::new_established(); + let mut stream = TestStream::new(); + stream.write_state = StreamState::WouldBlock; + ctx.set_stream(stream); + let data = &[1, 2, 3, 4]; + ctx.init_data_pkt(data); + ctx.send(); + assert!(!ctx.conn.tx_buf.is_empty()); + + // The host has already half-closed its send side. + ctx.conn.state = ConnState::LocalClosed(false, true); + + // The guest closes its send side too; termination is deferred while data is buffered. + ctx.init_pkt(uapi::VSOCK_OP_SHUTDOWN, 0) + .set_flags(uapi::VSOCK_FLAGS_SHUTDOWN_SEND); + ctx.send(); + assert_eq!(ctx.conn.state, ConnState::PeerClosed(true, true)); + assert!(!ctx.conn.has_pending_rx()); + + // Once the buffer drains, an RST is queued. + ctx.set_stream(TestStream::new()); + ctx.notify_epollout(); + assert!(ctx.conn.tx_buf.is_empty()); + assert!(ctx.conn.has_pending_rx()); + ctx.recv(); + assert_eq!(ctx.pkt.op(), uapi::VSOCK_OP_RST); + } + + #[test] + fn test_peer_send_shutdown_then_local_read_eof_terminates() { + // The reverse order of the above: the guest closes its send side first, then the host + // closes its send side. No data can flow either way, so this must terminate (RST) rather + // than reopen the connection for guest-to-host writes. + let mut ctx = CsmTestContext::new_established(); + ctx.init_pkt(uapi::VSOCK_OP_SHUTDOWN, 0) + .set_flags(uapi::VSOCK_FLAGS_SHUTDOWN_SEND); + ctx.send(); + assert_eq!(ctx.conn.state, ConnState::PeerClosed(false, true)); + + let mut stream = TestStream::new(); + stream.read_state = StreamState::Closed; + ctx.set_stream(stream); + ctx.notify_epollin(); + ctx.recv(); + assert_eq!(ctx.pkt.op(), uapi::VSOCK_OP_RST); + } + #[test] fn test_peer_close() { // Test that send/recv shutdown indications are handled correctly. diff --git a/virtio-devices/src/vsock/csm/mod.rs b/virtio-devices/src/vsock/csm/mod.rs index 590ba0a7b..183d5d607 100644 --- a/virtio-devices/src/vsock/csm/mod.rs +++ b/virtio-devices/src/vsock/csm/mod.rs @@ -55,8 +55,9 @@ pub enum ConnState { PeerInit, /// The connection handshake has been performed successfully, and data can now be exchanged. Established, - /// The host (AF_UNIX) socket was closed. - LocalClosed, + /// The host (AF_UNIX) socket has been (partially) shut down. The tuple represents the host + /// R/W indication: (will_not_recv_anymore_data, will_not_send_anymore_data). + LocalClosed(bool, bool), /// A VSOCK_OP_SHUTDOWN packet was received from the guest. The tuple represents the guest R/W /// indication: (will_not_recv_anymore_data, will_not_send_anymore_data). PeerClosed(bool, bool), diff --git a/virtio-devices/src/vsock/unix/muxer.rs b/virtio-devices/src/vsock/unix/muxer.rs index a3626c8ea..37a8806ce 100644 --- a/virtio-devices/src/vsock/unix/muxer.rs +++ b/virtio-devices/src/vsock/unix/muxer.rs @@ -889,6 +889,7 @@ mod unit_tests { use std::cmp::min; use std::fs; use std::io::Write; + use std::net::Shutdown; use std::path::{Path, PathBuf}; use virtio_queue::QueueOwnedT; @@ -1278,6 +1279,37 @@ mod unit_tests { assert!(!ctx.muxer.local_port_set.contains(&local_port)); } + #[test] + fn test_local_send_half_close() { + let peer_port = 1025; + let mut ctx = MuxerTestContext::new("local_send_half_close"); + let (mut stream, local_port) = ctx.local_connect(peer_port); + + stream.shutdown(Shutdown::Write).unwrap(); + ctx.notify_muxer(); + assert!(ctx.muxer.has_pending_rx()); + ctx.recv(); + assert_eq!(ctx.pkt.op(), uapi::VSOCK_OP_SHUTDOWN); + assert_ne!(ctx.pkt.flags() & uapi::VSOCK_FLAGS_SHUTDOWN_SEND, 0); + assert_eq!(ctx.pkt.flags() & uapi::VSOCK_FLAGS_SHUTDOWN_RCV, 0); + assert_eq!(ctx.pkt.src_port(), local_port); + assert_eq!(ctx.pkt.dst_port(), peer_port); + + let data = [1, 2, 3, 4]; + ctx.init_data_pkt(local_port, peer_port, &data); + ctx.send(); + + let mut buf = vec![0u8; data.len()]; + stream.read_exact(buf.as_mut_slice()).unwrap(); + assert_eq!(buf.as_slice(), &data); + + let key = ConnMapKey { + local_port, + peer_port, + }; + assert!(ctx.muxer.conn_map.contains_key(&key)); + } + #[test] fn test_peer_close() { let peer_port = 1025;