moirai_async/fs/file/
traits.rs

1//! The crate's async I/O traits for [`File`], and its positioned reads.
2
3use std::io;
4use std::pin::Pin;
5use std::sync::Arc;
6use std::task::{Context, Poll, ready};
7
8use moirai_pal::fs::File as Handle;
9
10use super::File;
11use crate::blocking::Abandoned;
12use crate::fs::pool;
13use crate::io::{AsyncLength, AsyncRead, AsyncReadAt, AsyncWrite};
14
15impl AsyncReadAt for File {
16    async fn read_at(&self, offset: u64, buf: &mut [u8]) -> io::Result<()> {
17        if buf.is_empty() {
18            return Ok(());
19        }
20
21        let requested = u64::try_from(buf.len()).map_err(|_| {
22            io::Error::new(io::ErrorKind::InvalidInput, "read length does not fit u64")
23        })?;
24        offset.checked_add(requested).ok_or_else(|| {
25            io::Error::new(io::ErrorKind::InvalidInput, "read range overflows u64")
26        })?;
27
28        self.fence.settled().await;
29        let handle = Arc::clone(&self.handle);
30        let len = buf.len();
31        let (data, filled) = pool()
32            .run(Abandoned::Skip, move || read_exact_at(&handle, len, offset))
33            .await?;
34        // Bytes read before a failure still reach the caller, as with a
35        // positioned read loop run in place.
36        buf[..data.len()].copy_from_slice(&data);
37        filled
38    }
39}
40
41/// Fill `len` bytes from `offset` with positioned reads, which leave the
42/// stream cursor alone. Returns the bytes read, and whether all `len` were.
43fn read_exact_at(handle: &Handle, len: usize, offset: u64) -> (Vec<u8>, io::Result<()>) {
44    let mut data = vec![0; len];
45    let mut filled = 0;
46    let result = loop {
47        if filled == len {
48            break Ok(());
49        }
50        let Some(position) = u64::try_from(filled)
51            .ok()
52            .and_then(|filled| offset.checked_add(filled))
53        else {
54            break Err(io::Error::new(
55                io::ErrorKind::InvalidInput,
56                "read range overflows u64",
57            ));
58        };
59        match handle.read_at(&mut data[filled..], position) {
60            Ok(0) => {
61                break Err(io::Error::new(
62                    io::ErrorKind::UnexpectedEof,
63                    "positioned read reached end of file before filling buffer",
64                ));
65            }
66            Ok(count) => filled += count,
67            Err(error) => break Err(error),
68        }
69    };
70    data.truncate(filled);
71    (data, result)
72}
73
74impl AsyncLength for File {
75    async fn len(&self) -> io::Result<u64> {
76        Ok(self.metadata().await?.len())
77    }
78}
79
80impl AsyncRead for File {
81    fn poll_read(
82        mut self: Pin<&mut Self>,
83        cx: &mut Context<'_>,
84        buf: &mut [u8],
85    ) -> Poll<io::Result<usize>> {
86        let n = ready!(self.poll_read_into(cx, buf))?;
87        self.stats.bytes_read += n as u64;
88        self.stats.read_operations += 1;
89        Poll::Ready(Ok(n))
90    }
91}
92
93impl AsyncWrite for File {
94    fn poll_write(
95        mut self: Pin<&mut Self>,
96        cx: &mut Context<'_>,
97        buf: &[u8],
98    ) -> Poll<io::Result<usize>> {
99        let n = ready!(self.poll_write_behind(cx, buf))?;
100        self.stats.bytes_written += n as u64;
101        self.stats.write_operations += 1;
102        Poll::Ready(Ok(n))
103    }
104
105    fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
106        self.poll_settled(cx)
107    }
108
109    fn poll_shutdown(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
110        self.poll_settled(cx)
111    }
112}
moirai_async/fs/file/traits.rs

moirai_async/fs/file/
traits.rs
