diff --git a/Sources/Containerization/LinuxContainer.swift b/Sources/Containerization/LinuxContainer.swift index d348a984..33801d12 100644 --- a/Sources/Containerization/LinuxContainer.swift +++ b/Sources/Containerization/LinuxContainer.swift @@ -57,8 +57,7 @@ public final class LinuxContainer: Container, Sendable { @SendableProperty private var state: State - @SendableProperty - private var config: Configuration + private let config: Mutex // Ports to be allocated from for stdio and for // unix socket relays that are sharing a guest // uds to the host. @@ -236,10 +235,11 @@ public final class LinuxContainer: Container, Sendable { self.guestVsockPorts = Atomic(0x1000_0000) self.rootfs = rootfs self.logger = logger - self.config = Configuration( + let configuration = Configuration( spec: Self.createDefaultRuntimeSpec(id), mounts: Self.createDefaultMounts() ) + self.config = Mutex(configuration) self.state = .initialized } @@ -280,16 +280,16 @@ public final class LinuxContainer: Container, Sendable { extension LinuxContainer { package var root: String { - self.config.spec.root!.path + config.withLock { $0.spec.root!.path } } /// Number of CPU cores allocated. public var cpus: Int { get { - config.cpus + config.withLock { $0.cpus } } set { - config.cpus = newValue + config.withLock { $0.cpus = newValue } } } @@ -297,27 +297,31 @@ extension LinuxContainer { /// This will be aligned to a 1MB boundary if it isn't already. public var memoryInBytes: UInt64 { get { - config.memoryInBytes + config.withLock { $0.memoryInBytes } } set { - config.memoryInBytes = newValue + config.withLock { $0.memoryInBytes = newValue } } } /// Network interfaces of the container. public var interfaces: [any Interface] { get { - config.interfaces + config.withLock { $0.interfaces } } set { - config.interfaces = newValue + config.withLock { $0.interfaces = newValue } } } /// DNS configuration for the container. public var dns: DNS? { - get { config.dns } - set { config.dns = newValue } + get { + config.withLock { $0.dns } + } + set { + config.withLock { $0.dns = newValue } + } } /// Unix sockets to share into or out of the container. @@ -328,146 +332,178 @@ extension LinuxContainer { /// set to `.unsupported`. public var sockets: [UnixSocketConfiguration] { get { - config.sockets + config.withLock { $0.sockets } } set { - config.sockets = newValue + config.withLock { $0.sockets = newValue } } } /// Enable/disable x86-64 emulation in the container. public var rosetta: Bool { get { - config.rosetta + config.withLock { $0.rosetta } } set { - config.rosetta = newValue + config.withLock { $0.rosetta = newValue } } } /// Enable/disable virtualization capabilities in the container. public var virtualization: Bool { get { - config.virtualization + config.withLock { $0.virtualization } } set { - config.virtualization = newValue + config.withLock { $0.virtualization = newValue } } } /// Filesystem mounts for the container. public var mounts: [Mount] { get { - config.mounts + config.withLock { $0.mounts } } set { - config.mounts = newValue + config.withLock { $0.mounts = newValue } } } /// Arguments passed to the container. public var arguments: [String] { get { - config.spec.process!.args + config.withLock { $0.spec.process!.args } } set { - config.spec.process!.args = newValue + config.withLock { $0.spec.process!.args = newValue } } } /// Environment variables for the container. public var environment: [String] { - get { config.spec.process!.env } - set { config.spec.process!.env = newValue } + get { + config.withLock { $0.spec.process!.env } + } + set { + config.withLock { $0.spec.process!.env = newValue } + } } /// The current working directory (cwd) for the container. public var workingDirectory: String { - get { config.spec.process!.cwd } - set { config.spec.process!.cwd = newValue } + get { + config.withLock { $0.spec.process!.cwd } + } + set { + config.withLock { $0.spec.process!.cwd = newValue } + } } /// The User the container should execute under. public var user: ContainerizationOCI.User { - get { config.spec.process!.user } - set { config.spec.process!.user = newValue } + get { + config.withLock { $0.spec.process!.user } + } + set { + config.withLock { $0.spec.process!.user = newValue } + } } /// Set the hostname for the container. public var hostname: String { - get { config.spec.hostname } - set { config.spec.hostname = newValue } + get { + config.withLock { $0.spec.hostname } + } + set { + config.withLock { $0.spec.hostname = newValue } + } } /// Set any sysctls for the container's environment. public var sysctl: [String: String]? { - get { config.spec.linux!.sysctl } - set { config.spec.linux!.sysctl = newValue } + get { + config.withLock { $0.spec.linux!.sysctl } + } + set { + config.withLock { $0.spec.linux!.sysctl = newValue } + } } /// Rlimits for the container. public var rlimits: [POSIXRlimit] { - get { config.spec.process!.rlimits } - set { config.spec.process!.rlimits = newValue } + get { + config.withLock { $0.spec.process!.rlimits } + } + set { + config.withLock { $0.spec.process!.rlimits = newValue } + } } /// Set a pty device as the container's stdio. This additionally will /// set the TERM=xterm environment variable, and the OCI runtime specs /// `process.terminal` field to true. public var terminalDevice: Terminal? { - get { config.terminal } + get { + config.withLock { $0.terminal } + } set { - config.spec.process!.terminal = newValue != nil ? true : false - config.terminal = newValue - config.spec.process!.env.append("TERM=xterm") - config.ioHandlers.stdin = newValue - config.ioHandlers.stdout = newValue - config.ioHandlers.stderr = nil + config.withLock { + $0.spec.process!.terminal = newValue != nil ? true : false + $0.terminal = newValue + $0.spec.process!.env.append("TERM=xterm") + $0.ioHandlers.stdin = newValue + $0.ioHandlers.stdout = newValue + $0.ioHandlers.stderr = nil + } } } /// If the container has a pty allocated. public var terminal: Bool { - get { config.spec.process!.terminal } + get { + config.withLock { $0.spec.process!.terminal } + } set { - config.spec.process!.terminal = newValue - config.spec.process!.env.append("TERM=xterm") + config.withLock { + $0.spec.process!.terminal = newValue + $0.spec.process!.env.append("TERM=xterm") + } } } /// Set the stdin stream for the initial process of the container. public var stdin: ReaderStream? { get { - config.ioHandlers.stdin + config.withLock { $0.ioHandlers.stdin } } set { - config.ioHandlers.stdin = newValue + config.withLock { $0.ioHandlers.stdin = newValue } } } /// Set the stdout handler for the initial process of the container. public var stdout: Writer? { get { - config.ioHandlers.stdout + config.withLock { $0.ioHandlers.stdout } } set { - config.ioHandlers.stdout = newValue + config.withLock { $0.ioHandlers.stdout = newValue } } } /// Set the stderr handler for the initial process of the container. public var stderr: Writer? { get { - config.ioHandlers.stderr + config.withLock { $0.ioHandlers.stderr } } set { - config.ioHandlers.stderr = newValue + config.withLock { $0.ioHandlers.stderr = newValue } } } public func setProcessConfig(from imageConfig: ImageConfig) { let process = ContainerizationOCI.Process(from: imageConfig) - self.config.spec.process = process + self.config.withLock { $0.spec.process = process } } /// Create the underlying container's virtual machine @@ -529,7 +565,7 @@ extension LinuxContainer { let agent = try await vm.dialAgent() do { - var specCopy = config.spec + var specCopy = config.withLock { $0.spec } // We don't need the rootfs, nor do OCI runtimes want it included. specCopy.mounts = vm.mounts.dropFirst().map { $0.to } @@ -619,7 +655,7 @@ extension LinuxContainer { try await startedState.vm.withAgent { agent in // First, we need to stop any unix socket relays as this will // keep the rootfs from being able to umount (EBUSY). - let sockets = self.config.sockets + let sockets = config.withLock { $0.sockets } if !sockets.isEmpty { guard let relayAgent = agent as? SocketRelayAgent else { throw ContainerizationError( @@ -687,7 +723,7 @@ extension LinuxContainer { ) async throws -> LinuxProcess { let state = try self.state.startedState("exec") - var specCopy = config.spec + var specCopy = config.withLock { $0.spec } specCopy.process = configuration let stdio = Self.setupIO( diff --git a/Sources/Containerization/NATNetworkInterface.swift b/Sources/Containerization/NATNetworkInterface.swift index b90c0932..e648802e 100644 --- a/Sources/Containerization/NATNetworkInterface.swift +++ b/Sources/Containerization/NATNetworkInterface.swift @@ -20,42 +20,53 @@ import vmnet import Virtualization import ContainerizationError import Foundation -import SendableProperty +import Synchronization /// An interface that uses NAT to provide an IP address for a given /// container/virtual machine. @available(macOS 26, *) public final class NATNetworkInterface: Interface, Sendable { public var address: String { - get { state.address } - set { state.address = newValue } + get { + state.withLock { $0.address } + } + set { + state.withLock { $0.address = newValue } + } } public var gateway: String? { - get { state.gateway } - set { state.gateway = newValue } + get { + state.withLock { $0.gateway } + } + set { + state.withLock { $0.gateway = newValue } + } } @available(macOS 26, *) public var reference: vmnet_network_ref { - state.reference + state.withLock { $0.reference } } public var macAddress: String? { - get { state.macAddress } - set { state.macAddress = newValue } + get { + state.withLock { $0.macAddress } + } + set { + state.withLock { $0.macAddress = newValue } + } } private struct State { - fileprivate var address: String - fileprivate var gateway: String? - fileprivate var reference: vmnet_network_ref! - fileprivate var macAddress: String? + var address: String + var gateway: String? + var reference: vmnet_network_ref! + var macAddress: String? } - @SendableProperty - private var state: State + private let state: Mutex @available(macOS 26, *) public init( @@ -64,12 +75,13 @@ public final class NATNetworkInterface: Interface, Sendable { reference: sending vmnet_network_ref, macAddress: String? = nil ) { - self.state = .init( + let state = State( address: address, gateway: gateway, reference: reference, macAddress: macAddress ) + self.state = Mutex(state) } @available(macOS, obsoleted: 26, message: "Use init(address:gateway:reference:macAddress:) instead") @@ -78,12 +90,13 @@ public final class NATNetworkInterface: Interface, Sendable { gateway: String?, macAddress: String? = nil ) { - self.state = .init( + let state = State( address: address, gateway: gateway, reference: nil, macAddress: macAddress ) + self.state = Mutex(state) } } diff --git a/Sources/ContainerizationOS/Socket/Socket.swift b/Sources/ContainerizationOS/Socket/Socket.swift index c1ca169c..d60cade2 100644 --- a/Sources/ContainerizationOS/Socket/Socket.swift +++ b/Sources/ContainerizationOS/Socket/Socket.swift @@ -15,7 +15,7 @@ //===----------------------------------------------------------------------===// import Foundation -import SendableProperty +import Synchronization #if canImport(Musl) import Musl @@ -73,11 +73,10 @@ public final class Socket: Sendable { private let _closeOnDeinit: Bool private let _queue: DispatchQueue - @SendableProperty - private var _state: State + private let state: Mutex public var fileDescriptor: Int32 { - guard let handle = _state.handle else { + guard let handle = state.withLock({ $0.handle }) else { return -1 } return handle.fileDescriptor @@ -94,12 +93,13 @@ public final class Socket: Sendable { init(fd: Int32, type: SocketType, closeOnDeinit: Bool) { _queue = DispatchQueue(label: "com.apple.containerization.socket") _closeOnDeinit = closeOnDeinit - _state = State( + let state = State( socketState: .created, handle: FileHandle(fileDescriptor: fd, closeOnDealloc: false), type: type, acceptSource: nil ) + self.state = Mutex(state) } deinit { @@ -115,84 +115,94 @@ extension Socket { } public func connect() throws { - guard let handle = _state.handle else { + guard let handle = state.withLock({ $0.handle }) else { throw SocketError.closed } - guard _state.socketState == .created else { + guard state.withLock({ $0.socketState }) == .created else { throw SocketError.invalidOperationOnSocket("connect") } var res: Int32 = 0 - try _state.type.withSockAddr { (ptr, length) in - res = Syscall.retrying { - sysConnect(handle.fileDescriptor, ptr, length) + try state.withLock { + try $0.type.withSockAddr { (ptr, length) in + res = Syscall.retrying { + sysConnect(handle.fileDescriptor, ptr, length) + } } } if res == -1 { - throw Socket.errnoToError(msg: "could not connect to socket \(_state.type)") + throw Socket.errnoToError(msg: "could not connect to socket \(state.withLock { $0.type })") + } + state.withLock { + $0 = State( + socketState: .connected, + handle: handle, + type: $0.type, + acceptSource: $0.acceptSource + ) } - _state = State( - socketState: .connected, - handle: handle, - type: _state.type, - acceptSource: _state.acceptSource - ) } public func listen() throws { - guard let handle = _state.handle else { + guard let handle = state.withLock({ $0.handle }) else { throw SocketError.closed } - guard _state.socketState == .created else { + guard state.withLock({ $0.socketState }) == .created else { throw SocketError.invalidOperationOnSocket("listen") } - try _state.type.beforeBind(fd: handle.fileDescriptor) + try state.withLock { try $0.type.beforeBind(fd: handle.fileDescriptor) } var rc: Int32 = 0 - try _state.type.withSockAddr { (ptr, length) in - rc = sysBind(handle.fileDescriptor, ptr, length) + try state.withLock { + try $0.type.withSockAddr { (ptr, length) in + rc = sysBind(handle.fileDescriptor, ptr, length) + } } if rc < 0 { - throw Socket.errnoToError(msg: "could not bind to \(_state.type)") + throw Socket.errnoToError(msg: "could not bind to \(state.withLock { $0.type })") } - try _state.type.beforeListen(fd: handle.fileDescriptor) + try state.withLock { try $0.type.beforeListen(fd: handle.fileDescriptor) } if sysListen(handle.fileDescriptor, SOMAXCONN) < 0 { - throw Socket.errnoToError(msg: "listen failed on \(_state.type)") + throw Socket.errnoToError(msg: "listen failed on \(state.withLock { $0.type })") + } + state.withLock { + $0 = State( + socketState: .listening, + handle: handle, + type: $0.type, + acceptSource: $0.acceptSource + ) } - _state = State( - socketState: .listening, - handle: handle, - type: _state.type, - acceptSource: _state.acceptSource - ) } public func close() throws { // Already closed. - guard let handle = _state.handle else { + guard let handle = state.withLock({ $0.handle }) else { return } - if let acceptSource = _state.acceptSource { + if let acceptSource = state.withLock({ $0.acceptSource }) { acceptSource.cancel() } try handle.close() - _state = State( - socketState: _state.socketState, - handle: nil, - type: _state.type, - acceptSource: nil - ) + state.withLock { + $0 = State( + socketState: $0.socketState, + handle: nil, + type: $0.type, + acceptSource: nil + ) + } } public func write(data: any DataProtocol) throws -> Int { - guard _state.socketState == .connected else { + guard state.withLock({ $0.socketState }) == .connected else { throw SocketError.invalidOperationOnSocket("write") } - guard let handle = _state.handle else { + guard let handle = state.withLock({ $0.handle }) else { throw SocketError.closed } @@ -205,28 +215,31 @@ extension Socket { } public func acceptStream(closeOnDeinit: Bool = true) throws -> AsyncThrowingStream { - guard _state.socketState == .listening else { + guard state.withLock({ $0.socketState }) == .listening else { throw SocketError.invalidOperationOnSocket("accept") } - guard let handle = _state.handle else { + guard let handle = state.withLock({ $0.handle }) else { throw SocketError.closed } - guard _state.acceptSource == nil else { + guard state.withLock({ $0.acceptSource }) == nil else { throw SocketError.acceptStreamExists } - let source = DispatchSource.makeReadSource( - fileDescriptor: handle.fileDescriptor, - queue: _queue - ) - _state = State( - socketState: _state.socketState, - handle: handle, - type: _state.type, - acceptSource: source - ) + let source = state.withLock { + let source = DispatchSource.makeReadSource( + fileDescriptor: handle.fileDescriptor, + queue: _queue + ) + $0 = State( + socketState: $0.socketState, + handle: handle, + type: $0.type, + acceptSource: source + ) + return source + } return AsyncThrowingStream { cont in source.setCancelHandler { @@ -253,15 +266,15 @@ extension Socket { } public func accept(closeOnDeinit: Bool = true) throws -> Socket { - guard _state.socketState == .listening else { + guard state.withLock({ $0.socketState }) == .listening else { throw SocketError.invalidOperationOnSocket("accept") } - guard let handle = _state.handle else { + guard let handle = state.withLock({ $0.handle }) else { throw SocketError.closed } - let (clientFD, socketType) = try _state.type.accept(fd: handle.fileDescriptor) + let (clientFD, socketType) = try state.withLock { try $0.type.accept(fd: handle.fileDescriptor) } return Socket( fd: clientFD, type: socketType, @@ -270,11 +283,11 @@ extension Socket { } public func read(buffer: inout Data) throws -> Int { - guard _state.socketState == .connected else { + guard state.withLock({ $0.socketState }) == .connected else { throw SocketError.invalidOperationOnSocket("read") } - guard let handle = _state.handle else { + guard let handle = state.withLock({ $0.handle }) else { throw SocketError.closed } @@ -298,7 +311,7 @@ extension Socket { } public func shutdown(how: ShutdownOption) throws { - guard let handle = _state.handle else { + guard let handle = state.withLock({ $0.handle }) else { throw SocketError.closed } @@ -318,7 +331,7 @@ extension Socket { } public func setSockOpt(sockOpt: Int32 = 0, ptr: UnsafeRawPointer, stride: UInt32) throws { - guard let handle = _state.handle else { + guard let handle = state.withLock({ $0.handle }) else { throw SocketError.closed } if setsockopt(handle.fileDescriptor, SOL_SOCKET, sockOpt, ptr, stride) < 0 { @@ -327,7 +340,7 @@ extension Socket { } public func setTimeout(option: TimeoutOption, seconds: Int) throws { - guard let handle = _state.handle else { + guard let handle = state.withLock({ $0.handle }) else { throw SocketError.closed } diff --git a/vminitd/Sources/vminitd/VsockProxy.swift b/vminitd/Sources/vminitd/VsockProxy.swift index a4073a70..70ea308c 100644 --- a/vminitd/Sources/vminitd/VsockProxy.swift +++ b/vminitd/Sources/vminitd/VsockProxy.swift @@ -55,17 +55,13 @@ final class VsockProxy: Sendable { private let log: Logger? @SendableProperty - private var state = State() - - private struct State { - var listener: Socket? - var task: Task<(), Never>? - } + private var listener: Socket? + private let task = Mutex?>(nil) } extension VsockProxy { func close() throws { - guard let listener = state.listener else { + guard let listener else { return } @@ -74,7 +70,8 @@ extension VsockProxy { if fm.fileExists(atPath: self.path.path) { try FileManager.default.removeItem(at: self.path) } - state.task?.cancel() + let task = task.withLock { $0 } + task?.cancel() } func start() throws { @@ -102,7 +99,7 @@ extension VsockProxy { ) let uds = try Socket(type: type) try uds.listen() - state.listener = uds + listener = uds try self.acceptLoop(socketType: .unix) } @@ -114,18 +111,18 @@ extension VsockProxy { ) let vsock = try Socket(type: type) try vsock.listen() - state.listener = vsock + listener = vsock try self.acceptLoop(socketType: .vsock) } private func acceptLoop(socketType: SocketType) throws { - guard let listener = state.listener else { + guard let listener else { return } let stream = try listener.acceptStream() - state.task = Task { + let task = Task { do { for try await conn in stream { Task { @@ -143,6 +140,7 @@ extension VsockProxy { self.log?.error("failed to accept connection: \(error)") } } + self.task.withLock { $0 = task } } private func handleConn(