vminitd: Turn ProcessSupervisor into a class (#433)

Really no reason for this to be an actor, and we're seeing runtime
errors seemingly because of it.

vminitd/Sources/vminitd/IOPair.swift

@@ -30,7 +30,6 @@ final class IOPair: Sendable {
         let to: IOCloser
         let buffer: UnsafeMutableBufferPointer<UInt8>
         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)
         }
     }
 }

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"

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

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",