vsock: bound the stdio port pool by concurrency, not VM lifetime

Sources/Containerization/CHStdioPortSlot.swift

@@ -0,0 +1,287 @@
+//===----------------------------------------------------------------------===//
+// Copyright © 2026 Apple Inc. and the Containerization project authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+//===----------------------------------------------------------------------===//
+
+#if os(Linux)
+import ContainerizationError
+import Foundation
+import Logging
+import Synchronization
+
+#if canImport(Musl)
+import Musl
+#elseif canImport(Glibc)
+import Glibc
+#endif
+
+/// A self-pipe used to break an accept loop out of `poll(2)` without closing
+/// the listening fd underneath it.
+///
+/// Closing an fd that another thread is parked in `poll(2)` on does not
+/// reliably wake that thread on Linux, and the fd number can be handed to an
+/// unrelated `open(2)` in the meantime — so the woken thread may go on to
+/// `accept(2)` somebody else's socket. Teardown therefore signals through a
+/// pipe and lets the loop close its own fds.
+final class WakePipe: Sendable {
+ private struct Ends {
+ let read: Int32
+ let write: Int32
+ }
+
+ private let ends: Mutex<Ends?>
+
+ init() throws {
+ var raw: [Int32] = [-1, -1]
+ let rc = raw.withUnsafeMutableBufferPointer { buf -> Int32 in
+ guard let base = buf.baseAddress else { return -1 }
+ return pipe(base)
+ }
+ guard rc == 0 else {
+ throw ContainerizationError(
+ .internalError,
+ message: "failed to create vsock accept-loop wake pipe (errno \(errno))"
+ )
+ }
+ self.ends = Mutex(Ends(read: raw[0], write: raw[1]))
+ }
+
+ /// The read end, for the loop's pollfd set. `nil` once closed.
+ var readFd: Int32? {
+ ends.withLock { $0?.read }
+ }
+
+ /// Ask the loop to exit. Safe to call repeatedly, and safe after
+ /// `closeEnds()`.
+ func signal() {
+ ends.withLock { state in
+ guard let state else { return }
+ var byte: UInt8 = 1
+ _ = write(state.write, &byte, 1)
+ }
+ }
+
+ /// Close both ends. Only the accept loop calls this, on its way out, so
+ /// `signal()` can never write into an fd number that has already been
+ /// reissued to something else.
+ func closeEnds() {
+ ends.withLock { state in
+ guard let current = state else { return }
+ _ = close(current.read)
+ _ = close(current.write)
+ state = nil
+ }
+ }
+}
+
+/// One guest→host vsock listening socket, owned by the VM for its entire
+/// lifetime and lent out to one `VsockListener` at a time.
+///
+/// The lifetime is the whole point. Cloud-hypervisor resolves a guest dial to
+/// port `P` against the host socket file `<base>_<P>`, freshly on every dial
+/// (`virtio-devices/src/vsock/unix/muxer.rs`), and on hosts where it cannot
+/// see files created after it forked — apple/container's `--virtualization`
+/// mode — that file must exist before the VMM starts. An AF_UNIX socket file
+/// whose last fd is closed is permanently dead: a later `connect(2)` gets
+/// ECONNREFUSED, and the inode cannot be revived by re-listening. Binding a
+/// replacement socket and renaming it over the path doesn't help either,
+/// because that is a new inode created after the fork, which is exactly what
+/// the VMM can't see.
+///
+/// So a slot that closed its fd when a process finished would be a slot that
+/// could never serve another process — which is what turned a pool sized for
+/// concurrent streams into a per-VM lifetime budget. Slots instead keep the
+/// fd and the path for as long as the VM lives, run a single accept loop, and
+/// hand accepted connections to whichever listener currently owns them.
+final class CHStdioPortSlot: Sendable {
+ let port: UInt32
+ let path: URL
+ let listenFd: Int32
+
+ private struct State {
+ /// The listener entitled to accepted connections right now.
+ var owner: VsockListener?
+ /// The running accept loop's wake pipe, or nil if no loop has been
+ /// started yet. Created lazily so an unused slot costs one fd (its
+ /// listening socket) rather than three.
+ var wake: WakePipe?
+ /// Set by `shutdown()`. Blocks further claims.
+ var closed: Bool
+ }
+
+ private enum ShutdownAction {
+ case alreadyClosed
+ case closeHere
+ case signal(WakePipe)
+ }
+
+ private let state: Mutex<State>
+
+ init(port: UInt32, path: URL, listenFd: Int32) {
+ self.port = port
+ self.path = path
+ self.listenFd = listenFd
+ self.state = Mutex(State(owner: nil, wake: nil, closed: false))
+ }
+
+ /// Lend the slot to `listener`.
+ ///
+ /// Throws if another listener still holds it. That case would otherwise
+ /// cross-wire two processes' stdio onto one port, so it is a hard error
+ /// rather than a wait — the port allocator is responsible for not handing
+ /// the same number to two live streams.
+ func claim(by listener: VsockListener) throws {
+ try state.withLock { state in
+ guard !state.closed else {
+ throw ContainerizationError(
+ .invalidState,
+ message: "vsock port \(port) is closed"
+ )
+ }
+ guard state.owner == nil else {
+ throw ContainerizationError(
+ .invalidState,
+ message: "vsock port \(port) is already being listened on"
+ )
+ }
+ state.owner = listener
+ }
+ }
+
+ /// Give up ownership without disturbing the socket. The accept loop keeps
+ /// running for the next tenant; connections that arrive in between are
+ /// closed by the loop.
+ func relinquish() {
+ state.withLock { $0.owner = nil }
+ }
+
+ /// Start the accept loop, if this is the slot's first tenant. The loop
+ /// then runs until `shutdown()`, so ownership handoff never has to stop
+ /// and restart it — which is what makes `relinquish()`/`claim(by:)` safe
+ /// back-to-back with no settling period.
+ func startAcceptingIfNeeded(logger: Logger?) throws {
+ let started = try state.withLock { state -> WakePipe? in
+ guard !state.closed, state.wake == nil else { return nil }
+ let wake = try WakePipe()
+ state.wake = wake
+ return wake
+ }
+ guard let started else { return }
+
+ // The accept loop blocks in poll(2)/accept(2), which is inappropriate
+ // for Swift's cooperative thread pool: a pool thread parked in a
+ // syscall can't service other tasks until it returns. With even a few
+ // of these, detached tasks queue behind the parked threads and never
+ // run, which shows up as the guest's dial never being seen by the
+ // host. libdispatch's global queue spawns OS threads on demand and is
+ // the right tool for a blocking syscall.
+ DispatchQueue.global(qos: .userInitiated).async { [self] in
+ self.acceptLoop(wake: started, logger: logger)
+ }
+ }
+
+ /// Stop the accept loop and release the socket. Idempotent, and safe to
+ /// call whether or not a loop was ever started.
+ func shutdown() {
+ let action = state.withLock { state -> ShutdownAction in
+ guard !state.closed else { return .alreadyClosed }
+ state.closed = true
+ state.owner = nil
+ // With a loop running, the loop owns the fds and closes them on
+ // its way out. With no loop, there is nobody else to do it.
+ if let wake = state.wake {
+ return .signal(wake)
+ }
+ return .closeHere
+ }
+ switch action {
+ case .alreadyClosed:
+ break
+ case .closeHere:
+ _ = close(listenFd)
+ case .signal(let wake):
+ wake.signal()
+ }
+ }
+
+ private func acceptLoop(wake: WakePipe, logger: Logger?) {
+ logger?.debug("vsock acceptLoop starting port=\(port) listenFd=\(listenFd)")
+ defer {
+ _ = close(listenFd)
+ wake.closeEnds()
+ logger?.debug("vsock acceptLoop exited port=\(port)")
+ }
+
+ while true {
+ guard let wakeFd = wake.readFd else { return }
+ var pfds = [
+ pollfd(fd: listenFd, events: Int16(POLLIN), revents: 0),
+ pollfd(fd: wakeFd, events: Int16(POLLIN), revents: 0),
+ ]
+ let rc = pfds.withUnsafeMutableBufferPointer { buf -> Int32 in
+ guard let base = buf.baseAddress else { return -1 }
+ return poll(base, 2, -1)
+ }
+ if rc < 0 {
+ let savedErrno = errno
+ if savedErrno == EINTR {
+ continue
+ }
+ logger?.error("vsock acceptLoop poll failed port=\(port) errno=\(savedErrno)")
+ return
+ }
+ // Shutdown wins over a pending connection: the VM is going away.
+ if pfds[1].revents != 0 {
+ return
+ }
+ guard pfds[0].revents & Int16(POLLIN) != 0 else {
+ // POLLERR / POLLNVAL on the listening socket — nothing to
+ // recover to.
+ if pfds[0].revents != 0 {
+ logger?.error("vsock acceptLoop listen socket error port=\(port) revents=\(pfds[0].revents)")
+ return
+ }
+ continue
+ }
+
+ let connFd = accept(listenFd, nil, nil)
+ if connFd < 0 {
+ let savedErrno = errno
+ if savedErrno == EINTR || savedErrno == EAGAIN || savedErrno == EWOULDBLOCK {
+ continue
+ }
+ logger?.error("vsock acceptLoop accept failed port=\(port) errno=\(savedErrno)")
+ return
+ }
+
+ let handle = FileHandle(fileDescriptor: connFd, closeOnDealloc: true)
+ guard let owner = state.withLock({ $0.owner }) else {
+ // A dial for a stream that already gave up the port — e.g. a
+ // guest that got to its connect(2) after the host stopped
+ // waiting for it. Dropping it here is what keeps it from
+ // being delivered to the port's next tenant.
+ logger?.warning("vsock dial on port \(port) with no listener; dropping")
+ try? handle.close()
+ continue
+ }
+ if case .terminated = owner.yield(handle) {
+ try? handle.close()
+ // That listener can never take another connection, so free
+ // the slot rather than wedging it, and keep accepting.
+ relinquish()
+ }
+ }
+ }
+}
+#endif