dolang_vfs/process/
mod.rs

1use std::{
2    fmt,
3    io::{self, IoSlice},
4    pin::Pin,
5    process::Stdio,
6    task::{Context, Poll, ready},
7    time::Duration,
8};
9
10use dolang_rpc::handle::DefaultHandle;
11use futures::io::{AsyncRead, AsyncWrite, AsyncWriteExt, BufReader};
12use serde::{Deserialize, Serialize};
13use tokio::{
14    io::{AsyncRead as TokioRead, AsyncWrite as TokioWrite, ReadBuf},
15    task::{JoinError, JoinHandle},
16};
17
18use crate::{
19    STREAM_CHUNK_SIZE, SessionMode, Vfs, VfsInner, client, direct,
20    error::{Error, ErrorKind, Result},
21    file::{File, FileInner},
22    path,
23    target::OperatingSystem,
24};
25
26mod foreign;
27
28pub(crate) use foreign::ProcessFamily;
29pub use foreign::{Process, ProcessExit, ProcessInfo, Processes, StartTime};
30
31/// Terminal status of a spawned process.
32#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
33pub enum ProcessStatus {
34    Exited(i32),
35    Signaled(i32),
36}
37
38impl ProcessStatus {
39    /// Returns whether the process exited successfully.
40    pub const fn success(self) -> bool {
41        matches!(self, Self::Exited(0))
42    }
43
44    /// Returns the numeric exit code, if the process exited normally.
45    pub const fn code(self) -> Option<i32> {
46        match self {
47            Self::Exited(code) => Some(code),
48            Self::Signaled(_) => None,
49        }
50    }
51
52    /// Returns the native signal number, if a signal terminated the process.
53    pub const fn signal(self) -> Option<i32> {
54        match self {
55            Self::Exited(_) => None,
56            Self::Signaled(signal) => Some(signal),
57        }
58    }
59
60    pub(crate) fn from_native(status: std::process::ExitStatus) -> io::Result<Self> {
61        if let Some(code) = status.code() {
62            return Ok(Self::Exited(code));
63        }
64        #[cfg(unix)]
65        {
66            use std::os::unix::process::ExitStatusExt;
67            if let Some(signal) = status.signal() {
68                return Ok(Self::Signaled(signal));
69            }
70        }
71        Err(io::Error::new(
72            io::ErrorKind::InvalidData,
73            "process returned an unrepresentable terminal status",
74        ))
75    }
76}
77
78/// Whether a spawned process is attached to the foreground process group.
79#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
80pub enum ProcessControl {
81    /// Run in the foreground process group.
82    Foreground,
83    /// Run without foreground process-group control.
84    Background,
85}
86
87/// A Unix process signal.
88#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
89#[non_exhaustive]
90pub enum Signal {
91    Hup,
92    Int,
93    Quit,
94    Ill,
95    Trap,
96    Abrt,
97    Emt,
98    Fpe,
99    Kill,
100    Bus,
101    Segv,
102    Sys,
103    Pipe,
104    Alrm,
105    Term,
106    Urg,
107    Stop,
108    Tstp,
109    Cont,
110    Chld,
111    Ttin,
112    Ttou,
113    Io,
114    Xcpu,
115    Xfsz,
116    Vtalrm,
117    Prof,
118    Winch,
119    Info,
120    Usr1,
121    Usr2,
122    Stkflt,
123    Pwr,
124    Thr,
125    Librt,
126    Number(i32),
127}
128
129impl Signal {
130    /// Returns whether this signal exists on an operating system.
131    pub fn is_supported(self, operating_system: OperatingSystem) -> bool {
132        use OperatingSystem::{FreeBsd, Linux, Macos, Windows};
133        match self {
134            Self::Emt | Self::Info => matches!(operating_system, FreeBsd | Macos),
135            Self::Stkflt | Self::Pwr => operating_system == Linux,
136            Self::Thr | Self::Librt => operating_system == FreeBsd,
137            Self::Number(_) => operating_system != Windows,
138            _ => operating_system != Windows,
139        }
140    }
141}
142enum CommandInner<'a> {
143    Client(client::Command<'a>),
144    Direct(direct::Command<'a>),
145}
146
147/// What a child reads as its standard input.
148pub enum ChildIn {
149    /// The read end of a pipe.
150    Pipe(PipeRecv),
151    /// A file, read from its cursor.
152    ///
153    /// Once the child has exited, the cursor is wherever the child left off,
154    /// so reading through the file picks up after what the child consumed.
155    /// Every clone shares the cursor, so it moves for all of them.
156    File(File),
157    /// A stream, copied into a pipe while the child runs.
158    ///
159    /// The copy may read past what the child consumed. Once the child exits,
160    /// the copy stops, and an error it met is reported by
161    /// [`Child::relay_error`] unless it is [`io::ErrorKind::BrokenPipe`].
162    Reader(Box<dyn AsyncRead + Send + Unpin>),
163}
164
165/// Where a child writes its standard output or standard error.
166pub enum ChildOut {
167    /// The write end of a pipe.
168    Pipe(PipeSend),
169    /// A file, written from its cursor.
170    ///
171    /// Once the child has exited, the cursor is wherever the child left off,
172    /// so writing through the file continues after the child's output. Every
173    /// clone shares the cursor, so it moves for all of them.
174    File(File),
175    /// A stream, copied from a pipe while the child runs.
176    ///
177    /// The stream is closed once the child's output ends, and
178    /// [`Child::wait`] waits for that. An error the copy met is reported by
179    /// [`Child::relay_error`].
180    Writer(Box<dyn AsyncWrite + Send + Unpin>),
181}
182
183impl ChildIn {
184    /// Creates a stream endpoint.
185    pub fn reader(reader: impl AsyncRead + Send + Unpin + 'static) -> Self {
186        Self::Reader(Box::new(reader))
187    }
188}
189
190impl ChildOut {
191    /// Creates a stream endpoint.
192    pub fn writer(writer: impl AsyncWrite + Send + Unpin + 'static) -> Self {
193        Self::Writer(Box::new(writer))
194    }
195}
196
197impl fmt::Debug for ChildIn {
198    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
199        match self {
200            Self::Pipe(pipe) => f.debug_tuple("Pipe").field(pipe).finish(),
201            Self::File(file) => f.debug_tuple("File").field(file).finish(),
202            Self::Reader(_) => f.debug_tuple("Reader").finish_non_exhaustive(),
203        }
204    }
205}
206
207impl fmt::Debug for ChildOut {
208    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
209        match self {
210            Self::Pipe(pipe) => f.debug_tuple("Pipe").field(pipe).finish(),
211            Self::File(file) => f.debug_tuple("File").field(file).finish(),
212            Self::Writer(_) => f.debug_tuple("Writer").finish_non_exhaustive(),
213        }
214    }
215}
216
217/// A standard input endpoint a spawn target takes as is, once
218/// [`classify_recv`] has relayed whatever it can't.
219pub(crate) enum SpawnIn {
220    Pipe(PipeRecv),
221    File(File),
222}
223
224/// A standard output or error endpoint a spawn target takes as is, once
225/// [`classify_send`] has relayed whatever it can't.
226pub(crate) enum SpawnOut {
227    Pipe(PipeSend),
228    File(File),
229}
230
231impl From<PipeRecv> for ChildIn {
232    fn from(pipe: PipeRecv) -> Self {
233        Self::Pipe(pipe)
234    }
235}
236
237impl From<File> for ChildIn {
238    fn from(file: File) -> Self {
239        Self::File(file)
240    }
241}
242
243impl From<PipeSend> for ChildOut {
244    fn from(pipe: PipeSend) -> Self {
245        Self::Pipe(pipe)
246    }
247}
248
249impl From<File> for ChildOut {
250    fn from(file: File) -> Self {
251        Self::File(file)
252    }
253}
254
255/// Builds a process to spawn, relaying stdio across VFS domains when a
256/// given endpoint cannot be consumed directly by the spawn target.
257pub struct Command<'a> {
258    inner: CommandInner<'a>,
259    vfs: Vfs,
260    stdin: Option<ChildIn>,
261    stdout: Option<ChildOut>,
262    stderr: Option<ChildOut>,
263}
264
265impl<'a> Command<'a> {
266    pub(crate) fn new(vfs: &'a Vfs, program: path::Path<'_>) -> Self {
267        let inner = match &vfs.inner {
268            VfsInner::Client(client) => CommandInner::Client(client.command(program)),
269            VfsInner::Direct(direct) => CommandInner::Direct(direct.command(program)),
270        };
271        Self {
272            inner,
273            vfs: vfs.clone(),
274            stdin: None,
275            stdout: None,
276            stderr: None,
277        }
278    }
279}
280
281enum ChildInner {
282    Client(client::Child),
283    Direct(Box<direct::Child>),
284}
285
286/// Tasks pumping bytes between a foreign-domain stdio endpoint and a pipe
287/// created in the spawn target's own domain, not yet started.
288///
289/// Relay tasks are only started once the underlying process has actually
290/// been spawned, so a failure setting up the spawn itself just drops the
291/// held endpoints (triggering their own best-effort cleanup) instead of
292/// leaking a running task.
293///
294/// A file endpoint is relayed through the file itself, so its cursor follows
295/// what the relay read or wrote.
296#[derive(Default)]
297struct PendingRelays {
298    inputs: Vec<(RelaySource, PipeSend)>,
299    outputs: Vec<(PipeRecv, RelaySink)>,
300}
301
302type RelaySource = Box<dyn AsyncRead + Send + Unpin>;
303type RelaySink = Box<dyn AsyncWrite + Send + Unpin>;
304
305impl PendingRelays {
306    fn start(self) -> ActiveRelays {
307        ActiveRelays {
308            inputs: self
309                .inputs
310                .into_iter()
311                .map(|(src, dst)| tokio::spawn(relay(src, dst)))
312                .collect(),
313            outputs: self
314                .outputs
315                .into_iter()
316                .map(|(src, dst)| tokio::spawn(relay(src, dst)))
317                .collect(),
318            error: None,
319        }
320    }
321}
322
323#[derive(Default)]
324struct ActiveRelays {
325    inputs: Vec<JoinHandle<io::Result<()>>>,
326    outputs: Vec<JoinHandle<io::Result<()>>>,
327    /// The first error a finished relay reported.
328    error: Option<io::Error>,
329}
330
331impl ActiveRelays {
332    fn abort_inputs(&mut self) {
333        for handle in self.inputs.drain(..) {
334            handle.abort();
335        }
336    }
337
338    /// Keeps `result`'s error if it is the first.
339    fn record(&mut self, result: std::result::Result<io::Result<()>, JoinError>) {
340        let error = match result {
341            Ok(Ok(())) => return,
342            Ok(Err(error)) => error,
343            Err(error) => match error.try_into_panic() {
344                Ok(panic) => std::panic::resume_unwind(panic),
345                Err(_) => return,
346            },
347        };
348        self.error.get_or_insert(error);
349    }
350
351    /// Waits for output relays to finish flushing everything the child
352    /// already produced before it exited. Input relays are aborted instead
353    /// of awaited, since no further input can matter once the child is gone.
354    ///
355    /// Once the child has exited, its end of each output pipe is closed, so
356    /// the relay's copy loop reaches EOF and returns on its own; this just
357    /// ensures the caller can't observe a partially-relayed destination.
358    ///
359    /// An input relay that already finished keeps its error, unless the
360    /// child closed its end of the pipe first. Every output relay keeps its
361    /// error.
362    async fn finish(&mut self) {
363        for handle in std::mem::take(&mut self.inputs) {
364            if !handle.is_finished() {
365                handle.abort();
366                continue;
367            }
368            match handle.await {
369                Ok(Err(error)) if error.kind() == io::ErrorKind::BrokenPipe => {}
370                result => self.record(result),
371            }
372        }
373        for handle in std::mem::take(&mut self.outputs) {
374            let result = handle.await;
375            self.record(result);
376        }
377    }
378
379    /// Aborts every relay task without waiting, for use when giving up on
380    /// the child without having confirmed it actually exited (e.g. `Drop`).
381    fn abandon(&mut self) {
382        self.abort_inputs();
383        for handle in self.outputs.drain(..) {
384            handle.abort();
385        }
386    }
387}
388
389impl Drop for ActiveRelays {
390    fn drop(&mut self) {
391        self.abandon();
392    }
393}
394
395/// A spawned process, possibly relaying stdio across VFS domains.
396pub struct Child {
397    inner: ChildInner,
398    relays: ActiveRelays,
399}
400
401impl Child {
402    /// Takes the first error a stream relay met, once [`wait`](Self::wait)
403    /// has finished the relays.
404    ///
405    /// The process status is reported apart from this, so a caller can
406    /// decide which matters more.
407    pub fn relay_error(&mut self) -> Option<io::Error> {
408        self.relays.error.take()
409    }
410
411    /// Waits for the process to exit.
412    pub async fn wait(&mut self) -> Result<ProcessStatus> {
413        let status = match &mut self.inner {
414            ChildInner::Client(child) => child.wait().await,
415            ChildInner::Direct(child) => child.wait().await,
416        }?;
417        self.relays.finish().await;
418        Ok(status)
419    }
420
421    /// Terminates the process and returns its status when it has exited.
422    pub async fn terminate(mut self) -> Result<Option<ProcessStatus>> {
423        self.relays.abort_inputs();
424        let result = match self.inner {
425            ChildInner::Client(child) => child.terminate().await,
426            ChildInner::Direct(child) => child.terminate().await,
427        };
428        if result.as_ref().is_ok_and(Option::is_some) {
429            self.relays.finish().await;
430        } else {
431            self.relays.abandon();
432        }
433        result
434    }
435}
436
437/// Returns whether `stdio` can be handed directly to a process spawned in
438/// `target`'s domain without relaying.
439fn is_direct_recv(target: &Vfs, stdio: &PipeRecv) -> bool {
440    match (&target.inner, &stdio.0) {
441        (VfsInner::Direct(_), PipeRecvInner::Native(_)) => true,
442        (VfsInner::Client(client), PipeRecvInner::Remote(remote)) => {
443            client.is_same_vfs(remote.client())
444        }
445        (VfsInner::Client(client), PipeRecvInner::Native(_)) => {
446            client.mode() == SessionMode::Native
447        }
448        _ => false,
449    }
450}
451
452/// Returns whether `stdio` can be handed directly to a process spawned in
453/// `target`'s domain without relaying.
454fn is_direct_send(target: &Vfs, stdio: &PipeSend) -> bool {
455    match (&target.inner, &stdio.0) {
456        (VfsInner::Direct(_), PipeSendInner::Native(_)) => true,
457        (VfsInner::Client(client), PipeSendInner::Remote(remote)) => {
458            client.is_same_vfs(remote.client())
459        }
460        (VfsInner::Client(client), PipeSendInner::Native(_)) => {
461            client.mode() == SessionMode::Native
462        }
463        _ => false,
464    }
465}
466
467/// Returns whether `file` can be handed directly to a process spawned in
468/// `target`'s domain without relaying.
469fn is_direct_file(target: &Vfs, file: &File) -> bool {
470    match (&target.inner, &file.inner) {
471        (VfsInner::Direct(_), FileInner::Direct(_)) => true,
472        (VfsInner::Client(client), FileInner::Client(file)) => client.is_same_vfs(file.client()),
473        (VfsInner::Client(client), FileInner::Direct(_)) => client.mode() == SessionMode::Native,
474        _ => false,
475    }
476}
477
478/// Classifies a stdin endpoint for a process about to be spawned in
479/// `target`'s domain, creating a relay pipe and queuing a pump task in
480/// `relays` if the endpoint cannot be consumed directly.
481async fn classify_recv(
482    target: &Vfs,
483    stdio: ChildIn,
484    relays: &mut PendingRelays,
485) -> Result<SpawnIn> {
486    let source: RelaySource = match stdio {
487        ChildIn::Pipe(pipe) if is_direct_recv(target, &pipe) => return Ok(SpawnIn::Pipe(pipe)),
488        ChildIn::File(file) if is_direct_file(target, &file) => return Ok(SpawnIn::File(file)),
489        ChildIn::Pipe(pipe) => Box::new(pipe),
490        ChildIn::File(file) => Box::new(file),
491        ChildIn::Reader(reader) => reader,
492    };
493    let (send, recv) = target.pipe(None).await?;
494    relays.inputs.push((source, send));
495    Ok(SpawnIn::Pipe(recv))
496}
497
498/// Classifies a stdout/stderr endpoint for a process about to be spawned in
499/// `target`'s domain, creating a relay pipe and queuing a pump task in
500/// `relays` if the endpoint cannot be consumed directly.
501async fn classify_send(
502    target: &Vfs,
503    stdio: ChildOut,
504    relays: &mut PendingRelays,
505) -> Result<SpawnOut> {
506    let sink: RelaySink = match stdio {
507        ChildOut::Pipe(pipe) if is_direct_send(target, &pipe) => return Ok(SpawnOut::Pipe(pipe)),
508        ChildOut::File(file) if is_direct_file(target, &file) => return Ok(SpawnOut::File(file)),
509        ChildOut::Pipe(pipe) => Box::new(pipe),
510        ChildOut::File(file) => Box::new(file),
511        ChildOut::Writer(writer) => writer,
512    };
513    let (send, recv) = target.pipe(None).await?;
514    relays.outputs.push((recv, sink));
515    Ok(SpawnOut::Pipe(send))
516}
517
518impl<'a> Command<'a> {
519    /// Appends an argument to the program invocation.
520    pub fn arg(&mut self, arg: &str) -> &mut Self {
521        match &mut self.inner {
522            CommandInner::Client(builder) => {
523                builder.arg(arg);
524            }
525            CommandInner::Direct(builder) => {
526                builder.arg(arg);
527            }
528        }
529        self
530    }
531
532    /// Sets an environment variable for the child.
533    pub fn env(&mut self, key: &str, val: &str) -> &mut Self {
534        match &mut self.inner {
535            CommandInner::Client(builder) => {
536                builder.env(key, val);
537            }
538            CommandInner::Direct(builder) => {
539                builder.env(key, val);
540            }
541        }
542        self
543    }
544
545    /// Removes an environment variable from the child.
546    pub fn env_remove(&mut self, key: &str) -> &mut Self {
547        match &mut self.inner {
548            CommandInner::Client(builder) => {
549                builder.env_remove(key);
550            }
551            CommandInner::Direct(builder) => {
552                builder.env_remove(key);
553            }
554        }
555        self
556    }
557
558    /// Sets the child's working directory.
559    pub fn current_dir(&mut self, dir: path::Path<'_>) -> &mut Self {
560        match &mut self.inner {
561            CommandInner::Client(builder) => {
562                builder.current_dir(dir);
563            }
564            CommandInner::Direct(builder) => {
565                builder.current_dir(dir);
566            }
567        }
568        self
569    }
570
571    /// Sets the child's standard input.
572    pub fn stdin(&mut self, stdio: impl Into<ChildIn>) -> Result<&mut Self> {
573        self.stdin = Some(stdio.into());
574        Ok(self)
575    }
576
577    /// Sets the child's standard output.
578    pub fn stdout(&mut self, stdio: impl Into<ChildOut>) -> Result<&mut Self> {
579        self.stdout = Some(stdio.into());
580        Ok(self)
581    }
582
583    /// Inherit the host process's standard input.
584    ///
585    /// Opaque remote clients treat terminal input as null because Tokio cannot
586    /// cancel an outstanding terminal read. Redirected input is relayed to the
587    /// remote process.
588    pub fn stdin_inherit(&mut self) -> Result<&mut Self> {
589        self.stdin = None;
590        match &mut self.inner {
591            CommandInner::Client(builder) => {
592                builder.stdin_inherit()?;
593            }
594            CommandInner::Direct(builder) => {
595                builder.stdin_inherit()?;
596            }
597        }
598        Ok(self)
599    }
600
601    /// Inherits the host process's standard output.
602    pub fn stdout_inherit(&mut self) -> Result<&mut Self> {
603        self.stdout = None;
604        match &mut self.inner {
605            CommandInner::Client(builder) => {
606                builder.stdout_inherit()?;
607            }
608            CommandInner::Direct(builder) => {
609                builder.stdout_inherit()?;
610            }
611        }
612        Ok(self)
613    }
614
615    /// Connects the child's standard output to the parent process's standard
616    /// error.
617    pub fn stdout_inherit_stderr(&mut self) -> Result<&mut Self> {
618        self.stdout = None;
619        match &mut self.inner {
620            CommandInner::Client(builder) => {
621                builder.stdout_inherit_stderr()?;
622            }
623            CommandInner::Direct(builder) => {
624                builder.stdout_inherit_stderr()?;
625            }
626        }
627        Ok(self)
628    }
629
630    /// Connects the child's standard input to the null device.
631    pub fn stdin_null(&mut self) -> &mut Self {
632        self.stdin = None;
633        match &mut self.inner {
634            CommandInner::Client(builder) => {
635                builder.stdin_null();
636            }
637            CommandInner::Direct(builder) => {
638                builder.stdin_null();
639            }
640        }
641        self
642    }
643
644    /// Connects the child's standard output to the null device.
645    pub fn stdout_null(&mut self) -> &mut Self {
646        self.stdout = None;
647        match &mut self.inner {
648            CommandInner::Client(builder) => {
649                builder.stdout_null();
650            }
651            CommandInner::Direct(builder) => {
652                builder.stdout_null();
653            }
654        }
655        self
656    }
657
658    /// Sets the child's standard error.
659    pub fn stderr(&mut self, stdio: impl Into<ChildOut>) -> Result<&mut Self> {
660        self.stderr = Some(stdio.into());
661        Ok(self)
662    }
663
664    /// Sets the child's standard input to a native handle, used as is.
665    ///
666    /// Only a local command can take one.
667    pub(crate) fn stdin_native(&mut self, stdio: Stdio) -> Result<&mut Self> {
668        self.stdin = None;
669        self.direct_inner()?.stdin_native(stdio);
670        Ok(self)
671    }
672
673    /// Sets the child's standard output to a native handle, used as is.
674    ///
675    /// Only a local command can take one.
676    pub(crate) fn stdout_native(&mut self, stdio: Stdio) -> Result<&mut Self> {
677        self.stdout = None;
678        self.direct_inner()?.stdout_native(stdio);
679        Ok(self)
680    }
681
682    /// Sets the child's standard error to a native handle, used as is.
683    ///
684    /// Only a local command can take one.
685    pub(crate) fn stderr_native(&mut self, stdio: Stdio) -> Result<&mut Self> {
686        self.stderr = None;
687        self.direct_inner()?.stderr_native(stdio);
688        Ok(self)
689    }
690
691    fn direct_inner(&mut self) -> Result<&mut direct::Command<'a>> {
692        match &mut self.inner {
693            CommandInner::Direct(builder) => Ok(builder),
694            CommandInner::Client(_) => Err(Error::new(
695                ErrorKind::InvalidInput,
696                "a native handle cannot be handed to a remote process",
697            )),
698        }
699    }
700
701    /// Inherits the host process's standard error.
702    pub fn stderr_inherit(&mut self) -> Result<&mut Self> {
703        self.stderr = None;
704        match &mut self.inner {
705            CommandInner::Client(builder) => {
706                builder.stderr_inherit()?;
707            }
708            CommandInner::Direct(builder) => {
709                builder.stderr_inherit()?;
710            }
711        }
712        Ok(self)
713    }
714
715    /// Connects the child's standard error to the same destination as its
716    /// configured standard output.
717    pub fn stderr_to_stdout(&mut self) -> Result<&mut Self> {
718        self.stderr = None;
719        match &mut self.inner {
720            CommandInner::Client(builder) => {
721                builder.stderr_to_stdout()?;
722            }
723            CommandInner::Direct(builder) => {
724                builder.stderr_to_stdout()?;
725            }
726        }
727        Ok(self)
728    }
729
730    /// Connects the child's standard error to the parent process's standard
731    /// output.
732    pub fn stderr_inherit_stdout(&mut self) -> Result<&mut Self> {
733        self.stderr = None;
734        match &mut self.inner {
735            CommandInner::Client(builder) => {
736                builder.stderr_inherit_stdout()?;
737            }
738            CommandInner::Direct(builder) => {
739                builder.stderr_inherit_stdout()?;
740            }
741        }
742        Ok(self)
743    }
744
745    /// Connects the child's standard error to the null device.
746    pub fn stderr_null(&mut self) -> &mut Self {
747        self.stderr = None;
748        match &mut self.inner {
749            CommandInner::Client(builder) => {
750                builder.stderr_null();
751            }
752            CommandInner::Direct(builder) => {
753                builder.stderr_null();
754            }
755        }
756        self
757    }
758
759    /// Sets foreground or background process control behavior.
760    pub fn process_control(&mut self, control: ProcessControl) -> &mut Self {
761        match &mut self.inner {
762            CommandInner::Client(builder) => {
763                builder.process_control(control);
764            }
765            CommandInner::Direct(builder) => {
766                builder.process_control(control);
767            }
768        }
769        self
770    }
771
772    /// Sets the policy used to terminate the child.
773    pub fn termination_policy(&mut self, policy: TerminationPolicy) -> &mut Self {
774        match &mut self.inner {
775            CommandInner::Client(builder) => {
776                builder.termination_policy(policy);
777            }
778            CommandInner::Direct(builder) => {
779                builder.termination_policy(policy);
780            }
781        }
782        self
783    }
784
785    /// Spawns the configured process.
786    pub async fn spawn(mut self) -> Result<Child> {
787        let mut relays = PendingRelays::default();
788        if let Some(stdio) = self.stdin.take() {
789            let stdio = classify_recv(&self.vfs, stdio, &mut relays).await?;
790            match &mut self.inner {
791                CommandInner::Client(builder) => {
792                    builder.stdin(stdio)?;
793                }
794                CommandInner::Direct(builder) => {
795                    builder.stdin(stdio)?;
796                }
797            }
798        }
799        if let Some(stdio) = self.stdout.take() {
800            let stdio = classify_send(&self.vfs, stdio, &mut relays).await?;
801            match &mut self.inner {
802                CommandInner::Client(builder) => {
803                    builder.stdout(stdio)?;
804                }
805                CommandInner::Direct(builder) => {
806                    builder.stdout(stdio)?;
807                }
808            }
809        }
810        if let Some(stdio) = self.stderr.take() {
811            let stdio = classify_send(&self.vfs, stdio, &mut relays).await?;
812            match &mut self.inner {
813                CommandInner::Client(builder) => {
814                    builder.stderr(stdio)?;
815                }
816                CommandInner::Direct(builder) => {
817                    builder.stderr(stdio)?;
818                }
819            }
820        }
821        let inner = match self.inner {
822            CommandInner::Client(builder) => builder.spawn().await.map(ChildInner::Client),
823            CommandInner::Direct(builder) => builder
824                .spawn()
825                .await
826                .map(|x| ChildInner::Direct(Box::new(x))),
827        }?;
828        Ok(Child {
829            inner,
830            relays: relays.start(),
831        })
832    }
833}
834
835/// Policy used to terminate a spawned process.
836#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
837pub struct TerminationPolicy {
838    /// Signal sent to request graceful termination.
839    pub(crate) signal: Signal,
840    /// Time to wait before optional forced termination.
841    pub(crate) grace: Duration,
842    /// Whether to force termination after the grace period.
843    pub(crate) force: bool,
844}
845
846impl TerminationPolicy {
847    /// Creates a process-termination policy.
848    pub const fn new(signal: Signal, grace: Duration, force: bool) -> Self {
849        Self {
850            signal,
851            grace,
852            force,
853        }
854    }
855    /// Returns the signal used to request graceful termination.
856    pub const fn signal(self) -> Signal {
857        self.signal
858    }
859    /// Returns the grace period before forced termination.
860    pub const fn grace(self) -> Duration {
861        self.grace
862    }
863    /// Returns whether the process is forcibly terminated after the grace period.
864    pub const fn force(self) -> bool {
865        self.force
866    }
867    /// Sets the graceful-termination signal.
868    pub fn set_signal(&mut self, signal: Signal) -> &mut Self {
869        self.signal = signal;
870        self
871    }
872    /// Sets the grace period before forced termination.
873    pub fn set_grace(&mut self, grace: Duration) -> &mut Self {
874        self.grace = grace;
875        self
876    }
877    /// Selects whether to force termination after the grace period.
878    pub fn set_force(&mut self, force: bool) -> &mut Self {
879        self.force = force;
880        self
881    }
882}
883
884impl Default for TerminationPolicy {
885    fn default() -> Self {
886        Self {
887            signal: Signal::Term,
888            grace: Duration::from_secs(5),
889            force: true,
890        }
891    }
892}
893
894#[cfg(test)]
895mod policy_tests {
896    use super::{Signal, TerminationPolicy};
897    use std::time::Duration;
898
899    #[test]
900    fn termination_policy_construction_and_mutation() {
901        let mut policy = TerminationPolicy::new(Signal::Int, Duration::from_secs(2), false);
902        assert_eq!(policy.signal(), Signal::Int);
903        assert_eq!(policy.grace(), Duration::from_secs(2));
904        assert!(!policy.force());
905        policy
906            .set_signal(Signal::Term)
907            .set_grace(Duration::from_secs(3))
908            .set_force(true);
909        assert_eq!(
910            policy,
911            TerminationPolicy::new(Signal::Term, Duration::from_secs(3), true)
912        );
913    }
914}
915
916#[cfg(unix)]
917use std::os::fd::{AsFd, OwnedFd};
918#[cfg(windows)]
919use std::{
920    io::{PipeReader, PipeWriter, Read as _, Write as _},
921    os::windows::io::OwnedHandle,
922    sync::Arc,
923};
924/// A writable standard-I/O endpoint from either a local or remote VFS.
925#[derive(Debug)]
926pub struct PipeSend(pub(crate) PipeSendInner);
927
928#[derive(Debug)]
929pub(crate) enum PipeSendInner {
930    /// Endpoint backed by a local OS resource.
931    Native(NativePipeSend),
932    /// Endpoint retained by a remote VFS session.
933    Remote(crate::client::RemotePipeSend),
934}
935
936/// A readable standard-I/O endpoint from either a local or remote VFS.
937#[derive(Debug)]
938pub struct PipeRecv(pub(crate) PipeRecvInner);
939
940#[derive(Debug)]
941pub(crate) enum PipeRecvInner {
942    /// Endpoint backed by a local OS resource.
943    Native(NativePipeRecv),
944    /// Endpoint retained by a remote VFS session.
945    Remote(crate::client::RemotePipeRecv),
946}
947
948/// Local pipe sender.
949#[cfg(unix)]
950pub(crate) type NativePipeSend = tokio::net::unix::pipe::Sender;
951
952/// Local pipe receiver.
953#[cfg(unix)]
954pub(crate) type NativePipeRecv = tokio::net::unix::pipe::Receiver;
955
956/// Local pipe sender, written on blocking workers.
957#[cfg(windows)]
958#[derive(Debug)]
959pub(crate) struct NativePipeSend {
960    inner: Arc<PipeWriter>,
961    pending: Option<JoinHandle<io::Result<usize>>>,
962}
963
964/// Local pipe receiver, read on blocking workers.
965#[cfg(windows)]
966#[derive(Debug)]
967pub(crate) struct NativePipeRecv {
968    inner: Arc<PipeReader>,
969    pending: Option<JoinHandle<(Vec<u8>, io::Result<usize>)>>,
970    ready: Option<(Vec<u8>, usize)>,
971}
972
973/// Pumps bytes from `src` to `dst` until EOF or error, then closes `dst`,
974/// returning the first error.
975pub(crate) async fn relay<R, W>(src: R, mut dst: W) -> io::Result<()>
976where
977    R: AsyncRead + Unpin,
978    W: AsyncWrite + Unpin,
979{
980    let mut src = BufReader::with_capacity(STREAM_CHUNK_SIZE, src);
981    let copied = futures::io::copy_buf(&mut src, &mut dst).await;
982    let closed = dst.close().await;
983    copied?;
984    closed
985}
986
987/// Creates a native OS pipe, optionally hinting a kernel buffer size.
988///
989/// The hint is best-effort: on platforms/configurations where it can't be
990/// honored, the pipe is created with its default buffer size instead of
991/// failing outright.
992pub(crate) fn pipe(buf_size: Option<usize>) -> io::Result<(PipeSend, PipeRecv)> {
993    #[cfg(unix)]
994    {
995        let (send, recv) = tokio::net::unix::pipe::pipe()?;
996        if let Some(size) = buf_size {
997            use std::os::fd::AsRawFd;
998            set_pipe_buffer_size(send.as_raw_fd(), size);
999        }
1000        Ok((
1001            PipeSend(PipeSendInner::Native(send)),
1002            PipeRecv(PipeRecvInner::Native(recv)),
1003        ))
1004    }
1005    #[cfg(windows)]
1006    {
1007        let (recv, send) = match buf_size {
1008            Some(size) => create_pipe_sized(size)?,
1009            None => std::io::pipe()?,
1010        };
1011        Ok((
1012            PipeSend(PipeSendInner::Native(NativePipeSend {
1013                inner: Arc::new(send),
1014                pending: None,
1015            })),
1016            PipeRecv(PipeRecvInner::Native(NativePipeRecv {
1017                inner: Arc::new(recv),
1018                pending: None,
1019                ready: None,
1020            })),
1021        ))
1022    }
1023}
1024
1025/// Resizes a pipe's kernel buffer via `fcntl(F_SETPIPE_SZ)`, applied to
1026/// either end (the buffer is shared by both). No raw-handle bypass is
1027/// needed: tokio's pipe types expose the fd directly via `AsRawFd`.
1028#[cfg(target_os = "linux")]
1029pub(crate) fn set_pipe_buffer_size(fd: std::os::fd::RawFd, size: usize) {
1030    let size = i32::try_from(size).unwrap_or(i32::MAX);
1031    // Best-effort: failure (e.g. exceeding /proc/sys/fs/pipe-max-size
1032    // without CAP_SYS_RESOURCE) just leaves the default buffer size.
1033    unsafe {
1034        libc::fcntl(fd, libc::F_SETPIPE_SZ, size);
1035    }
1036}
1037
1038#[cfg(all(unix, not(target_os = "linux")))]
1039pub(crate) fn set_pipe_buffer_size(_fd: std::os::fd::RawFd, _size: usize) {}
1040
1041/// Creates an anonymous pipe with a requested kernel buffer size.
1042///
1043/// `std::io::pipe` has no size parameter, so this bypasses it and calls
1044/// `CreatePipe` directly with `bInheritHandle = FALSE`, matching the
1045/// non-inheritable handles `std::io::pipe` itself produces.
1046#[cfg(windows)]
1047fn create_pipe_sized(size: usize) -> io::Result<(std::io::PipeReader, std::io::PipeWriter)> {
1048    use std::os::windows::io::FromRawHandle;
1049
1050    use windows_sys::Win32::{
1051        Foundation::HANDLE, Security::SECURITY_ATTRIBUTES, System::Pipes::CreatePipe,
1052    };
1053
1054    let mut read_handle: HANDLE = std::ptr::null_mut();
1055    let mut write_handle: HANDLE = std::ptr::null_mut();
1056    let attrs = SECURITY_ATTRIBUTES {
1057        nLength: size_of::<SECURITY_ATTRIBUTES>() as u32,
1058        lpSecurityDescriptor: std::ptr::null_mut(),
1059        bInheritHandle: 0,
1060    };
1061    let size = u32::try_from(size).unwrap_or(u32::MAX);
1062    // SAFETY: `read_handle`/`write_handle` are valid out-params for the
1063    // duration of this call; `attrs` lives on the stack until it returns.
1064    let ok = unsafe { CreatePipe(&mut read_handle, &mut write_handle, &attrs, size) };
1065    if ok == 0 {
1066        return Err(io::Error::last_os_error());
1067    }
1068    // SAFETY: `read_handle`/`write_handle` are freshly created, uniquely
1069    // owned handles from a successful `CreatePipe` call.
1070    let reader = unsafe { std::io::PipeReader::from_raw_handle(read_handle as _) };
1071    let writer = unsafe { std::io::PipeWriter::from_raw_handle(write_handle as _) };
1072    Ok((reader, writer))
1073}
1074
1075impl PipeSend {
1076    pub(crate) fn remote(remote: crate::client::RemotePipeSend) -> Self {
1077        Self(PipeSendInner::Remote(remote))
1078    }
1079
1080    /// Clones this writable child-stdio endpoint.
1081    pub async fn try_clone(&self) -> Result<Self> {
1082        match &self.0 {
1083            #[cfg(unix)]
1084            PipeSendInner::Native(pipe) => {
1085                let fd = pipe.as_fd().try_clone_to_owned()?;
1086                Ok(Self(PipeSendInner::Native(
1087                    tokio::net::unix::pipe::Sender::from_owned_fd_unchecked(fd)?,
1088                )))
1089            }
1090            #[cfg(windows)]
1091            PipeSendInner::Native(NativePipeSend { inner, .. }) => {
1092                Ok(Self(PipeSendInner::Native(NativePipeSend {
1093                    inner: Arc::new(inner.try_clone()?),
1094                    pending: None,
1095                })))
1096            }
1097            PipeSendInner::Remote(remote) => Ok(Self::remote(remote.try_clone().await?)),
1098        }
1099    }
1100
1101    /// Converts this endpoint into native process stdio.
1102    ///
1103    /// Remote endpoints cannot be converted.
1104    pub async fn into_stdio(self) -> Result<Stdio> {
1105        match self.0 {
1106            #[cfg(unix)]
1107            PipeSendInner::Native(pipe) => {
1108                let fd: OwnedFd = pipe.into_blocking_fd()?;
1109                Ok(Stdio::from(fd))
1110            }
1111            #[cfg(windows)]
1112            PipeSendInner::Native(NativePipeSend { inner, pending }) => {
1113                if pending.is_some() {
1114                    return Err(Error::new(
1115                        ErrorKind::ResourceBusy,
1116                        "cannot convert PipeSend while an async write is in flight",
1117                    ));
1118                }
1119                Ok(Arc::try_unwrap(inner)
1120                    .or_else(|inner| inner.try_clone())
1121                    .map(Stdio::from)?)
1122            }
1123            PipeSendInner::Remote(_) => Err(Error::new(
1124                ErrorKind::InvalidInput,
1125                "remote stdio cannot be converted to a native handle",
1126            )),
1127        }
1128    }
1129
1130    pub(crate) async fn into_blocking_handle(self) -> io::Result<DefaultHandle> {
1131        match self.0 {
1132            #[cfg(unix)]
1133            PipeSendInner::Native(pipe) => pipe.into_blocking_fd(),
1134            #[cfg(windows)]
1135            PipeSendInner::Native(NativePipeSend { inner, pending }) => {
1136                if pending.is_some() {
1137                    return Err(io::Error::other(
1138                        "cannot convert PipeSend while an async write is in flight",
1139                    ));
1140                }
1141                let pipe = Arc::try_unwrap(inner).or_else(|inner| inner.try_clone())?;
1142                Ok(OwnedHandle::from(pipe))
1143            }
1144            PipeSendInner::Remote(_) => Err(io::Error::new(
1145                io::ErrorKind::InvalidInput,
1146                "remote stdio has no native handle",
1147            )),
1148        }
1149    }
1150}
1151
1152impl PipeRecv {
1153    pub(crate) fn remote(remote: crate::client::RemotePipeRecv) -> Self {
1154        Self(PipeRecvInner::Remote(remote))
1155    }
1156
1157    /// Clones this readable child-stdio endpoint.
1158    pub async fn try_clone(&self) -> Result<Self> {
1159        match &self.0 {
1160            #[cfg(unix)]
1161            PipeRecvInner::Native(pipe) => {
1162                let fd = pipe.as_fd().try_clone_to_owned()?;
1163                Ok(Self(PipeRecvInner::Native(
1164                    tokio::net::unix::pipe::Receiver::from_owned_fd_unchecked(fd)?,
1165                )))
1166            }
1167            #[cfg(windows)]
1168            PipeRecvInner::Native(NativePipeRecv { inner, .. }) => {
1169                Ok(Self(PipeRecvInner::Native(NativePipeRecv {
1170                    inner: Arc::new(inner.try_clone()?),
1171                    pending: None,
1172                    ready: None,
1173                })))
1174            }
1175            PipeRecvInner::Remote(remote) => Ok(Self::remote(remote.try_clone().await?)),
1176        }
1177    }
1178
1179    /// Converts this endpoint into native process stdio.
1180    ///
1181    /// Remote endpoints cannot be converted.
1182    pub async fn into_stdio(self) -> Result<Stdio> {
1183        match self.0 {
1184            #[cfg(unix)]
1185            PipeRecvInner::Native(pipe) => {
1186                let fd: OwnedFd = pipe.into_blocking_fd()?;
1187                Ok(Stdio::from(fd))
1188            }
1189            #[cfg(windows)]
1190            PipeRecvInner::Native(NativePipeRecv { inner, pending, .. }) => {
1191                if pending.is_some() {
1192                    return Err(Error::new(
1193                        ErrorKind::ResourceBusy,
1194                        "cannot convert PipeRecv while an async read is in flight",
1195                    ));
1196                }
1197                Ok(Arc::try_unwrap(inner)
1198                    .or_else(|inner| inner.try_clone())
1199                    .map(Stdio::from)?)
1200            }
1201            PipeRecvInner::Remote(_) => Err(Error::new(
1202                ErrorKind::InvalidInput,
1203                "remote stdio cannot be converted to a native handle",
1204            )),
1205        }
1206    }
1207
1208    pub(crate) async fn into_blocking_handle(self) -> io::Result<DefaultHandle> {
1209        match self.0 {
1210            #[cfg(unix)]
1211            PipeRecvInner::Native(pipe) => pipe.into_blocking_fd(),
1212            #[cfg(windows)]
1213            PipeRecvInner::Native(NativePipeRecv { inner, pending, .. }) => {
1214                if pending.is_some() {
1215                    return Err(io::Error::other(
1216                        "cannot convert PipeRecv while an async read is in flight",
1217                    ));
1218                }
1219                let pipe = Arc::try_unwrap(inner).or_else(|inner| inner.try_clone())?;
1220                Ok(OwnedHandle::from(pipe))
1221            }
1222            PipeRecvInner::Remote(_) => Err(io::Error::new(
1223                io::ErrorKind::InvalidInput,
1224                "remote stdio has no native handle",
1225            )),
1226        }
1227    }
1228}
1229
1230impl AsyncWrite for PipeSend {
1231    fn poll_write(
1232        mut self: Pin<&mut Self>,
1233        cx: &mut Context<'_>,
1234        buf: &[u8],
1235    ) -> Poll<io::Result<usize>> {
1236        match &mut self.0 {
1237            PipeSendInner::Native(native) => TokioWrite::poll_write(Pin::new(native), cx, buf),
1238            PipeSendInner::Remote(remote) => Pin::new(remote).poll_write(cx, buf),
1239        }
1240    }
1241
1242    fn poll_write_vectored(
1243        mut self: Pin<&mut Self>,
1244        cx: &mut Context<'_>,
1245        bufs: &[IoSlice<'_>],
1246    ) -> Poll<io::Result<usize>> {
1247        match &mut self.0 {
1248            PipeSendInner::Native(native) => {
1249                TokioWrite::poll_write_vectored(Pin::new(native), cx, bufs)
1250            }
1251            PipeSendInner::Remote(remote) => Pin::new(remote).poll_write_vectored(cx, bufs),
1252        }
1253    }
1254
1255    fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
1256        match &mut self.0 {
1257            PipeSendInner::Native(native) => TokioWrite::poll_flush(Pin::new(native), cx),
1258            PipeSendInner::Remote(remote) => Pin::new(remote).poll_flush(cx),
1259        }
1260    }
1261
1262    fn poll_close(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
1263        match &mut self.0 {
1264            PipeSendInner::Native(native) => TokioWrite::poll_shutdown(Pin::new(native), cx),
1265            PipeSendInner::Remote(remote) => Pin::new(remote).poll_close(cx),
1266        }
1267    }
1268}
1269
1270impl AsyncRead for PipeRecv {
1271    fn poll_read(
1272        mut self: Pin<&mut Self>,
1273        cx: &mut Context<'_>,
1274        buf: &mut [u8],
1275    ) -> Poll<io::Result<usize>> {
1276        match &mut self.0 {
1277            PipeRecvInner::Native(native) => {
1278                let mut buf = ReadBuf::new(buf);
1279                ready!(TokioRead::poll_read(Pin::new(native), cx, &mut buf))?;
1280                Poll::Ready(Ok(buf.filled().len()))
1281            }
1282            PipeRecvInner::Remote(remote) => Pin::new(remote).poll_read(cx, buf),
1283        }
1284    }
1285}
1286
1287#[cfg(windows)]
1288impl TokioWrite for NativePipeSend {
1289    fn poll_write(
1290        mut self: Pin<&mut Self>,
1291        cx: &mut Context<'_>,
1292        buf: &[u8],
1293    ) -> Poll<io::Result<usize>> {
1294        let Self { inner, pending } = &mut *self;
1295        if let Some(task) = pending {
1296            return match Pin::new(task).poll(cx) {
1297                Poll::Pending => Poll::Pending,
1298                Poll::Ready(Ok(result)) => {
1299                    *pending = None;
1300                    Poll::Ready(result)
1301                }
1302                Poll::Ready(Err(error)) => {
1303                    *pending = None;
1304                    Poll::Ready(Err(io::Error::other(error)))
1305                }
1306            };
1307        }
1308        let inner = Arc::clone(inner);
1309        let data = buf.to_vec();
1310        *pending = Some(tokio::task::spawn_blocking(move || (&*inner).write(&data)));
1311        self.poll_write(cx, &[])
1312    }
1313    fn poll_write_vectored(
1314        mut self: Pin<&mut Self>,
1315        cx: &mut Context<'_>,
1316        bufs: &[IoSlice<'_>],
1317    ) -> Poll<io::Result<usize>> {
1318        let Self { inner, pending } = &mut *self;
1319        if let Some(task) = pending {
1320            return match Pin::new(task).poll(cx) {
1321                Poll::Pending => Poll::Pending,
1322                Poll::Ready(Ok(result)) => {
1323                    *pending = None;
1324                    Poll::Ready(result)
1325                }
1326                Poll::Ready(Err(error)) => {
1327                    *pending = None;
1328                    Poll::Ready(Err(io::Error::other(error)))
1329                }
1330            };
1331        }
1332        let mut data = Vec::new();
1333        for buf in bufs {
1334            data.extend_from_slice(buf);
1335        }
1336        if data.is_empty() {
1337            return Poll::Ready(Ok(0));
1338        }
1339        let inner = Arc::clone(inner);
1340        *pending = Some(tokio::task::spawn_blocking(move || (&*inner).write(&data)));
1341        self.poll_write_vectored(cx, &[])
1342    }
1343    fn is_write_vectored(&self) -> bool {
1344        true
1345    }
1346    fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
1347        match self.as_mut().poll_write(cx, &[]) {
1348            Poll::Pending => Poll::Pending,
1349            Poll::Ready(Ok(_)) => Poll::Ready(Ok(())),
1350            Poll::Ready(Err(e)) => Poll::Ready(Err(e)),
1351        }
1352    }
1353    fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
1354        self.poll_flush(cx)
1355    }
1356}
1357
1358#[cfg(windows)]
1359impl TokioRead for NativePipeRecv {
1360    fn poll_read(
1361        mut self: Pin<&mut Self>,
1362        cx: &mut Context<'_>,
1363        buf: &mut ReadBuf<'_>,
1364    ) -> Poll<io::Result<()>> {
1365        let Self {
1366            inner,
1367            pending,
1368            ready,
1369        } = &mut *self;
1370        if let Some((data, len)) = ready {
1371            let n = (*len).min(buf.remaining());
1372            buf.put_slice(&data[..n]);
1373            if n == *len {
1374                *ready = None;
1375            } else {
1376                data.drain(..n);
1377                *len -= n;
1378            }
1379            return Poll::Ready(Ok(()));
1380        }
1381        if pending.is_none() {
1382            if buf.remaining() == 0 {
1383                return Poll::Ready(Ok(()));
1384            }
1385            let inner = Arc::clone(inner);
1386            let cap = buf.remaining();
1387            *pending = Some(tokio::task::spawn_blocking(move || {
1388                let mut data = vec![0; cap];
1389                let result = (&*inner).read(&mut data);
1390                (data, result)
1391            }));
1392        }
1393        match Pin::new(pending.as_mut().unwrap()).poll(cx) {
1394            Poll::Pending => Poll::Pending,
1395            Poll::Ready(Ok((data, Ok(len)))) => {
1396                *pending = None;
1397                let n = len.min(buf.remaining());
1398                buf.put_slice(&data[..n]);
1399                if n < len {
1400                    *ready = Some((data[n..len].to_vec(), len - n));
1401                }
1402                Poll::Ready(Ok(()))
1403            }
1404            Poll::Ready(Ok((_, Err(error)))) => {
1405                *pending = None;
1406                Poll::Ready(Err(error))
1407            }
1408            Poll::Ready(Err(error)) => {
1409                *pending = None;
1410                Poll::Ready(Err(io::Error::other(error)))
1411            }
1412        }
1413    }
1414}
dolang_vfs/process/mod.rs

dolang_vfs/process/
mod.rs
