diff --git a/Packages/macOS/CmuxFoundation/Sources/CmuxFoundation/ProcessPipe/ProcessPipeEndRead.swift b/Packages/macOS/CmuxFoundation/Sources/CmuxFoundation/ProcessPipe/ProcessPipeEndRead.swift index a1108085dadf..e9b08247722c 100644 --- a/Packages/macOS/CmuxFoundation/Sources/CmuxFoundation/ProcessPipe/ProcessPipeEndRead.swift +++ b/Packages/macOS/CmuxFoundation/Sources/CmuxFoundation/ProcessPipe/ProcessPipeEndRead.swift @@ -62,4 +62,5 @@ public struct ProcessPipeEndRead: Equatable, Sendable { } } } + } diff --git a/Packages/macOS/CmuxFoundation/Sources/CmuxFoundation/ProcessPipe/ProcessPipeStopAwareReader.swift b/Packages/macOS/CmuxFoundation/Sources/CmuxFoundation/ProcessPipe/ProcessPipeStopAwareReader.swift new file mode 100644 index 000000000000..2d4da8d0dcbc --- /dev/null +++ b/Packages/macOS/CmuxFoundation/Sources/CmuxFoundation/ProcessPipe/ProcessPipeStopAwareReader.swift @@ -0,0 +1,144 @@ +import Darwin +public import Foundation + +/// Drains a process pipe until EOF or a separate stop descriptor fires. +public struct ProcessPipeStopAwareReader: Sendable { + // Darwin's FIONREAD macro is not imported into Swift because it expands + // through `_IOR`. This is `_IOR('f', 127, Int32)` from sys/filio.h. + private static let bytesAvailableIOControlRequest = + UInt(0x4000_0000) + | (UInt(MemoryLayout.size) << 16) + | UInt(0x66_7f) + + private let fileDescriptor: Int32 + private let chunkSize: Int + private let stopFileDescriptor: Int32 + + /// Creates a stop-aware pipe reader. + /// + /// - Parameters: + /// - fileDescriptor: The process-pipe descriptor to drain. + /// - chunkSize: The requested maximum bytes read per operation. + /// Non-positive values are normalized to one byte. + /// - stopFileDescriptor: A descriptor whose readiness ends the drain + /// after bytes already buffered in `fileDescriptor` are preserved. + public init( + fileDescriptor: Int32, + chunkSize: Int = FileHandle.processPipeReadChunkSize, + stopFileDescriptor: Int32 + ) { + self.fileDescriptor = fileDescriptor + self.chunkSize = max(1, chunkSize) + self.stopFileDescriptor = stopFileDescriptor + } + + /// Drains the pipe, preserving bytes buffered when the stop signal fires. + /// + /// - Returns: The captured bytes and any read failure. + public func readToEnd() -> ProcessPipeEndRead { + var data = Data() + + while true { + var descriptors = [ + pollfd( + fd: fileDescriptor, + events: Int16(POLLIN | POLLERR | POLLHUP), + revents: 0 + ), + pollfd( + fd: stopFileDescriptor, + events: Int16(POLLIN | POLLERR | POLLHUP), + revents: 0 + ), + ] + let pollResult = Darwin.poll( + &descriptors, + nfds_t(descriptors.count), + -1 + ) + if pollResult < 0 { + let code = errno + if code == EINTR { + continue + } + return ProcessPipeEndRead( + data: data, + readError: ProcessPipeReadError( + operation: "readDataToEndOfFile.poll", + errnoCode: code + ) + ) + } + if (descriptors[0].revents & Int16(POLLNVAL)) != 0 { + return ProcessPipeEndRead( + data: data, + readError: ProcessPipeReadError( + operation: "readDataToEndOfFile.poll", + errnoCode: EBADF + ) + ) + } + + let isStopping = descriptors[1].revents != 0 + var bytesRemainingAtStop: Int? + if isStopping { + var availableBytes: Int32 = 0 + guard Darwin.ioctl( + fileDescriptor, + Self.bytesAvailableIOControlRequest, + &availableBytes + ) == 0 else { + return ProcessPipeEndRead( + data: data, + readError: ProcessPipeReadError( + operation: "readDataToEndOfFile.ioctl", + errnoCode: errno + ) + ) + } + bytesRemainingAtStop = Int(availableBytes) + if availableBytes == 0 { + return ProcessPipeEndRead(data: data, readError: nil) + } + } + + repeat { + let maximumReadLength = bytesRemainingAtStop.map { + min(chunkSize, $0) + } ?? chunkSize + switch ProcessPipeAvailableRead.readOnceIfReady( + fileDescriptor: fileDescriptor, + maxLength: maximumReadLength, + operation: "readDataToEndOfFile" + ) { + case .success(.data(let chunk)): + data.append(chunk) + if let remaining = bytesRemainingAtStop.map({ + $0 - chunk.count + }) { + if remaining <= 0 { + return ProcessPipeEndRead( + data: data, + readError: nil + ) + } + bytesRemainingAtStop = remaining + } + case .success(.wouldBlock): + if isStopping { + return ProcessPipeEndRead(data: data, readError: nil) + } + break + case .success(.endOfFile): + return ProcessPipeEndRead(data: data, readError: nil) + case .failure(let error): + return ProcessPipeEndRead(data: data, readError: error) + } + } while isStopping + + if isStopping { + return ProcessPipeEndRead(data: data, readError: nil) + } + } + } +} diff --git a/Packages/macOS/CmuxFoundation/Sources/CmuxFoundation/ProcessPipe/ProcessPipeStopSignal.swift b/Packages/macOS/CmuxFoundation/Sources/CmuxFoundation/ProcessPipe/ProcessPipeStopSignal.swift new file mode 100644 index 000000000000..eafb205b0840 --- /dev/null +++ b/Packages/macOS/CmuxFoundation/Sources/CmuxFoundation/ProcessPipe/ProcessPipeStopSignal.swift @@ -0,0 +1,52 @@ +import Darwin +import Foundation + +/// A one-shot, pollable signal for stopping process-pipe I/O. +/// +/// Closing the signal's write end leaves its read end permanently hung up, so +/// every reader polling the same descriptor wakes. Retain the signal until all +/// polling operations have returned; the read descriptor closes at deinit. +public final class ProcessPipeStopSignal: Sendable { + /// The stable descriptor callers include in their `poll(2)` descriptor set. + /// + /// The signal owns this descriptor. Callers must not close it and must stop + /// polling before releasing the signal. + public let readFileDescriptor: Int32 + + private let writeFileDescriptor: Int32 + private let didSignal = AtomicBooleanGate(false) + + /// Creates a close-on-exec pipe used exclusively as a stop signal. + /// + /// - Throws: A ``POSIXError`` when the pipe or close-on-exec flags cannot + /// be created. + public init() throws { + var descriptors: [Int32] = [-1, -1] + guard Darwin.pipe(&descriptors) == 0 else { + throw POSIXError(POSIXErrorCode(rawValue: errno) ?? .EIO) + } + + guard fcntl(descriptors[0], F_SETFD, FD_CLOEXEC) == 0, + fcntl(descriptors[1], F_SETFD, FD_CLOEXEC) == 0 else { + let code = errno + Darwin.close(descriptors[0]) + Darwin.close(descriptors[1]) + throw POSIXError(POSIXErrorCode(rawValue: code) ?? .EIO) + } + readFileDescriptor = descriptors[0] + writeFileDescriptor = descriptors[1] + } + + /// Wakes every poller. Repeated calls are safe and have no effect. + public func signal() { + guard didSignal.compareExchange(expected: false, desired: true) else { + return + } + Darwin.close(writeFileDescriptor) + } + + deinit { + signal() + Darwin.close(readFileDescriptor) + } +} diff --git a/Packages/macOS/CmuxFoundation/Tests/CmuxFoundationTests/FileHandleProcessPipeReadingTests.swift b/Packages/macOS/CmuxFoundation/Tests/CmuxFoundationTests/FileHandleProcessPipeReadingTests.swift index 5a8e4fc3e6ee..9444fe87043a 100644 --- a/Packages/macOS/CmuxFoundation/Tests/CmuxFoundationTests/FileHandleProcessPipeReadingTests.swift +++ b/Packages/macOS/CmuxFoundation/Tests/CmuxFoundationTests/FileHandleProcessPipeReadingTests.swift @@ -64,6 +64,109 @@ struct FileHandleProcessPipeReadingTests { #expect(result.readError == readError) } + @Test("A stopped end read drains buffered bytes without waiting for EOF") + func stoppedEndReadDrainsBufferedBytesWithoutEOF() throws { + let pipe = Pipe() + let stopSignal = try ProcessPipeStopSignal() + defer { + try? pipe.fileHandleForWriting.close() + try? pipe.fileHandleForReading.close() + } + try pipe.fileHandleForWriting.write(contentsOf: Data("buffered".utf8)) + stopSignal.signal() + + let result = ProcessPipeStopAwareReader( + fileDescriptor: pipe.fileHandleForReading.fileDescriptor, + stopFileDescriptor: stopSignal.readFileDescriptor + ).readToEnd() + + #expect(result.data == Data("buffered".utf8)) + #expect(result.readError == nil) + } + + @Test("A stopped end read drains every byte present at the stop snapshot") + func stoppedEndReadDrainsSnapshotLargerThanSixteenChunks() throws { + let pipe = Pipe() + let stopSignal = try ProcessPipeStopSignal() + let payload = Data((0..<128).map { UInt8($0) }) + defer { + try? pipe.fileHandleForWriting.close() + try? pipe.fileHandleForReading.close() + } + try pipe.fileHandleForWriting.write(contentsOf: payload) + stopSignal.signal() + + let result = ProcessPipeStopAwareReader( + fileDescriptor: pipe.fileHandleForReading.fileDescriptor, + chunkSize: 1, + stopFileDescriptor: stopSignal.readFileDescriptor + ).readToEnd() + + #expect(result.data == payload) + #expect(result.readError == nil) + } + + @Test( + "A non-positive chunk size still drains the stopped snapshot", + arguments: [0, -1] + ) + func nonPositiveChunkSizeDrainsStoppedSnapshot( + chunkSize: Int + ) throws { + let pipe = Pipe() + let stopSignal = try ProcessPipeStopSignal() + let payload = Data("buffered".utf8) + defer { + try? pipe.fileHandleForWriting.close() + try? pipe.fileHandleForReading.close() + } + try pipe.fileHandleForWriting.write(contentsOf: payload) + stopSignal.signal() + + let result = ProcessPipeStopAwareReader( + fileDescriptor: pipe.fileHandleForReading.fileDescriptor, + chunkSize: chunkSize, + stopFileDescriptor: stopSignal.readFileDescriptor + ).readToEnd() + + #expect(result.data == payload) + #expect(result.readError == nil) + } + + @Test("One stop signal wakes every blocked pipe reader") + func stopSignalBroadcastsToBlockedReaders() throws { + let firstPipe = Pipe() + let secondPipe = Pipe() + let stopSignal = try ProcessPipeStopSignal() + defer { + try? firstPipe.fileHandleForWriting.close() + try? firstPipe.fileHandleForReading.close() + try? secondPipe.fileHandleForWriting.close() + try? secondPipe.fileHandleForReading.close() + } + + stopSignal.signal() + let firstFinished = DispatchSemaphore(value: 0) + let secondFinished = DispatchSemaphore(value: 0) + DispatchQueue.global(qos: .utility).async { + _ = ProcessPipeStopAwareReader( + fileDescriptor: firstPipe.fileHandleForReading.fileDescriptor, + stopFileDescriptor: stopSignal.readFileDescriptor + ).readToEnd() + firstFinished.signal() + } + DispatchQueue.global(qos: .utility).async { + _ = ProcessPipeStopAwareReader( + fileDescriptor: secondPipe.fileHandleForReading.fileDescriptor, + stopFileDescriptor: stopSignal.readFileDescriptor + ).readToEnd() + secondFinished.signal() + } + + #expect(firstFinished.wait(timeout: .now() + 1) == .success) + #expect(secondFinished.wait(timeout: .now() + 1) == .success) + } + @Test("readDataToEndOfFileOrEmpty drains a closed pipe") func drainsClosedPipe() throws { let pipe = Pipe() diff --git a/Packages/macOS/CmuxRemoteSession/Sources/CmuxRemoteSession/Process/RemoteProcessCompletionWaitOutcome.swift b/Packages/macOS/CmuxRemoteSession/Sources/CmuxRemoteSession/Process/RemoteProcessCompletionWaitOutcome.swift new file mode 100644 index 000000000000..b834723ed776 --- /dev/null +++ b/Packages/macOS/CmuxRemoteSession/Sources/CmuxRemoteSession/Process/RemoteProcessCompletionWaitOutcome.swift @@ -0,0 +1,6 @@ +enum RemoteProcessCompletionWaitOutcome: Equatable, Sendable { + case processExited + case stdinWriteFailed + case timedOut + case waitFailed(Int32) +} diff --git a/Packages/macOS/CmuxRemoteSession/Sources/CmuxRemoteSession/Process/RemoteProcessCompletionWaiter.swift b/Packages/macOS/CmuxRemoteSession/Sources/CmuxRemoteSession/Process/RemoteProcessCompletionWaiter.swift new file mode 100644 index 000000000000..1f5d552d44cd --- /dev/null +++ b/Packages/macOS/CmuxRemoteSession/Sources/CmuxRemoteSession/Process/RemoteProcessCompletionWaiter.swift @@ -0,0 +1,57 @@ +internal import Darwin +internal import Foundation + +struct RemoteProcessCompletionWaiter: Sendable { + let processExitFileDescriptor: Int32 + let stdinFailureFileDescriptor: Int32 + + func wait( + until deadline: DispatchTime + ) -> RemoteProcessCompletionWaitOutcome { + while true { + let now = DispatchTime.now().uptimeNanoseconds + guard now < deadline.uptimeNanoseconds else { + return .timedOut + } + let remainingNanoseconds = deadline.uptimeNanoseconds - now + let wholeMilliseconds = remainingNanoseconds / 1_000_000 + let roundedMilliseconds = wholeMilliseconds + + (remainingNanoseconds % 1_000_000 == 0 ? 0 : 1) + let timeoutMilliseconds = Int32( + min(roundedMilliseconds, UInt64(Int32.max)) + ) + var descriptors = [ + pollfd( + fd: processExitFileDescriptor, + events: Int16(POLLIN | POLLERR | POLLHUP), + revents: 0 + ), + pollfd( + fd: stdinFailureFileDescriptor, + events: Int16(POLLIN | POLLERR | POLLHUP), + revents: 0 + ), + ] + let result = Darwin.poll( + &descriptors, + nfds_t(descriptors.count), + timeoutMilliseconds + ) + if result > 0 { + if descriptors[1].revents != 0 { + return .stdinWriteFailed + } + if descriptors[0].revents != 0 { + return .processExited + } + continue + } + if result == 0 { + return .timedOut + } + if errno != EINTR { + return .waitFailed(errno) + } + } + } +} diff --git a/Packages/macOS/CmuxRemoteSession/Sources/CmuxRemoteSession/Process/RemoteProcessExitSignal.swift b/Packages/macOS/CmuxRemoteSession/Sources/CmuxRemoteSession/Process/RemoteProcessExitSignal.swift new file mode 100644 index 000000000000..dca6d6d6eb9f --- /dev/null +++ b/Packages/macOS/CmuxRemoteSession/Sources/CmuxRemoteSession/Process/RemoteProcessExitSignal.swift @@ -0,0 +1,34 @@ +internal import CmuxFoundation +internal import Foundation + +/// One-shot process-exit notification with deadline-aware blocking waits. +/// +/// Foundation can report `isRunning == false` just before its termination +/// callback runs. `recordExit()` is therefore idempotent so the blocking owner +/// may publish that observed exit without racing the callback. +final class RemoteProcessExitSignal: Sendable { + private let didExit = AtomicBooleanGate(false) + private let group = DispatchGroup() + private let pollSignal: ProcessPipeStopSignal + + init() throws { + pollSignal = try ProcessPipeStopSignal() + group.enter() + } + + var readFileDescriptor: Int32 { + pollSignal.readFileDescriptor + } + + func recordExit() { + guard didExit.compareExchange(expected: false, desired: true) else { + return + } + pollSignal.signal() + group.leave() + } + + func wait(until deadline: DispatchTime) -> Bool { + group.wait(timeout: deadline) == .success + } +} diff --git a/Packages/macOS/CmuxRemoteSession/Sources/CmuxRemoteSession/Process/RemoteProcessPipeCaptureState.swift b/Packages/macOS/CmuxRemoteSession/Sources/CmuxRemoteSession/Process/RemoteProcessPipeCaptureState.swift new file mode 100644 index 000000000000..efe5ced08ab8 --- /dev/null +++ b/Packages/macOS/CmuxRemoteSession/Sources/CmuxRemoteSession/Process/RemoteProcessPipeCaptureState.swift @@ -0,0 +1,10 @@ +internal import Foundation + +/// Process-local capture storage whose mutations are confined to the runner's +/// serial capture queue and read only after its capture group completes. +final class RemoteProcessPipeCaptureState: @unchecked Sendable { + var stdoutData = Data() + var stderrData = Data() + var stdoutReadError: (any Error)? + var stderrReadError: (any Error)? +} diff --git a/Packages/macOS/CmuxRemoteSession/Sources/CmuxRemoteSession/Process/RemoteProcessStdinWriteState.swift b/Packages/macOS/CmuxRemoteSession/Sources/CmuxRemoteSession/Process/RemoteProcessStdinWriteState.swift new file mode 100644 index 000000000000..3c4e82b081cb --- /dev/null +++ b/Packages/macOS/CmuxRemoteSession/Sources/CmuxRemoteSession/Process/RemoteProcessStdinWriteState.swift @@ -0,0 +1,20 @@ +internal import CmuxFoundation + +/// Lock-free publication of the stdin writer's terminal error. +/// +/// The writer stores one immutable error before release-publishing the gate. +/// The blocking owner uses an acquire load before reading that value. +final class RemoteProcessStdinWriteState: @unchecked Sendable { + private let hasRecordedError = AtomicBooleanGate(false) + nonisolated(unsafe) private var error: (any Error)? + + func record(error: any Error) { + self.error = error + hasRecordedError.storeRelease(true) + } + + var recordedError: (any Error)? { + guard hasRecordedError.loadAcquire() else { return nil } + return error + } +} diff --git a/Packages/macOS/CmuxRemoteSession/Sources/CmuxRemoteSession/Process/RemoteProcessStdinWriter.swift b/Packages/macOS/CmuxRemoteSession/Sources/CmuxRemoteSession/Process/RemoteProcessStdinWriter.swift new file mode 100644 index 000000000000..77cd43d980cc --- /dev/null +++ b/Packages/macOS/CmuxRemoteSession/Sources/CmuxRemoteSession/Process/RemoteProcessStdinWriter.swift @@ -0,0 +1,75 @@ +internal import Darwin +internal import Foundation + +struct RemoteProcessStdinWriter: RemoteProcessStdinWriting { + func write( + _ data: Data, + to handle: FileHandle, + stopFileDescriptor: Int32 + ) throws { + let descriptor = handle.fileDescriptor + guard fcntl(descriptor, F_SETNOSIGPIPE, 1) == 0 else { + throw POSIXError(POSIXErrorCode(rawValue: errno) ?? .EIO) + } + let originalFlags = fcntl(descriptor, F_GETFL) + guard originalFlags >= 0 else { + throw POSIXError(POSIXErrorCode(rawValue: errno) ?? .EIO) + } + guard fcntl(descriptor, F_SETFL, originalFlags | O_NONBLOCK) == 0 else { + throw POSIXError(POSIXErrorCode(rawValue: errno) ?? .EIO) + } + defer { _ = fcntl(descriptor, F_SETFL, originalFlags) } + + try data.withUnsafeBytes { bytes in + guard let baseAddress = bytes.baseAddress else { return } + var offset = 0 + while offset < bytes.count { + var readiness = [ + pollfd( + fd: descriptor, + events: Int16(POLLOUT | POLLERR | POLLHUP), + revents: 0 + ), + pollfd( + fd: stopFileDescriptor, + events: Int16(POLLIN | POLLERR | POLLHUP), + revents: 0 + ), + ] + let pollResult = Darwin.poll(&readiness, nfds_t(readiness.count), -1) + if pollResult < 0 { + let code = errno + if code == EINTR { + continue + } + throw POSIXError(POSIXErrorCode(rawValue: code) ?? .EIO) + } + if readiness[1].revents != 0 { + return + } + if (readiness[0].revents & Int16(POLLNVAL)) != 0 { + throw POSIXError(.EBADF) + } + + let written = Darwin.write( + descriptor, + baseAddress.advanced(by: offset), + bytes.count - offset + ) + if written > 0 { + offset += written + continue + } + + let code = written == 0 ? EIO : errno + if code == EINTR { + continue + } + if code == EAGAIN || code == EWOULDBLOCK { + continue + } + throw POSIXError(POSIXErrorCode(rawValue: code) ?? .EIO) + } + } + } +} diff --git a/Packages/macOS/CmuxRemoteSession/Sources/CmuxRemoteSession/Process/RemoteProcessStdinWriting.swift b/Packages/macOS/CmuxRemoteSession/Sources/CmuxRemoteSession/Process/RemoteProcessStdinWriting.swift new file mode 100644 index 000000000000..67ce56146e98 --- /dev/null +++ b/Packages/macOS/CmuxRemoteSession/Sources/CmuxRemoteSession/Process/RemoteProcessStdinWriting.swift @@ -0,0 +1,10 @@ +internal import CmuxFoundation +internal import Foundation + +protocol RemoteProcessStdinWriting: Sendable { + func write( + _ data: Data, + to handle: FileHandle, + stopFileDescriptor: Int32 + ) throws +} diff --git a/Packages/macOS/CmuxRemoteSession/Sources/CmuxRemoteSession/Process/RemoteSessionProcessRunner.swift b/Packages/macOS/CmuxRemoteSession/Sources/CmuxRemoteSession/Process/RemoteSessionProcessRunner.swift index 05b46faf9129..70de409feb9d 100644 --- a/Packages/macOS/CmuxRemoteSession/Sources/CmuxRemoteSession/Process/RemoteSessionProcessRunner.swift +++ b/Packages/macOS/CmuxRemoteSession/Sources/CmuxRemoteSession/Process/RemoteSessionProcessRunner.swift @@ -1,6 +1,6 @@ internal import CmuxFoundation internal import Darwin -public import Foundation +internal import Foundation #if DEBUG internal import CMUXDebugLog #endif @@ -14,11 +14,12 @@ internal import CMUXDebugLog /// and timeout NSError domain/codes/messages, the capture/close ordering, /// stdin handling, and the debug-log lines are all pinned behavior. /// -/// Isolation: the struct is stateless (one immutable test hook); each `run` -/// call owns its process-local state. The capture readers run on the global -/// utility pool and hand their results through a small queue-confined box, -/// synchronized by the capture `DispatchGroup` before any read-back, exactly -/// like the legacy local-variable captures. +/// Isolation: the struct owns immutable process-input behavior and one optional +/// read-handle observation hook; each `run` call owns its process-local state. +/// The capture readers run on the global utility pool and hand their results +/// through a small queue-confined box, synchronized by the capture +/// `DispatchGroup` before any read-back, exactly like the legacy local-variable +/// captures. public struct RemoteSessionProcessRunner: RemoteSessionProcessRunning { /// Test observation seam (package tests only): invoked right after the /// stdout/stderr capture readers are installed, with the pipe read @@ -27,27 +28,20 @@ public struct RemoteSessionProcessRunner: RemoteSessionProcessRunning { /// The capture-survives-teardown regression test uses that to prove /// `run` still completes; production constructs the runner without a hook. let readHandlesDidInstall: (@Sendable (FileHandle, FileHandle) -> Bool)? + private let stdinWriter: any RemoteProcessStdinWriting /// Creates the production runner. public init() { self.readHandlesDidInstall = nil + self.stdinWriter = RemoteProcessStdinWriter() } - init(readHandlesDidInstall: (@Sendable (FileHandle, FileHandle) -> Bool)?) { + init( + readHandlesDidInstall: (@Sendable (FileHandle, FileHandle) -> Bool)? = nil, + stdinWriter: any RemoteProcessStdinWriting = RemoteProcessStdinWriter() + ) { self.readHandlesDidInstall = readHandlesDidInstall - } - - // Mutable capture-state shared between the two background pipe readers - // and the blocking caller. Writes are confined to the serial - // `captureQueue`; the caller only reads after `captureGroup.wait()` - // ordered every writer before it. `@unchecked Sendable` because the - // compiler cannot see that confinement (the legacy code expressed the - // same contract with captured local `var`s). - private final class PipeCaptureState: @unchecked Sendable { - var stdoutData = Data() - var stderrData = Data() - var stdoutReadError: (any Error)? - var stderrReadError: (any Error)? + self.stdinWriter = stdinWriter } /// Runs the request to completion on the calling thread; see @@ -87,9 +81,10 @@ public struct RemoteSessionProcessRunner: RemoteSessionProcessRunning { process.standardOutput = stdoutPipe process.standardError = stderrPipe + let writesInlineStdin = stdinFileHandle == nil && stdin != nil if let stdinFileHandle { process.standardInput = stdinFileHandle - } else if stdin != nil { + } else if writesInlineStdin { process.standardInput = Pipe() } else { process.standardInput = FileHandle.nullDevice @@ -98,11 +93,19 @@ public struct RemoteSessionProcessRunner: RemoteSessionProcessRunning { let stdoutHandle = stdoutPipe.fileHandleForReading let stderrHandle = stderrPipe.fileHandleForReading let captureQueue = DispatchQueue(label: "cmux.remote.process.capture") - let exitSemaphore = DispatchSemaphore(value: 0) - let captureState = PipeCaptureState() + let processExitSignal = try RemoteProcessExitSignal() + let captureStopSignal = try ProcessPipeStopSignal() + let stdinStopSignal = try ProcessPipeStopSignal() + let stdinFailureSignal = try ProcessPipeStopSignal() + let stdinWriteState = RemoteProcessStdinWriteState() + let captureState = RemoteProcessPipeCaptureState() let captureGroup = DispatchGroup() + let noteProcessExit: @Sendable () -> Void = { + captureStopSignal.signal() + processExitSignal.recordExit() + } process.terminationHandler = { _ in - exitSemaphore.signal() + noteProcessExit() } // Duplicate the descriptors on the calling thread, while the handles // are guaranteed open, and drain the duplicates. The contract (pinned @@ -122,7 +125,10 @@ public struct RemoteSessionProcessRunner: RemoteSessionProcessRunning { DispatchQueue.global(qos: .utility).async { defer { captureGroup.leave() } defer { _ = Darwin.close(stdoutDescriptor) } - let result = ProcessPipeEndRead.reading(fileDescriptor: stdoutDescriptor) + let result = ProcessPipeStopAwareReader( + fileDescriptor: stdoutDescriptor, + stopFileDescriptor: captureStopSignal.readFileDescriptor + ).readToEnd() captureQueue.sync { captureState.stdoutData = result.data captureState.stdoutReadError = result.readError @@ -132,7 +138,10 @@ public struct RemoteSessionProcessRunner: RemoteSessionProcessRunning { DispatchQueue.global(qos: .utility).async { defer { captureGroup.leave() } defer { _ = Darwin.close(stderrDescriptor) } - let result = ProcessPipeEndRead.reading(fileDescriptor: stderrDescriptor) + let result = ProcessPipeStopAwareReader( + fileDescriptor: stderrDescriptor, + stopFileDescriptor: captureStopSignal.readFileDescriptor + ).readToEnd() captureQueue.sync { captureState.stderrData = result.data captureState.stderrReadError = result.readError @@ -167,8 +176,10 @@ public struct RemoteSessionProcessRunner: RemoteSessionProcessRunning { try operation?.throwIfCancelled() try process.run() } catch { + processExitSignal.recordExit() try? stdoutPipe.fileHandleForWriting.close() try? stderrPipe.fileHandleForWriting.close() + captureStopSignal.signal() finishCaptureAndCloseReadHandles() debugLog( "remote.proc.launchFailed exec=\(URL(fileURLWithPath: executable).lastPathComponent) " + @@ -181,34 +192,121 @@ public struct RemoteSessionProcessRunner: RemoteSessionProcessRunning { try? stdoutPipe.fileHandleForWriting.close() try? stderrPipe.fileHandleForWriting.close() operation?.installCancellationHandler { + stdinStopSignal.signal() if process.isRunning { process.terminate() } } defer { operation?.clearCancellationHandler() } - if let stdin, let pipe = process.standardInput as? Pipe { - pipe.fileHandleForWriting.write(stdin) - try? pipe.fileHandleForWriting.close() - } - func terminateProcessAndWait() { process.terminate() - let terminatedGracefully = exitSemaphore.wait(timeout: .now() + 2.0) == .success + let terminatedGracefully = processExitSignal.wait(until: .now() + 2.0) if !terminatedGracefully, process.isRunning { _ = Darwin.kill(process.processIdentifier, SIGKILL) process.waitUntilExit() + noteProcessExit() } } - let didExitBeforeTimeout = exitSemaphore.wait(timeout: .now() + max(0, timeout)) == .success - if !didExitBeforeTimeout, process.isRunning { - if let operation, operation.isCancelled { + func stdinWriteFailure(_ error: any Error) -> NSError { + debugLog( + "remote.proc.stdinWriteFailed exec=\(URL(fileURLWithPath: executable).lastPathComponent) " + + "error=\(error.localizedDescription)" + ) + return NSError(domain: "cmux.remote.process", code: 3, userInfo: [ + NSLocalizedDescriptionKey: "Failed to write stdin for \(URL(fileURLWithPath: executable).lastPathComponent): \(error.localizedDescription)", + NSUnderlyingErrorKey: error, + ]) + } + + let stdinWriteGroup = DispatchGroup() + if let stdin, let pipe = process.standardInput as? Pipe { + let inputHandle = pipe.fileHandleForWriting + stdinWriteGroup.enter() + DispatchQueue.global(qos: .utility).async { + defer { + try? inputHandle.close() + stdinWriteGroup.leave() + } + do { + try stdinWriter.write( + stdin, + to: inputHandle, + stopFileDescriptor: stdinStopSignal.readFileDescriptor + ) + } catch let error as POSIXError where error.code == .EPIPE { + // A child may exit before consuming optional stdin. The + // POSIX helper turns that race into EPIPE instead of SIGPIPE. + } catch { + stdinWriteState.record(error: error) + stdinFailureSignal.signal() + if process.isRunning { + process.terminate() + } + } + } + } + + let timeoutDeadline = DispatchTime.now() + max(0, timeout) + let completionOutcome = RemoteProcessCompletionWaiter( + processExitFileDescriptor: processExitSignal.readFileDescriptor, + stdinFailureFileDescriptor: stdinFailureSignal.readFileDescriptor + ).wait(until: timeoutDeadline) + var didTimeOut = false + switch completionOutcome { + case .processExited: + break + case .stdinWriteFailed: + let stdinWriteError = stdinWriteState.recordedError ?? POSIXError(.EIO) + if process.isRunning { terminateProcessAndWait() + } + stdinWriteGroup.wait() + finishCaptureAndCloseReadHandles() + throw stdinWriteFailure(stdinWriteError) + case .timedOut: + didTimeOut = true + case .waitFailed(let code): + stdinStopSignal.signal() + if process.isRunning { + terminateProcessAndWait() + } + stdinWriteGroup.wait() + finishCaptureAndCloseReadHandles() + throw POSIXError(POSIXErrorCode(rawValue: code) ?? .EIO) + } + if didTimeOut, !process.isRunning { + process.waitUntilExit() + noteProcessExit() + didTimeOut = false + } + if !didTimeOut { + didTimeOut = stdinWriteGroup.wait(timeout: timeoutDeadline) == .timedOut + } + + if didTimeOut { + if let stdinWriteError = stdinWriteState.recordedError { + if process.isRunning { + terminateProcessAndWait() + } + stdinWriteGroup.wait() + finishCaptureAndCloseReadHandles() + throw stdinWriteFailure(stdinWriteError) + } + stdinStopSignal.signal() + if let operation, operation.isCancelled { + if process.isRunning { + terminateProcessAndWait() + } + stdinWriteGroup.wait() finishCaptureAndCloseReadHandles() throw operation.cancellationError } - terminateProcessAndWait() + if process.isRunning { + terminateProcessAndWait() + } + stdinWriteGroup.wait() finishCaptureAndCloseReadHandles() debugLog( "remote.proc.timeout exec=\(URL(fileURLWithPath: executable).lastPathComponent) " + @@ -219,6 +317,10 @@ public struct RemoteSessionProcessRunner: RemoteSessionProcessRunning { ]) } + if let stdinWriteError = stdinWriteState.recordedError { + finishCaptureAndCloseReadHandles() + throw stdinWriteFailure(stdinWriteError) + } finishCaptureAndCloseReadHandles() let stdout = String(data: captureState.stdoutData, encoding: .utf8) ?? "" let stderr = String(data: captureState.stderrData, encoding: .utf8) ?? "" diff --git a/Packages/macOS/CmuxRemoteSession/Sources/CmuxRemoteSession/Process/RemoteSessionProcessRunning.swift b/Packages/macOS/CmuxRemoteSession/Sources/CmuxRemoteSession/Process/RemoteSessionProcessRunning.swift index 327b0629e69d..b2897963091a 100644 --- a/Packages/macOS/CmuxRemoteSession/Sources/CmuxRemoteSession/Process/RemoteSessionProcessRunning.swift +++ b/Packages/macOS/CmuxRemoteSession/Sources/CmuxRemoteSession/Process/RemoteSessionProcessRunning.swift @@ -15,8 +15,8 @@ public protocol RemoteSessionProcessRunning: Sendable { /// - operation: Optional transfer-cancellation token; when it cancels, /// the process is terminated and `operation.cancellationError` is /// thrown. - /// - Throws: The legacy launch-failure (`cmux.remote.process` code 1) and - /// timeout (code 2) errors, or `operation.cancellationError`. + /// - Throws: Launch-failure (`cmux.remote.process` code 1), timeout + /// (code 2), stdin-write (code 3), or `operation.cancellationError`. func run( _ request: RemoteProcessRequest, operation: (any RemoteTransferCancelling)? diff --git a/Packages/macOS/CmuxRemoteSession/Tests/CmuxRemoteSessionTests/MarkerGatedFailingRemoteProcessStdinWriter.swift b/Packages/macOS/CmuxRemoteSession/Tests/CmuxRemoteSessionTests/MarkerGatedFailingRemoteProcessStdinWriter.swift new file mode 100644 index 000000000000..e3d7972e0b6a --- /dev/null +++ b/Packages/macOS/CmuxRemoteSession/Tests/CmuxRemoteSessionTests/MarkerGatedFailingRemoteProcessStdinWriter.swift @@ -0,0 +1,40 @@ +import Darwin +import Foundation +@testable import CmuxRemoteSession + +struct MarkerGatedFailingRemoteProcessStdinWriter: RemoteProcessStdinWriting { + let readyFIFOURL: URL + let exitFIFOURL: URL + let markerURL: URL + let gate: RemoteProcessStdinWriterFailureGate + + func write( + _ data: Data, + to handle: FileHandle, + stopFileDescriptor: Int32 + ) throws { + let processIdentifier = try readProcessIdentifier() + try Data("\(processIdentifier)".utf8).write(to: markerURL) + if gate == .exited { + let exitHandle = try FileHandle(forReadingFrom: exitFIFOURL) + defer { try? exitHandle.close() } + _ = exitHandle.readDataToEndOfFile() + } + throw POSIXError(.EIO) + } + + private func readProcessIdentifier() throws -> pid_t { + let readyHandle = try FileHandle(forReadingFrom: readyFIFOURL) + defer { try? readyHandle.close() } + let contents = String( + decoding: readyHandle.readDataToEndOfFile(), + as: UTF8.self + ) + guard let processIdentifier = pid_t( + contents.trimmingCharacters(in: .whitespacesAndNewlines) + ) else { + throw POSIXError(.EINVAL) + } + return processIdentifier + } +} diff --git a/Packages/macOS/CmuxRemoteSession/Tests/CmuxRemoteSessionTests/ProcessMarkerFixture.swift b/Packages/macOS/CmuxRemoteSession/Tests/CmuxRemoteSessionTests/ProcessMarkerFixture.swift new file mode 100644 index 000000000000..42c2dc479c7d --- /dev/null +++ b/Packages/macOS/CmuxRemoteSession/Tests/CmuxRemoteSessionTests/ProcessMarkerFixture.swift @@ -0,0 +1,50 @@ +import Darwin +import Foundation +import Testing + +final class ProcessMarkerFixture { + let readyFIFOURL: URL + let exitFIFOURL: URL + let markerURL: URL + + init() throws { + let temporaryDirectory = FileManager.default.temporaryDirectory + let identifier = UUID().uuidString + readyFIFOURL = temporaryDirectory.appendingPathComponent( + "cmux-remote-process-\(identifier).ready", + isDirectory: false + ) + exitFIFOURL = temporaryDirectory.appendingPathComponent( + "cmux-remote-process-\(identifier).exit", + isDirectory: false + ) + markerURL = temporaryDirectory.appendingPathComponent( + "cmux-remote-process-\(identifier).pid", + isDirectory: false + ) + + guard Darwin.mkfifo(readyFIFOURL.path, 0o600) == 0 else { + throw POSIXError(POSIXErrorCode(rawValue: errno) ?? .EIO) + } + guard Darwin.mkfifo(exitFIFOURL.path, 0o600) == 0 else { + let code = POSIXErrorCode(rawValue: errno) ?? .EIO + try? FileManager.default.removeItem(at: readyFIFOURL) + throw POSIXError(code) + } + } + + deinit { + try? FileManager.default.removeItem(at: readyFIFOURL) + try? FileManager.default.removeItem(at: exitFIFOURL) + try? FileManager.default.removeItem(at: markerURL) + } + + func recordedProcessHasExited() throws -> Bool { + let contents = try String(contentsOf: markerURL, encoding: .utf8) + let processIdentifier = try #require( + pid_t(contents.trimmingCharacters(in: .whitespacesAndNewlines)) + ) + errno = 0 + return Darwin.kill(processIdentifier, 0) == -1 && errno == ESRCH + } +} diff --git a/Packages/macOS/CmuxRemoteSession/Tests/CmuxRemoteSessionTests/ProcessRunEvent.swift b/Packages/macOS/CmuxRemoteSession/Tests/CmuxRemoteSessionTests/ProcessRunEvent.swift new file mode 100644 index 000000000000..a0abfdd368b8 --- /dev/null +++ b/Packages/macOS/CmuxRemoteSession/Tests/CmuxRemoteSessionTests/ProcessRunEvent.swift @@ -0,0 +1,7 @@ +import Foundation +@testable import CmuxRemoteSession + +enum ProcessRunEvent: Sendable { + case completed(Result) + case deadline +} diff --git a/Packages/macOS/CmuxRemoteSession/Tests/CmuxRemoteSessionTests/RemoteProcessStdinWriterFailureGate.swift b/Packages/macOS/CmuxRemoteSession/Tests/CmuxRemoteSessionTests/RemoteProcessStdinWriterFailureGate.swift new file mode 100644 index 000000000000..b3962c0c561a --- /dev/null +++ b/Packages/macOS/CmuxRemoteSession/Tests/CmuxRemoteSessionTests/RemoteProcessStdinWriterFailureGate.swift @@ -0,0 +1,6 @@ +import Foundation + +enum RemoteProcessStdinWriterFailureGate: Equatable, Sendable { + case launched + case exited +} diff --git a/Packages/macOS/CmuxRemoteSession/Tests/CmuxRemoteSessionTests/RemoteSessionProcessRunnerTests.swift b/Packages/macOS/CmuxRemoteSession/Tests/CmuxRemoteSessionTests/RemoteSessionProcessRunnerTests.swift index fbcc0665e2ca..23ace48664a6 100644 --- a/Packages/macOS/CmuxRemoteSession/Tests/CmuxRemoteSessionTests/RemoteSessionProcessRunnerTests.swift +++ b/Packages/macOS/CmuxRemoteSession/Tests/CmuxRemoteSessionTests/RemoteSessionProcessRunnerTests.swift @@ -70,6 +70,193 @@ struct RemoteSessionProcessRunnerTests { #expect(result.stdout == "hello-stdin") } + @Test("A child closing stdin before a large write does not abort the runner") + func childClosingStdinDoesNotAbortRunner() throws { + let runner = RemoteSessionProcessRunner() + let result = try runner.run( + RemoteProcessRequest( + executable: "/usr/bin/true", + arguments: [], + stdin: Data(repeating: 0x41, count: 1_048_576), + timeout: 5 + ), + operation: nil + ) + + #expect(result.status == 0) + } + + @Test("An unexpected stdin write error terminates the child and keeps the pinned runner error") + func unexpectedStdinWriteErrorTerminatesChildAndKeepsPinnedRunnerError() throws { + let marker = try ProcessMarkerFixture() + let runner = RemoteSessionProcessRunner( + stdinWriter: MarkerGatedFailingRemoteProcessStdinWriter( + readyFIFOURL: marker.readyFIFOURL, + exitFIFOURL: marker.exitFIFOURL, + markerURL: marker.markerURL, + gate: .launched + ) + ) + + #expect { + try runner.run( + RemoteProcessRequest( + executable: "/bin/sh", + arguments: [ + "-c", + "echo $$ > \"$CMUX_TEST_PROCESS_READY\"; exec /bin/sleep 30", + ], + environment: [ + "CMUX_TEST_PROCESS_READY": marker.readyFIFOURL.path, + ], + stdin: Data("payload".utf8), + timeout: 5 + ), + operation: nil + ) + } throws: { error in + let nsError = error as NSError + let underlyingError = nsError.userInfo[NSUnderlyingErrorKey] as? POSIXError + return nsError.domain == "cmux.remote.process" + && nsError.code == 3 + && nsError.localizedDescription.hasPrefix("Failed to write stdin for sh:") + && underlyingError?.code == .EIO + } + #expect(try marker.recordedProcessHasExited()) + } + + @Test("A stdin write error wins when the child ignores termination") + func stdinWriteErrorWinsWhenChildIgnoresTermination() async throws { + let marker = try ProcessMarkerFixture() + let runner = RemoteSessionProcessRunner( + stdinWriter: MarkerGatedFailingRemoteProcessStdinWriter( + readyFIFOURL: marker.readyFIFOURL, + exitFIFOURL: marker.exitFIFOURL, + markerURL: marker.markerURL, + gate: .launched + ) + ) + + let outcome = try #require(await runProcess( + runner, + request: RemoteProcessRequest( + executable: "/bin/sh", + arguments: [ + "-c", + "trap '' TERM; echo $$ > \"$CMUX_TEST_PROCESS_READY\"; exec /bin/sleep 30", + ], + environment: [ + "CMUX_TEST_PROCESS_READY": marker.readyFIFOURL.path, + ], + stdin: Data("payload".utf8), + timeout: 10 + ), + completingWithin: 4 + )) + + guard case .failure(let error) = outcome else { + Issue.record("Expected the stdin write to fail") + return + } + let nsError = error as NSError + let underlyingError = nsError.userInfo[NSUnderlyingErrorKey] as? POSIXError + #expect(nsError.domain == "cmux.remote.process") + #expect(nsError.code == 3) + #expect(underlyingError?.code == .EIO) + } + + @Test("A late stdin write error wins over an earlier child exit") + func lateStdinWriteErrorWinsOverChildExit() throws { + let marker = try ProcessMarkerFixture() + let runner = RemoteSessionProcessRunner( + stdinWriter: MarkerGatedFailingRemoteProcessStdinWriter( + readyFIFOURL: marker.readyFIFOURL, + exitFIFOURL: marker.exitFIFOURL, + markerURL: marker.markerURL, + gate: .exited + ) + ) + + #expect { + try runner.run( + RemoteProcessRequest( + executable: "/bin/sh", + arguments: [ + "-c", + "echo $$ > \"$CMUX_TEST_PROCESS_READY\"; exec 3>\"$CMUX_TEST_PROCESS_EXIT\"; exit 0", + ], + environment: [ + "CMUX_TEST_PROCESS_READY": marker.readyFIFOURL.path, + "CMUX_TEST_PROCESS_EXIT": marker.exitFIFOURL.path, + ], + stdin: Data("payload".utf8), + timeout: 5 + ), + operation: nil + ) + } throws: { error in + let nsError = error as NSError + let underlyingError = nsError.userInfo[NSUnderlyingErrorKey] as? POSIXError + return nsError.domain == "cmux.remote.process" + && nsError.code == 3 + && underlyingError?.code == .EIO + } + } + + @Test("A child retaining unread stdin cannot delay the process timeout") + func unreadStdinCannotDelayProcessTimeout() async throws { + let runner = RemoteSessionProcessRunner() + let outcome = try #require(await runProcess( + runner, + request: RemoteProcessRequest( + executable: "/bin/sleep", + arguments: ["5"], + stdin: Data(repeating: 0x41, count: 1_048_576), + timeout: 1 + ), + completingWithin: 4 + )) + + switch outcome { + case .success: + Issue.record("Expected the unread stdin request to time out") + case .failure(let error): + let nsError = error as NSError + #expect( + nsError.domain == "cmux.remote.process" + && nsError.code == 2 + && nsError.localizedDescription == "sleep timed out after 1s" + ) + } + } + + @Test("A descendant retaining unread stdin cannot outlive the process timeout") + func inheritedUnreadStdinCannotOutliveProcessTimeout() async throws { + let runner = RemoteSessionProcessRunner() + let outcome = try #require(await runProcess( + runner, + request: RemoteProcessRequest( + executable: "/bin/sh", + arguments: ["-c", "sleep 5 <&0 &"], + stdin: Data(repeating: 0x41, count: 1_048_576), + timeout: 1 + ), + completingWithin: 4 + )) + + switch outcome { + case .success: + Issue.record("Expected the inherited stdin request to time out") + case .failure(let error): + let nsError = error as NSError + #expect( + nsError.domain == "cmux.remote.process" + && nsError.code == 2 + && nsError.localizedDescription == "sh timed out after 1s" + ) + } + } + @Test("Streams a local file through stdin") func streamsFileStdin() throws { let fileManager = FileManager.default @@ -134,5 +321,39 @@ struct RemoteSessionProcessRunnerTests { && nsError.localizedDescription == "sh timed out after 1s" } } + + private func runProcess( + _ runner: RemoteSessionProcessRunner, + request: RemoteProcessRequest, + completingWithin timeout: TimeInterval + ) async -> Result? { + await withTaskGroup( + of: ProcessRunEvent.self, + returning: Result?.self + ) { group in + group.addTask { + .completed(Result { + try runner.run(request, operation: nil) + }) + } + group.addTask { + try? await Task.sleep(for: .seconds(timeout)) + return .deadline + } + + guard let firstEvent = await group.next() else { + return nil + } + group.cancelAll() + switch firstEvent { + case .completed(let result): + return result + case .deadline: + // Structured concurrency reaps the test-owned blocking task + // before leaving this scope and starting another subprocess. + return nil + } + } + } } }