nuclei/fs/file/
many.rs

1//! Many positional reads in one call: [`Handle::read_many_at`]. The reads
2//! wait once for the cursor writes accepted before the call (as each
3//! positional read does), then go out together: on io_uring every read the
4//! window holds is an entry of the shard's ring, handed to the kernel in the
5//! shard's one submission; elsewhere each is a job of the blocking pool. At
6//! most [`MANY_AT_ONCE`] are out at a time, the next going out as one ends,
7//! so a call of any length holds a bounded share of the ring and the pool.
8//! The reads read into the caller's own buffers, which the runtime's buffer
9//! cap does not count. `proofs/tla/many/ReadMany.tla` checks the window.
10
11use core::future::poll_fn;
12use core::task::Poll;
13use std::fs::File as StdFile;
14use std::io;
15use std::sync::Arc;
16
17use super::FileIo;
18use super::work::{Work, tally_read};
19use crate::buf::owned::{IoBufMut, OwnedBuf};
20use crate::io::Handle;
21use crate::io::source::Failure;
22
23/// The most reads of one call out at once.
24pub(crate) const MANY_AT_ONCE: usize = 64;
25
26/// A read's end: its result, and the caller's buffer with what it read.
27type End = (io::Result<usize>, OwnedBuf);
28
29impl FileIo {
30    pub(crate) async fn read_many_at<B>(
31        self: &Arc<Self>,
32        reads: Vec<(B, u64)>,
33    ) -> Vec<(io::Result<usize>, B)>
34    where
35        B: IoBufMut,
36    {
37        poll_fn(|cx| self.poll_ordered(cx)).await;
38        let mut ends: Vec<Option<End>> = (0..reads.len()).map(|_| None).collect();
39        let mut waiting = reads.into_iter().enumerate();
40        let mut out: Vec<(usize, Work<End>)> = Vec::with_capacity(MANY_AT_ONCE.min(ends.len()));
41        poll_fn(|cx| {
42            loop {
43                while out.len() < MANY_AT_ONCE {
44                    let Some((i, (buf, at))) = waiting.next() else {
45                        break;
46                    };
47                    let buf = OwnedBuf::filled(buf);
48                    if buf.len() == buf.capacity() {
49                        ends[i] = Some((Ok(0), buf));
50                        continue;
51                    }
52                    out.push((i, self.submit_read_owned(&mut Some(buf), at, cx.waker())));
53                }
54                let before = out.len();
55                out.retain_mut(|(i, work)| match (work.pooled(), work.poll(cx.waker())) {
56                    (pooled, Poll::Ready(end)) => {
57                        tally_read(pooled, &end.0, |n| *n);
58                        ends[*i] = Some(end);
59                        false
60                    }
61                    (_, Poll::Pending) => true,
62                });
63                if out.is_empty() && waiting.len() == 0 {
64                    return Poll::Ready(());
65                }
66                if out.len() == before {
67                    return Poll::Pending;
68                }
69            }
70        })
71        .await;
72        ends.into_iter()
73            .map(|end| {
74                let (result, buf) = end.unwrap_or_else(|| {
75                    (
76                        Err(io::Error::other("the read lost its buffer")),
77                        OwnedBuf::default(),
78                    )
79                });
80                (result, buf.into_filled())
81            })
82            .collect()
83    }
84}
85
86impl Handle<StdFile> {
87    /// Reads at each offset of `reads` into the spare capacity of its
88    /// buffer, after the bytes it holds, and gives every buffer back in the
89    /// order of `reads` with its read's result, as
90    /// [`read_at_owned`](Self::read_at_owned) answers one: the bytes that
91    /// came, zero at the end of the file or for a buffer with no spare
92    /// capacity. The reads are made at once, in any order: on io_uring
93    /// together, in the shard ring's one submission, at most 64 out at a
94    /// time; elsewhere as jobs of the blocking pool, as many at a time.
95    ///
96    /// The reads see every write accepted at the cursor before the call, and
97    /// every positional write that returned before it. Each read is one call
98    /// of the system's positional read into its buffer itself, never copied.
99    /// If the future is dropped, the reads still out are left to end by
100    /// themselves, and their buffers go with them.
101    ///
102    /// # Errors
103    ///
104    /// Each read's own, as `read_at_owned`'s. On a stream (a pipe, a
105    /// terminal) the file has no offset to read at, and every read answers
106    /// the system's error for that.
107    pub async fn read_many_at<B>(&self, reads: Vec<(B, u64)>) -> Vec<(io::Result<usize>, B)>
108    where
109        B: IoBufMut,
110    {
111        match self.file() {
112            Ok(file) => file.read_many_at(reads).await,
113            Err(err) => {
114                let failure = Failure::of(&err);
115                reads
116                    .into_iter()
117                    .map(|(buf, _)| (Err(failure.error()), buf))
118                    .collect()
119            }
120        }
121    }
122}
nuclei/fs/file/many.rs

nuclei/fs/file/
many.rs
