diff --git a/Sources/PortScanner+Process.swift b/Sources/PortScanner+Process.swift index 88638782de9f..757ad8b03529 100644 --- a/Sources/PortScanner+Process.swift +++ b/Sources/PortScanner+Process.swift @@ -536,6 +536,22 @@ extension PortScanner { } func runLsof(pidsCsv: String) async -> PortLsofScanResult { + let pids = pidsCsv.split(separator: ",").compactMap { Int($0) } + guard pids.count > 256 else { return await runLsofChunk(pidsCsv: pidsCsv) } + var values: [Int: Set] = [:] + var incomplete: Set = [] + var complete = true + for start in stride(from: 0, to: pids.count, by: 256) { + let chunk = pids[start.. PortLsofScanResult { let result = await commandRunner.run( directory: "/", executable: "/usr/sbin/lsof", diff --git a/Sources/PortScanner.swift b/Sources/PortScanner.swift index 70c639ab36fc..df562bb41e7d 100644 --- a/Sources/PortScanner.swift +++ b/Sources/PortScanner.swift @@ -67,6 +67,15 @@ final class PortScanner: @unchecked Sendable { /// Whether a burst sequence is currently running. private var burstActive = false + /// Generation invalidates callbacks that were queued before a panel + /// lifecycle changed. The queue is the sole owner, so cancellation and + /// generation checks are deterministic and race-free. + private var burstGeneration: UInt64 = 0 + /// There is at most one future callback for the active burst. The callback + /// schedules its successor after it starts, so completed work cannot + /// accumulate in a history array. + private var scheduledBurstWorkItem: DispatchWorkItem? + private var coalesceTimer: DispatchSourceTimer? /// Periodic timer for agent-owned process trees that aren't attached to a TTY. @@ -139,6 +148,11 @@ final class PortScanner: @unchecked Sendable { } panelPortSnapshot.remove(keys: [key]) panelPortOwnersByKey.removeValue(forKey: key) + if ttyNames.isEmpty { + cancelBurstScheduling() + } else if !pendingKicks.isEmpty, !burstActive { + startCoalesce() + } } } @@ -224,11 +238,12 @@ final class PortScanner: @unchecked Sendable { guard !pendingKicks.isEmpty else { return } burstActive = true - runBurst(index: 0) + runBurst(index: 0, generation: burstGeneration) } - private func runBurst(index: Int, burstStart: DispatchTime? = nil) { + private func runBurst(index: Int, burstStart: DispatchTime? = nil, generation: UInt64) { // Already on `queue`. + guard generation == burstGeneration else { return } guard index < Self.burstOffsets.count else { burstActive = false // If new kicks arrived during the burst, start a new coalesce cycle. @@ -240,11 +255,28 @@ final class PortScanner: @unchecked Sendable { let start = burstStart ?? .now() let deadline = start + Self.burstOffsets[index] - queue.asyncAfter(deadline: deadline) { [weak self] in + let workItem = DispatchWorkItem { [weak self] in guard let self else { return } + guard generation == self.burstGeneration else { return } + // Clear the completed callback before scheduling its successor. + self.scheduledBurstWorkItem = nil self.runScan() - self.runBurst(index: index + 1, burstStart: start) + self.runBurst(index: index + 1, burstStart: start, generation: generation) } + scheduledBurstWorkItem = workItem + queue.asyncAfter(deadline: deadline, execute: workItem) + } + + /// Cancels the global burst scheduler after its last panel is removed. + /// Individual panel removal must leave a burst alive for the remaining + /// panels because each scan snapshots the current `ttyNames` map. + private func cancelBurstScheduling() { + burstGeneration &+= 1 + scheduledBurstWorkItem?.cancel() + scheduledBurstWorkItem = nil + burstActive = false + coalesceTimer?.cancel() + coalesceTimer = nil } // MARK: - Scan @@ -556,7 +588,11 @@ final class PortScanner: @unchecked Sendable { inspectedPIDs: inspectedPIDs, requestID: requestID ) - if hasPendingScan { + // A pending attempt can represent a burst callback that overlapped + // this scan after the latest kick was already paid down. Preserve it + // for any remaining panel, but do not restart work after the final + // panel has been unregistered. + if hasPendingScan, !ttyNames.isEmpty { runScan() } } diff --git a/cmuxTests/PortScannerTests.swift b/cmuxTests/PortScannerTests.swift index e91eaac46a4f..62ac08dbe150 100644 --- a/cmuxTests/PortScannerTests.swift +++ b/cmuxTests/PortScannerTests.swift @@ -1076,6 +1076,129 @@ private actor ScriptedCommandRunner: CommandRunning { } } +/// Returns an empty, successful `ps` result and signals each command call. +/// The lifecycle tests use the signal to position an unregister between burst +/// scans without depending on a wall-clock sleep. +private actor BurstInvocationCommandRunner: CommandRunning { + private(set) var invocationCount = 0 + private var invocationStreams: [UUID: AsyncStream.Continuation] = [:] + + func run( + directory: String, + executable: String, + arguments: [String], + timeout: TimeInterval? + ) async -> CommandResult { + invocationCount += 1 + for continuation in invocationStreams.values { + continuation.yield(invocationCount) + } + return CommandResult( + stdout: "", + stderr: "", + exitStatus: 0, + timedOut: false, + executionError: nil + ) + } + + func waitForInvocation(_ target: Int, timeout: Duration) async -> Bool { + guard invocationCount < target else { return true } + + let streamID = UUID() + var streamContinuation: AsyncStream.Continuation! + let stream = AsyncStream { continuation in + streamContinuation = continuation + } + invocationStreams[streamID] = streamContinuation + defer { + invocationStreams.removeValue(forKey: streamID)?.finish() + } + + return await withTaskGroup(of: Bool.self) { group in + group.addTask { + for await count in stream where count >= target { + return true + } + return false + } + group.addTask { + do { + try await Task.sleep(for: timeout) + return false + } catch { + return false + } + } + let result = await group.next() ?? false + group.cancelAll() + return result + } + } +} + +private func waitForScannerQueue(_ scanner: PortScanner) async { + await withCheckedContinuation { continuation in + scanner.queue.async { continuation.resume() } + } +} + +@Suite("Port scanner lifecycle") +struct PortScannerLifecycleTests { + @Test("Unregister cancels a pending coalesce and burst generation") + func unregisterCancelsPendingBurst() async { + let runner = ScriptedCommandRunner(results: []) + let scanner = PortScanner(commandRunner: runner) + let workspaceID = UUID() + let panelID = UUID() + await MainActor.run { + scanner.registerTTY(workspaceId: workspaceID, panelId: panelID, ttyName: "ttys999") + } + scanner.kick(workspaceId: workspaceID, panelId: panelID) + await MainActor.run { + scanner.unregisterPanel(workspaceId: workspaceID, panelId: panelID) + } + await waitForScannerQueue(scanner) + let calls = await runner.recordedArguments + #expect(calls.isEmpty) + } + + @Test("Unregistering one panel preserves another panel's active burst") + func unregisteringOnePanelPreservesOtherBurst() async { + let runner = BurstInvocationCommandRunner() + let scanner = PortScanner(commandRunner: runner) + let workspaceID = UUID() + let removedPanelID = UUID() + let retainedPanelID = UUID() + + await MainActor.run { + scanner.registerTTY( + workspaceId: workspaceID, + panelId: removedPanelID, + ttyName: "ttys998" + ) + scanner.registerTTY( + workspaceId: workspaceID, + panelId: retainedPanelID, + ttyName: "ttys997" + ) + } + scanner.kick(workspaceId: workspaceID, panelId: removedPanelID) + scanner.kick(workspaceId: workspaceID, panelId: retainedPanelID) + + // The third scan consumes the pending-kick budget. The original PR + // then cancels the remaining global burst when the first panel closes. + #expect(await runner.waitForInvocation(3, timeout: .seconds(6))) + await MainActor.run { + scanner.unregisterPanel(workspaceId: workspaceID, panelId: removedPanelID) + } + #expect(await runner.waitForInvocation(4, timeout: .seconds(7))) + + let calls = await runner.invocationCount + #expect(calls >= 4) + } +} + @Suite("Port scanner retirement end to end") struct PortScannerPortRetirementTests { /// Drives the whole scanner — TTY registration, kick, coalesce, burst,