compio_fs/
async_fd.rs

1#[cfg(unix)]
2use std::os::fd::FromRawFd;
3#[cfg(windows)]
4use std::os::windows::io::{
5    AsRawHandle, AsRawSocket, FromRawHandle, FromRawSocket, RawHandle, RawSocket,
6};
7use std::{io, ops::Deref};
8
9use compio_buf::{BufResult, IntoInner, IoBuf, IoBufMut};
10use compio_driver::{
11    AsFd, AsRawFd, BorrowedFd, RawFd, SharedFd, ToSharedFd,
12    op::{BufResultExt, Read, ReadManaged, ResultTakeBuffer, Write},
13};
14use compio_io::{AsyncRead, AsyncReadManaged, AsyncWrite};
15use compio_runtime::{Attacher, BorrowedBuffer, BufferPool};
16#[cfg(unix)]
17use {
18    compio_buf::{IoVectoredBuf, IoVectoredBufMut},
19    compio_driver::op::{ReadVectored, WriteVectored},
20};
21
22/// A wrapper for IO source, providing implementations for [`AsyncRead`] and
23/// [`AsyncWrite`].
24#[derive(Debug)]
25pub struct AsyncFd<T: AsFd> {
26    inner: Attacher<T>,
27}
28
29impl<T: AsFd> AsyncFd<T> {
30    /// Create [`AsyncFd`] and attach the source to the current runtime.
31    pub fn new(source: T) -> io::Result<Self> {
32        Ok(Self {
33            inner: Attacher::new(source)?,
34        })
35    }
36
37    /// Create [`AsyncFd`] without attaching the source.
38    ///
39    /// # Safety
40    ///
41    /// * The user should handle the attachment correctly.
42    /// * `T` should be an owned fd.
43    pub unsafe fn new_unchecked(source: T) -> Self {
44        Self {
45            inner: unsafe { Attacher::new_unchecked(source) },
46        }
47    }
48}
49
50impl<T: AsFd + 'static> AsyncRead for AsyncFd<T> {
51    #[inline]
52    async fn read<B: IoBufMut>(&mut self, buf: B) -> BufResult<usize, B> {
53        (&*self).read(buf).await
54    }
55
56    #[cfg(unix)]
57    #[inline]
58    async fn read_vectored<V: IoVectoredBufMut>(&mut self, buf: V) -> BufResult<usize, V> {
59        (&*self).read_vectored(buf).await
60    }
61}
62
63impl<T: AsFd + 'static> AsyncReadManaged for AsyncFd<T> {
64    type Buffer<'a> = BorrowedBuffer<'a>;
65    type BufferPool = BufferPool;
66
67    async fn read_managed<'a>(
68        &mut self,
69        buffer_pool: &'a Self::BufferPool,
70        len: usize,
71    ) -> io::Result<Self::Buffer<'a>> {
72        (&*self).read_managed(buffer_pool, len).await
73    }
74}
75
76impl<T: AsFd + 'static> AsyncReadManaged for &AsyncFd<T> {
77    type Buffer<'a> = BorrowedBuffer<'a>;
78    type BufferPool = BufferPool;
79
80    async fn read_managed<'a>(
81        &mut self,
82        buffer_pool: &'a Self::BufferPool,
83        len: usize,
84    ) -> io::Result<Self::Buffer<'a>> {
85        let fd = self.to_shared_fd();
86        let buffer_pool = buffer_pool.try_inner()?;
87        let op = ReadManaged::new(fd, buffer_pool, len)?;
88        compio_runtime::submit(op)
89            .with_extra()
90            .await
91            .take_buffer(buffer_pool)
92    }
93}
94
95impl<T: AsFd + 'static> AsyncRead for &AsyncFd<T> {
96    async fn read<B: IoBufMut>(&mut self, buf: B) -> BufResult<usize, B> {
97        let fd = self.inner.to_shared_fd();
98        let op = Read::new(fd, buf);
99        let res = compio_runtime::submit(op).await.into_inner();
100        unsafe { res.map_advanced() }
101    }
102
103    #[cfg(unix)]
104    async fn read_vectored<V: IoVectoredBufMut>(&mut self, buf: V) -> BufResult<usize, V> {
105        use compio_driver::op::VecBufResultExt;
106
107        let fd = self.inner.to_shared_fd();
108        let op = ReadVectored::new(fd, buf);
109        let res = compio_runtime::submit(op).await.into_inner();
110        unsafe { res.map_vec_advanced() }
111    }
112}
113
114impl<T: AsFd + 'static> AsyncWrite for AsyncFd<T> {
115    #[inline]
116    async fn write<B: IoBuf>(&mut self, buf: B) -> BufResult<usize, B> {
117        (&*self).write(buf).await
118    }
119
120    #[cfg(unix)]
121    #[inline]
122    async fn write_vectored<V: IoVectoredBuf>(&mut self, buf: V) -> BufResult<usize, V> {
123        (&*self).write_vectored(buf).await
124    }
125
126    #[inline]
127    async fn flush(&mut self) -> io::Result<()> {
128        (&*self).flush().await
129    }
130
131    #[inline]
132    async fn shutdown(&mut self) -> io::Result<()> {
133        (&*self).shutdown().await
134    }
135}
136
137impl<T: AsFd + 'static> AsyncWrite for &AsyncFd<T> {
138    async fn write<B: IoBuf>(&mut self, buf: B) -> BufResult<usize, B> {
139        let fd = self.inner.to_shared_fd();
140        let op = Write::new(fd, buf);
141        compio_runtime::submit(op).await.into_inner()
142    }
143
144    #[cfg(unix)]
145    async fn write_vectored<V: IoVectoredBuf>(&mut self, buf: V) -> BufResult<usize, V> {
146        let fd = self.inner.to_shared_fd();
147        let op = WriteVectored::new(fd, buf);
148        compio_runtime::submit(op).await.into_inner()
149    }
150
151    async fn flush(&mut self) -> io::Result<()> {
152        Ok(())
153    }
154
155    async fn shutdown(&mut self) -> io::Result<()> {
156        Ok(())
157    }
158}
159
160impl<T: AsFd> IntoInner for AsyncFd<T> {
161    type Inner = SharedFd<T>;
162
163    fn into_inner(self) -> Self::Inner {
164        self.inner.into_inner()
165    }
166}
167
168impl<T: AsFd> AsFd for AsyncFd<T> {
169    fn as_fd(&self) -> BorrowedFd<'_> {
170        self.inner.as_fd()
171    }
172}
173
174impl<T: AsFd> AsRawFd for AsyncFd<T> {
175    fn as_raw_fd(&self) -> RawFd {
176        self.inner.as_fd().as_raw_fd()
177    }
178}
179
180#[cfg(windows)]
181impl<T: AsFd + AsRawHandle> AsRawHandle for AsyncFd<T> {
182    fn as_raw_handle(&self) -> RawHandle {
183        self.inner.as_raw_handle()
184    }
185}
186
187#[cfg(windows)]
188impl<T: AsFd + AsRawSocket> AsRawSocket for AsyncFd<T> {
189    fn as_raw_socket(&self) -> RawSocket {
190        self.inner.as_raw_socket()
191    }
192}
193
194impl<T: AsFd> ToSharedFd<T> for AsyncFd<T> {
195    fn to_shared_fd(&self) -> SharedFd<T> {
196        self.inner.to_shared_fd()
197    }
198}
199
200impl<T: AsFd> Clone for AsyncFd<T> {
201    fn clone(&self) -> Self {
202        Self {
203            inner: self.inner.clone(),
204        }
205    }
206}
207
208#[cfg(unix)]
209impl<T: AsFd + FromRawFd> FromRawFd for AsyncFd<T> {
210    unsafe fn from_raw_fd(fd: RawFd) -> Self {
211        unsafe { Self::new_unchecked(FromRawFd::from_raw_fd(fd)) }
212    }
213}
214
215#[cfg(windows)]
216impl<T: AsFd + FromRawHandle> FromRawHandle for AsyncFd<T> {
217    unsafe fn from_raw_handle(handle: RawHandle) -> Self {
218        unsafe { Self::new_unchecked(FromRawHandle::from_raw_handle(handle)) }
219    }
220}
221
222#[cfg(windows)]
223impl<T: AsFd + FromRawSocket> FromRawSocket for AsyncFd<T> {
224    unsafe fn from_raw_socket(sock: RawSocket) -> Self {
225        unsafe { Self::new_unchecked(FromRawSocket::from_raw_socket(sock)) }
226    }
227}
228
229impl<T: AsFd> Deref for AsyncFd<T> {
230    type Target = T;
231
232    fn deref(&self) -> &Self::Target {
233        &self.inner
234    }
235}
compio_fs/async_fd.rs

compio_fs/
async_fd.rs
