diff --git a/src/lazy_init.mbt b/src/lazy_init.mbt index 9739b0937..f358ff424 100644 --- a/src/lazy_init.mbt +++ b/src/lazy_init.mbt @@ -81,6 +81,18 @@ pub async fn[X] Lazy::wait(self : Lazy[X]) -> X { } } +///| +/// Return the result of a lazy value if is already computed, +/// or return `None` if the result is not immediately available. +/// Fail with the same error if the initialization process failed. +pub fn[X] Lazy::try_wait(self : Lazy[X]) -> X? raise { + match self.state { + Done(result) => Some(result) + Fail(err) => raise err + Running(_) | Uninitialized => None + } +} + ///| /// Create a new lazily initialized value by passing an async initialization function. /// The function `f` will be started in the background automatically diff --git a/src/os_error/errno_linux.mbt b/src/os_error/errno_linux.mbt index 8b64fc1f6..65f472495 100644 --- a/src/os_error/errno_linux.mbt +++ b/src/os_error/errno_linux.mbt @@ -41,3 +41,6 @@ let linux_EINPROGRESS : Int = 115 ///| let linux_EWOULDBLOCK : Int = linux_EAGAIN + +///| +let linux_EPIPE : Int = 32 diff --git a/src/os_error/errno_macos.mbt b/src/os_error/errno_macos.mbt index 40f16db6a..431621668 100644 --- a/src/os_error/errno_macos.mbt +++ b/src/os_error/errno_macos.mbt @@ -41,3 +41,6 @@ let macos_EINPROGRESS : Int = 36 ///| let macos_EWOULDBLOCK : Int = macos_EAGAIN + +///| +let macos_EPIPE : Int = 32 diff --git a/src/os_error/errno_windows.mbt b/src/os_error/errno_windows.mbt index 887df1183..68afcae2a 100644 --- a/src/os_error/errno_windows.mbt +++ b/src/os_error/errno_windows.mbt @@ -44,3 +44,6 @@ let windows_ERROR_DIRECTORY : Int = 267 ///| let windows_ERROR_NOT_SUPPORTED : Int = 50 + +///| +let windows_ERROR_BROKEN_PIPE : Int = 109 diff --git a/src/os_error/error.mbt b/src/os_error/error.mbt index 4d3114bc7..ae0db1dc6 100644 --- a/src/os_error/error.mbt +++ b/src/os_error/error.mbt @@ -182,3 +182,10 @@ pub let errno_ENOTSUP : Int = match @env_util.platform { MacOS => macos_ENOTSUP Windows => windows_ERROR_NOT_SUPPORTED } + +///| +pub let errno_EPIPE : Int = match @env_util.platform { + Linux => linux_EPIPE + MacOS => macos_EPIPE + Windows => windows_ERROR_BROKEN_PIPE +} diff --git a/src/os_error/pkg.generated.mbti b/src/os_error/pkg.generated.mbti index a4a0a196e..a8070beb4 100644 --- a/src/os_error/pkg.generated.mbti +++ b/src/os_error/pkg.generated.mbti @@ -8,6 +8,8 @@ pub let errno_ENOTDIR : Int pub let errno_ENOTSUP : Int +pub let errno_EPIPE : Int + pub fn errno_to_string(Int) -> String pub fn get_errno() -> Int diff --git a/src/pkg.generated.mbti b/src/pkg.generated.mbti index 0387b816b..3be42b530 100644 --- a/src/pkg.generated.mbti +++ b/src/pkg.generated.mbti @@ -60,6 +60,7 @@ type Lazy[X] #as_free_fn(lazy_init, deprecated) #callsite(autofill(loc)) pub fn[X] Lazy::Lazy(async () -> X, loc~ : SourceLoc) -> Self[X] +pub fn[X] Lazy::try_wait(Self[X]) -> X? raise pub async fn[X] Lazy::wait(Self[X]) -> X type Mutex diff --git a/src/process/pkg.generated.mbti b/src/process/pkg.generated.mbti index f208ecef2..568123d7c 100644 --- a/src/process/pkg.generated.mbti +++ b/src/process/pkg.generated.mbti @@ -23,13 +23,13 @@ pub fn graceful_cancel(timeout~ : Int, signal? : @signal.Signal) -> Cancellation pub fn hard_cancel() -> CancellationHandler -pub fn pipe() -> (&ProcessInput, &ProcessOutput) raise +pub fn pipe() -> (&ProcessInput, &ProcessOutput) -pub fn read_from_process(shared? : Bool) -> (ReadFromProcess, &ProcessOutput) raise +pub fn read_from_process(shared? : Bool) -> (ReadFromProcess, &ProcessOutput) -pub async fn redirect_from_file(String) -> &ProcessInput +pub fn redirect_from_file(String) -> &ProcessInput -pub async fn redirect_to_file(String, append? : Bool, create_mode? : @fs.CreateMode, permission? : Int, shared? : Bool) -> &ProcessOutput +pub fn redirect_to_file(String, append? : Bool, create_mode? : @fs.CreateMode, permission? : Int, shared? : Bool) -> &ProcessOutput pub async fn run(StringView, ArrayView[String], extra_env? : Map[String, String], inherit_env? : Bool, stdin? : &ProcessInput, stdout? : &ProcessOutput, stderr? : &ProcessOutput, cwd? : StringView, no_console_window? : Bool, cancel_handler? : CancellationHandler) -> Int @@ -39,7 +39,7 @@ pub async fn spawn_orphan(StringView, ArrayView[String], extra_env? : Map[String pub async fn wait_pid(Int) -> Int -pub fn write_to_process() -> (&ProcessInput, WriteToProcess) raise +pub fn write_to_process() -> (&ProcessInput, WriteToProcess) // Errors diff --git a/src/process/redirect.mbt b/src/process/redirect.mbt index 091fbd5a2..4e3e6137f 100644 --- a/src/process/redirect.mbt +++ b/src/process/redirect.mbt @@ -12,16 +12,289 @@ // See the License for the specific language governing permissions and // limitations under the License. +///| +/// An entity that can be used to redirect stdin of a process +trait ProcessInput { + async fn fd(Self) -> @fd_util.Fd + // invoked when `spawn` succeeded using this channel + fn on_success(Self) -> Unit + // invoked when `spawn` failed using this channel + fn on_failure(Self) -> Unit +} + +///| +/// An entity that can be used to redirect stdout/stderr of a process +trait ProcessOutput { + async fn fd(Self) -> @fd_util.Fd + fn is_shared(Self) -> Bool + // invoked when `spawn` succeeded using this channel + fn on_success(Self) -> Unit + // invoked when `spawn` failed using this channel + fn on_failure(Self) -> Unit +} + +///| +#deprecated("use `@process.write_to_process()` or `@process.pipe()` instead") +pub impl ProcessInput for @pipe.PipeRead + +///| +pub impl ProcessInput for @pipe.PipeRead with fn fd(self) { + self.fd() +} + +///| +pub impl ProcessInput for @pipe.PipeRead with fn on_success(_) { + +} + +///| +pub impl ProcessInput for @pipe.PipeRead with fn on_failure(_) { + +} + +///| +#deprecated("use `@process.read_from_process(shared~)` or `@process.pipe()` instead") +pub impl ProcessOutput for @pipe.PipeWrite + +///| +pub impl ProcessOutput for @pipe.PipeWrite with fn fd(self) { + self.fd() +} + +///| +pub impl ProcessOutput for @pipe.PipeWrite with fn is_shared(_) { + false +} + +///| +pub impl ProcessOutput for @pipe.PipeWrite with fn on_success(_) { + +} + +///| +pub impl ProcessOutput for @pipe.PipeWrite with fn on_failure(_) { + +} + +///| +pub impl ProcessInput for @stdio.Input with fn fd(self) { + self.fd() +} + +///| +pub impl ProcessInput for @stdio.Input with fn on_success(_) { + +} + +///| +pub impl ProcessInput for @stdio.Input with fn on_failure(_) { + +} + +///| +pub impl ProcessOutput for @stdio.Output with fn fd(self) { + self.fd() +} + +///| +pub impl ProcessOutput for @stdio.Output with fn is_shared(_) { + false +} + +///| +pub impl ProcessOutput for @stdio.Output with fn on_success(_) { + +} + +///| +pub impl ProcessOutput for @stdio.Output with fn on_failure(_) { + +} + +///| +/// Close an input channel that will not be passed to a child process. +/// A channel is closed automatically once it is passed to a child process, +/// so this is only needed when a channel has been created +/// but spawning the process that would consume it fails or is abandoned. +/// Closing is idempotent, and a no-op for channels that only borrow +/// their file descriptor, such as `@stdio.stdin`. +#doc(hidden) +#deprecated +pub fn &ProcessInput::close(self : &ProcessInput) -> Unit { + self.on_success() +} + +///| +/// Close a shared channel for redirecting process output. +/// Once closed, the channel can no longer be passed to child process. +/// Note that unique channels are automatically closed after being passed to child process, +/// so there is no need to manually call `.close()` on a unique channel. +/// See `@process.read_from_process` for more details. +pub fn &ProcessOutput::close(self : &ProcessOutput) -> Unit { + self.on_success() +} + +///| +priv enum TempPipeKind { + ReadFromProcess + ReadFromProcessShared + WriteToProcess + Pipe +} + +///| +priv enum TempPipeEndpoint { + Uninitialized + Active(@event_loop.IoHandle) + Closed +} + +///| +priv struct TempPipe { + mut r : TempPipeEndpoint + mut w : TempPipeEndpoint + /// how this pipe is created, for error reporting + kind : TempPipeKind + mut waiter : @coroutine.Coroutine? +} + ///| /// A temporary pipe used to read output from a spawned process struct ReadFromProcess { - io : @event_loop.IoHandle + pipe : TempPipe read_buf : @io.ReaderBuffer } ///| /// A temporary pipe used to write data to a spawned process -struct WriteToProcess(@event_loop.IoHandle) +struct WriteToProcess(TempPipe) + +///| +fn TempPipe::TempPipe(kind : TempPipeKind) -> TempPipe { + { r: Uninitialized, w: Uninitialized, kind, waiter: None } +} + +///| +fn TempPipe::init(self : TempPipe) -> Unit raise { + let (read_end_is_async, write_end_is_async, context) = match self.kind { + ReadFromProcess | ReadFromProcessShared => + (true, false, "@process.read_from_process()") + WriteToProcess => (false, true, "@process.write_to_process()") + Pipe => (false, false, "@process.pipe()") + } + if self.waiter is Some(coro) { + coro.wake() + } + let (r, w) = @fd_util.pipe(read_end_is_async~, write_end_is_async~, context~) + let r = { + errdefer @fd_util.close(w, kind=Pipe, context~) + @event_loop.IoHandle::from_fd(r, kind=Pipe, is_async=read_end_is_async) + } + let w = @event_loop.IoHandle::from_fd( + w, + kind=Pipe, + is_async=write_end_is_async, + ) + self.r = Active(r) + self.w = Active(w) +} + +///| +fn TempPipe::close_read(self : TempPipe) -> Unit { + if self.r is Active(r) { + r.close() + } + self.r = Closed + if self.waiter is Some(coro) { + coro.wake() + } + if self.w is Uninitialized { + self.w = Closed + } +} + +///| +fn TempPipe::close_write(self : TempPipe) -> Unit { + if self.w is Active(w) { + w.close() + } + self.w = Closed + if self.waiter is Some(coro) { + coro.wake() + } + if self.r is Uninitialized { + self.r = Closed + } +} + +///| +impl ProcessInput for TempPipe with fn fd(self) { + match self.r { + Uninitialized => self.init() + Active(_) => () + Closed => { + let context = match self.kind { + ReadFromProcess | ReadFromProcessShared => + "@process.read_from_process()" + WriteToProcess => "@process.write_to_process()" + Pipe => "@process.pipe()" + } + raise @os_error.OSError(@os_error.errno_EPIPE, context~) + } + } + guard! self.r is Active(r) + r.fd() +} + +///| +impl ProcessInput for TempPipe with fn on_success(self) { + self.close_read() +} + +///| +impl ProcessInput for TempPipe with fn on_failure(self) { + self.close_read() + if self.kind is Pipe { + self.close_write() + } +} + +///| +impl ProcessOutput for TempPipe with fn fd(self) { + match self.w { + Uninitialized => self.init() + Active(_) => () + Closed => { + let context = match self.kind { + ReadFromProcess | ReadFromProcessShared => + "@process.read_from_process()" + WriteToProcess => "@process.write_to_process()" + Pipe => "@process.pipe()" + } + raise @os_error.OSError(@os_error.errno_EPIPE, context~) + } + } + guard! self.w is Active(w) + w.fd() +} + +///| +impl ProcessOutput for TempPipe with fn is_shared(self) { + self.kind is ReadFromProcessShared +} + +///| +impl ProcessOutput for TempPipe with fn on_success(self) { + self.close_write() +} + +///| +impl ProcessOutput for TempPipe with fn on_failure(self) { + self.close_write() + if self.kind is Pipe { + self.close_read() + } +} ///| /// Create a temporary pipe for reading from stdout/stderr of a process. @@ -32,74 +305,79 @@ struct WriteToProcess(@event_loop.IoHandle) /// `w` is temporary: it can only be passed to one `@process.run` call. /// However, it is safe to pass `w` to both `stdout` and `stderr` of the same process. /// +/// The pipe is lazy initialized: it will only get created +/// after being passed to the first child process. +/// So there will be no resource leak if the pipe never get used. +/// /// If `shared` is `false` (the default), the created pipe is unique. /// In this case, `w` can only be passed to one child process /// (but it is ok to pass `w` to both stdout and stderr of the same process), -/// and `w` will be closed automatically after the child process started. +/// but there is no need to close `w` manually in any circumstance: +/// - `w` is lazy initialized: the pipe will only get created +/// when being passed to the child process. +/// So there will be no resource leak if the pipe never get used. +/// - `w` will be closed automatically after the child process started +/// or if child process spawning failed /// /// If `shared` is `true`, the created pipe is shared: /// `w` can be passed to multiple children process to merge and collect their output. /// In this case, `w` should be closed manually via `w.close()` /// after being passed to the last child process. -/// Note that `w` can be closed immediately after the last child process *started*, +/// Note that `w` should be closed immediately after the last child process *started*, /// there is no need to wait for the termination of the child process. /// If you forgot to close `w` or close it too late, /// the read end `r` will fail to observe EOF from children process. pub fn read_from_process( shared? : Bool = false, -) -> (ReadFromProcess, &ProcessOutput) raise { - let context = "@process.read_from_process()" - let (r, w) = @fd_util.pipe( - read_end_is_async=true, - write_end_is_async=false, - context~, - ) - let r = @event_loop.IoHandle::from_fd(r, kind=Pipe) - let w = @event_loop.IoHandle::from_fd(w, kind=Pipe, is_async=false) - ( - { io: r, read_buf: @io.ReaderBuffer::new() }, - TempPipeWrite::{ pipe: w, shared, closed: false }, +) -> (ReadFromProcess, &ProcessOutput) { + let pipe = TempPipe( + if shared { + ReadFromProcessShared + } else { + ReadFromProcess + }, ) + ({ pipe, read_buf: @io.ReaderBuffer::new() }, pipe) } ///| /// Create a temporary pipe for writing to stdin of a process. /// The return value is a pair `(r, w)`, /// where `w` is a temporary pipe that can be used to write to process output, -/// and `r` should be passed to `@process.run`. -pub fn write_to_process() -> (&ProcessInput, WriteToProcess) raise { - let context = "@process.write_to_process()" - let (r, w) = @fd_util.pipe( - read_end_is_async=false, - write_end_is_async=true, - context~, - ) - let r = @event_loop.IoHandle::from_fd(r, kind=Pipe, is_async=false) - let w = @event_loop.IoHandle::from_fd(w, kind=Pipe) - (TempPipeRead::{ pipe: r, closed: false }, w) +/// and `r` should be passed to `@process.run +pub fn write_to_process() -> (&ProcessInput, WriteToProcess) { + let pipe = TempPipe(WriteToProcess) + (pipe, pipe) } ///| /// Create a temporary pipe for connecting the output of /// one child process to the input of another child process. -pub fn pipe() -> (&ProcessInput, &ProcessOutput) raise { - let context = "@process.pipe()" - let (r, w) = @fd_util.pipe( - read_end_is_async=false, - write_end_is_async=false, - context~, - ) - let r = @event_loop.IoHandle::from_fd(r, kind=Pipe, is_async=false) - let w = @event_loop.IoHandle::from_fd(w, kind=Pipe, is_async=false) - ( - TempPipeRead::{ pipe: r, closed: false }, - TempPipeWrite::{ pipe: w, shared: false, closed: false }, - ) +/// The two pipe endpoints returned can be passed to only one child process. +/// +/// In most cases, there is no need to manually close the pipe: +/// +/// - the pipe is lazy initialized. If both ends of pipe are never used, +/// for example the program fail before spawning the first child, +/// the pipe will not get created at all, so there will be no resource leak +/// - when the child process is spawned successfully, +/// it will automatically close the end it receives. +/// - when the child process fail to spawn, +/// *both* ends of pipe will be closed automatically. +/// +/// The only case where the pipe need to be closed manually is when the program +/// fail *after* the first child is spawned successfully, +/// but *before* invoking spawning operation on the second child. +/// Users should avoid this pattern and perform necessary initialization +/// before spawning the first child. +pub fn pipe() -> (&ProcessInput, &ProcessOutput) { + let pipe = TempPipe(Pipe) + (pipe, pipe) } ///| pub fn ReadFromProcess::close(self : ReadFromProcess) -> Unit { - self.io.close() + self.pipe.close_read() } ///| @@ -114,12 +392,16 @@ pub impl @io.Reader for ReadFromProcess with fn _direct_read( offset~, max_len~, ) { - self.io.read( - buf, - offset~, - len=max_len, - context="@process.ReadFromProcess::read()", - ) + if self.pipe.r is Uninitialized { + guard! self.pipe.waiter is None + self.pipe.waiter = Some(@coroutine.current_coroutine()) + defer { + self.pipe.waiter = None + } + @coroutine.suspend() + } + guard self.pipe.r is Active(io) else { 0 } + io.read(buf, offset~, len=max_len, context="@process.ReadFromProcess::read()") } ///| @@ -134,8 +416,8 @@ pub extend ReadFromProcess with @io.Reader::{ ///| pub fn WriteToProcess::close(self : WriteToProcess) -> Unit { - let WriteToProcess(io) = self - io.close() + let WriteToProcess(pipe) = self + pipe.close_write() } ///| @@ -145,8 +427,20 @@ pub impl @io.Writer for WriteToProcess with fn write_once( offset~, len~, ) { - let WriteToProcess(io) = self - io.write(buf, offset~, len~, context="@process.WriteToProcess::write()") + let context = "@process.WriteToProcess::write()" + let WriteToProcess(pipe) = self + if pipe.w is Uninitialized { + guard! pipe.waiter is None + pipe.waiter = Some(@coroutine.current_coroutine()) + defer { + pipe.waiter = None + } + @coroutine.suspend() + } + guard pipe.w is Active(io) else { + raise @os_error.OSError(@os_error.errno_EPIPE, context~) + } + io.write(buf, offset~, len~, context~) } ///| @@ -165,7 +459,12 @@ fn @fs.CreateMode::to_int(self : @fs.CreateMode) -> Int = "%identity" /// If `shared` is `false` (the default), the returned channel `w` is unique. /// In this case, `w` can only be passed to one child process /// (but it is ok to pass `w` to both stdout and stderr of the same process), -/// and `w` will be closed automatically after the child process started. +/// but there is no need to manually close `w` in any circumstance: +/// - `w` is lazy initialized: the file will only get opened +/// when being passed to the child process. +/// So there will be no resource leak if the channel never get used. +/// - `w` will be closed automatically after the child process started +/// or if child process spawning failed /// /// If `shared` is `true`, the returned channel `w` is shared: /// it can be passed to multiple children process to merge and collect their output. @@ -173,223 +472,115 @@ fn @fs.CreateMode::to_int(self : @fs.CreateMode) -> Int = "%identity" /// after being passed to the last child process. /// Note that `w` can be closed immediately after the last child process *started*, /// there is no need to wait for the termination of the child process. -pub async fn redirect_to_file( +pub fn redirect_to_file( path : String, append? : Bool = false, create_mode? : @fs.CreateMode = CreateOrTruncate, permission? : Int = 0o644, shared? : Bool = false, ) -> &ProcessOutput { - let (file, _) = @event_loop.open( - path, - 1, // write only - create=create_mode.to_int(), - append~, - sync=0, - mode=permission, - context="@process.redirect_to_file()", - ) - RedirectToFile::{ io: file, shared, closed: false } + let closed = Ref(false) + let file = @async.Lazy <| () => { + let context = "@process.redirect_to_file()" + if closed.val { + raise @os_error.OSError(@os_error.errno_EPIPE, context~) + } + let (file, _) = @event_loop.open( + path, + 1, // write only + create=create_mode.to_int(), + append~, + sync=0, + mode=permission, + context~, + ) + if closed.val { + file.close() + raise @os_error.OSError(@os_error.errno_EPIPE, context~) + } + file + } + RedirectToFile::{ file, shared, closed } } ///| /// Redirect the content of a file at `path` to the stdin of a process. -pub async fn redirect_from_file(path : String) -> &ProcessInput { - let (file, _) = @event_loop.open( - path, - 0, // read only - create=0, // `OpenExisting` - append=false, - sync=0, - mode=0, - context="@process.redirect_from_file()", - ) - RedirectToFile::{ io: file, shared: false, closed: false } -} - -///| -/// An entity that can be used to redirect stdin of a process -trait ProcessInput { - fn fd(Self) -> @fd_util.Fd raise - /// Close the input channel. - /// May be called automatically via `after_spawn`, or manually via `.close()` - fn do_close(Self) -> Unit = _ - fn after_spawn(Self) -> Unit = _ -} - -///| -/// Close an input channel that will not be passed to a child process. -/// A channel is closed automatically once it is passed to a child process, -/// so this is only needed when a channel has been created -/// but spawning the process that would consume it fails or is abandoned. -/// Closing is idempotent, and a no-op for channels that only borrow -/// their file descriptor, such as `@stdio.stdin`. -#doc(hidden) -#deprecated -pub fn &ProcessInput::close(self : &ProcessInput) -> Unit { - self.do_close() -} - -///| -impl ProcessInput with fn do_close(_) { - () -} - -///| -impl ProcessInput with fn after_spawn(_) { - () -} - -///| -#deprecated("use `@process.write_to_process()` or `@process.pipe()` instead") -pub impl ProcessInput for @pipe.PipeRead - -///| -pub impl ProcessInput for @pipe.PipeRead with fn fd(self) { - self.fd() -} - -///| -pub impl ProcessInput for @stdio.Input with fn fd(self) { - self.fd() -} - -///| -/// An entity that can be used to redirect stdout/stderr of a process -trait ProcessOutput { - fn fd(Self) -> @fd_util.Fd raise - /// Close the output channel. - /// May be called automatically via `after_spawn`, or manually via `.close()` - fn do_close(Self) -> Unit = _ - fn after_spawn(Self) -> Unit = _ -} - -///| -/// Close a shared channel for redirecting process output. -/// Once closed, the channel can no longer be passed to child process. -/// Note that unique channels are automatically closed after being passed to child process, -/// so there is no need to manually call `.close()` on a unique channel. -/// See `@process.read_from_process` for more details. -pub fn &ProcessOutput::close(self : &ProcessOutput) -> Unit { - self.do_close() -} - -///| -impl ProcessOutput with fn do_close(_) { - () -} - -///| -impl ProcessOutput with fn after_spawn(_) { - () -} - -///| -#deprecated("use `@process.read_from_process(shared~)` or `@process.pipe()` instead") -pub impl ProcessOutput for @pipe.PipeWrite - -///| -pub impl ProcessOutput for @pipe.PipeWrite with fn fd(self) { - self.fd() -} - -///| -pub impl ProcessOutput for @stdio.Output with fn fd(self) { - self.fd() -} - -///| -priv struct TempPipeRead { - pipe : @event_loop.IoHandle - mut closed : Bool +pub fn redirect_from_file(path : String) -> &ProcessInput { + let closed = Ref(false) + let file = @async.Lazy <| () => { + let context = "@process.redirect_from_file()" + if closed.val { + raise @os_error.OSError(@os_error.errno_EPIPE, context~) + } + let (file, _) = @event_loop.open( + path, + 0, // read only + create=0, // `OpenExisting` + append=false, + sync=0, + mode=0, + context~, + ) + if closed.val { + file.close() + raise @os_error.OSError(@os_error.errno_EPIPE, context~) + } + file + } + RedirectToFile::{ file, shared: false, closed } } ///| -priv struct TempPipeWrite { - pipe : @event_loop.IoHandle +priv struct RedirectToFile { + file : @async.Lazy[@event_loop.IoHandle] shared : Bool - mut closed : Bool + closed : Ref[Bool] } ///| -impl ProcessOutput for TempPipeWrite with fn fd(self) { - self.pipe.fd() -} - -///| -impl ProcessOutput for TempPipeWrite with fn do_close(self) { - if !self.closed { - self.closed = true - self.pipe.close() - } -} - -///| -impl ProcessOutput for TempPipeWrite with fn after_spawn(self) { - if !self.shared { - ProcessOutput::do_close(self) +fn RedirectToFile::close(self : RedirectToFile) -> Unit { + if !self.closed.val { + self.closed.val = true + try self.file.try_wait() catch { + _ => () + } noraise { + Some(file) => file.close() + None => () + } } } ///| -impl ProcessInput for TempPipeRead with fn fd(self) { - self.pipe.fd() -} - -///| -impl ProcessInput for TempPipeRead with fn do_close(self) { - if !self.closed { - self.closed = true - self.pipe.close() - } +impl ProcessInput for RedirectToFile with fn fd(self) { + self.file.wait().fd() } ///| -impl ProcessInput for TempPipeRead with fn after_spawn(self) { - ProcessInput::do_close(self) +impl ProcessInput for RedirectToFile with fn on_success(self) { + self.close() } ///| -priv struct RedirectToFile { - io : @event_loop.IoHandle - shared : Bool - mut closed : Bool +impl ProcessInput for RedirectToFile with fn on_failure(self) { + self.close() } ///| impl ProcessOutput for RedirectToFile with fn fd(self) { - self.io.fd() -} - -///| -impl ProcessOutput for RedirectToFile with fn do_close(self) { - if !self.closed { - self.closed = true - self.io.close() - } -} - -///| -impl ProcessOutput for RedirectToFile with fn after_spawn(self) { - if !self.shared { - ProcessOutput::do_close(self) - } + self.file.wait().fd() } ///| -impl ProcessInput for RedirectToFile with fn fd(self) { - self.io.fd() +impl ProcessOutput for RedirectToFile with fn is_shared(self) { + self.shared } ///| -impl ProcessInput for RedirectToFile with fn do_close(self) { - if !self.closed { - self.closed = true - self.io.close() - } +impl ProcessOutput for RedirectToFile with fn on_success(self) { + self.close() } ///| -impl ProcessInput for RedirectToFile with fn after_spawn(self) { - ProcessInput::do_close(self) +impl ProcessOutput for RedirectToFile with fn on_failure(self) { + self.close() } diff --git a/src/process/redirect_test.mbt b/src/process/redirect_test.mbt index c907223dd..eebb6371d 100644 --- a/src/process/redirect_test.mbt +++ b/src/process/redirect_test.mbt @@ -154,3 +154,48 @@ async test "pipe" { "writing: abcd", "received: abcd", "writing: xyzw", "received: xyzw", ]) } + +///| +async test "failure before spawn" { + let cat = cat.wait() + let file = "_build/does_not_exist" + let (r, w) = @process.write_to_process() + defer w.close() + let err = @test_util.expect_error_async <| () => { + @process.run( + cat, + [], + stdin=r, + stdout=@process.redirect_to_file(file, create_mode=TruncateExisting), + ) + } + assert_true(err is (@os_error.OSError(_) as err) && err.is_ENOENT()) +} + +///| +async test "pipe failure on first spawn" { + let cat = cat.wait() + let (r, w) = @process.pipe() + defer w.close() + let err = @test_util.expect_error_async <| () => { + @async.with_task_group <| group => { + let _ = @process.spawn(group, "no_such_program", [], stdout=w) + let _ = @process.spawn(group, cat, [], stdin=r) + } + } + assert_true(err is (@os_error.OSError(_) as err) && err.is_ENOENT()) +} + +///| +async test "pipe failure on second spawn" { + let cat = cat.wait() + let (r, w) = @process.pipe() + defer w.close() + let err = @test_util.expect_error_async <| () => { + @async.with_task_group <| group => { + let _ = @process.spawn(group, cat, [], stdin=r) + let _ = @process.spawn(group, "no_such_program", [], stdout=w) + } + } + assert_true(err is (@os_error.OSError(_) as err) && err.is_ENOENT()) +} diff --git a/src/process/unix.mbt b/src/process/unix.mbt index ada3fd2f7..a67075f7a 100644 --- a/src/process/unix.mbt +++ b/src/process/unix.mbt @@ -128,26 +128,26 @@ async fn raw_spawn_unix( for i, arg in args { argv.add_entry(i + 1, arg) } - defer { + errdefer { if stdin is Some(p) { - p.after_spawn() + p.on_failure() } - if stdout is Some(p) { - p.after_spawn() + if stdout is Some(p) && !p.is_shared() { + p.on_failure() } - if stderr is Some(p) { - p.after_spawn() + if stderr is Some(p) && !p.is_shared() { + p.on_failure() } } - let stdin = match stdin { + let stdin_fd = match stdin { Some(pipe) => pipe.fd() None => @fd_util.invalid_fd } - let stdout = match stdout { + let stdout_fd = match stdout { Some(pipe) => pipe.fd() None => @fd_util.invalid_fd } - let stderr = match stderr { + let stderr_fd = match stderr { Some(pipe) => pipe.fd() None => @fd_util.invalid_fd } @@ -155,17 +155,27 @@ async fn raw_spawn_unix( // so `is_orphan` makes no difference on actual spawning ignore(is_orphan) ignore(no_console_window) - @event_loop.spawn_unix( + let process = @event_loop.spawn_unix( cmd, argv.0, env=OsEnv::make(extra_env, inherit_env~).0, - stdin~, - stdout~, - stderr~, + stdin=stdin_fd, + stdout=stdout_fd, + stderr=stderr_fd, is_orphan~, cwd~, context~, ) + if stdin is Some(p) { + p.on_success() + } + if stdout is Some(p) && !p.is_shared() { + p.on_success() + } + if stderr is Some(p) && !p.is_shared() { + p.on_success() + } + process } ///| diff --git a/src/process/windows.mbt b/src/process/windows.mbt index 34e28bfd0..c26e55a80 100644 --- a/src/process/windows.mbt +++ b/src/process/windows.mbt @@ -164,40 +164,50 @@ async fn raw_spawn_windows( None => None Some(cwd) => Some(@os_string.encode(cwd)) } - defer { + errdefer { if stdin is Some(p) { - p.after_spawn() + p.on_failure() } - if stdout is Some(p) { - p.after_spawn() + if stdout is Some(p) && !p.is_shared() { + p.on_failure() } - if stderr is Some(p) { - p.after_spawn() + if stderr is Some(p) && !p.is_shared() { + p.on_failure() } } - let stdin = match stdin { + let stdin_fd = match stdin { Some(pipe) => pipe.fd() None => @fd_util.invalid_fd } - let stdout = match stdout { + let stdout_fd = match stdout { Some(pipe) => pipe.fd() None => @fd_util.invalid_fd } - let stderr = match stderr { + let stderr_fd = match stderr { Some(pipe) => pipe.fd() None => @fd_util.invalid_fd } - @event_loop.spawn_windows( + let process = @event_loop.spawn_windows( command_line, env=OsEnv::make(extra_env, inherit_env~).0, - stdin~, - stdout~, - stderr~, + stdin=stdin_fd, + stdout=stdout_fd, + stderr=stderr_fd, cwd~, no_console_window~, is_orphan~, context~, ) + if stdin is Some(p) { + p.on_success() + } + if stdout is Some(p) && !p.is_shared() { + p.on_success() + } + if stderr is Some(p) && !p.is_shared() { + p.on_success() + } + process } ///| diff --git a/src/shell/execute.mbt b/src/shell/execute.mbt index 26f5f733b..bdfb4d026 100644 --- a/src/shell/execute.mbt +++ b/src/shell/execute.mbt @@ -302,39 +302,23 @@ fn validate_output_limit(limit : Int) -> Unit raise ShellError { /// `None` means the child inherits the parent's descriptor. `capture` is the /// descriptor this run reads from, and is absent when the caller keeps nothing. /// -/// A file opened here is recorded in `created`, so that if a later setup step -/// fails before the spawn that would consume it, the handle is closed rather -/// than leaked. The spawn itself closes it through `after_spawn`. -async fn resolve_redirect( +/// File redirections are lazy process descriptors: the file is opened by +/// `@process.spawn`, and spawn closes the parent copy on success or failure. +fn resolve_redirect( redirect : Redirect, capture : &@process.ProcessOutput?, - created : Array[&@process.ProcessOutput], ) -> &@process.ProcessOutput? { match redirect { Capture => capture Inherit => None - ToFile(path) => { - let file = @process.redirect_to_file(path, create_mode=CreateOrTruncate) - created.push(file) - Some(file) - } - AppendToFile(path) => { - let file = @process.redirect_to_file( - path, - append=true, - create_mode=OpenOrCreate, - ) - created.push(file) - Some(file) - } - Discard => { - let file = @process.redirect_to_file( - null_device, - create_mode=OpenExisting, + ToFile(path) => + Some(@process.redirect_to_file(path, create_mode=CreateOrTruncate)) + AppendToFile(path) => + Some( + @process.redirect_to_file(path, append=true, create_mode=OpenOrCreate), ) - created.push(file) - Some(file) - } + Discard => + Some(@process.redirect_to_file(null_device, create_mode=OpenExisting)) } } @@ -503,12 +487,6 @@ fn take_line( @utf8.decode_lossy(bytes) } -///| -#warnings("-deprecated") -fn close_process_input(input : &@process.ProcessInput) -> Unit { - input.close() -} - ///| async fn run_pipeline(pipeline : Pipeline, sink : Sink) -> Output { let commands = pipeline.commands @@ -524,10 +502,10 @@ async fn run_pipeline(pipeline : Pipeline, sink : Sink) -> Output { // instead of blocking. `read_from_process` and `write_to_process` keep the // child's end blocking, which is the whole point of using them here. // - // Passing a process-owned end to `spawn` transfers that end: `after_spawn` - // closes the parent's copy on success or failure. A junction is created - // immediately before its producer is spawned, keeping any unconsumed input - // end local to that one junction if the producer itself fails to spawn. + // Redirection descriptors are lazy: a pipe or file is only opened while + // `spawn` is preparing the child, and `spawn` closes the parent's child-side + // copy on success or failure. This lets setup build descriptors freely + // without a failure cleanup list for unconsumed child redirections. let capture_stderr = sink is Buffer(_) @async.with_task_group() <| group => { let captured = Ref(0) @@ -537,26 +515,6 @@ async fn run_pipeline(pipeline : Pipeline, sink : Sink) -> Output { } let mut stdout_task : @async.Task[Bytes]? = None let stderr_tasks : Array[@async.Task[Bytes]?] = [] - // - // A child-side end exists before the spawn that consumes it, so a failing - // setup step — an output file that cannot be opened, a pipe that cannot - // be created — would otherwise leak every end already made. Closing an - // end is idempotent, and a no-op for a borrowed one such as the parent's - // own stdin, so the cleanup is simply: record every end ever created, and - // on failure close them all. Ends a spawn has consumed were already - // closed by its `after_spawn` hook — on success and on failure alike — - // and closing them again does nothing. - let created_inputs : Array[&@process.ProcessInput] = [] - let created_outputs : Array[&@process.ProcessOutput] = [] - errdefer { - for input in created_inputs { - close_process_input(input) - } - for output in created_outputs { - output.close() - } - } - // // Standard input for the first stage: a file the child reads itself, the // parent's own descriptor, or a pipe this task fills. let first_input : &@process.ProcessInput = match commands[0].stdin { @@ -570,7 +528,6 @@ async fn run_pipeline(pipeline : Pipeline, sink : Sink) -> Output { child_end } } - created_inputs.push(first_input) let processes : Array[@process.Process] = [] let mut next_stdin : &@process.ProcessInput? = None for index, cmd in commands { @@ -590,13 +547,10 @@ async fn run_pipeline(pipeline : Pipeline, sink : Sink) -> Output { let stdout = if index + 1 < count { let (child_stdin, child_stdout) = @process.pipe() next_stdin = Some(child_stdin) - created_inputs.push(child_stdin) - created_outputs.push(child_stdout) Some(child_stdout) } else { let final_capture = if !(sink is Inherited) && cmd.stdout is Capture { let (reader, child_stdout) = @process.read_from_process() - created_outputs.push(child_stdout) match sink { Buffer(_) => stdout_task = Some( @@ -616,13 +570,12 @@ async fn run_pipeline(pipeline : Pipeline, sink : Sink) -> Output { } else { None } - resolve_redirect(cmd.stdout, final_capture, created_outputs) + resolve_redirect(cmd.stdout, final_capture) } // // Each reader starts before its child, so a full pipe never stalls one. let stderr_capture = if capture_stderr && cmd.stderr is Capture { let (reader, child_stderr) = @process.read_from_process() - created_outputs.push(child_stderr) stderr_tasks.push( Some( group.spawn(() => { @@ -636,7 +589,7 @@ async fn run_pipeline(pipeline : Pipeline, sink : Sink) -> Output { stderr_tasks.push(None) None } - let stderr = resolve_redirect(cmd.stderr, stderr_capture, created_outputs) + let stderr = resolve_redirect(cmd.stderr, stderr_capture) processes.push(spawn_stage(group, cmd, stdin, stdout, stderr)) } let exit_codes = [ for child in processes => child.wait() ]