fix: 🥵 100% CPU usage in reconnectd when changing port settings (#474)

This one is painfully nuanced and boils down to how I was managing ncp
thread lifecycles. It turns out (perhaps unsurprisingly) that
mixing-and-matching `pthread` and `Thread` APIs doesn't go smoothly, and
`pthread_join` was not working as expected with `Thread` instances. This
meant that I was failing to wait for ncp threads to terminate before
starting new ones which, coupled with some lingering global state in
plptools (in multi-device scenarios) meant that reconnectd could get
into an active spin trying to use an unavailable TCP port.

Almost all of the work to fix this has been done in plptools see
jbmorley/plptools#1), moving the ncp lifecycle
management into the C/C++ code to ensure it uses a singe consistent API.

Author: jbmorley

macos/Reconnect/Model/DeviceModel.swift

@@ -241,8 +241,10 @@ class DeviceModel: Identifiable, Equatable, @unchecked Sendable {
                 }
             } catch {
                 print("Failed to get owner info with error \(error).")
-                let alert = NSAlert(error: error)
-                alert.runModal()
+                DispatchQueue.main.async {
+                    let alert = NSAlert(error: error)
+                    alert.runModal()
+                }
             }
         }
     }

macos/ReconnectCore/Package.resolved

@@ -28,7 +28,7 @@ let package = Package(
             dependencies: [
                 .product(name: "Diligence", package: "diligence"),
                 .product(name: "OpoLuaCore", package: "opolua"),
-                .product(name: "ncp", package: "plptools"),
+                .product(name: "plptools", package: "plptools"),
             ],
             resources: [
                 .process("Resources"),

macos/ReconnectCore/Package.swift

@@ -18,7 +18,7 @@
 
 import Foundation
 
-import ncp
+import plptools
 
 extension PLPToolsError {
 

macos/ReconnectCore/Sources/ReconnectCore/Extensions/rfsv.errs.swift

@@ -18,7 +18,7 @@
 
 import Foundation
 
-import ncp
+import plptools
 
 public enum ReconnectError: Error {
     case unknown

macos/ReconnectCore/Sources/ReconnectCore/Model/ReconnectError.swift

@@ -18,7 +18,7 @@
 
 import Foundation
 
-import ncp
+import plptools
 import OpoLuaCore
 
 // Thread-safe FileServer implementation.

macos/ReconnectCore/Sources/ReconnectCore/PLP/FileServer.swift

@@ -19,7 +19,7 @@
 import os
 import Foundation
 
-import ncp
+import plptools
 
 // Callbacks on main.
 public protocol NCPDelegate: NSObject {
@@ -29,6 +29,7 @@ public protocol NCPDelegate: NSObject {
 }
 
 // TODO: NCPSession?
+
 public class NCP {
 
     public struct DeviceConfiguration: Equatable, Hashable {
@@ -48,77 +49,54 @@ public class NCP {
     public let device: DeviceConfiguration
     public let port: Int32
 
-    private let lock = NSLock()
     private let logger = Logger(subsystem: "PLP", category: "Server")
+    private var session: UnsafeMutableRawPointer?
 
-    // Synchronized with lock.
-    private var threadID: pthread_t? = nil
-    private var isCancelled: Bool = false
-
-    func threadEntryPoint() {
-
-        setup_signal_handlers()
+    private var callback: NCPStatusCallback?
 
-        lock.withLock {
-            threadID = pthread_self()
-        }
+    public func start() {
+        dispatchPrecondition(condition: .notOnQueue(.main))
 
+        // Set up the callback.
         let context = Unmanaged.passRetained(self).toOpaque()
-        let callback: statusCallback_t = { context, status in
+        callback = { context, status in
             guard let context else {
                 return
             }
             let ncp = Unmanaged<NCP>.fromOpaque(context).takeUnretainedValue()
-            DispatchQueue.main.sync {
+            // We dispatch async here to ensure we can't deadlock against `stop` calls on the main queue.
+            DispatchQueue.main.async {
                 let isConnected = status == 1 ? true : false
                 ncp.delegate?.ncp(ncp, didChangeConnectionState: isConnected)
             }
         }
 
-        while true {
-            logger.notice("Starting NCP for device '\(self.device.path)' baud rate \(self.device.baudRate)...")
-            ncpd(port, device.baudRate, "127.0.0.1", device.path, 0x0000, callback, context)
-            DispatchQueue.main.async {
-                self.delegate?.ncp(self, didChangeConnectionState: false)
-            }
-            logger.notice("NCP session ended.")
+        logger.notice("Starting NCP for device '\(self.device.path)' baud rate \(self.device.baudRate)...")
+        session = ncp_session_init(port, device.baudRate, "127.0.0.1", device.path, false, 0, callback, context)
 
-            // Check to see if we need to exit.
-            let isCancelled: Bool = lock.withLock {
-                return self.isCancelled
-            }
-            if isCancelled {
-                logger.notice("NCP thread exiting.")
-                return
-            }
+        // While it's pretty grim, but we to the main thread to start each ncp session. This ensures we inherit the
+        // correct thread priority when the session is started using `pthread_create`, without having to change the
+        // internal plptools implementation. Thankfully, this does almost no work except for setting up the thread state
+        // and starting it.
+        DispatchQueue.main.sync {
+            ncp_session_start(self.session)
         }
+
     }
 
     public init(device: DeviceConfiguration, port: Int32) {
         self.device = device
         self.port = port
-
-    }
-
-    public func start() {
-        let thread = Thread(block: threadEntryPoint)
-        thread.start()
     }
 
     public func stop() {
-
-        // Cancel the thread and get its id.
-        let threadID = lock.withLock {
-            isCancelled = true
-            return self.threadID
-        }
-        guard let threadID = threadID else {
+        dispatchPrecondition(condition: .notOnQueue(.main))
+        guard let session else {
             return
         }
-
-        // Signal the thread to stop it and then join it to ensure it's stopped.
-        pthread_kill(threadID, SIGINT)
-        pthread_join(threadID, nil)
+        ncp_session_cancel(session)
+        ncp_session_wait(session)
+        self.session = nil
     }
 
 }

macos/ReconnectCore/Sources/ReconnectCore/PLP/NCP.swift

@@ -18,7 +18,7 @@
 
 import Foundation
 
-import ncp
+import plptools
 
 public typealias PLPToolsError = rfsv.errs
 

macos/ReconnectCore/Sources/ReconnectCore/PLP/PLPToolsError.swift

@@ -18,7 +18,7 @@
 
 import Foundation
 
-import ncp
+import plptools
 
 public class RemoteCommandServicesClient {
 

macos/ReconnectCore/Sources/ReconnectCore/PLP/RemoteCommandServicesClient.swift

@@ -22,8 +22,8 @@ import IOKit.serial
 
 public protocol SerialDeviceMonitorDelegate: NSObject {
 
-    func serialDeviceMonitor(serialDeviceMonitor: SerialDeviceMonitor, didAddDevice device: String)
-    func serialDeviceMonitor(serialDeviceMonitor: SerialDeviceMonitor, didRemoveDevice device: String)
+    func serialDeviceMonitor(serialDeviceMonitor: SerialDeviceMonitor, didAddDevices devices: [String])
+    func serialDeviceMonitor(serialDeviceMonitor: SerialDeviceMonitor, didRemoveDevices devices: [String])
 
 }
 
@@ -36,6 +36,8 @@ public class SerialDeviceMonitor {
     }
 
     public func start() {
+        dispatchPrecondition(condition: .onQueue(.main))
+
         let matchingDict = IOServiceMatching(kIOSerialBSDServiceValue)
         var notifyPort: IONotificationPortRef?
         var addedIterator: io_iterator_t = 0
@@ -88,15 +90,33 @@ public class SerialDeviceMonitor {
         // https://developer.apple.com/documentation/iokit/1514362-ioserviceaddmatchingnotification
 
         // Handle existing removals.
+        var removals: [String] = []
         while case let service = IOIteratorNext(removedIterator), service != 0 {
-            deviceRemoved(service: service)
+            defer { IOObjectRelease(service) }
+            guard let deviceName = name(for: service) else {
+                continue
+            }
+            removals.append(deviceName)
         }
+        delegate?.serialDeviceMonitor(serialDeviceMonitor: self, didRemoveDevices: removals)
 
         // Handle existing additions.
+        var additions: [String] = []
         while case let service = IOIteratorNext(addedIterator), service != 0 {
-            deviceAdded(service: service)
+            defer { IOObjectRelease(service) }
+            guard let deviceName = name(for: service) else {
+                continue
+            }
+            additions.append(deviceName)
         }
+        delegate?.serialDeviceMonitor(serialDeviceMonitor: self, didAddDevices: additions)
+
+    }
 
+    func name(for service: io_object_t) -> String? {
+        return IORegistryEntryCreateCFProperty(service,
+                                               kIOCalloutDeviceKey as CFString,
+                                               kCFAllocatorDefault, 0)?.takeUnretainedValue() as? String
     }
 
     func stop() {
@@ -105,26 +125,20 @@ public class SerialDeviceMonitor {
 
     func deviceAdded(service: io_object_t) {
         dispatchPrecondition(condition: .onQueue(.main))
-        defer {
-            IOObjectRelease(service)
-        }
-        if let deviceName = IORegistryEntryCreateCFProperty(service,
-                                                            kIOCalloutDeviceKey as CFString,
-                                                            kCFAllocatorDefault, 0)?.takeUnretainedValue() as? String {
-            delegate?.serialDeviceMonitor(serialDeviceMonitor: self, didAddDevice: deviceName)
+        defer { IOObjectRelease(service) }
+        guard let deviceName = name(for: service) else {
+            return
         }
+        delegate?.serialDeviceMonitor(serialDeviceMonitor: self, didAddDevices: [deviceName])
     }
 
     func deviceRemoved(service: io_object_t) {
         dispatchPrecondition(condition: .onQueue(.main))
-        defer {
-            IOObjectRelease(service)
-        }
-        if let deviceName = IORegistryEntryCreateCFProperty(service,
-                                                            kIOCalloutDeviceKey as CFString,
-                                                            kCFAllocatorDefault, 0)?.takeUnretainedValue() as? String {
-            delegate?.serialDeviceMonitor(serialDeviceMonitor: self, didRemoveDevice: deviceName)
+        defer { IOObjectRelease(service) }
+        guard let deviceName = name(for: service) else {
+            return
         }
+        delegate?.serialDeviceMonitor(serialDeviceMonitor: self, didRemoveDevices: [deviceName])
     }
 
 }