peridot_native_io/
adapter.rs

1//! Adapter structs for other crate's traits.
2
3use pin_project::pin_project;
4use std::io::{Read, Seek};
5
6pub struct RandomBlobReadSeekAdapter<R> {
7    inner: R,
8    pos: u64,
9}
10impl<R> RandomBlobReadSeekAdapter<R> {
11    #[inline(always)]
12    pub const fn new(inner: R) -> Self {
13        Self { inner, pos: 0 }
14    }
15
16    #[inline(always)]
17    pub const fn readpos(&self) -> u64 {
18        self.pos
19    }
20
21    #[inline(always)]
22    pub fn into_inner(self) -> R {
23        self.inner
24    }
25}
26impl<R> Read for RandomBlobReadSeekAdapter<R>
27where
28    R: crate::RandomReadBlob,
29{
30    #[inline]
31    fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
32        let transferred = self.inner.read(self.pos, unsafe {
33            core::mem::transmute::<&mut [_], &mut [core::mem::MaybeUninit<_>]>(buf)
34        })?;
35        self.pos += transferred as u64;
36
37        Ok(transferred)
38    }
39}
40impl<R> Seek for RandomBlobReadSeekAdapter<R>
41where
42    R: crate::BlobMetadata,
43{
44    #[inline]
45    fn seek(&mut self, pos: std::io::SeekFrom) -> std::io::Result<u64> {
46        let new_pos = match pos {
47            std::io::SeekFrom::Start(x) => x,
48            std::io::SeekFrom::Current(x) => (self.pos as i64 + x) as _,
49            std::io::SeekFrom::End(x) => (self.inner.byte_length()? as i64 + x) as _,
50        };
51        self.pos = new_pos;
52        Ok(new_pos)
53    }
54}
55
56#[pin_project(project = RandomBlobAsyncReadSeekAdapterStateProjected)]
57enum RandomBlobAsyncReadSeekAdapterState<'r, R: crate::RandomReadBlobAsync + 'r> {
58    Idle,
59    Reading(#[pin] R::ReadFuture<'r, 'static>),
60    ReadingVec(#[pin] R::ReadVecFuture<'r, 'static, 'static>),
61}
62
63#[pin_project]
64pub struct RandomBlobAsyncReadSeekAdapter<'r, R: crate::RandomReadBlobAsync + 'r> {
65    inner: &'r R,
66    pos: u64,
67    #[pin]
68    state: RandomBlobAsyncReadSeekAdapterState<'r, R>,
69}
70impl<'r, R: crate::RandomReadBlobAsync + 'r> RandomBlobAsyncReadSeekAdapter<'r, R> {
71    #[inline(always)]
72    pub const fn new(inner: &'r R) -> Self {
73        Self::with_pos(inner, 0)
74    }
75
76    #[inline(always)]
77    pub const fn with_pos(inner: &'r R, pos: u64) -> Self {
78        Self {
79            inner,
80            pos,
81            state: RandomBlobAsyncReadSeekAdapterState::Idle,
82        }
83    }
84
85    #[inline(always)]
86    pub const fn readpos(&self) -> u64 {
87        self.pos
88    }
89
90    /// # Safety
91    ///
92    /// read/seek operation may be in progress.
93    #[inline(always)]
94    pub unsafe fn into_inner_unchecked(self) -> &'r R {
95        self.inner
96    }
97}
98impl<'r, R: crate::RandomReadBlobAsync + 'r> futures_io::AsyncRead
99    for RandomBlobAsyncReadSeekAdapter<'r, R>
100{
101    fn poll_read(
102        self: std::pin::Pin<&mut Self>,
103        cx: &mut std::task::Context<'_>,
104        buf: &mut [u8],
105    ) -> std::task::Poll<std::io::Result<usize>> {
106        let mut this = self.project();
107
108        loop {
109            match this.state.as_mut().project() {
110                RandomBlobAsyncReadSeekAdapterStateProjected::Idle => {
111                    this.state.set(RandomBlobAsyncReadSeekAdapterState::Reading(
112                        this.inner.read_async(*this.pos, unsafe {
113                            core::mem::transmute::<&mut [_], &mut [core::mem::MaybeUninit<_>]>(buf)
114                        }),
115                    ));
116                }
117                RandomBlobAsyncReadSeekAdapterStateProjected::Reading(f) => {
118                    let r = std::task::ready!(f.poll(cx))?;
119                    *this.pos += r as u64;
120                    this.state.set(RandomBlobAsyncReadSeekAdapterState::Idle);
121                    return core::task::Poll::Ready(Ok(r));
122                }
123                RandomBlobAsyncReadSeekAdapterStateProjected::ReadingVec(_) => {
124                    panic!("poll_read called but poll_read_vectored is ongoing");
125                }
126            }
127        }
128    }
129
130    fn poll_read_vectored(
131        self: std::pin::Pin<&mut Self>,
132        cx: &mut std::task::Context<'_>,
133        bufs: &mut [std::io::IoSliceMut<'_>],
134    ) -> std::task::Poll<std::io::Result<usize>> {
135        let mut this = self.project();
136
137        loop {
138            match this.state.as_mut().project() {
139                RandomBlobAsyncReadSeekAdapterStateProjected::Idle => {
140                    this.state
141                        .set(RandomBlobAsyncReadSeekAdapterState::ReadingVec(
142                            this.inner.readv_async(*this.pos, unsafe {
143                                core::mem::transmute::<&mut [_], &mut [_]>(&mut *bufs)
144                            }),
145                        ));
146                }
147                RandomBlobAsyncReadSeekAdapterStateProjected::ReadingVec(f) => {
148                    let r = std::task::ready!(f.poll(cx))?;
149                    *this.pos += r as u64;
150                    this.state.set(RandomBlobAsyncReadSeekAdapterState::Idle);
151                    break std::task::Poll::Ready(Ok(r));
152                }
153                RandomBlobAsyncReadSeekAdapterStateProjected::Reading(_) => {
154                    panic!("poll_read_vectored called but poll_read is ongoing");
155                }
156            }
157        }
158    }
159}
160impl<'r, R: crate::RandomReadBlobAsync + 'r> futures_io::AsyncSeek
161    for RandomBlobAsyncReadSeekAdapter<'r, R>
162{
163    fn poll_seek(
164        self: std::pin::Pin<&mut Self>,
165        _cx: &mut std::task::Context<'_>,
166        pos: std::io::SeekFrom,
167    ) -> std::task::Poll<std::io::Result<u64>> {
168        // これは同期的にできる
169        let pref = self.project().pos;
170        *pref = match pos {
171            std::io::SeekFrom::Start(x) => x,
172            std::io::SeekFrom::Current(x) => pref.checked_add_signed(x).ok_or_else(|| {
173                std::io::Error::new(
174                    std::io::ErrorKind::InvalidInput,
175                    "resulting file pointer is out of range!",
176                )
177            })?,
178            std::io::SeekFrom::End(x) => pref.checked_add_signed(x).ok_or_else(|| {
179                std::io::Error::new(
180                    std::io::ErrorKind::InvalidInput,
181                    "resulting file pointer is out of range!",
182                )
183            })?,
184        };
185
186        std::task::Poll::Ready(Ok(*pref))
187    }
188}
peridot_native_io/adapter.rs

peridot_native_io/
adapter.rs
