diff --git a/Sources/Containerization/LinuxContainer.swift b/Sources/Containerization/LinuxContainer.swift index 7ad21cf5..b8a32b1f 100644 --- a/Sources/Containerization/LinuxContainer.swift +++ b/Sources/Containerization/LinuxContainer.swift @@ -316,7 +316,7 @@ extension LinuxContainer { try await vm.start() do { - let relayManager = UnixSocketRelayManager(vm: vm) + let relayManager = UnixSocketRelayManager(vm: vm, log: self.logger) try await vm.withAgent { agent in try await agent.standardSetup() diff --git a/Sources/Containerization/UnixSocketRelay.swift b/Sources/Containerization/UnixSocketRelay.swift index d59bac61..3f3184b3 100644 --- a/Sources/Containerization/UnixSocketRelay.swift +++ b/Sources/Containerization/UnixSocketRelay.swift @@ -159,6 +159,12 @@ extension SocketRelay { let hostSocket = try Socket(type: socketType) try hostSocket.listen() + log?.info( + "listening on host UDS", + metadata: [ + "path": "\(hostConn.path)", + "vport": "\(self.port)", + ]) let connectionStream = try hostSocket.acceptStream(closeOnDeinit: false) self.state.withLock { $0.t = Task { @@ -185,6 +191,12 @@ extension SocketRelay { let log = self.log let connectionStream = try self.vm.listen(self.port) + log?.info( + "listening on guest vsock", + metadata: [ + "path": "\(hostPath)", + "vport": "\(port)", + ]) self.state.withLock { $0.t = Task { do { @@ -212,6 +224,13 @@ extension SocketRelay { ) async throws { do { let guestConn = try await vm.dial(port) + log?.info( + "initiating connection from host to guest", + metadata: [ + "vport": "\(port)", + "hostFd": "\(guestConn.fileDescriptor)", + "guestFd": "\(hostConn.fileDescriptor)", + ]) try await self.relay( hostConn: hostConn, guestFd: guestConn.fileDescriptor @@ -234,6 +253,13 @@ extension SocketRelay { type: socketType, closeOnDeinit: false ) + log?.info( + "initiating connection from host to guest", + metadata: [ + "vport": "\(port)", + "hostFd": "\(hostSocket.fileDescriptor)", + "guestFd": "\(vsockConn.fileDescriptor)", + ]) try hostSocket.connect() do { @@ -250,15 +276,19 @@ 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 ) + + // set up the source for guest to host transfers let vsockConnectionSource = DispatchSource.makeReadSource( fileDescriptor: guestFd, queue: self.q ) + // add the sources to the connection map let pairID = UUID().uuidString self.state.withLock { $0.relaySources[pairID] = ConnectionSources( @@ -267,46 +297,72 @@ extension SocketRelay { ) } - // `buf1` isn't used concurrently. + // `buf1` is thread-safe because it is only used when servicing a serial dispatch queue nonisolated(unsafe) let buf1 = UnsafeMutableBufferPointer.allocate(capacity: Int(getpagesize())) connSource.setEventHandler { Self.fdCopyHandler( buffer: buf1, source: connSource, from: hostConn.fileDescriptor, - to: guestFd + 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.allocate(capacity: Int(getpagesize())) - // `buf2` isn't used concurrently. vsockConnectionSource.setEventHandler { Self.fdCopyHandler( buffer: buf2, source: vsockConnectionSource, from: guestFd, - to: hostConn.fileDescriptor + to: hostConn.fileDescriptor, + log: self.log ) } connSource.setCancelHandler { - if !connSource.isCancelled { + self.log?.info( + "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 { + try? hostConn.close() + close(guestFd) + } } - if !vsockConnectionSource.isCancelled { - vsockConnectionSource.cancel() - } - try? hostConn.close() } vsockConnectionSource.setCancelHandler { - if !vsockConnectionSource.isCancelled { + self.log?.info( + "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) + } } - if !connSource.isCancelled { - connSource.cancel() - } - close(guestFd) } connSource.activate() @@ -321,13 +377,42 @@ extension SocketRelay { log: Logger? = nil ) { if source.data == 0 { + log?.info( + "source EOF", + metadata: [ + "sourceFd": "\(sourceFd)", + "dstFd": "\(destinationFd)", + ]) if !source.isCancelled { + log?.info( + "canceling DispatchSourceRead", + metadata: [ + "sourceFd": "\(sourceFd)", + "dstFd": "\(destinationFd)", + ]) source.cancel() + if shutdown(destinationFd, SHUT_WR) != 0 { + log?.info( + "failed to shut down reads", + metadata: [ + "errno": "\(errno)", + "sourceFd": "\(sourceFd)", + "dstFd": "\(destinationFd)", + ] + ) + } } return } do { + log?.debug( + "source copy", + metadata: [ + "sourceFd": "\(sourceFd)", + "dstFd": "\(destinationFd)", + "size": "\(source.data)", + ]) try self.fileDescriptorCopy( buffer: buffer, size: source.data, @@ -338,6 +423,16 @@ extension SocketRelay { log?.error("file descriptor copy failed \(error)") if !source.isCancelled { source.cancel() + if shutdown(destinationFd, SHUT_RDWR) != 0 { + log?.info( + "failed to shut down destination", + metadata: [ + "errno": "\(errno)", + "sourceFd": "\(sourceFd)", + "dstFd": "\(destinationFd)", + ] + ) + } } } } @@ -374,7 +469,7 @@ extension SocketRelay { if writeResult <= 0 { throw ContainerizationError( .internalError, - message: "zero byte write or error in socket relay" + message: "zero byte write or error in socket relay: fd \(destinationFd), result \(writeResult)" ) } writeBytesRemaining -= writeResult diff --git a/Sources/ContainerizationOS/Linux/Epoll.swift b/Sources/ContainerizationOS/Linux/Epoll.swift index 60fd9126..9a54ced9 100644 --- a/Sources/ContainerizationOS/Linux/Epoll.swift +++ b/Sources/ContainerizationOS/Linux/Epoll.swift @@ -169,7 +169,11 @@ public final class Epoll: Sendable { extension Epoll.Mask { public var isHangup: Bool { - (self & (EPOLLHUP | EPOLLERR | EPOLLRDHUP)) != 0 + (self & (EPOLLHUP | EPOLLERR)) != 0 + } + + public var isRhangup: Bool { + (self & EPOLLRDHUP) != 0 } public var readyToRead: Bool { diff --git a/vminitd/Sources/vminitd/VsockProxy.swift b/vminitd/Sources/vminitd/VsockProxy.swift index a0722a36..a2fb92b5 100644 --- a/vminitd/Sources/vminitd/VsockProxy.swift +++ b/vminitd/Sources/vminitd/VsockProxy.swift @@ -63,6 +63,13 @@ extension VsockProxy { return } + log?.info( + "stopping proxy", + metadata: [ + "vport": "\(port)", + "uds": "\(path)", + "action": "\(action)", + ]) try listener.close() let fm = FileManager.default if fm.fileExists(atPath: self.path.path) { @@ -76,6 +83,14 @@ extension VsockProxy { guard listener == nil else { return } + + log?.info( + "starting proxy", + metadata: [ + "vport": "\(port)", + "uds": "\(path)", + "action": "\(action)", + ]) switch self.action { case .dial: try dialHost() @@ -127,6 +142,14 @@ extension VsockProxy { do { for try await conn in stream { Task { + log?.info( + "accepting connection", + metadata: [ + "vport": "\(port)", + "uds": "\(path)", + "action": "\(action)", + "socketType": "\(socketType)", + ]) do { try await handleConn( conn: conn, @@ -175,10 +198,31 @@ extension VsockProxy { // `clientFile` isn't used concurrently. nonisolated(unsafe) var clientFile = OSFile.SpliceFile(fd: conn.fileDescriptor) + nonisolated(unsafe) var eofFromClient = false // `serverFile` isn't used concurrently. nonisolated(unsafe) var serverFile = OSFile.SpliceFile(fd: relayTo.fileDescriptor) + nonisolated(unsafe) var eofFromServer = false + + // clean up when any of these conditions apply: + // - the client has completely hung up or errored + // - the server has completely hung up or errored + // - both the client and server have half closed via: + // - read hangup on epoll + // - EOF on splice + let cleanup = { @Sendable [log, port, path, action] in + log?.info( + "cleaning up", + metadata: [ + "vport": "\(port)", + "uds": "\(path)", + "action": "\(action)", + "eofFromClient": "\(eofFromClient)", + "eofFromServer": "\(eofFromServer)", + "clientFd": "\(clientFile.fileDescriptor)", + "serverFd": "\(serverFile.fileDescriptor)", + ] + ) - let cleanup = { @Sendable in do { try ProcessSupervisor.default.poller.delete(clientFile.fileDescriptor) try ProcessSupervisor.default.poller.delete(serverFile.fileDescriptor) @@ -191,57 +235,105 @@ extension VsockProxy { } try! ProcessSupervisor.default.poller.add(clientFile.fileDescriptor, mask: EPOLLIN | EPOLLOUT) { mask in - if mask.readyToRead { - do { - let (_, _, action) = try OSFile.splice(from: &clientFile, to: &serverFile) - if action == .eof || action == .brokenPipe { - return cleanup() - } - } catch { - return cleanup() - } + if mask.readyToRead && !eofFromClient { + let (fromEof, toEof) = Self.transferData( + fromFile: &clientFile, + toFile: &serverFile, + description: "readyToRead:toServer", + log: self.log + ) + eofFromClient = eofFromClient || fromEof + eofFromServer = eofFromServer || toEof } - if mask.readyToWrite { - do { - let (_, _, action) = try OSFile.splice(from: &serverFile, to: &clientFile) - if action == .eof || action == .brokenPipe { - return cleanup() - } - } catch { - return cleanup() - } + if mask.readyToWrite && !eofFromServer { + let (fromEof, toEof) = Self.transferData( + fromFile: &serverFile, + toFile: &clientFile, + description: "readyToWrite:toClient", + log: self.log + ) + eofFromClient = eofFromClient || toEof + eofFromServer = eofFromServer || fromEof } if mask.isHangup { + eofFromClient = true + eofFromServer = true + } else if mask.isRhangup && !eofFromClient { + // half close, shut down client to server transfer + // we should see no more EPOLLIN events on the client fd + // and no more EPOLLOUT events on the server fd + eofFromClient = true + if shutdown(serverFile.fileDescriptor, SHUT_WR) != 0 { + self.log?.info( + "failed to shut down client reads", + metadata: [ + "vport": "\(self.port)", + "uds": "\(self.path)", + "errno": "\(errno)", + "eofFromClient": "\(eofFromClient)", + "eofFromServer": "\(eofFromServer)", + "clientFd": "\(clientFile.fileDescriptor)", + "serverFd": "\(serverFile.fileDescriptor)", + ] + ) + } + } + + if eofFromClient && eofFromServer { return cleanup() } } try! ProcessSupervisor.default.poller.add(serverFile.fileDescriptor, mask: EPOLLIN | EPOLLOUT) { mask in - if mask.readyToRead { - do { - let (_, _, action) = try OSFile.splice(from: &serverFile, to: &clientFile) - if action == .eof || action == .brokenPipe { - return cleanup() - } - } catch { - return cleanup() - } + if mask.readyToRead && !eofFromServer { + let (fromEof, toEof) = Self.transferData( + fromFile: &serverFile, + toFile: &clientFile, + description: "readyToRead:toClient", + log: self.log + ) + eofFromClient = eofFromClient || toEof + eofFromServer = eofFromServer || fromEof } - if mask.readyToWrite { - do { - let (_, _, action) = try OSFile.splice(from: &clientFile, to: &serverFile) - if action == .eof || action == .brokenPipe { - return cleanup() - } - } catch { - return cleanup() - } + if mask.readyToWrite && !eofFromClient { + let (fromEof, toEof) = Self.transferData( + fromFile: &clientFile, + toFile: &serverFile, + description: "readyToWrite:toServer", + log: self.log + ) + eofFromClient = eofFromClient || fromEof + eofFromServer = eofFromServer || toEof } if mask.isHangup { + eofFromClient = true + eofFromServer = true + } else if mask.isRhangup && !eofFromServer { + // half close, shut down server to client transfer + // we should see no more EPOLLIN events on the server fd + // and no more EPOLLOUT events on the client fd + eofFromServer = true + if shutdown(clientFile.fileDescriptor, SHUT_WR) != 0 { + self.log?.info( + "failed to shut down server reads", + metadata: [ + "vport": "\(self.port)", + "uds": "\(self.path)", + "errno": "\(errno)", + "eofFromClient": "\(eofFromClient)", + "eofFromServer": "\(eofFromServer)", + "clientFd": "\(clientFile.fileDescriptor)", + "serverFd": "\(serverFile.fileDescriptor)", + ] + ) + } + } + + if eofFromClient && eofFromServer { return cleanup() } } @@ -250,4 +342,51 @@ extension VsockProxy { } } } + + private static func transferData( + fromFile: inout OSFile.SpliceFile, + toFile: inout OSFile.SpliceFile, + description: String, + log: Logger? + ) -> (Bool, Bool) { + do { + let (readBytes, writeBytes, action) = try OSFile.splice(from: &fromFile, to: &toFile) + log?.debug( + "transferred data", + metadata: [ + "description": "\(description)", + "action": "\(action)", + "readBytes": "\(readBytes)", + "writeBytes": "\(writeBytes)", + "fromFd": "\(fromFile.fileDescriptor)", + "toFd": "\(toFile.fileDescriptor)", + ] + ) + if action == .eof { + // half close, shut down client to server transfer + // we should see no more EPOLLIN events on the client fd + // and no more EPOLLOUT events on the server fd + if shutdown(toFile.fileDescriptor, SHUT_WR) != 0 { + log?.info( + "failed to shut down reads", + metadata: [ + "description": "\(description)", + "errno": "\(errno)", + "action": "\(action)", + "readBytes": "\(readBytes)", + "writeBytes": "\(writeBytes)", + "fromFd": "\(fromFile.fileDescriptor)", + "toFd": "\(toFile.fileDescriptor)", + ] + ) + } + return (true, false) + } else if action == .brokenPipe { + return (true, true) + } + return (false, false) + } catch { + return (true, true) + } + } }