Use Mutex for thread-safe access to structs (#208)

Changes in this PR prevent a race caused by an implicit call to a
computed property getter when updating the property value.

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<Configuration>
     // 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<UInt32>(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(

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<State>
 
     @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)
     }
 }
 

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<State>
 
     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<Socket, Swift.Error> {
-        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
         }
 

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<Task<(), Never>?>(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(