make FilesystemEventWorker properly sendable and fix kernel panic when vminitd runs as PID 1

Author: realrajaryan

vminitd/Sources/vminitd/Application.swift

@@ -48,7 +48,8 @@ struct Application: AsyncParsableCommand {
     static let standardErrorLock = NSLock()
 
     static func runInForeground(_ log: Logger) throws {
-        log.info("running vminitd under pid1")
+        precondition(getpid() != 1, "runInForeground must not be called as PID 1")
+        log.info("running vminitd under pid1 wrapper")
 
         var command = Command("/sbin/vminitd")
         command.attrs = .init(setsid: true)
@@ -111,17 +112,21 @@ extension Application {
             #if DEBUG
             let environment = ProcessInfo.processInfo.environment
             let foreground = environment[Application.foregroundEnvVar]
-            log.info("checking for shim var \(Application.foregroundEnvVar)=\(String(describing: foreground))")
+            let isPid1 = (getpid() == 1)
+            log.info("checking for shim var \(Application.foregroundEnvVar)=\(String(describing: foreground)); pid=\(getpid())")
 
-            if foreground == nil {
+            // only use the FOREGROUND shim when we're not PID 1
+            // if we are PID 1 (fresh VM boot), skip the shim to avoid exiting init
+            if foreground == nil && !isPid1 {
                 try Application.runInForeground(log)
-                Application.exit(0)
+                Application.exit(0)  // parent is not PID 1; safe to exit after child completes
             }
 
-            // since we are not running as pid1 in this mode we must set ourselves
-            // as a subpreaper so that all child processes are reaped by us and not
-            // passed onto our parent.
-            CZ_set_sub_reaper()
+            // we only need to be a subreaper when we're not PID 1
+            // (when PID 1, the kernel already reaps children)
+            if !isPid1 {
+                CZ_set_sub_reaper()
+            }
             #endif
 
             signal(SIGPIPE, SIG_IGN)

vminitd/Sources/vminitd/FilesystemEventWorker.swift

@@ -28,28 +28,38 @@ import Musl
 import Glibc
 #endif
 
-final class FilesystemEventWorker: @unchecked Sendable {
+final class FilesystemEventWorker: Sendable {
     private static let handshakeReady: UInt8 = 0xAA
     private static let handshakeFailure: UInt8 = 0xFF
 
     private let containerID: String
     private let containerPID: Int32
-    private var childPID: Int32?
-    private var parentSocket: Int32?
-    private var channel: Channel?
     private let eventLoop: EventLoop
-    private var eventIDCounter: UInt32 = 0
-    private let pendingEvents: Mutex<[UInt32: CheckedContinuation<Void, Error>]> = Mutex([:])
     private let shouldStop: Atomic<Bool> = Atomic(false)
 
+    // Cross-thread state (accessed from any thread)
+    private struct State {
+        var childPID: Int32?
+        var parentSocket: Int32?
+    }
+    private let state: Mutex<State> = Mutex(State(childPID: nil, parentSocket: nil))
+
+    // Event-loop confined state (only accessed on channel.eventLoop)
+    private final class ELState: @unchecked Sendable {
+        var channel: Channel?
+        var eventIDCounter: UInt32 = 0
+        var pendingEvents: [UInt32: CheckedContinuation<Void, Error>] = [:]
+    }
+    private let elState = ELState()
+
     init(containerID: String, containerPID: Int32, eventLoop: EventLoop) {
         self.containerID = containerID
         self.containerPID = containerPID
         self.eventLoop = eventLoop
     }
 
     func start() throws {
-        guard childPID == nil else {
+        guard state.withLock({ $0.childPID }) == nil else {
             throw ContainerizationError(.invalidState, message: "FilesystemEventWorker already started")
         }
 
@@ -94,25 +104,27 @@ final class FilesystemEventWorker: @unchecked Sendable {
         let pid = command.pid
         close(childSocket)
         close(errorWriteFD)  // Close write end in parent
-        self.childPID = pid
-        self.parentSocket = parentSocket
+        state.withLock {
+            $0.childPID = pid
+            $0.parentSocket = parentSocket
+        }
 
         var handshake: UInt8 = 0
         let readResult = read(parentSocket, &handshake, 1)
 
         if readResult != 1 {
             close(parentSocket)
-            self.parentSocket = nil
+            state.withLock { $0.parentSocket = nil }
             close(errorReadFD)
             var status: Int32 = 0
             waitpid(pid, &status, 0)
-            self.childPID = nil
+            state.withLock { $0.childPID = nil }
             throw ContainerizationError(.internalError, message: "Child process failed to start")
         }
 
         if handshake == Self.handshakeFailure {
             close(parentSocket)
-            self.parentSocket = nil
+            state.withLock { $0.parentSocket = nil }
 
             // Read error message from child
             var errorBuffer = [UInt8](repeating: 0, count: 1024)
@@ -121,7 +133,7 @@ final class FilesystemEventWorker: @unchecked Sendable {
 
             var status: Int32 = 0
             waitpid(pid, &status, 0)
-            self.childPID = nil
+            state.withLock { $0.childPID = nil }
 
             let errorMsg =
                 bytesRead > 0
@@ -132,11 +144,11 @@ final class FilesystemEventWorker: @unchecked Sendable {
 
         if handshake != Self.handshakeReady {
             close(parentSocket)
-            self.parentSocket = nil
+            state.withLock { $0.parentSocket = nil }
             close(errorReadFD)
             var status: Int32 = 0
             waitpid(pid, &status, 0)
-            self.childPID = nil
+            state.withLock { $0.childPID = nil }
             throw ContainerizationError(.internalError, message: "Child process sent unexpected handshake: \(handshake)")
         }
 
@@ -149,68 +161,73 @@ final class FilesystemEventWorker: @unchecked Sendable {
                     let handler = ResponseHandler(worker: self)
                     return channel.pipeline.addHandler(handler)
                 }
-            self.channel = try bootstrap.takingOwnershipOfDescriptor(inputOutput: parentSocket).wait()
+            self.elState.channel = try bootstrap.takingOwnershipOfDescriptor(inputOutput: parentSocket).wait()
         } catch {
             close(parentSocket)
-            self.parentSocket = nil
+            state.withLock { $0.parentSocket = nil }
             var status: Int32 = 0
             waitpid(pid, &status, 0)
-            self.childPID = nil
+            state.withLock { $0.childPID = nil }
             throw ContainerizationError(.internalError, message: "Failed to setup NIO channel: \(error)")
         }
     }
 
     func enqueueEvent(path: String, eventType: Com_Apple_Containerization_Sandbox_V3_FileSystemEventType) async throws {
-        guard let socket = parentSocket, !shouldStop.load(ordering: .relaxed) else {
-            throw ContainerizationError(.invalidState, message: "FilesystemEventWorker not running")
+        let socket = try state.withLock { state throws -> Int32 in
+            guard let socket = state.parentSocket, !shouldStop.load(ordering: .relaxed) else {
+                throw ContainerizationError(.invalidState, message: "FilesystemEventWorker not running")
+            }
+            return socket
         }
 
-        let eventID = eventIDCounter
-        eventIDCounter += 1
-
+        // Use continuation to bridge between async and event loop
         try await withCheckedThrowingContinuation { (continuation: CheckedContinuation<Void, Error>) in
-            pendingEvents.withLock { events in
-                events[eventID] = continuation
-            }
-
-            do {
-                try sendEventToChild(socket: socket, eventID: eventID, path: path, eventType: eventType)
-            } catch {
-                _ = pendingEvents.withLock { events in
-                    events.removeValue(forKey: eventID)
+            // Hop to event loop to access event ID counter and store continuation
+            eventLoop.execute { [elState] in
+                let eventID = elState.eventIDCounter
+                elState.eventIDCounter += 1
+                elState.pendingEvents[eventID] = continuation
+
+                do {
+                    try self.sendEventToChild(socket: socket, eventID: eventID, path: path, eventType: eventType)
+                } catch {
+                    // Remove continuation and resume with error on send failure
+                    _ = elState.pendingEvents.removeValue(forKey: eventID)
+                    continuation.resume(throwing: error)
                 }
-                continuation.resume(throwing: error)
             }
         }
     }
 
     func stop() {
         shouldStop.store(true, ordering: .relaxed)
 
-        if let channel = self.channel {
-            try? channel.close().wait()
-            self.channel = nil
+        // Close channel and clean up pending events on event loop
+        eventLoop.execute { [elState] in
+            elState.channel?.close(promise: nil)
+            elState.channel = nil
+
+            for (_, continuation) in elState.pendingEvents {
+                continuation.resume(throwing: ContainerizationError(.cancelled, message: "FilesystemEventWorker stopped"))
+            }
+            elState.pendingEvents.removeAll()
         }
 
-        self.parentSocket = nil
+        // Kill child process
+        state.withLock { state in
+            state.parentSocket = nil
 
-        if let pid = childPID {
-            #if canImport(Musl)
-            Musl.kill(pid, SIGTERM)
-            #elseif canImport(Glibc)
-            Glibc.kill(pid, SIGTERM)
-            #endif
+            if let pid = state.childPID {
+                #if canImport(Musl)
+                Musl.kill(pid, SIGTERM)
+                #elseif canImport(Glibc)
+                Glibc.kill(pid, SIGTERM)
+                #endif
 
-            var status: Int32 = 0
-            waitpid(pid, &status, 0)
-            childPID = nil
-        }
-
-        pendingEvents.withLock { events in
-            for (_, continuation) in events {
-                continuation.resume(throwing: ContainerizationError(.cancelled, message: "FilesystemEventWorker stopped"))
+                var status: Int32 = 0
+                waitpid(pid, &status, 0)
+                state.childPID = nil
             }
-            events.removeAll()
         }
     }
 
@@ -254,25 +271,23 @@ final class FilesystemEventWorker: @unchecked Sendable {
                     break
                 }
 
-                worker.pendingEvents.withLock { events in
-                    if let continuation = events.removeValue(forKey: eventID) {
-                        if success == 1 {
-                            continuation.resume()
-                        } else {
-                            continuation.resume(throwing: ContainerizationError(.internalError, message: "Child process failed to process filesystem event"))
-                        }
+                // ResponseHandler runs on event loop, so can access elState.pendingEvents directly
+                if let continuation = worker.elState.pendingEvents.removeValue(forKey: eventID) {
+                    if success == 1 {
+                        continuation.resume()
+                    } else {
+                        continuation.resume(throwing: ContainerizationError(.internalError, message: "Child process failed to process filesystem event"))
                     }
                 }
             }
         }
 
         func errorCaught(context: ChannelHandlerContext, error: Error) {
-            worker.pendingEvents.withLock { events in
-                for (_, continuation) in events {
-                    continuation.resume(throwing: error)
-                }
-                events.removeAll()
+            // ResponseHandler runs on event loop, so can access elState.pendingEvents directly
+            for (_, continuation) in worker.elState.pendingEvents {
+                continuation.resume(throwing: error)
             }
+            worker.elState.pendingEvents.removeAll()
         }
     }
 }