compio_process/
unix.rs

1use std::{io, process};
2
3use compio_buf::{BufResult, IntoInner, IoBuf, IoBufMut};
4use compio_driver::{
5    BufferRef, ResultTakeBuffer, ToSharedFd,
6    op::{BufResultExt, Read, ReadManaged, Write},
7};
8use compio_io::{AsyncRead, AsyncReadManaged, AsyncWrite};
9use compio_runtime::{ResumeUnwind, Runtime, SpawnMeta};
10
11use crate::{ChildStderr, ChildStdin, ChildStdout};
12
13pub async fn child_wait(mut child: process::Child) -> io::Result<process::ExitStatus> {
14    // Name the task: its location points here rather than into the code that
15    // waited for the child, since this is an `async fn`.
16    let meta = SpawnMeta::capture().named("process::wait");
17    compio_runtime::spawn_blocking_at(move || child.wait(), meta)
18        .await
19        .resume_unwind()
20        .expect("shouldn't be cancelled")
21}
22
23impl AsyncRead for ChildStdout {
24    async fn read<B: IoBufMut>(&mut self, buffer: B) -> BufResult<usize, B> {
25        let fd = self.to_shared_fd();
26        let op = Read::new(fd, buffer);
27        let res = compio_runtime::submit(op).await.into_inner();
28        unsafe { res.map_advanced() }
29    }
30}
31
32impl AsyncReadManaged for ChildStdout {
33    type Buffer = BufferRef;
34
35    async fn read_managed(&mut self, len: usize) -> io::Result<Option<Self::Buffer>> {
36        let fd = self.to_shared_fd();
37        let res = Runtime::with_current(|rt| {
38            let buffer_pool = rt.buffer_pool()?;
39            let op = ReadManaged::new(fd, &buffer_pool, len)?;
40            io::Result::Ok(rt.submit(op))
41        })?
42        .await;
43        unsafe { res.take_buffer() }
44    }
45}
46
47impl AsyncRead for ChildStderr {
48    async fn read<B: IoBufMut>(&mut self, buffer: B) -> BufResult<usize, B> {
49        let fd = self.to_shared_fd();
50        let op = Read::new(fd, buffer);
51        let res = compio_runtime::submit(op).await.into_inner();
52        unsafe { res.map_advanced() }
53    }
54}
55
56impl AsyncReadManaged for ChildStderr {
57    type Buffer = BufferRef;
58
59    async fn read_managed(&mut self, len: usize) -> io::Result<Option<Self::Buffer>> {
60        let fd = self.to_shared_fd();
61        let res = Runtime::with_current(|rt| {
62            let buffer_pool = rt.buffer_pool()?;
63            let op = ReadManaged::new(fd, &buffer_pool, len)?;
64            io::Result::Ok(rt.submit(op))
65        })?
66        .await;
67        unsafe { res.take_buffer() }
68    }
69}
70
71impl AsyncWrite for ChildStdin {
72    async fn write<T: IoBuf>(&mut self, buffer: T) -> BufResult<usize, T> {
73        let fd = self.to_shared_fd();
74        let op = Write::new(fd, buffer);
75        compio_runtime::submit(op).await.into_inner()
76    }
77
78    async fn flush(&mut self) -> io::Result<()> {
79        Ok(())
80    }
81
82    async fn shutdown(&mut self) -> io::Result<()> {
83        Ok(())
84    }
85}
compio_process/unix.rs

compio_process/
unix.rs
