From 977552849564be442f127072305ad04de52c2e1d Mon Sep 17 00:00:00 2001 From: Danny Canter Date: Thu, 3 Jul 2025 14:11:29 -0700 Subject: [PATCH] Vminitd: Rework ManagedProcess.IO protocols (#194) I think the way these types were structured made reasoning about them a bit difficult. I don't think this fully solves the problem, but this change aims to make things a bit simpler by hoisting a lot of the logic for the IO relays to a new IOPair type thats goal is to simply take in a reader and writer and handle their resource cleanup after a relay finishes. Bundled in with this is also the buffer we'll use to copy between them. This change: - Alters ManagedProcess.IO `start` to take in the process to alter instead of this just being a side effect of the constructors. - Gets rid of `close()` in favor of `CloseStdin`. The IO will get closed when the relays finish, which will naturally happen if the process exits or just closes its side of the pipes/pty. This makes it so that the one special case (a client wants to signal no more input is coming) is still sane. - Move all relay logic and resource cleanup to a new IOPair type that takes in protocols that are easily conformable by all of our various io types ( Socket, Terminal, FileHandle). --- .../Sources/vminitd/IOCloser+Extensions.swift | 28 ++ vminitd/Sources/vminitd/IOCloser.swift | 21 ++ vminitd/Sources/vminitd/IOPair.swift | 124 +++++++++ .../Sources/vminitd/ManagedContainer.swift | 5 - vminitd/Sources/vminitd/ManagedProcess.swift | 21 +- vminitd/Sources/vminitd/StandardIO.swift | 261 ++++++------------ vminitd/Sources/vminitd/TerminalIO.swift | 169 ++++-------- 7 files changed, 321 insertions(+), 308 deletions(-) create mode 100644 vminitd/Sources/vminitd/IOCloser+Extensions.swift create mode 100644 vminitd/Sources/vminitd/IOCloser.swift create mode 100644 vminitd/Sources/vminitd/IOPair.swift diff --git a/vminitd/Sources/vminitd/IOCloser+Extensions.swift b/vminitd/Sources/vminitd/IOCloser+Extensions.swift new file mode 100644 index 00000000..d3d0397c --- /dev/null +++ b/vminitd/Sources/vminitd/IOCloser+Extensions.swift @@ -0,0 +1,28 @@ +//===----------------------------------------------------------------------===// +// Copyright © 2025 Apple Inc. and the Containerization project authors. All rights reserved. +// +// 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 ContainerizationOS +import Foundation + +extension Socket: IOCloser {} + +extension Terminal: IOCloser { + var fileDescriptor: Int32 { + self.handle.fileDescriptor + } +} + +extension FileHandle: IOCloser {} diff --git a/vminitd/Sources/vminitd/IOCloser.swift b/vminitd/Sources/vminitd/IOCloser.swift new file mode 100644 index 00000000..cf8c78b8 --- /dev/null +++ b/vminitd/Sources/vminitd/IOCloser.swift @@ -0,0 +1,21 @@ +//===----------------------------------------------------------------------===// +// Copyright © 2025 Apple Inc. and the Containerization project authors. All rights reserved. +// +// 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. +//===----------------------------------------------------------------------===// + +protocol IOCloser: Sendable { + var fileDescriptor: Int32 { get } + + func close() throws +} diff --git a/vminitd/Sources/vminitd/IOPair.swift b/vminitd/Sources/vminitd/IOPair.swift new file mode 100644 index 00000000..0c277782 --- /dev/null +++ b/vminitd/Sources/vminitd/IOPair.swift @@ -0,0 +1,124 @@ +//===----------------------------------------------------------------------===// +// Copyright © 2025 Apple Inc. and the Containerization project authors. All rights reserved. +// +// 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 ContainerizationOS +import Foundation +import Logging +import Synchronization + +final class IOPair: Sendable { + let readFrom: IOCloser + let writeTo: IOCloser + nonisolated(unsafe) let buffer: UnsafeMutableBufferPointer + private let logger: Logger? + + private let done: Atomic + + init(readFrom: IOCloser, writeTo: IOCloser, logger: Logger? = nil) { + self.readFrom = readFrom + self.writeTo = writeTo + self.done = Atomic(false) + self.buffer = UnsafeMutableBufferPointer.allocate(capacity: Int(getpagesize())) + self.logger = logger + } + + func relay() throws { + let readFromFd = self.readFrom.fileDescriptor + let writeToFd = self.writeTo.fileDescriptor + + let readFrom = OSFile(fd: readFromFd) + let writeTo = OSFile(fd: writeToFd) + + try ProcessSupervisor.default.poller.add(readFromFd, mask: EPOLLIN) { mask in + if mask.isHangup && !mask.readyToRead { + self.close() + return + } + // Loop so that in the case that someone wrote > buf.count down the pipe + // we properly will drain it fully. + while true { + let r = readFrom.read(self.buffer) + if r.read > 0 { + let view = UnsafeMutableBufferPointer( + start: self.buffer.baseAddress, + count: r.read + ) + + let w = writeTo.write(view) + if w.wrote != r.read { + self.logger?.error("stopping relay: short write for stdio") + self.close() + return + } + } + + switch r.action { + case .error(let errno): + self.logger?.error("failed with errno \(errno) while reading for fd \(readFromFd)") + fallthrough + case .eof: + self.close() + self.logger?.debug("closing relay for \(readFromFd)") + return + case .again: + // We read all we could, exit. + if mask.isHangup { + self.close() + } + return + default: + break + } + } + } + } + + func close() { + guard + self.done.compareExchange( + expected: false, + desired: true, + successOrdering: .acquiringAndReleasing, + failureOrdering: .acquiring + ).exchanged + else { + return + } + + self.buffer.deallocate() + + let readFromFd = self.readFrom.fileDescriptor + // Remove the fd from our global epoll instance first. + do { + try ProcessSupervisor.default.poller.delete(readFromFd) + } catch { + self.logger?.error("failed to delete fd from epoll \(readFromFd): \(error)") + } + + do { + try self.readFrom.close() + } catch { + self.logger?.error("failed to close reader fd for IOPair: \(error)") + } + + do { + try self.writeTo.close() + } catch { + self.logger?.error("failed to close writer fd for IOPair: \(error)") + } + } +} diff --git a/vminitd/Sources/vminitd/ManagedContainer.swift b/vminitd/Sources/vminitd/ManagedContainer.swift index 484ea510..1d16261a 100644 --- a/vminitd/Sources/vminitd/ManagedContainer.swift +++ b/vminitd/Sources/vminitd/ManagedContainer.swift @@ -111,11 +111,6 @@ extension ManagedContainer { try proc.resize(size: size) } - func close(execID: String) throws { - let proc = try self.getExecOrInit(execID: execID) - try proc.close() - } - func deleteExec(id: String) throws { try ensureExecExists(id) do { diff --git a/vminitd/Sources/vminitd/ManagedProcess.swift b/vminitd/Sources/vminitd/ManagedProcess.swift index 18369f7f..11412467 100644 --- a/vminitd/Sources/vminitd/ManagedProcess.swift +++ b/vminitd/Sources/vminitd/ManagedProcess.swift @@ -40,7 +40,6 @@ final class ManagedProcess: Sendable { let io: IO var waiters: [CheckedContinuation] = [] var exitStatus: Int32? = nil - var closed: Bool = false var pid: Int32 = 0 } @@ -52,10 +51,10 @@ final class ManagedProcess: Sendable { // swiftlint: disable type_name protocol IO { - func start() throws + func start(process: inout Command) throws func closeAfterExec() throws func resize(size: Terminal.Size) throws - func close() throws + func closeStdin() throws } // swiftlint: enable type_name @@ -105,14 +104,12 @@ final class ManagedProcess: Sendable { let attrs = Command.Attrs(setsid: false, setctty: false) process.attrs = attrs io = try TerminalIO( - process: &process, stdio: stdio, log: log ) } else { process.attrs = .init(setsid: false) io = StandardIO( - process: &process, stdio: stdio, log: log ) @@ -121,7 +118,7 @@ final class ManagedProcess: Sendable { log.info("starting io") // Setup IO early. We expect the host to be listening already. - try io.start() + try io.start(process: &process) self.process = process self.lock = Mutex(State(io: io)) @@ -213,20 +210,10 @@ extension ManagedProcess { func resize(size: Terminal.Size) throws { try self.lock.withLock { - if $0.closed { + guard $0.exitStatus == nil else { return } try $0.io.resize(size: size) } } - - func close() throws { - try self.lock.withLock { - if $0.closed { - return - } - try $0.io.close() - $0.closed = true - } - } } diff --git a/vminitd/Sources/vminitd/StandardIO.swift b/vminitd/Sources/vminitd/StandardIO.swift index 7aba046f..dd1f908d 100644 --- a/vminitd/Sources/vminitd/StandardIO.swift +++ b/vminitd/Sources/vminitd/StandardIO.swift @@ -17,204 +17,125 @@ import ContainerizationOS import Foundation import Logging -import SendableProperty import Synchronization final class StandardIO: ManagedProcess.IO & Sendable { + private struct State { + var stdin: IOPair? + var stdout: IOPair? + var stderr: IOPair? + + var stdinPipe: Pipe? + var stdoutPipe: Pipe? + var stderrPipe: Pipe? + } + private let log: Logger? - - private let stdio: HostStdio - private let stdinPipe: Pipe? - private let stdoutPipe: Pipe? - private let stderrPipe: Pipe? - - @SendableProperty - private var stdinSocket: Socket? - @SendableProperty - private var stdoutSocket: Socket? - @SendableProperty - private var stderrSocket: Socket? + private let hostStdio: HostStdio + private let state: Mutex init( - process: inout Command, stdio: HostStdio, log: Logger? ) { - self.stdio = stdio + self.hostStdio = stdio self.log = log - - if stdio.stdin != nil { - let inPipe = Pipe() - process.stdin = inPipe.fileHandleForReading - self.stdinPipe = inPipe - } else { - process.stdin = nil - self.stdinPipe = nil - } - - if stdio.stdout != nil { - let outPipe = Pipe() - process.stdout = outPipe.fileHandleForWriting - self.stdoutPipe = outPipe - } else { - process.stdout = nil - self.stdoutPipe = nil - } - - if stdio.stderr != nil { - let errPipe = Pipe() - process.stderr = errPipe.fileHandleForWriting - self.stderrPipe = errPipe - } else { - process.stderr = nil - self.stderrPipe = nil - } + self.state = Mutex(State()) } - func start() throws { - if let stdinPort = self.stdio.stdin { - let type = VsockType( - port: stdinPort, - cid: VsockType.hostCID - ) - let stdinSocket = try Socket(type: type, closeOnDeinit: false) - try stdinSocket.connect() - self.stdinSocket = stdinSocket + func start(process: inout Command) throws { + try self.state.withLock { + if let stdinPort = self.hostStdio.stdin { + let inPipe = Pipe() + process.stdin = inPipe.fileHandleForReading + $0.stdinPipe = inPipe - try relay( - readFromFd: stdinSocket.fileDescriptor, - writeToFd: self.stdinPipe!.fileHandleForWriting.fileDescriptor - ) - } + let type = VsockType( + port: stdinPort, + cid: VsockType.hostCID + ) + let stdinSocket = try Socket(type: type, closeOnDeinit: false) + try stdinSocket.connect() - if let stdoutPort = self.stdio.stdout { - let type = VsockType( - port: stdoutPort, - cid: VsockType.hostCID - ) - // These fd's get closed when cleanupRelay is called - let stdoutSocket = try Socket(type: type, closeOnDeinit: false) - try stdoutSocket.connect() - self.stdoutSocket = stdoutSocket + let pair = IOPair( + readFrom: stdinSocket, + writeTo: inPipe.fileHandleForWriting, + logger: log + ) + $0.stdin = pair - try relay( - readFromFd: self.stdoutPipe!.fileHandleForReading.fileDescriptor, - writeToFd: stdoutSocket.fileDescriptor - ) - } + try pair.relay() + } - if let stderrPort = self.stdio.stderr { - let type = VsockType( - port: stderrPort, - cid: VsockType.hostCID - ) - let stderrSocket = try Socket(type: type, closeOnDeinit: false) - try stderrSocket.connect() - self.stderrSocket = stderrSocket + if let stdoutPort = self.hostStdio.stdout { + let outPipe = Pipe() + process.stdout = outPipe.fileHandleForWriting + $0.stdoutPipe = outPipe - try relay( - readFromFd: self.stderrPipe!.fileHandleForReading.fileDescriptor, - writeToFd: stderrSocket.fileDescriptor - ) + let type = VsockType( + port: stdoutPort, + cid: VsockType.hostCID + ) + let stdoutSocket = try Socket(type: type, closeOnDeinit: false) + try stdoutSocket.connect() + + let pair = IOPair( + readFrom: outPipe.fileHandleForReading, + writeTo: stdoutSocket, + logger: log + ) + $0.stdout = pair + + try pair.relay() + } + + if let stderrPort = self.hostStdio.stderr { + let errPipe = Pipe() + process.stderr = errPipe.fileHandleForWriting + $0.stderrPipe = errPipe + + let type = VsockType( + port: stderrPort, + cid: VsockType.hostCID + ) + let stderrSocket = try Socket(type: type, closeOnDeinit: false) + try stderrSocket.connect() + + let pair = IOPair( + readFrom: errPipe.fileHandleForReading, + writeTo: stderrSocket, + logger: log + ) + $0.stderr = pair + + try pair.relay() + } } } // NOP func resize(size: Terminal.Size) throws {} - func relay(readFromFd: Int32, writeToFd: Int32) throws { - let readFrom = OSFile(fd: readFromFd) - let writeTo = OSFile(fd: writeToFd) - // `buf` and `didCleanup` aren't used concurrently. - nonisolated(unsafe) let buf = UnsafeMutableBufferPointer.allocate(capacity: Int(getpagesize())) - nonisolated(unsafe) var didCleanup = false - - let cleanupRelay: @Sendable () -> Void = { - if didCleanup { return } - didCleanup = true - self.cleanupRelay(readFd: readFromFd, writeFd: writeToFd, buffer: buf, log: self.log) - } - - try ProcessSupervisor.default.poller.add(readFromFd, mask: EPOLLIN) { mask in - if mask.isHangup && !mask.readyToRead { - cleanupRelay() - return + func closeStdin() throws { + self.state.withLock { + if let stdin = $0.stdin { + stdin.close() + $0.stdin = nil } - // Loop so that in the case that someone wrote > buf.count down the pipe - // we properly will drain it fully. - while true { - let r = readFrom.read(buf) - if r.read > 0 { - let view = UnsafeMutableBufferPointer( - start: buf.baseAddress, - count: r.read - ) - - let w = writeTo.write(view) - if w.wrote != r.read { - self.log?.error("stopping relay: short write for stdio") - cleanupRelay() - return - } - } - - switch r.action { - case .error(let errno): - self.log?.error("failed with errno \(errno) while reading for fd \(readFromFd)") - fallthrough - case .eof: - cleanupRelay() - self.log?.debug("closing relay for \(readFromFd)") - return - case .again: - // We read all we could, exit. - if mask.isHangup { - cleanupRelay() - } - return - default: - break - } - } - } - } - - func cleanupRelay(readFd: Int32, writeFd: Int32, buffer: UnsafeMutableBufferPointer, log: Logger?) { - do { - // We could alternatively just allocate buffers in the constructor, and free them - // on close(). - buffer.deallocate() - try ProcessSupervisor.default.poller.delete(readFd) - } catch { - self.log?.error("failed to delete pipe fd from epoll \(readFd): \(error)") - } - if Foundation.close(writeFd) != 0 { - let err = POSIXError.fromErrno() - self.log?.error("failed to close write fd for StandardIO relay: \(String(describing:err))") - } - } - - func close() throws { - if let stdin = self.stdinPipe { - try stdin.fileHandleForWriting.close() - } - if let stdout = self.stdoutPipe { - try stdout.fileHandleForReading.close() - } - if let stderr = self.stderrPipe { - try stderr.fileHandleForReading.close() } } func closeAfterExec() throws { - if let stdin = self.stdinPipe { - try stdin.fileHandleForReading.close() - } - if let stdout = self.stdoutPipe { - try stdout.fileHandleForWriting.close() - } - if let stderr = self.stderrPipe { - try stderr.fileHandleForWriting.close() + try self.state.withLock { + if let stdin = $0.stdinPipe { + try stdin.fileHandleForReading.close() + } + if let stdout = $0.stdoutPipe { + try stdout.fileHandleForWriting.close() + } + if let stderr = $0.stderrPipe { + try stderr.fileHandleForWriting.close() + } } } } diff --git a/vminitd/Sources/vminitd/TerminalIO.swift b/vminitd/Sources/vminitd/TerminalIO.swift index f67d1b03..c8815848 100644 --- a/vminitd/Sources/vminitd/TerminalIO.swift +++ b/vminitd/Sources/vminitd/TerminalIO.swift @@ -17,153 +17,90 @@ import ContainerizationOS import Foundation import Logging -import SendableProperty +import Synchronization final class TerminalIO: ManagedProcess.IO & Sendable { + private struct State { + var stdin: IOPair? + var stdout: IOPair? + } + private let parent: Terminal private let child: Terminal private let log: Logger? - - private let stdio: HostStdio - @SendableProperty - private var stdinSocket: Socket? - @SendableProperty - private var stdoutSocket: Socket? + private let hostStdio: HostStdio + private let state: Mutex init( - process: inout Command, stdio: HostStdio, log: Logger? ) throws { let pair = try Terminal.create() self.parent = pair.parent self.child = pair.child - self.stdio = stdio + self.state = Mutex(State()) + self.hostStdio = stdio self.log = log - - let ptyHandle = child.handle - let useHandles = stdio.stdin != nil || stdio.stdout != nil - // We currently set stdin to the controlling terminal always, so - // it must be a valid pty descriptor. - process.stdin = useHandles ? ptyHandle : nil - - let stdoutHandle = useHandles ? ptyHandle : nil - process.stdout = stdoutHandle - process.stderr = stdoutHandle } func resize(size: Terminal.Size) throws { - if self.stdio.stdin != nil { - try parent.resize(size: size) - } + try parent.resize(size: size) } - func start() throws { - if let stdinPort = self.stdio.stdin { - let type = VsockType( - port: stdinPort, - cid: VsockType.hostCID - ) - let stdinSocket = try Socket(type: type, closeOnDeinit: false) - try stdinSocket.connect() - self.stdinSocket = stdinSocket + func start(process: inout Command) throws { + try self.state.withLock { + let ptyHandle = self.child.handle + let useHandles = self.hostStdio.stdin != nil || self.hostStdio.stdout != nil + // We currently set stdin to the controlling terminal always, so + // it must be a valid pty descriptor. + process.stdin = useHandles ? ptyHandle : nil - try relay( - readFromFd: stdinSocket.fileDescriptor, - writeToFd: self.parent.handle.fileDescriptor - ) - } + let stdoutHandle = useHandles ? ptyHandle : nil + process.stdout = stdoutHandle + process.stderr = stdoutHandle - if let stdoutPort = self.stdio.stdout { - let type = VsockType( - port: stdoutPort, - cid: VsockType.hostCID - ) - let stdoutSocket = try Socket(type: type, closeOnDeinit: false) - try stdoutSocket.connect() - self.stdoutSocket = stdoutSocket + if let stdinPort = self.hostStdio.stdin { + let type = VsockType( + port: stdinPort, + cid: VsockType.hostCID + ) + let stdinSocket = try Socket(type: type, closeOnDeinit: false) + try stdinSocket.connect() - try relay( - readFromFd: self.parent.handle.fileDescriptor, - writeToFd: stdoutSocket.fileDescriptor - ) - } - } + let pair = IOPair( + readFrom: stdinSocket, + writeTo: self.parent.handle, + logger: self.log + ) + $0.stdin = pair - func relay(readFromFd: Int32, writeToFd: Int32) throws { - let readFrom = OSFile(fd: readFromFd) - let writeTo = OSFile(fd: writeToFd) - // `buf` and `didCleanup` aren't used concurrently. - nonisolated(unsafe) let buf = UnsafeMutableBufferPointer.allocate(capacity: Int(getpagesize())) - nonisolated(unsafe) var didCleanup = false - - let cleanupRelay: @Sendable () -> Void = { - if didCleanup { return } - didCleanup = true - self.cleanupRelay(readFd: readFromFd, writeFd: writeToFd, buffer: buf, log: self.log) - } - - try ProcessSupervisor.default.poller.add(readFromFd, mask: EPOLLIN) { mask in - if mask.isHangup && !mask.readyToRead { - cleanupRelay() - return + try pair.relay() } - // Loop so that in the case that someone wrote > buf.count down the pipe - // we properly will drain it fully. - while true { - let r = readFrom.read(buf) - if r.read > 0 { - let view = UnsafeMutableBufferPointer( - start: buf.baseAddress, - count: r.read - ) - let w = writeTo.write(view) - if w.wrote != r.read { - self.log?.error("stopping relay: short write for stdio") - cleanupRelay() - return - } - } + if let stdoutPort = self.hostStdio.stdout { + let type = VsockType( + port: stdoutPort, + cid: VsockType.hostCID + ) + let stdoutSocket = try Socket(type: type, closeOnDeinit: false) + try stdoutSocket.connect() - switch r.action { - case .error(let errno): - self.log?.error("failed with errno \(errno) while reading for fd \(readFromFd)") - fallthrough - case .eof: - cleanupRelay() - self.log?.debug("closing relay for \(readFromFd)") - return - case .again: - // We read all we could, exit. - if mask.isHangup { - cleanupRelay() - } - return - default: - break - } + let pair = IOPair( + readFrom: self.parent.handle, + writeTo: stdoutSocket, + logger: self.log + ) + $0.stdout = pair + + try pair.relay() } } } - func cleanupRelay(readFd: Int32, writeFd: Int32, buffer: UnsafeMutableBufferPointer, log: Logger?) { - do { - // We could alternatively just allocate buffers in the constructor, and free them - // on close(). - buffer.deallocate() - try ProcessSupervisor.default.poller.delete(readFd) - } catch { - self.log?.error("failed to delete pipe fd from epoll \(readFd): \(error)") + func closeStdin() throws { + self.state.withLock { + $0.stdin?.close() } - if Foundation.close(writeFd) != 0 { - let err = POSIXError.fromErrno() - self.log?.error("failed to close write fd for TerminalIO relay: \(String(describing:err))") - } - } - - func close() throws { - try parent.close() } func closeAfterExec() throws {