Merge branch 'main' into feature/ipv6-support

Author: yordsel

Sources/Containerization/FileMount.swift

@@ -230,7 +230,7 @@ extension FileMountContext {
 
 extension FileMountContext {
     /// Clean up temp directories.
-    func cleanup() {
+    func cleanUp() {
         let fm = FileManager.default
         for prepared in preparedMounts {
             try? fm.removeItem(at: prepared.tempDirectory)

Sources/Containerization/Image/ImageStore/ImageStore.swift

@@ -124,7 +124,7 @@ extension ImageStore {
         try await self.lock.withLock { lockCtx in
             try await self.referenceManager.delete(reference: reference)
             if performCleanup {
-                try await self._cleanupOrphanedBlobs(lockCtx)
+                try await self._cleanUpOrphanedBlobs(lockCtx)
             }
         }
     }
@@ -135,9 +135,9 @@ extension ImageStore {
     ///   `deleted` :  A  list of the names of the content items that were deleted from the `ContentStore`,
     ///   `freed` : The total size of the items that were deleted.
     @discardableResult
-    public func cleanupOrphanedBlobs() async throws -> (deleted: [String], freed: UInt64) {
+    public func cleanUpOrphanedBlobs() async throws -> (deleted: [String], freed: UInt64) {
         try await self.lock.withLock { lockCtx in
-            try await self._cleanupOrphanedBlobs(lockCtx)
+            try await self._cleanUpOrphanedBlobs(lockCtx)
         }
     }
 
@@ -151,7 +151,7 @@ extension ImageStore {
     }
 
     @discardableResult
-    private func _cleanupOrphanedBlobs(_ lock: AsyncLock.Context) async throws -> (deleted: [String], freed: UInt64) {
+    private func _cleanUpOrphanedBlobs(_ lock: AsyncLock.Context) async throws -> (deleted: [String], freed: UInt64) {
         let images = try await self.list()
         var referenced: [String] = []
         for image in images {

Sources/Containerization/LinuxContainer.swift

@@ -703,7 +703,7 @@ extension LinuxContainer {
             }
 
             // Clean up file mount temporary directories.
-            fileMountContext.cleanup()
+            fileMountContext.cleanUp()
 
             do {
                 try await vm.stop()

Sources/Containerization/LinuxPod.swift

@@ -679,7 +679,7 @@ extension LinuxPod {
                         container.state = .stopped
 
                         // Clean up file mount temporary directories.
-                        container.fileMountContext.cleanup()
+                        container.fileMountContext.cleanUp()
 
                         state.containers[containerID] = container
                     }

Sources/Containerization/UnixSocketConfiguration.swift

@@ -20,7 +20,8 @@ import SystemPackage
 /// Represents a UnixSocket that can be shared into or out of a container/guest.
 public struct UnixSocketConfiguration: Sendable {
     // TODO: Realistically, we can just hash this struct and use it as the "id".
-    package var id: String {
+    /// The unique identifier for this socket configuration.
+    public var id: String {
         _id
     }
 

Sources/Containerization/UnixSocketRelay.swift

@@ -86,17 +86,11 @@ package final class SocketRelay: Sendable {
     private let state: Mutex<State>
 
     private struct State {
-        var relaySources: [String: ConnectionSources] = [:]
+        var activeRelays: [String: BidirectionalRelay] = [:]
         var t: Task<(), Never>? = nil
         var listener: VsockListener? = nil
     }
 
-    // `DispatchSourceRead` is thread-safe.
-    private struct ConnectionSources: @unchecked Sendable {
-        let hostSource: DispatchSourceRead
-        let guestSource: DispatchSourceRead
-    }
-
     init(
         port: UInt32,
         socket: UnixSocketConfiguration,
@@ -137,7 +131,10 @@ extension SocketRelay {
             }
             t.cancel()
             $0.t = nil
-            $0.relaySources.removeAll()
+            for (_, relay) in $0.activeRelays {
+                relay.stop()
+            }
+            $0.activeRelays.removeAll()
 
             switch configuration.direction {
             case .outOf:
@@ -227,7 +224,7 @@ extension SocketRelay {
     ) async throws {
         do {
             let guestConn = try await vm.dial(port)
-            log?.info(
+            log?.debug(
                 "initiating connection from host to guest",
                 metadata: [
                     "vport": "\(port)",
@@ -256,7 +253,7 @@ extension SocketRelay {
             type: socketType,
             closeOnDeinit: false
         )
-        log?.info(
+        log?.debug(
             "initiating connection from guest to host",
             metadata: [
                 "vport": "\(port)",
@@ -279,210 +276,20 @@ extension SocketRelay {
         hostConn: Socket,
         guestFd: Int32
     ) async throws {
-        // set up the source for host to guest transfers
-        let connSource = DispatchSource.makeReadSource(
-            fileDescriptor: hostConn.fileDescriptor,
-            queue: self.q
-        )
+        let hostFd = hostConn.fileDescriptor
 
-        // set up the source for guest to host transfers
-        let vsockConnectionSource = DispatchSource.makeReadSource(
-            fileDescriptor: guestFd,
-            queue: self.q
+        let relayID = UUID().uuidString
+        let relay = BidirectionalRelay(
+            fd1: hostFd,
+            fd2: guestFd,
+            queue: self.q,
+            log: self.log
         )
 
-        // add the sources to the connection map
-        let pairID = UUID().uuidString
         self.state.withLock {
-            $0.relaySources[pairID] = ConnectionSources(
-                hostSource: connSource,
-                guestSource: vsockConnectionSource
-            )
-        }
-
-        // `buf1` is thread-safe because it is only used when servicing a serial dispatch queue
-        nonisolated(unsafe) let buf1 = UnsafeMutableBufferPointer<UInt8>.allocate(capacity: Int(getpagesize()))
-        connSource.setEventHandler {
-            Self.fdCopyHandler(
-                buffer: buf1,
-                source: connSource,
-                from: hostConn.fileDescriptor,
-                to: guestFd,
-                log: self.log
-            )
-        }
-
-        // `buf2` is thread-safe because it is only used when servicing a serial dispatch queue
-        nonisolated(unsafe) let buf2 = UnsafeMutableBufferPointer<UInt8>.allocate(capacity: Int(getpagesize()))
-        vsockConnectionSource.setEventHandler {
-            Self.fdCopyHandler(
-                buffer: buf2,
-                source: vsockConnectionSource,
-                from: guestFd,
-                to: hostConn.fileDescriptor,
-                log: self.log
-            )
-        }
-
-        connSource.setCancelHandler {
-            self.log?.debug(
-                "host cancel received",
-                metadata: [
-                    "hostFd": "\(hostConn.fileDescriptor)",
-                    "guestFd": "\(guestFd)",
-                ])
-
-            // only close underlying fds when both sources are at EOF
-            // ensure that one of the cancel handlers will see both sources cancelled
-            self.state.withLock { _ in
-                connSource.cancel()
-                if vsockConnectionSource.isCancelled {
-                    self.log?.info(
-                        "close file descriptors",
-                        metadata: [
-                            "hostFd": "\(hostConn.fileDescriptor)",
-                            "guestFd": "\(guestFd)",
-                        ])
-                    try? hostConn.close()
-                    close(guestFd)
-                }
-            }
-        }
-
-        vsockConnectionSource.setCancelHandler {
-            self.log?.debug(
-                "guest cancel received",
-                metadata: [
-                    "hostFd": "\(hostConn.fileDescriptor)",
-                    "guestFd": "\(guestFd)",
-                ])
-
-            // only close underlying fds when both sources are at EOF
-            // ensure that one of the cancel handlers will see both sources cancelled
-            self.state.withLock { _ in
-                vsockConnectionSource.cancel()
-                if connSource.isCancelled {
-                    self.log?.info(
-                        "close file descriptors",
-                        metadata: [
-                            "hostFd": "\(hostConn.fileDescriptor)",
-                            "guestFd": "\(guestFd)",
-                        ])
-                    try? hostConn.close()
-                    close(guestFd)
-                }
-            }
+            $0.activeRelays[relayID] = relay
         }
 
-        connSource.activate()
-        vsockConnectionSource.activate()
-    }
-
-    private static func fdCopyHandler(
-        buffer: UnsafeMutableBufferPointer<UInt8>,
-        source: DispatchSourceRead,
-        from sourceFd: Int32,
-        to destinationFd: Int32,
-        log: Logger? = nil
-    ) {
-        if source.data == 0 {
-            log?.debug(
-                "source EOF",
-                metadata: [
-                    "sourceFd": "\(sourceFd)",
-                    "dstFd": "\(destinationFd)",
-                ])
-            if !source.isCancelled {
-                log?.debug(
-                    "canceling DispatchSourceRead",
-                    metadata: [
-                        "sourceFd": "\(sourceFd)",
-                        "dstFd": "\(destinationFd)",
-                    ])
-                source.cancel()
-                if shutdown(destinationFd, Int32(SHUT_WR)) != 0 {
-                    log?.warning(
-                        "failed to shut down reads",
-                        metadata: [
-                            "errno": "\(errno)",
-                            "sourceFd": "\(sourceFd)",
-                            "dstFd": "\(destinationFd)",
-                        ]
-                    )
-                }
-            }
-            return
-        }
-
-        do {
-            log?.trace(
-                "source copy",
-                metadata: [
-                    "sourceFd": "\(sourceFd)",
-                    "dstFd": "\(destinationFd)",
-                    "size": "\(source.data)",
-                ])
-            try self.fileDescriptorCopy(
-                buffer: buffer,
-                size: source.data,
-                from: sourceFd,
-                to: destinationFd
-            )
-        } catch {
-            log?.error("file descriptor copy failed \(error)")
-            if !source.isCancelled {
-                source.cancel()
-                if shutdown(destinationFd, Int32(SHUT_RDWR)) != 0 {
-                    log?.warning(
-                        "failed to shut down destination after I/O error",
-                        metadata: [
-                            "errno": "\(errno)",
-                            "sourceFd": "\(sourceFd)",
-                            "dstFd": "\(destinationFd)",
-                        ]
-                    )
-                }
-            }
-        }
-    }
-
-    private static func fileDescriptorCopy(
-        buffer: UnsafeMutableBufferPointer<UInt8>,
-        size: UInt,
-        from sourceFd: Int32,
-        to destinationFd: Int32
-    ) throws {
-        let bufferSize = buffer.count
-        var readBytesRemaining = min(Int(size), bufferSize)
-
-        guard let baseAddr = buffer.baseAddress else {
-            throw ContainerizationError(
-                .invalidState,
-                message: "buffer has no base address"
-            )
-        }
-
-        while readBytesRemaining > 0 {
-            let readResult = read(sourceFd, baseAddr, min(bufferSize, readBytesRemaining))
-            if readResult <= 0 {
-                throw ContainerizationError(
-                    .internalError,
-                    message: "missing pointer base address"
-                )
-            }
-            readBytesRemaining -= readResult
-
-            var writeBytesRemaining = readResult
-            while writeBytesRemaining > 0 {
-                let writeResult = write(destinationFd, baseAddr, writeBytesRemaining)
-                if writeResult <= 0 {
-                    throw ContainerizationError(
-                        .internalError,
-                        message: "zero byte write or error in socket relay: fd \(destinationFd), result \(writeResult)"
-                    )
-                }
-                writeBytesRemaining -= writeResult
-            }
-        }
+        relay.start()
     }
 }