From af0850a785b98cfdfde9fe9808d1f6319b0e4966 Mon Sep 17 00:00:00 2001 From: michael_crosby Date: Thu, 27 Aug 2026 13:46:08 -0400 Subject: [PATCH] vsock: bound the stdio port pool by concurrency, not VM lifetime Signed-off-by: michael_crosby --- .../Containerization/CHStdioPortSlot.swift | 287 ++++++++++++++++++ .../CHVirtualMachineInstance.swift | 202 ++++++------ Sources/Containerization/LinuxContainer.swift | 37 ++- Sources/Containerization/LinuxPod.swift | 14 +- Sources/Containerization/LinuxProcess.swift | 179 +++++++++-- .../Containerization/VsockPortAllocator.swift | 90 ++++++ Sources/Integration/ContainerTests.swift | 65 ++++ Sources/Integration/Suite.swift | 1 + .../CHStdioPortSlotTests.swift | 199 ++++++++++++ .../LinuxProcessStdioTests.swift | 268 ++++++++++++++++ .../VsockPortAllocatorTests.swift | 113 +++++++ 11 files changed, 1307 insertions(+), 148 deletions(-) create mode 100644 Sources/Containerization/CHStdioPortSlot.swift create mode 100644 Sources/Containerization/VsockPortAllocator.swift create mode 100644 Tests/ContainerizationTests/CHStdioPortSlotTests.swift create mode 100644 Tests/ContainerizationTests/LinuxProcessStdioTests.swift create mode 100644 Tests/ContainerizationTests/VsockPortAllocatorTests.swift diff --git a/Sources/Containerization/CHStdioPortSlot.swift b/Sources/Containerization/CHStdioPortSlot.swift new file mode 100644 index 000000000..0a41d9c44 --- /dev/null +++ b/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 + + 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 `_

`, 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 + + 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 diff --git a/Sources/Containerization/CHVirtualMachineInstance.swift b/Sources/Containerization/CHVirtualMachineInstance.swift index 122f592a2..1e6d5513d 100644 --- a/Sources/Containerization/CHVirtualMachineInstance.swift +++ b/Sources/Containerization/CHVirtualMachineInstance.swift @@ -111,20 +111,32 @@ public final class CHVirtualMachineInstance: Sendable { /// snapshotted filesystem view at fork time, so files written under the /// per-VM workDir AFTER cloud-hypervisor starts are invisible to CH. /// We work around this by binding a fixed range of `vsock.sock_` - /// listener files BEFORE launching CH; `vm.listen(_:)` then consumes - /// pre-bound entries from this pool instead of binding on demand. + /// listener files BEFORE launching CH; `vm.listen(_:)` then lends out + /// pre-bound slots from this pool instead of binding on demand. + /// + /// The pool bounds *concurrent* streams, not the VM's lifetime total: a + /// slot outlives each of its tenants (see `CHStdioPortSlot`) and the host + /// port allocator reuses released numbers (see `VsockPortAllocator`). /// Range covers `LinuxContainer.hostVsockPorts` initial value /// (`0x10000000`) through the next `stdioPoolSize` sequential ports — - /// enough for `[stdin,stdout,stderr] x N` processes per VM. Bump - /// `stdioPoolSize` if you need more concurrent stdio streams than that. + /// enough for `[stdin,stdout,stderr] x N` concurrent processes per VM. + /// Sized for a pod's worth of containers plus concurrent `exec`s (probes, + /// `kubectl exec`) rather than for one container: at three ports per + /// process, 64 covers 21 concurrent processes. Idle slots cost one fd + /// each, so the headroom is cheap. + /// + /// A `listen(_:)` for a port outside the range still works — it binds on + /// demand, which is correct on any host where CH can see socket files + /// created after it forked, and warns because that is exactly what the + /// dev container cannot do. static let stdioPoolBase: UInt32 = 0x1000_0000 - static let stdioPoolSize: Int = 16 - private struct PreboundListener: Sendable { - let port: UInt32 - let listenFd: Int32 - let path: URL - } - private let _stdioPool: Mutex<[UInt32: PreboundListener]> + static let stdioPoolSize: Int = 64 + /// Pre-bound slots, keyed by port. Populated once in `start()` and held + /// until `stop()` — entries are lent out, never removed. + private let _stdioPool: Mutex<[UInt32: CHStdioPortSlot]> + /// Slots bound on demand for ports outside the pre-bound range. Removed + /// when their listener finishes. + private let _dynamicStdioSlots: Mutex<[CHStdioPortSlot]> public convenience init( group: (any EventLoopGroup)? = nil, @@ -220,6 +232,7 @@ public final class CHVirtualMachineInstance: Sendable { self.timeSyncer = .init(logger: logger) self._state = Mutex(.stopped) self._stdioPool = Mutex([:]) + self._dynamicStdioSlots = Mutex([]) } /// Mutate the mount registry. Forwards to the hotplug provider, which @@ -294,16 +307,9 @@ extension CHVirtualMachineInstance: VirtualMachineInstance { // by an in-flight hotplug. Empty if neither ran. await self.hotplug.shutdown() - // Close pre-bound stdio listener fds the start path opened in - // prebindStdioPool. Files unlink with workDir below. - let leftover = self._stdioPool.withLock { pool -> [PreboundListener] in - let entries = Array(pool.values) - pool.removeAll() - return entries - } - for entry in leftover { - _ = close(entry.listenFd) - } + // Stop the stdio accept loops and close their listening fds. The + // socket files unlink with workDir below. + self.shutdownStdioSlots() try? FileManager.default.removeItem(at: self.workDir) @@ -348,17 +354,10 @@ extension CHVirtualMachineInstance: VirtualMachineInstance { // before `group.shutdownGracefully()` below. try? await self.client.shutdown() - // Close any listening fds for stdio ports the test never - // consumed. The files themselves are removed when workDir is + // Stop every stdio accept loop and close its listening socket. + // The socket files themselves are removed when workDir is // unlinked below. - let leftover = self._stdioPool.withLock { pool -> [PreboundListener] in - let entries = Array(pool.values) - pool.removeAll() - return entries - } - for entry in leftover { - _ = close(entry.listenFd) - } + self.shutdownStdioSlots() if self.ownsGroup { try? await self.group.shutdownGracefully() @@ -406,67 +405,83 @@ extension CHVirtualMachineInstance: VirtualMachineInstance { } public func listen(_ port: UInt32) throws -> VsockListener { - // Consume from the pre-bound pool (see `_stdioPool` doc). - let prebound = _stdioPool.withLock { $0.removeValue(forKey: port) } - guard let prebound else { - throw ContainerizationError( - .invalidArgument, - message: "vsock port \(port) was not pre-bound; only ports " - + "\(Self.stdioPoolBase)..<\(Self.stdioPoolBase + UInt32(Self.stdioPoolSize)) " - + "are available for stdio. Increase CHVirtualMachineInstance.stdioPoolSize " - + "if you need more concurrent stdio streams per VM." - ) + // Borrow a pre-bound slot when the port is in range (see `_stdioPool`). + // Ports outside it bind on demand — correct wherever CH can see socket + // files created after it forked, which is everywhere except the + // `--virtualization` dev container. + if let slot = _stdioPool.withLock({ $0[port] }) { + logger?.debug("vsock listen claiming pool slot port=\(port) path=\(slot.path.path)") + let listener = VsockListener(port: port) { [slot, logger] _ in + logger?.debug("vsock listen releasing pool slot port=\(port)") + // Socket and accept loop both stay up for the next tenant. + slot.relinquish() + } + try slot.claim(by: listener) + do { + try slot.startAcceptingIfNeeded(logger: logger) + } catch { + slot.relinquish() + throw error + } + return listener } - let listenFd = prebound.listenFd - let path = prebound.path - logger?.debug("vsock listen consuming pool entry port=\(port) path=\(path.path)") - let listener = VsockListener(port: port) { [path, listenFd, logger] _ in - logger?.debug("vsock listen finishing port=\(port) closing listenFd=\(listenFd)") - _ = close(listenFd) - try? FileManager.default.removeItem(at: path) + + let poolEnd = Self.stdioPoolBase + UInt32(Self.stdioPoolSize) + logger?.warning( + """ + vsock port \(port) is outside the pre-bound range \(Self.stdioPoolBase)..<\(poolEnd); \ + binding on demand. Under apple/container --virtualization cloud-hypervisor cannot see \ + socket files created after it forked, so the guest's dial to this port will be reset — \ + raise CHVirtualMachineInstance.stdioPoolSize if this host needs more concurrent streams. + """ + ) + let base = workDir.appendingPathComponent("vsock.sock") + let path = chVsockListenSocketPath(baseSocket: base, port: port) + let listenFd = try chVsockBindListener(at: path) + let slot = CHStdioPortSlot(port: port, path: path, listenFd: listenFd) + // Nothing else can reclaim an on-demand slot, so it is torn down for + // real when its listener finishes. The socket file is deliberately + // left in place: unlinking it races a successor that has already + // rebound the same path, and workDir removal cleans it up anyway. + let listener = VsockListener(port: port) { [slot, logger, weak self] _ in + logger?.debug("vsock listen closing on-demand socket port=\(port)") + slot.shutdown() + self?._dynamicStdioSlots.withLock { $0.removeAll { $0 === slot } } } - let acceptLogger = logger - // The accept loop calls a blocking accept() syscall, which is - // inappropriate for Swift's cooperative thread pool: a pool thread - // pinned to accept() can't service other tasks until the syscall - // returns. With even a few leaked accept loops (e.g. when a test's - // setupIO times out and the listener is finished only when the - // 30s timer fires), Task.detached'd accept loops queue behind the - // pinned threads and never run, manifesting as the "vsock acceptLoop - // starting" log being silent and the dial-back never being seen by - // the host. Use libdispatch's global queue instead — it spawns - // OS threads on demand and is the right tool for blocking syscalls. - DispatchQueue.global(qos: .userInitiated).async { [listener, listenFd] in - acceptLogger?.debug("vsock acceptLoop starting port=\(listener.port) listenFd=\(listenFd)") - Self.acceptLoop(listenFd: listenFd, into: listener, logger: acceptLogger) - acceptLogger?.debug("vsock acceptLoop exited port=\(listener.port)") + do { + try slot.claim(by: listener) + try slot.startAcceptingIfNeeded(logger: logger) + } catch { + slot.shutdown() + throw error } + _dynamicStdioSlots.withLock { $0.append(slot) } return listener } /// Bind every port in `stdioPoolBase../vsock.sock_`. Must run before /// `chProcess.start()` so the files end up in CH's snapshot view of - /// the workDir. Files for ports never consumed are removed during - /// `stop()` along with the rest of `workDir`; the listening fds are - /// closed there too. + /// the workDir. The sockets stay bound for the life of the VM; `stop()` + /// shuts the slots down and removes `workDir` with the files in it. private func prebindStdioPool() throws { let base = workDir.appendingPathComponent("vsock.sock") - var pool: [UInt32: PreboundListener] = [:] + var pool: [UInt32: CHStdioPortSlot] = [:] pool.reserveCapacity(Self.stdioPoolSize) do { for offset in 0.. [CHStdioPortSlot] in + let slots = Array(pool.values) + pool.removeAll() + return slots + } + let dynamic = self._dynamicStdioSlots.withLock { slots -> [CHStdioPortSlot] in + let all = slots + slots.removeAll() + return all + } + for slot in pooled + dynamic { + slot.shutdown() } } } diff --git a/Sources/Containerization/LinuxContainer.swift b/Sources/Containerization/LinuxContainer.swift index 34964fcdc..6e59b82b8 100644 --- a/Sources/Containerization/LinuxContainer.swift +++ b/Sources/Containerization/LinuxContainer.swift @@ -146,8 +146,9 @@ public final class LinuxContainer: Container, Sendable { // Ports to be allocated from for stdio and for // unix socket relays that are sharing a guest - // uds to the host. - private let hostVsockPorts: Atomic + // uds to the host. Released ports are reused — see + // `VsockPortAllocator` for why that matters. + private let hostVsockPorts: VsockPortAllocator // Ports we request the guest to allocate for unix socket relays from // the host. private let guestVsockPorts: Atomic @@ -365,7 +366,7 @@ public final class LinuxContainer: Container, Sendable { } self.id = id self.vmm = vmm - self.hostVsockPorts = Atomic(0x1000_0000) + self.hostVsockPorts = VsockPortAllocator(base: 0x1000_0000) self.guestVsockPorts = Atomic(0x1000_0000) self.logger = logger self.config = configuration @@ -827,6 +828,7 @@ extension LinuxContainer { containerID: self.id, spec: spec, io: stdio, + portAllocator: self.hostVsockPorts, ociRuntimePath: self.config.ociRuntimePath, agent: agent, vm: createdState.vm, @@ -1011,6 +1013,7 @@ extension LinuxContainer { containerID: self.id, spec: spec, io: stdio, + portAllocator: self.hostVsockPorts, ociRuntimePath: self.config.ociRuntimePath, agent: agent, vm: startedState.vm, @@ -1048,6 +1051,7 @@ extension LinuxContainer { containerID: self.id, spec: spec, io: stdio, + portAllocator: self.hostVsockPorts, ociRuntimePath: self.config.ociRuntimePath, agent: agent, vm: state.vm, @@ -1157,7 +1161,9 @@ extension LinuxContainer { let port: UInt32 if socket.direction == .into { - port = self.hostVsockPorts.wrappingAdd(1, ordering: .relaxed).oldValue + // Held for the lifetime of the relay, so it is deliberately never + // released — the relay manager outlives this call. + port = self.hostVsockPorts.allocate() socket.destination = URL(filePath: Self.guestSocketStagingPath(socket.id)) } else { port = self.guestVsockPorts.wrappingAdd(1, ordering: .relaxed).oldValue @@ -1205,8 +1211,13 @@ extension LinuxContainer { ) } - let port = self.hostVsockPorts.wrappingAdd(1, ordering: .relaxed).oldValue + let port = self.hostVsockPorts.allocate() + // Deferred LIFO: hand the listener back before the port number, so + // a caller that reuses the number immediately finds the port free + // to listen on again. + defer { self.hostVsockPorts.release(port) } let listener = try state.vm.listen(port) + defer { try? listener.finish() } try await withThrowingTaskGroup(of: Void.self) { group in group.addTask { @@ -1349,8 +1360,11 @@ extension LinuxContainer { } let guestPath = URL(filePath: self.root).appending(path: source.path) - let port = self.hostVsockPorts.wrappingAdd(1, ordering: .relaxed).oldValue + let port = self.hostVsockPorts.allocate() + // Deferred LIFO: listener back first, then the port number. + defer { self.hostVsockPorts.release(port) } let listener = try state.vm.listen(port) + defer { try? listener.finish() } let (metadataStream, metadataCont) = AsyncStream.makeStream(of: Vminitd.CopyMetadata.self) @@ -1491,34 +1505,31 @@ func sortMountsByDestinationDepth(_ mounts: [ContainerizationOCI.Mount]) -> [Con struct IOUtil { static func setup( - portAllocator: borrowing Atomic, + portAllocator: VsockPortAllocator, stdin: ReaderStream?, stdout: Writer?, stderr: Writer? ) -> LinuxProcess.Stdio { var stdinSetup: LinuxProcess.StdioReaderSetup? = nil if let reader = stdin { - let ret = portAllocator.wrappingAdd(1, ordering: .relaxed) stdinSetup = .init( - port: ret.oldValue, + port: portAllocator.allocate(), reader: reader ) } var stdoutSetup: LinuxProcess.StdioSetup? = nil if let writer = stdout { - let ret = portAllocator.wrappingAdd(1, ordering: .relaxed) stdoutSetup = LinuxProcess.StdioSetup( - port: ret.oldValue, + port: portAllocator.allocate(), writer: writer ) } var stderrSetup: LinuxProcess.StdioSetup? = nil if let writer = stderr { - let ret = portAllocator.wrappingAdd(1, ordering: .relaxed) stderrSetup = LinuxProcess.StdioSetup( - port: ret.oldValue, + port: portAllocator.allocate(), writer: writer ) } diff --git a/Sources/Containerization/LinuxPod.swift b/Sources/Containerization/LinuxPod.swift index 6275a4d49..cb0b88ed4 100644 --- a/Sources/Containerization/LinuxPod.swift +++ b/Sources/Containerization/LinuxPod.swift @@ -185,8 +185,9 @@ public final class LinuxPod: Sendable { // Ports to be allocated from for stdio and for // unix socket relays that are sharing a guest - // uds to the host. - private let hostVsockPorts: Atomic + // uds to the host. Released ports are reused — see + // `VsockPortAllocator` for why that matters. + private let hostVsockPorts: VsockPortAllocator // Ports we request the guest to allocate for unix socket relays from // the host. private let guestVsockPorts: Atomic @@ -265,7 +266,7 @@ public final class LinuxPod: Sendable { } self.id = id self.vmm = vmm - self.hostVsockPorts = Atomic(0x1000_0000) + self.hostVsockPorts = VsockPortAllocator(base: 0x1000_0000) self.guestVsockPorts = Atomic(0x1000_0000) self.logger = logger @@ -714,6 +715,7 @@ extension LinuxPod { containerID: pauseID, spec: pauseSpec, io: LinuxProcess.Stdio(stdin: nil, stdout: nil, stderr: nil), + portAllocator: self.hostVsockPorts, ociRuntimePath: nil, agent: agent, vm: vm, @@ -962,6 +964,7 @@ extension LinuxPod { containerID: containerID, spec: spec, io: stdio, + portAllocator: self.hostVsockPorts, ociRuntimePath: nil, agent: agent, vm: createdState.vm, @@ -1209,6 +1212,7 @@ extension LinuxPod { containerID: containerID, spec: spec, io: stdio, + portAllocator: self.hostVsockPorts, ociRuntimePath: nil, agent: agent, vm: createdState.vm, @@ -1347,7 +1351,9 @@ extension LinuxPod { let port: UInt32 if socket.direction == .into { - port = self.hostVsockPorts.wrappingAdd(1, ordering: .relaxed).oldValue + // Held for the lifetime of the relay, so it is deliberately never + // released — the relay manager outlives this call. + port = self.hostVsockPorts.allocate() socket.destination = URL(filePath: Self.guestSocketStagingPath(socket.id)) } else { port = self.guestVsockPorts.wrappingAdd(1, ordering: .relaxed).oldValue diff --git a/Sources/Containerization/LinuxProcess.swift b/Sources/Containerization/LinuxProcess.swift index a77cbcfc1..21ceac74a 100644 --- a/Sources/Containerization/LinuxProcess.swift +++ b/Sources/Containerization/LinuxProcess.swift @@ -100,21 +100,35 @@ public final class LinuxProcess: Sendable { private let logger: Logger? private let onDelete: (@Sendable () async -> Void)? + /// The allocator that `io`'s ports came from. This process gives them + /// back when it is deleted; see `VsockPortAllocator` for why they can't + /// simply be abandoned. + private let portAllocator: VsockPortAllocator + + /// How long to wait for the guest to dial back for each stdio stream. + /// Defaults to the process-wide `stdioDialBackTimeout`; injectable so + /// tests don't have to wait it out. + private let stdioTimeout: UInt32 + init( _ id: String, containerID: String? = nil, spec: Spec, io: Stdio, + portAllocator: VsockPortAllocator, ociRuntimePath: String?, agent: any VirtualMachineAgent, vm: any VirtualMachineInstance, logger: Logger?, + stdioTimeoutSeconds: UInt32 = LinuxProcess.stdioDialBackTimeout, onDelete: (@Sendable () async -> Void)? = nil ) { self.id = id self.owningContainer = containerID self.state = Mutex(.init(spec: spec, pid: -1, stdio: StdioHandles())) self.ioSetup = io + self.portAllocator = portAllocator + self.stdioTimeout = stdioTimeoutSeconds self.agent = agent self.ociRuntimePath = ociRuntimePath self.vm = vm @@ -124,28 +138,81 @@ public final class LinuxProcess: Sendable { } extension LinuxProcess { - func setupIO(listeners: [VsockListener?]) async throws -> [FileHandle?] { - let ioTimeout: UInt32 = 30 + /// Seconds to wait for the guest to dial back on each stdio vsock port. + /// The guest only connects once `createProcess` has reached it, so this + /// window has to cover a loaded guest getting from that RPC to its + /// `connect(2)` — on a busy host running many VMs that is far more than + /// the few hundred milliseconds it takes when idle. + /// + /// Env: `CONTAINERIZATION_STDIO_TIMEOUT` (seconds, default 30). Values + /// that aren't a positive integer are ignored. + /// + /// The asymmetry is what sets the default high: too long only delays + /// reporting a guest that was never coming back, while too short + /// actively corrupts the failure. Expiring tears down the host + /// listeners, so the guest's own connect is then answered with a reset + /// and the error surfaced to the caller names a vsock connection + /// problem instead of a slow guest. + static let stdioDialBackTimeout: UInt32 = { + guard let raw = ProcessInfo.processInfo.environment["CONTAINERIZATION_STDIO_TIMEOUT"], + let seconds = UInt32(raw), + seconds > 0 + else { + return 30 + } + return seconds + }() - let handles = try await Timeout.run(seconds: ioTimeout) { - try await withThrowingTaskGroup(of: (Int, FileHandle?).self) { group in - var results = [FileHandle?](repeating: nil, count: 3) + /// Stream names by `setupIO` listener index, for error messages. + private static let stdioNames = ["stdin", "stdout", "stderr"] - for (index, listener) in listeners.enumerated() { - guard let listener else { continue } + /// What each sibling task in `start()` reports back. + private enum StartStep: Sendable { + case stdio([FileHandle?]) + case processCreated + } - group.addTask { - let first = await listener.first(where: { _ in true }) - try listener.finish() - return (index, first) + func setupIO(listeners: [VsockListener?]) async throws -> [FileHandle?] { + let timeout = self.stdioTimeout + let handles = try await withThrowingTaskGroup(of: (Int, FileHandle?).self) { group in + var results = [FileHandle?](repeating: nil, count: 3) + + for (index, listener) in listeners.enumerated() { + guard let listener else { continue } + let name = Self.stdioNames.indices.contains(index) ? Self.stdioNames[index] : "stdio[\(index)]" + + group.addTask { + let first: FileHandle? + // Timed per stream rather than around the whole group so + // the error can name the stream the guest never dialed. + do { + first = try await Timeout.run(seconds: timeout) { + await listener.first(where: { _ in true }) + } + } catch { + try? listener.finish() + // A cancelled sibling — or a cancelled caller — lands + // here too. That error isn't ours to relabel, and the + // group reports whichever failure came first anyway. + if Task.isCancelled { + throw error + } + throw ContainerizationError( + .timeout, + message: "guest did not dial back for \(name) (vsock port \(listener.port)) " + + "within \(timeout)s", + cause: error + ) } + try listener.finish() + return (index, first) } + } - for try await (index, fileHandle) in group { - results[index] = fileHandle - } - return results + for try await (index, fileHandle) in group { + results[index] = fileHandle } + return results } // Note: stdin relay is started separately via startStdinRelay() after @@ -239,14 +306,14 @@ extension LinuxProcess { /// Start the process. public func start() async throws { + var pending = [VsockListener?](repeating: nil, count: 3) do { let spec = self.state.withLock { $0.spec } - var listeners = [VsockListener?](repeating: nil, count: 3) if let stdin = self.ioSetup.stdin { - listeners[0] = try self.vm.listen(stdin.port) + pending[0] = try self.vm.listen(stdin.port) } if let stdout = self.ioSetup.stdout { - listeners[1] = try self.vm.listen(stdout.port) + pending[1] = try self.vm.listen(stdout.port) } if let stderr = self.ioSetup.stderr { if spec.process!.terminal { @@ -255,25 +322,48 @@ extension LinuxProcess { message: "stderr should not be configured with terminal=true" ) } - listeners[2] = try self.vm.listen(stderr.port) + pending[2] = try self.vm.listen(stderr.port) } + let listeners = pending + + // setupIO and createProcess must run concurrently: the guest only + // dials back for stdio once createProcess has reached it. Run them + // as siblings so whichever fails *first* is the error we report. + // That ordering is the point. A stdio dial-back timeout tears down + // the host listeners, after which the guest's own connect is + // answered with a reset — so if createProcess's error won the race + // the caller was told "connection reset by peer" for a port that + // the host itself had stopped listening on moments earlier, which + // sends every investigation to the wrong layer. + let result = try await withThrowingTaskGroup(of: StartStep.self, returning: [FileHandle?].self) { group in + group.addTask { + let handles = try await self.setupIO(listeners: listeners) + return .stdio(handles) + } - let t = Task { - try await self.setupIO(listeners: listeners) - } + group.addTask { + try await self.agent.createProcess( + id: self.id, + containerID: self.owningContainer, + stdinPort: self.ioSetup.stdin?.port, + stdoutPort: self.ioSetup.stdout?.port, + stderrPort: self.ioSetup.stderr?.port, + ociRuntimePath: self.ociRuntimePath, + configuration: spec, + options: nil + ) + return .processCreated + } - try await agent.createProcess( - id: self.id, - containerID: self.owningContainer, - stdinPort: self.ioSetup.stdin?.port, - stdoutPort: self.ioSetup.stdout?.port, - stderrPort: self.ioSetup.stderr?.port, - ociRuntimePath: self.ociRuntimePath, - configuration: spec, - options: nil - ) + var handles = [FileHandle?](repeating: nil, count: 3) + for try await step in group { + if case .stdio(let stdio) = step { + handles = stdio + } + } + return handles + } - let result = try await t.value let pid = try await self.agent.startProcess( id: self.id, containerID: self.owningContainer @@ -294,6 +384,13 @@ extension LinuxProcess { $0.pid = pid } } catch { + // Release any listener this failure path left open — e.g. a later + // vm.listen(_:) throwing after earlier ports were already claimed. + // finish() is idempotent, so streams setupIO already finished are + // unaffected. + for listener in pending { + try? listener?.finish() + } if let err = error as? ContainerizationError { throw err } @@ -434,6 +531,13 @@ extension LinuxProcess { } private func performDeletion() async throws { + // Runs after the paths below have closed the host stdio handles, and + // only once (performDeletion is guarded by state.deletionTask). Ports + // deliberately come back at delete rather than at process exit: a + // straggling guest dial for a finished stream must not be handed to + // whichever process reuses the number next. + defer { self.releaseStdioPorts() } + do { try await self.agent.deleteProcess( id: self.id, @@ -476,4 +580,13 @@ extension LinuxProcess { ) } } + + /// Return this process's stdio vsock ports to the allocator. `release` is + /// idempotent, so overlapping teardown paths don't need to coordinate. + private func releaseStdioPorts() { + for port in [self.ioSetup.stdin?.port, self.ioSetup.stdout?.port, self.ioSetup.stderr?.port] { + guard let port else { continue } + self.portAllocator.release(port) + } + } } diff --git a/Sources/Containerization/VsockPortAllocator.swift b/Sources/Containerization/VsockPortAllocator.swift new file mode 100644 index 000000000..e02308ee0 --- /dev/null +++ b/Sources/Containerization/VsockPortAllocator.swift @@ -0,0 +1,90 @@ +//===----------------------------------------------------------------------===// +// 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. +//===----------------------------------------------------------------------===// + +import Synchronization + +/// Hands out host-side vsock port numbers for guest→host streams, and takes +/// them back when the stream is gone. +/// +/// Taking them back is the whole reason this type exists instead of a bare +/// counter. Cloud-hypervisor's hybrid vsock resolves a guest dial to port +/// `P` against the host socket file `_

`, freshly per dial, so a +/// port number is only usable if a socket file for exactly that number +/// exists. On hosts where cloud-hypervisor cannot see files created after it +/// forked — apple/container's `--virtualization` mode, see +/// `CHVirtualMachineInstance.stdioPoolSize` — those files have to be bound +/// before the VMM starts, which makes the set of usable numbers finite. +/// +/// A monotonically increasing port number turns that finite set into a +/// *lifetime* budget for the VM rather than a concurrency one: at three +/// ports per process, a pool of 16 is spent after five processes even though +/// none of their streams are still open, and every process started after +/// that fails — an `exec` liveness probe every 10s bricks the sandbox in +/// well under a minute. Reusing released numbers makes the pre-bound set +/// bound *concurrent* streams, which is what it was sized for. +/// +/// Ports come back when the owning process is deleted rather than when it +/// exits, so a late dial for a finished stream cannot be delivered to a +/// newer process that reused the number. +package final class VsockPortAllocator: Sendable { + private struct State { + /// Lowest port never yet handed out. + var next: UInt32 + /// Ports handed out and given back, available for reuse. + var free: Set + } + + private let base: UInt32 + private let state: Mutex + + package init(base: UInt32) { + self.base = base + self.state = Mutex(State(next: base, free: [])) + } + + /// Take a port. Prefers the lowest released port, which keeps + /// allocations packed at the bottom of the range so they stay inside a + /// pre-bound window for as long as that window covers the concurrent + /// stream count. + package func allocate() -> UInt32 { + state.withLock { state in + if let reused = state.free.min() { + state.free.remove(reused) + return reused + } + let port = state.next + state.next = state.next &+ 1 + return port + } + } + + /// Give a port back. Ports this allocator never handed out are ignored, + /// and releasing the same port twice is a no-op, so callers on + /// overlapping teardown paths don't have to coordinate. + package func release(_ port: UInt32) { + state.withLock { state in + guard port >= self.base, port < state.next else { + return + } + state.free.insert(port) + } + } + + /// Ports currently handed out. Test seam. + package var outstandingCount: Int { + state.withLock { Int($0.next - self.base) - $0.free.count } + } +} diff --git a/Sources/Integration/ContainerTests.swift b/Sources/Integration/ContainerTests.swift index 9c8edd880..3d5cda4ba 100644 --- a/Sources/Integration/ContainerTests.swift +++ b/Sources/Integration/ContainerTests.swift @@ -1479,6 +1479,71 @@ extension IntegrationSuite { } } + /// Sequential `exec`s with stdio must not run a VM out of vsock ports. + /// + /// The cloud-hypervisor backend pre-binds a fixed pool of host stdio + /// sockets (it has to: the VMM can't see socket files created after it + /// forked). While host port numbers were handed out by a fetch-add that + /// never reused one, and a finished stream destroyed its pool entry, that + /// pool was a *lifetime* budget for the VM rather than a concurrency one — + /// a one-container pod accepted exactly four sequential execs and the + /// fifth failed with "vsock port … was not pre-bound". An `exec` liveness + /// probe every 10s therefore bricked a pod in well under a minute. + /// + /// 40 execs at two stdio ports each is 80 allocations, comfortably past + /// the pre-bound pool, so this only passes if ports and pool entries are + /// both recycled. Each exec also checks its *own* output, which is what + /// catches the failure mode recycling introduces: a straggling dial for a + /// finished stream getting delivered to whichever process reused the port. + func testSequentialExecsReuseStdioPorts() async throws { + let id = "test-sequential-execs-reuse-stdio-ports" + + let bs = try await bootstrap(id) + let container = try LinuxContainer(id, rootfs: bs.rootfs, vmm: bs.vmm) { config in + config.process.arguments = ["/bin/sleep", "1000"] + config.bootLog = bs.bootLog + } + + do { + try await container.create() + try await container.start() + + for index in 0..<40 { + let expected = "exec-\(index)" + let stdout = BufferWriter() + let stderr = BufferWriter() + let exec = try await container.exec("seq-\(index)") { config in + config.arguments = ["/bin/echo", expected] + config.stdout = stdout + config.stderr = stderr + } + try await exec.start() + let status = try await exec.wait() + try await exec.delete() + + guard status.exitCode == 0 else { + throw IntegrationError.assert(msg: "exec \(index) status \(status) != 0") + } + let got = String(data: stdout.data, encoding: .utf8) ?? "" + guard got == "\(expected)\n" else { + throw IntegrationError.assert( + msg: "exec \(index) stdout '\(got)' != '\(expected)\\n' — a recycled port may be cross-wired") + } + let err = String(data: stderr.data, encoding: .utf8) ?? "" + guard err.isEmpty else { + throw IntegrationError.assert(msg: "exec \(index) stderr should be empty, got '\(err)'") + } + } + + try await container.kill(.kill) + try await container.wait() + try await container.stop() + } catch { + try? await container.stop() + throw error + } + } + func testNonExistentBinary() async throws { let id = "test-non-existent-binary" diff --git a/Sources/Integration/Suite.swift b/Sources/Integration/Suite.swift index bf11e1cef..6002a8f53 100644 --- a/Sources/Integration/Suite.swift +++ b/Sources/Integration/Suite.swift @@ -447,6 +447,7 @@ struct IntegrationSuite: AsyncParsableCommand { Test("container bootlog using filehandle", testBootLogFileHandle), Test("process delete idempotency", testProcessDeleteIdempotency), Test("multiple execs without delete", testMultipleExecsWithoutDelete), + Test("sequential execs reuse stdio ports", testSequentialExecsReuseStdioPorts), // Capabilities Test("container capabilities sys admin", testCapabilitiesSysAdmin), diff --git a/Tests/ContainerizationTests/CHStdioPortSlotTests.swift b/Tests/ContainerizationTests/CHStdioPortSlotTests.swift new file mode 100644 index 000000000..e3cda051a --- /dev/null +++ b/Tests/ContainerizationTests/CHStdioPortSlotTests.swift @@ -0,0 +1,199 @@ +//===----------------------------------------------------------------------===// +// 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 ContainerizationExtras +import ContainerizationOS +import Foundation +import Testing + +@testable import Containerization + +struct CHStdioPortSlotTests { + /// The regression this whole design exists for. + /// + /// The old `listen(_:)` closed the listening fd and unlinked the socket + /// file when a stream finished, which permanently killed the port: an + /// AF_UNIX socket whose last fd is gone answers `connect(2)` with + /// ECONNREFUSED, and the file cannot be revived in a way cloud-hypervisor + /// can see. A slot must instead hand off between tenants with the socket + /// left intact, or a pre-bound pool is a per-VM lifetime budget. + @Test func slotServesSuccessiveTenantsOnTheSameSocket() async throws { + let harness = try SlotHarness() + defer { harness.cleanup() } + + // First tenant: dial, get accepted, finish. + let first = harness.makeListener() + try harness.slot.claim(by: first) + try harness.slot.startAcceptingIfNeeded(logger: nil) + + let firstClient = try harness.dial() + let firstAccepted = try await harness.accept(on: first) + #expect(firstAccepted != nil) + try? firstClient.close() + try first.finish() + + // The socket file must still be there — unlinking it is what turned a + // "host stopped listening" into the guest's "connection reset". + #expect(FileManager.default.fileExists(atPath: harness.path.path)) + + // Second tenant: same slot, same socket, no rebinding. + let second = harness.makeListener() + try harness.slot.claim(by: second) + try harness.slot.startAcceptingIfNeeded(logger: nil) + + let secondClient = try harness.dial() + let secondAccepted = try await harness.accept(on: second) + #expect(secondAccepted != nil) + try? secondClient.close() + try second.finish() + } + + /// The cumulative half, without needing a VM: one slot must serve far more + /// tenants than a pool has ports. The old code gave each tenant its own + /// socket and destroyed it on finish, so `stdioPoolSize / 3` processes was + /// the hard ceiling for a VM's entire life. + @Test func slotServesFarMoreTenantsThanAPoolHasPorts() async throws { + let harness = try SlotHarness() + defer { harness.cleanup() } + + for round in 0..<25 { + let tenant = harness.makeListener() + try harness.slot.claim(by: tenant) + try harness.slot.startAcceptingIfNeeded(logger: nil) + + let client = try harness.dial() + let accepted = try await harness.accept(on: tenant) + #expect(accepted != nil, "tenant \(round) never got its connection") + try? client.close() + try tenant.finish() + } + + #expect(FileManager.default.fileExists(atPath: harness.path.path)) + } + + /// A dial that lands while the port has no tenant — the guest reaching its + /// `connect(2)` after the host gave up waiting — must be dropped rather + /// than delivered to the port's next tenant, and must not kill the accept + /// loop. + @Test func ownerlessDialIsDroppedAndTheLoopKeepsServing() async throws { + let harness = try SlotHarness() + defer { harness.cleanup() } + + let abandoned = harness.makeListener() + try harness.slot.claim(by: abandoned) + try harness.slot.startAcceptingIfNeeded(logger: nil) + try abandoned.finish() + + // Late dial with nobody home. + let straggler = try harness.dial() + try? await Task.sleep(for: .milliseconds(200)) + try? straggler.close() + + // The next tenant still gets its own connection. + let next = harness.makeListener() + try harness.slot.claim(by: next) + let client = try harness.dial() + let accepted = try await harness.accept(on: next) + #expect(accepted != nil) + try? client.close() + try next.finish() + } + + /// Two live listeners on one port would cross-wire two processes' stdio, + /// so the second claim is a hard error rather than a queue. + @Test func claimingABusySlotFails() throws { + let harness = try SlotHarness() + defer { harness.cleanup() } + + let held = harness.makeListener() + try harness.slot.claim(by: held) + #expect(throws: ContainerizationError.self) { + try harness.slot.claim(by: harness.makeListener()) + } + + // Once released, the slot is claimable again. + try held.finish() + try harness.slot.claim(by: harness.makeListener()) + } + + @Test func shutdownRefusesFurtherClaims() throws { + let harness = try SlotHarness() + defer { harness.cleanup() } + + harness.slot.shutdown() + // Idempotent. + harness.slot.shutdown() + #expect(throws: ContainerizationError.self) { + try harness.slot.claim(by: harness.makeListener()) + } + } +} + +/// A slot bound on a real UDS in a temp directory, plus the client side. +private struct SlotHarness { + static let port: UInt32 = 42 + + let directory: URL + let path: URL + let slot: CHStdioPortSlot + + init() throws { + self.directory = FileManager.default.temporaryDirectory + .appendingPathComponent("ch-slot-\(UUID().uuidString)") + try FileManager.default.createDirectory(at: directory, withIntermediateDirectories: true) + self.path = chVsockListenSocketPath( + baseSocket: directory.appendingPathComponent("vsock.sock"), + port: Self.port + ) + let fd = try chVsockBindListener(at: path) + self.slot = CHStdioPortSlot(port: Self.port, path: path, listenFd: fd) + } + + func makeListener() -> VsockListener { + let slot = self.slot + return VsockListener(port: Self.port) { _ in slot.relinquish() } + } + + func dial() throws -> Socket { + let unix = try UnixType(path: path.path) + let socket = try Socket(type: unix, closeOnDeinit: false) + do { + try socket.connect() + } catch { + try? socket.close() + throw error + } + return socket + } + + /// Wait for the accept loop to hand a connection to `listener`. The bound + /// only exists so a broken loop fails the test instead of hanging it, so + /// it is generous — the whole suite runs in parallel and scheduling delay + /// alone can be seconds. + func accept(on listener: VsockListener) async throws -> FileHandle? { + try await Timeout.run(seconds: 60) { + await listener.first(where: { _ in true }) + } + } + + func cleanup() { + slot.shutdown() + try? FileManager.default.removeItem(at: directory) + } +} +#endif diff --git a/Tests/ContainerizationTests/LinuxProcessStdioTests.swift b/Tests/ContainerizationTests/LinuxProcessStdioTests.swift new file mode 100644 index 000000000..9af091c15 --- /dev/null +++ b/Tests/ContainerizationTests/LinuxProcessStdioTests.swift @@ -0,0 +1,268 @@ +//===----------------------------------------------------------------------===// +// 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. +//===----------------------------------------------------------------------===// + +import ContainerizationError +import ContainerizationExtras +import ContainerizationOCI +import Foundation +import Synchronization +import Testing + +@testable import Containerization + +#if canImport(Darwin) +import Darwin +#elseif canImport(Glibc) +import Glibc +#endif + +struct LinuxProcessStdioTests { + private static let stdoutPort: UInt32 = 0x1000_0000 + + /// The guest is late, so the host's dial-back window expires first. The + /// error the caller sees must be the host's own timeout, naming the stream + /// and port — not the guest's `ECONNRESET`, which is only a consequence of + /// the host having stopped listening. Reporting the guest's symptom sent a + /// downstream consumer hunting a phantom port-allocator bug for weeks. + @Test func reportsTheHostTimeoutRatherThanTheGuestsReset() async throws { + let allocator = VsockPortAllocator(base: Self.stdoutPort) + let vm = StubVirtualMachineInstance() + let agent = StubVirtualMachineAgent(onCreateProcess: { + // The guest gets to its connect(2) after the host gave up, and + // cloud-hypervisor answers it with a reset. This used to be the + // only thing the caller was ever told. The delay is far longer + // than the 1s dial-back window below so the ordering can't invert + // under a loaded CI machine's scheduling; it is cancelled as soon + // as the timeout wins, so it costs nothing. + try await Task.sleep(for: .seconds(60)) + throw ContainerizationError( + .internalError, + message: "createProcess: socket: error could not connect to socket 2:268435456 (Connection reset by peer)" + ) + }) + + let stdio = IOUtil.setup(portAllocator: allocator, stdin: nil, stdout: DiscardWriter(), stderr: nil) + let process = LinuxProcess( + "late-guest", + containerID: "c", + spec: Self.spec(), + io: stdio, + portAllocator: allocator, + ociRuntimePath: nil, + agent: agent, + vm: vm, + logger: nil, + stdioTimeoutSeconds: 1 + ) + + do { + try await process.start() + Issue.record("start() should have failed") + } catch let error as ContainerizationError { + #expect(error.isCode(.timeout)) + #expect(error.message.contains("guest did not dial back for stdout")) + #expect(error.message.contains("vsock port \(Self.stdoutPort)")) + #expect(error.message.contains("within 1s")) + #expect(!"\(error)".contains("Connection reset by peer")) + } + } + + /// A `createProcess` failure that has nothing to do with stdio must still + /// be reported as itself, and promptly — not swallowed by, or made to wait + /// out, the dial-back window. + @Test func reportsCreateProcessFailureWhenItComesFirst() async throws { + let allocator = VsockPortAllocator(base: Self.stdoutPort) + let vm = StubVirtualMachineInstance() + let agent = StubVirtualMachineAgent(onCreateProcess: { + throw ContainerizationError(.invalidArgument, message: "no such executable") + }) + + let stdio = IOUtil.setup(portAllocator: allocator, stdin: nil, stdout: DiscardWriter(), stderr: nil) + let process = LinuxProcess( + "bad-spec", + containerID: "c", + spec: Self.spec(), + io: stdio, + portAllocator: allocator, + ociRuntimePath: nil, + agent: agent, + vm: vm, + logger: nil, + stdioTimeoutSeconds: 300 + ) + + let clock = ContinuousClock() + let started = clock.now + do { + try await process.start() + Issue.record("start() should have failed") + } catch let error as ContainerizationError { + #expect(error.message.contains("no such executable")) + } + // Must not have waited out the dial-back window: a correct start() + // cancels the stdio wait as soon as createProcess fails. The bound is + // deliberately far from both ends — a whole parallel test suite's + // scheduling delay is seconds, and the regression would be 300 — so + // this measures cancellation, not the CI machine's load. + #expect(clock.now - started < .seconds(60)) + } + + /// The happy path, plus the port accounting: a deleted process gives its + /// stdio ports back, which is what lets a pre-bound pool serve an + /// unbounded number of sequential processes. + @Test func releasesPortsOnDeleteSoTheyCanBeReused() async throws { + let allocator = VsockPortAllocator(base: Self.stdoutPort) + let vm = StubVirtualMachineInstance() + + var pipeFds: [Int32] = [-1, -1] + let rc = pipeFds.withUnsafeMutableBufferPointer { buf -> Int32 in + guard let base = buf.baseAddress else { return -1 } + return pipe(base) + } + try #require(rc == 0) + let writeEnd = pipeFds[1] + let readEnd = pipeFds[0] + defer { _ = close(writeEnd) } + + let agent = StubVirtualMachineAgent( + pid: 4242, + onCreateProcess: { + // Stand in for the guest dialing back on the stdout port. + guard let listener = vm.listener(forPort: Self.stdoutPort) else { + throw ContainerizationError(.internalError, message: "no listener for the stdout port") + } + _ = listener.yield(FileHandle(fileDescriptor: readEnd, closeOnDealloc: false)) + } + ) + + let stdio = IOUtil.setup(portAllocator: allocator, stdin: nil, stdout: DiscardWriter(), stderr: nil) + #expect(allocator.outstandingCount == 1) + + let process = LinuxProcess( + "good", + containerID: "c", + spec: Self.spec(), + io: stdio, + portAllocator: allocator, + ociRuntimePath: nil, + agent: agent, + vm: vm, + logger: nil, + stdioTimeoutSeconds: 30 + ) + + try await process.start() + #expect(process.pid == 4242) + + try await process.delete() + #expect(allocator.outstandingCount == 0) + #expect(allocator.allocate() == Self.stdoutPort) + } + + private static func spec() -> ContainerizationOCI.Spec { + var spec = ContainerizationOCI.Spec() + spec.process = LinuxProcessConfiguration(arguments: ["/bin/true"]).toOCI() + return spec + } +} + +// MARK: - Stubs + +private final class DiscardWriter: Writer { + func write(_ data: Data) throws {} + func close() throws {} +} + +/// Minimal `VirtualMachineInstance` that hands out real `VsockListener`s and +/// keeps them addressable, so a stub agent can play the guest's dial-back. +private final class StubVirtualMachineInstance: VirtualMachineInstance { + typealias Agent = StubVirtualMachineAgent + + private let listeners = Mutex<[UInt32: VsockListener]>([:]) + + var state: VirtualMachineInstanceState { .running } + var mounts: [String: [AttachedFilesystem]] { [:] } + + func listener(forPort port: UInt32) -> VsockListener? { + listeners.withLock { $0[port] } + } + + func listen(_ port: UInt32) throws -> VsockListener { + let listener = VsockListener(port: port) { _ in } + listeners.withLock { $0[port] = listener } + return listener + } + + func dialAgent() async throws -> StubVirtualMachineAgent { + throw ContainerizationError(.unsupported, message: "dialAgent") + } + func dial(_ port: UInt32) async throws -> FileHandle { + throw ContainerizationError(.unsupported, message: "dial") + } + func start() async throws {} + func stop() async throws {} +} + +/// `VirtualMachineAgent` stub whose `createProcess` is injectable so a test +/// can decide whether the guest dials back, is late, or fails outright. +/// Everything the tests don't touch is a no-op. +private final class StubVirtualMachineAgent: VirtualMachineAgent { + private let pid: Int32 + private let onCreateProcess: (@Sendable () async throws -> Void)? + + init(pid: Int32 = 1, onCreateProcess: (@Sendable () async throws -> Void)? = nil) { + self.pid = pid + self.onCreateProcess = onCreateProcess + } + + func createProcess( + id: String, + containerID: String?, + stdinPort: UInt32?, + stdoutPort: UInt32?, + stderrPort: UInt32?, + ociRuntimePath: String?, + configuration: ContainerizationOCI.Spec, + options: Data? + ) async throws { + try await onCreateProcess?() + } + + func startProcess(id: String, containerID: String?) async throws -> Int32 { pid } + func deleteProcess(id: String, containerID: String?) async throws {} + func close() async throws {} + + func standardSetup() async throws {} + func filesystemOperation(operation: FilesystemOperation, path: String) async throws {} + func getenv(key: String) async throws -> String { "" } + func setenv(key: String, value: String) async throws {} + func mount(_ mount: ContainerizationOCI.Mount) async throws {} + func umount(path: String, flags: Int32) async throws {} + func mkdir(path: String, all: Bool, perms: UInt32) async throws {} + @discardableResult + func kill(pid: Int32, signal: Int32) async throws -> Int32 { 0 } + func signalProcess(id: String, containerID: String?, signal: Int32) async throws {} + func resizeProcess(id: String, containerID: String?, columns: UInt32, rows: UInt32) async throws {} + func waitProcess(id: String, containerID: String?, timeoutInSeconds: Int64?) async throws -> Containerization.ExitStatus { + Containerization.ExitStatus(exitCode: 0) + } + func up(name: String, mtu: UInt32?) async throws {} + func down(name: String) async throws {} + func addressAdd(name: String, address: InterfaceAddress) async throws {} + func routeAddLink(name: String, route: LinkRoute) async throws {} + func routeAddDefault(name: String, route: DefaultRoute) async throws {} + func configureDNS(config: DNS, location: String) async throws {} +} diff --git a/Tests/ContainerizationTests/VsockPortAllocatorTests.swift b/Tests/ContainerizationTests/VsockPortAllocatorTests.swift new file mode 100644 index 000000000..42ecaa72f --- /dev/null +++ b/Tests/ContainerizationTests/VsockPortAllocatorTests.swift @@ -0,0 +1,113 @@ +//===----------------------------------------------------------------------===// +// 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. +//===----------------------------------------------------------------------===// + +import Testing + +@testable import Containerization + +struct VsockPortAllocatorTests { + private let base: UInt32 = 0x1000_0000 + + @Test func allocatesSequentiallyWhenNothingIsReleased() { + let allocator = VsockPortAllocator(base: base) + #expect(allocator.allocate() == base) + #expect(allocator.allocate() == base + 1) + #expect(allocator.allocate() == base + 2) + #expect(allocator.outstandingCount == 3) + } + + @Test func reusesTheLowestReleasedPort() { + let allocator = VsockPortAllocator(base: base) + let first = allocator.allocate() + let second = allocator.allocate() + _ = allocator.allocate() + + allocator.release(second) + allocator.release(first) + + // Lowest-first keeps allocations packed at the bottom of the range, so + // they stay inside a pre-bound window. + #expect(allocator.allocate() == first) + #expect(allocator.allocate() == second) + #expect(allocator.allocate() == base + 3) + } + + @Test func releaseIsIdempotentAndIgnoresUnknownPorts() { + let allocator = VsockPortAllocator(base: base) + let port = allocator.allocate() + + allocator.release(port) + allocator.release(port) + // Never handed out, and below the base: neither may enter the free set. + allocator.release(base + 99) + allocator.release(base - 1) + + #expect(allocator.allocate() == port) + #expect(allocator.allocate() == base + 1) + #expect(allocator.outstandingCount == 2) + } + + /// The exec-ceiling regression, at the allocator level. + /// + /// A pre-bound vsock pool can only serve port numbers it bound up front, + /// so a monotonically increasing number turned a pool of 16 into a budget + /// of five processes for the VM's entire life — the fifth `exec` in a + /// one-container pod asked for `base + 16` and failed. Sequential + /// allocate/release cycles must stay inside the window instead. + @Test func sequentialProcessesStayInsideAPreBoundWindow() { + let allocator = VsockPortAllocator(base: base) + let poolSize: UInt32 = 16 + let window = base..<(base + poolSize) + + // The container init holds three ports for the whole run. + let held = [allocator.allocate(), allocator.allocate(), allocator.allocate()] + + // Then 50 sequential execs, each taking and returning three. + for _ in 0..<50 { + let ports = [allocator.allocate(), allocator.allocate(), allocator.allocate()] + for port in ports { + #expect(window.contains(port), "port \(port) fell outside the pre-bound window") + } + for port in ports { + allocator.release(port) + } + } + + for port in held { + #expect(window.contains(port)) + } + #expect(allocator.outstandingCount == 3) + } + + @Test func concurrentAllocationsNeverCollide() async { + let allocator = VsockPortAllocator(base: base) + let count = 200 + + let ports = await withTaskGroup(of: UInt32.self, returning: [UInt32].self) { group in + for _ in 0..