Add per-PTY dispatch-source I/O path as alternative to TaskNotifier select loop

ModernTests/DispatchSources/DispatchSourceConcurrencyTests.swift

@@ -0,0 +1,157 @@
+//
+//  DispatchSourceConcurrencyTests.swift
+//  ModernTests
+//
+//  Regression tests for data races in dispatch source management.
+//  These verify that concurrent operations on PTYTaskIOHandler don't crash.
+//
+
+import XCTest
+@testable import iTerm2SharedARC
+
+// MARK: - Coprocess Setup Serialization Tests
+
+/// Regression test: setupCoprocessSources must serialize source
+/// reference assignments on the ioQueue to prevent data races with event
+/// handlers and updateCoprocess*SourceState.
+final class DispatchSourceCoprocessSerializationTests: XCTestCase {
+
+    /// Rapid sequential setup/teardown of coprocess sources while the primary
+    /// sources are active. Tests that the syncOnIOQueue wrapper properly
+    /// serializes reference assignments with event handlers.
+    func testRapidCoprocessSetupTeardownCycles() {
+        guard let task = PTYTask() else {
+            XCTFail("Failed to create PTYTask")
+            return
+        }
+
+        guard let ptyPipe = createTestPipe() else {
+            XCTFail("Failed to create PTY pipe")
+            return
+        }
+        defer { closeTestPipe(ptyPipe) }
+
+        task.testSetFd(ptyPipe.readFd)
+        task.testSetupDispatchSourcesForTesting()
+        task.testWaitForIOQueue()
+
+        guard let coprocessPipe = createTestPipe() else {
+            XCTFail("Failed to create coprocess pipe")
+            task.testTeardownDispatchSourcesForTesting()
+            return
+        }
+        defer { closeTestPipe(coprocessPipe) }
+
+        // Rapid setup/teardown cycles — each must properly cancel old sources
+        // and assign new references atomically on ioQueue.
+        for _ in 0..<20 {
+            task.testSetupCoprocessSources(withReadFd: coprocessPipe.readFd,
+                                           writeFd: coprocessPipe.writeFd)
+            task.testWaitForIOQueue()
+
+            XCTAssertTrue(task.testHasCoprocessReadSource(),
+                          "Coprocess read source should exist after setup")
+            XCTAssertTrue(task.testHasCoprocessWriteSource(),
+                          "Coprocess write source should exist after setup")
+
+            task.testTeardownCoprocessSources()
+            task.testWaitForIOQueue()
+
+            XCTAssertFalse(task.testHasCoprocessReadSource(),
+                           "Coprocess read source should be nil after teardown")
+            XCTAssertFalse(task.testHasCoprocessWriteSource(),
+                           "Coprocess write source should be nil after teardown")
+        }
+
+        task.testTeardownDispatchSourcesForTesting()
+    }
+
+}
+
+// MARK: - Concurrent Teardown + Update Tests
+
+/// Regression test: teardown() concurrent with
+/// updateReadSourceState/updateWriteSourceState must not crash.
+/// Before the centralized helpers, source references were read from the
+/// calling queue and then mutated on ioQueue, creating a TOCTOU race.
+final class DispatchSourceConcurrentTeardownUpdateTests: XCTestCase {
+
+    func testConcurrentTeardownAndUpdateDoesNotCrash() {
+        for _ in 0..<5 {
+            guard let task = PTYTask() else {
+                XCTFail("Failed to create PTYTask")
+                return
+            }
+
+            guard let pipe = createTestPipe() else {
+                XCTFail("Failed to create test pipe")
+                return
+            }
+            defer { closeTestPipe(pipe) }
+
+            task.testSetFd(pipe.readFd)
+            task.testIoAllowedOverride = NSNumber(value: true)
+            task.paused = false
+
+            task.testSetupDispatchSourcesForTesting(withPid: getpid())
+            task.testWaitForIOQueue()
+
+            let group = DispatchGroup()
+
+            // Spam update calls from multiple threads (uses ioQueue.async internally)
+            for _ in 0..<10 {
+                group.enter()
+                DispatchQueue.global().async {
+                    task.perform(NSSelectorFromString("updateReadSourceState"))
+                    task.perform(NSSelectorFromString("updateWriteSourceState"))
+                    group.leave()
+                }
+            }
+
+            // Concurrently tear down (uses ioQueue.sync internally)
+            group.enter()
+            DispatchQueue.global().async {
+                task.testTeardownDispatchSourcesForTesting()
+                group.leave()
+            }
+
+            let result = group.wait(timeout: .now() + 5.0)
+            XCTAssertEqual(result, .success, "Concurrent teardown+update timed out")
+
+            XCTAssertFalse(task.testHasReadSource(),
+                           "Read source should be nil after teardown")
+            XCTAssertFalse(task.testHasWriteSource(),
+                           "Write source should be nil after teardown")
+        }
+    }
+
+    func testRapidSetupTeardownCyclesDoNotCrash() {
+        guard let task = PTYTask() else {
+            XCTFail("Failed to create PTYTask")
+            return
+        }
+
+        guard let pipe = createTestPipe() else {
+            XCTFail("Failed to create test pipe")
+            return
+        }
+        defer { closeTestPipe(pipe) }
+
+        task.testSetFd(pipe.readFd)
+        task.testIoAllowedOverride = NSNumber(value: true)
+
+        // Rapid setup/teardown cycles from the same thread
+        for _ in 0..<20 {
+            task.testSetupDispatchSourcesForTesting(withPid: getpid())
+            task.testWaitForIOQueue()
+            task.paused = true
+            task.paused = false
+            task.testTeardownDispatchSourcesForTesting()
+        }
+
+        XCTAssertFalse(task.testHasReadSource(),
+                       "Read source should be nil after final teardown")
+        XCTAssertFalse(task.testHasWriteSource(),
+                       "Write source should be nil after final teardown")
+    }
+}

ModernTests/DispatchSources/DispatchSourceCoprocessTests.swift

@@ -0,0 +1,160 @@
+//
+//  DispatchSourceCoprocessTests.swift
+//  ModernTests
+//
+//  Tests for coprocess dispatch source lifecycle: setup, teardown,
+//  suspend/resume state, and interaction with primary source teardown.
+//
+
+import XCTest
+@testable import iTerm2SharedARC
+
+// MARK: - Coprocess Dispatch Source Tests
+
+/// Tests for coprocess dispatch source setup, teardown, and state transitions.
+final class DispatchSourceCoprocessTests: XCTestCase {
+
+    var task: PTYTask!
+    /// Primary PTY pipe — needed for testSetupDispatchSourcesForTesting.
+    var ptyPipe: (readFd: Int32, writeFd: Int32)!
+    /// Coprocess pipe — used as the coprocess read/write fds.
+    var coprocessPipe: (readFd: Int32, writeFd: Int32)!
+
+    override func setUp() {
+        super.setUp()
+        task = PTYTask()
+
+        ptyPipe = createTestPipe()
+        XCTAssertNotNil(ptyPipe, "Failed to create PTY pipe")
+
+        coprocessPipe = createTestPipe()
+        XCTAssertNotNil(coprocessPipe, "Failed to create coprocess pipe")
+
+        task.testSetFd(ptyPipe.readFd)
+        task.testSetupDispatchSourcesForTesting()
+        task.testWaitForIOQueue()
+    }
+
+    override func tearDown() {
+        task.testTeardownDispatchSourcesForTesting()
+        if ptyPipe != nil { closeTestPipe(ptyPipe) }
+        if coprocessPipe != nil { closeTestPipe(coprocessPipe) }
+        task = nil
+        super.tearDown()
+    }
+
+    // MARK: - Setup
+
+    func testSetupCoprocessSourcesCreatesSources() {
+        XCTAssertFalse(task.testHasCoprocessReadSource(), "No coprocess read source before setup")
+        XCTAssertFalse(task.testHasCoprocessWriteSource(), "No coprocess write source before setup")
+
+        task.testSetupCoprocessSources(withReadFd: coprocessPipe.readFd, writeFd: coprocessPipe.writeFd)
+        task.testWaitForIOQueue()
+
+        XCTAssertTrue(task.testHasCoprocessReadSource(), "Coprocess read source should exist after setup")
+        XCTAssertTrue(task.testHasCoprocessWriteSource(), "Coprocess write source should exist after setup")
+    }
+
+    func testCoprocessSourcesStartSuspended() {
+        task.testSetupCoprocessSources(withReadFd: coprocessPipe.readFd, writeFd: coprocessPipe.writeFd)
+        task.testWaitForIOQueue()
+
+        XCTAssertTrue(task.testIsCoprocessReadSourceSuspended(),
+                      "Coprocess read source should start suspended")
+        XCTAssertTrue(task.testIsCoprocessWriteSourceSuspended(),
+                      "Coprocess write source should start suspended")
+    }
+
+    // MARK: - Teardown
+
+    func testTeardownCoprocessSourcesCleansUp() {
+        task.testSetupCoprocessSources(withReadFd: coprocessPipe.readFd, writeFd: coprocessPipe.writeFd)
+        task.testWaitForIOQueue()
+
+        XCTAssertTrue(task.testHasCoprocessReadSource(), "Coprocess read source should exist")
+        XCTAssertTrue(task.testHasCoprocessWriteSource(), "Coprocess write source should exist")
+
+        task.testTeardownCoprocessSources()
+        task.testWaitForIOQueue()
+
+        XCTAssertFalse(task.testHasCoprocessReadSource(), "Coprocess read source should be nil after teardown")
+        XCTAssertFalse(task.testHasCoprocessWriteSource(), "Coprocess write source should be nil after teardown")
+    }
+
+    func testTeardownCoprocessSourcesSafeWithoutSetup() {
+        XCTAssertFalse(task.testHasCoprocessReadSource(), "No coprocess read source before setup")
+        XCTAssertFalse(task.testHasCoprocessWriteSource(), "No coprocess write source before setup")
+
+        // Should not crash
+        task.testTeardownCoprocessSources()
+        task.testWaitForIOQueue()
+
+        XCTAssertFalse(task.testHasCoprocessReadSource(), "Still no coprocess read source")
+        XCTAssertFalse(task.testHasCoprocessWriteSource(), "Still no coprocess write source")
+    }
+
+    func testDoubleTeardownCoprocessSourcesSafe() {
+        task.testSetupCoprocessSources(withReadFd: coprocessPipe.readFd, writeFd: coprocessPipe.writeFd)
+        task.testWaitForIOQueue()
+
+        task.testTeardownCoprocessSources()
+        task.testWaitForIOQueue()
+
+        // Second teardown should not crash
+        task.testTeardownCoprocessSources()
+        task.testWaitForIOQueue()
+
+        XCTAssertFalse(task.testHasCoprocessReadSource(), "No coprocess read source after double teardown")
+        XCTAssertFalse(task.testHasCoprocessWriteSource(), "No coprocess write source after double teardown")
+    }
+
+    // MARK: - Replacement
+
+    /// Regression: setupCoprocessSources must tear down existing sources before creating new ones.
+    func testSetupCoprocessSourcesTearsDownExisting() {
+        task.testSetupCoprocessSources(withReadFd: coprocessPipe.readFd, writeFd: coprocessPipe.writeFd)
+        task.testWaitForIOQueue()
+
+        XCTAssertTrue(task.testHasCoprocessReadSource(), "First coprocess read source should exist")
+
+        // Create a second pipe for the replacement coprocess
+        guard let secondPipe = createTestPipe() else {
+            XCTFail("Failed to create second coprocess pipe")
+            return
+        }
+        defer { closeTestPipe(secondPipe) }
+
+        // Second setup should tear down first sources, then create new ones — no crash
+        task.testSetupCoprocessSources(withReadFd: secondPipe.readFd, writeFd: secondPipe.writeFd)
+        task.testWaitForIOQueue()
+
+        XCTAssertTrue(task.testHasCoprocessReadSource(), "Replacement coprocess read source should exist")
+        XCTAssertTrue(task.testHasCoprocessWriteSource(), "Replacement coprocess write source should exist")
+    }
+
+    // MARK: - Primary teardown interaction
+
+    /// teardown() on PTYTaskIOHandler calls teardownCoprocessSources(), so tearing
+    /// down primary sources must also clean up coprocess sources.
+    func testPrimaryTeardownAlsoTearsDownCoprocessSources() {
+        task.testSetupCoprocessSources(withReadFd: coprocessPipe.readFd, writeFd: coprocessPipe.writeFd)
+        task.testWaitForIOQueue()
+
+        XCTAssertTrue(task.testHasCoprocessReadSource(), "Coprocess read source should exist")
+        XCTAssertTrue(task.testHasCoprocessWriteSource(), "Coprocess write source should exist")
+
+        // Primary teardown should also clean up coprocess sources
+        task.testTeardownDispatchSourcesForTesting()
+        task.testWaitForIOQueue()
+
+        XCTAssertFalse(task.testHasCoprocessReadSource(),
+                       "Coprocess read source should be nil after primary teardown")
+        XCTAssertFalse(task.testHasCoprocessWriteSource(),
+                       "Coprocess write source should be nil after primary teardown")
+
+        // Re-setup primary sources so tearDown() doesn't double-teardown
+        task.testSetFd(ptyPipe.readFd)
+        task.testSetupDispatchSourcesForTesting()
+    }
+}