diff --git a/vminitd/Sources/vminitd/IOPair.swift b/vminitd/Sources/vminitd/IOPair.swift index b3dacdcb..5749fa90 100644 --- a/vminitd/Sources/vminitd/IOPair.swift +++ b/vminitd/Sources/vminitd/IOPair.swift @@ -30,7 +30,6 @@ final class IOPair: Sendable { let to: IOCloser let buffer: UnsafeMutableBufferPointer var closed: Bool - let logger: Logger? func drain() { let readFrom = OSFile(fd: from.fileDescriptor) @@ -59,7 +58,7 @@ final class IOPair: Sendable { } } - mutating func close() { + mutating func close(logger: Logger?) { if self.closed { return } @@ -72,19 +71,19 @@ final class IOPair: Sendable { do { try ProcessSupervisor.default.poller.delete(readFromFd) } catch { - self.logger?.error("failed to delete fd from epoll \(readFromFd): \(error)") + logger?.error("failed to delete fd from epoll \(readFromFd): \(error)") } do { try self.from.close() } catch { - self.logger?.error("failed to close reader fd for IOPair: \(error)") + logger?.error("failed to close reader fd for IOPair: \(error)") } do { try self.to.close() } catch { - self.logger?.error("failed to close writer fd for IOPair: \(error)") + logger?.error("failed to close writer fd for IOPair: \(error)") } self.buffer.deallocate() self.closed = true @@ -103,8 +102,7 @@ final class IOPair: Sendable { from: readFrom, to: writeTo, buffer: buffer, - closed: false, - logger: logger + closed: false )) self.reason = reason self.logger = logger @@ -129,7 +127,7 @@ final class IOPair: Sendable { if mask.isHangup && !mask.readyToRead { self.logger?.debug("received EPOLLHUP with no EPOLLIN") if !ignoreHup { - io.close() + io.close(logger: self.logger) } return } @@ -146,7 +144,7 @@ final class IOPair: Sendable { let w = writeTo.write(view) if w.wrote != r.read { self.logger?.error("stopping relay: short write for stdio") - io.close() + io.close(logger: self.logger) return } } @@ -157,7 +155,7 @@ final class IOPair: Sendable { fallthrough case .eof: self.logger?.debug("closing relay for \(readFromFd)") - io.close() + io.close(logger: self.logger) return case .again: if mask.isHangup && !ignoreHup { @@ -176,7 +174,7 @@ final class IOPair: Sendable { func close() { self.io.withLock { io in self.logger?.info("closing relay for \(reason)") - io.close() + io.close(logger: self.logger) } } } diff --git a/vminitd/Sources/vminitd/ManagedContainer.swift b/vminitd/Sources/vminitd/ManagedContainer.swift index 8d1d91ae..d60973cb 100644 --- a/vminitd/Sources/vminitd/ManagedContainer.swift +++ b/vminitd/Sources/vminitd/ManagedContainer.swift @@ -68,7 +68,7 @@ actor ManagedContainer { if let runtimePath = ociRuntimePath { // Use runc runtime - let runc = await ProcessSupervisor.default.getRuncWithReaper( + let runc = ProcessSupervisor.default.getRuncWithReaper( Runc( command: runtimePath, root: "/run/runc" diff --git a/vminitd/Sources/vminitd/ProcessSupervisor.swift b/vminitd/Sources/vminitd/ProcessSupervisor.swift index 0b43d4ac..fe8e0d7f 100644 --- a/vminitd/Sources/vminitd/ProcessSupervisor.swift +++ b/vminitd/Sources/vminitd/ProcessSupervisor.swift @@ -17,20 +17,25 @@ import ContainerizationOS import Foundation import Logging +import Synchronization -actor ProcessSupervisor { +final class ProcessSupervisor: Sendable { let poller: Epoll private let queue: DispatchQueue // `DispatchSourceSignal` is thread-safe. private nonisolated(unsafe) let source: DispatchSourceSignal - private var processes = [any ContainerProcess]() + + private struct State { + var processes: [any ContainerProcess] = [] + var log: Logger? + } + + private let state: Mutex private let reaperCommandRunner = ReaperCommandRunner() - var log: Logger? - func setLog(_ log: Logger?) { - self.log = log + self.state.withLock { $0.log = log } } static let `default` = ProcessSupervisor() @@ -40,6 +45,7 @@ actor ProcessSupervisor { self.source = DispatchSource.makeSignalSource(signal: SIGCHLD, queue: queue) self.queue = queue self.poller = try! Epoll() + self.state = Mutex(State()) let t = Thread { try! self.poller.run() } @@ -48,64 +54,62 @@ actor ProcessSupervisor { func ready() { self.source.setEventHandler { - do { - self.log?.debug("received SIGCHLD, reaping processes") - try self.handleSignal() - } catch { - self.log?.error("reaping processes failed", metadata: ["error": "\(error)"]) - } + self.handleSignal() } self.source.resume() } - private func handleSignal() throws { + private func handleSignal() { dispatchPrecondition(condition: .onQueue(queue)) - self.log?.debug("starting to wait4 processes") let exited = Reaper.reap() - self.log?.debug("finished wait4 of \(exited.count) processes") for (pid, status) in exited { reaperCommandRunner.notifyExit(pid: pid, status: status) } - self.log?.debug("checking for exit of managed process", metadata: ["exits": "\(exited)", "processes": "\(processes.count)"]) - let exitedProcesses = self.processes.filter { proc in - exited.contains { pid, _ in - proc.pid == pid - } - } + self.state.withLock { state in + state.log?.debug("received SIGCHLD, reaping processes") + state.log?.debug("finished wait4 of \(exited.count) processes") + state.log?.debug("checking for exit of managed process", metadata: ["exits": "\(exited)", "processes": "\(state.processes.count)"]) - for proc in exitedProcesses { - guard let pid = proc.pid else { - continue + let exitedProcesses = state.processes.filter { proc in + exited.contains { pid, _ in + proc.pid == pid + } } - if let status = exited[pid] { - self.log?.debug( - "managed process exited", - metadata: [ - "pid": "\(pid)", - "status": "\(status)", - "count": "\(processes.count - 1)", - ]) - proc.setExit(status) - self.processes.removeAll(where: { $0.pid == pid }) + for proc in exitedProcesses { + guard let pid = proc.pid else { + continue + } + + if let status = exited[pid] { + state.log?.debug( + "managed process exited", + metadata: [ + "pid": "\(pid)", + "status": "\(status)", + "count": "\(state.processes.count - 1)", + ]) + proc.setExit(status) + state.processes.removeAll(where: { $0.pid == pid }) + } } } } func start(process: any ContainerProcess) async throws -> Int32 { - self.log?.debug("in supervisor lock to start process") - defer { - self.log?.debug("out of supervisor lock to start process") + self.state.withLock { state in + state.log?.debug("in supervisor lock to start process") + state.processes.append(process) } - do { - self.processes.append(process) return try await process.start() } catch { - self.log?.error("process start failed \(error)", metadata: ["process-id": "\(process.id)"]) + self.state.withLock { state in + state.processes.removeAll(where: { $0.id == process.id }) + } throw error } } @@ -118,7 +122,7 @@ actor ProcessSupervisor { } deinit { - self.log?.info("process supervisor deinit") source.cancel() + try? poller.shutdown() } } diff --git a/vminitd/Sources/vminitd/Server.swift b/vminitd/Sources/vminitd/Server.swift index 59fb57d8..7da5ee45 100644 --- a/vminitd/Sources/vminitd/Server.swift +++ b/vminitd/Sources/vminitd/Server.swift @@ -90,8 +90,8 @@ final class Initd: Sendable { try await withThrowingTaskGroup(of: Void.self) { group in log.debug("starting process supervisor") - await ProcessSupervisor.default.setLog(self.log) - await ProcessSupervisor.default.ready() + ProcessSupervisor.default.setLog(self.log) + ProcessSupervisor.default.ready() log.info( "booting gRPC server on vsock",