Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 16 additions & 0 deletions Sources/PortScanner+Process.swift
Original file line number Diff line number Diff line change
Expand Up @@ -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<Int>] = [:]
var incomplete: Set<Int> = []
var complete = true
for start in stride(from: 0, to: pids.count, by: 256) {
let chunk = pids[start..<min(start + 256, pids.count)]
let result = await runLsofChunk(pidsCsv: chunk.map(String.init).joined(separator: ","))
for (pid, ports) in result.values { values[pid, default: []].formUnion(ports) }
incomplete.formUnion(result.incompletePIDs)
complete = complete && result.globallyComplete
}
return PortLsofScanResult(values: values, globallyComplete: complete, incompletePIDs: incomplete)
}

private func runLsofChunk(pidsCsv: String) async -> PortLsofScanResult {
let result = await commandRunner.run(
directory: "/",
executable: "/usr/sbin/lsof",
Expand Down
46 changes: 41 additions & 5 deletions Sources/PortScanner.swift
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down Expand Up @@ -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()
}
}
}

Expand Down Expand Up @@ -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.
Expand All @@ -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
Expand Down Expand Up @@ -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()
}
}
Expand Down
123 changes: 123 additions & 0 deletions cmuxTests/PortScannerTests.swift
Original file line number Diff line number Diff line change
Expand Up @@ -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<Int>.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<Int>.Continuation!
let stream = AsyncStream<Int> { 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,
Expand Down
Loading