fuchsia_fs/file/
async_read_at.rs

1// Copyright 2021 The Fuchsia Authors. All rights reserved.
2// Use of this source code is governed by a BSD-style license that can be
3// found in the LICENSE file.
4
5use fidl::client::QueryResponseFut;
6use flex_client::Dialect;
7use flex_fuchsia_io as fio;
8use futures::FutureExt as _;
9use futures::future::Future;
10use futures::io::{AsyncRead, AsyncSeek, SeekFrom};
11use futures::lock::Mutex;
12use std::cmp::min;
13use std::convert::TryInto as _;
14use std::io;
15use std::pin::Pin;
16use std::sync::Arc;
17use std::task::{Context, Poll};
18use zx_status_ext::StatusExt;
19
20/// Trait for reading at a given offset asynchronously.
21/// This is basically `futures::io::AsyncRead` with an extra offset.
22pub trait AsyncReadAt {
23    /// Attempt to read at most `buf.len()` bytes starting at `offset` into `buf`. On success
24    /// returns the number of bytes read.
25    /// Contents of `buf` are only altered on success.
26    /// Reads of more than zero but fewer than `buf.len()` bytes do NOT indicate EOF.
27    /// Reads of zero bytes only occur if `buf.len() == 0` or EOF.
28    fn poll_read_at(
29        self: Pin<&mut Self>,
30        cx: &mut Context<'_>,
31        offset: u64,
32        buf: &mut [u8],
33    ) -> Poll<io::Result<usize>>;
34}
35
36/// Trait for getting the size of the file asynchronously.
37pub trait AsyncGetSize {
38    /// Attempt to get the size of the file, on success returns the file size.
39    fn poll_get_size(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<u64>>;
40}
41
42impl<T: AsyncReadAt + Unpin> AsyncReadAt for Arc<Mutex<T>> {
43    fn poll_read_at(
44        self: Pin<&mut Self>,
45        cx: &mut Context<'_>,
46        offset: u64,
47        buf: &mut [u8],
48    ) -> Poll<Result<usize, std::io::Error>> {
49        let mut guard = futures::ready!(self.lock().poll_unpin(cx));
50        Pin::new(&mut *guard).poll_read_at(cx, offset, buf)
51    }
52}
53
54impl<T: AsyncGetSize + Unpin> AsyncGetSize for Arc<Mutex<T>> {
55    fn poll_get_size(
56        self: Pin<&mut Self>,
57        cx: &mut Context<'_>,
58    ) -> Poll<Result<u64, std::io::Error>> {
59        let mut guard = futures::ready!(self.lock().poll_unpin(cx));
60        Pin::new(&mut *guard).poll_get_size(cx)
61    }
62}
63
64/// An extension trait which adds utility methods to AsyncGetSize.
65pub trait AsyncGetSizeExt: AsyncGetSize {
66    /// Returns a future that will return the file size on success.
67    fn get_size<'a>(&'a mut self) -> GetSize<'a, Self>
68    where
69        Self: Unpin,
70    {
71        GetSize { size_getter: self }
72    }
73}
74
75impl<T: AsyncGetSize + ?Sized> AsyncGetSizeExt for T {}
76
77/// Future for the [`get_size`](AsyncGetSizeExt::get_size) method.
78#[derive(Debug)]
79#[must_use = "futures do nothing unless you `.await` or poll them"]
80pub struct GetSize<'a, R: ?Sized> {
81    size_getter: &'a mut R,
82}
83
84impl<R: ?Sized + Unpin> Unpin for GetSize<'_, R> {}
85
86impl<R: AsyncGetSize + ?Sized + Unpin> Future for GetSize<'_, R> {
87    type Output = io::Result<u64>;
88
89    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
90        let this = &mut *self;
91        Pin::new(&mut *this.size_getter).poll_get_size(cx)
92    }
93}
94
95/// Wraps a `fidl_fuchsia_io::FileProxy` and implements `AsyncReadAt` and `AsyncGetSize`, which
96/// allows one to perform asynchronous file reads that don't block the current thread while waiting
97/// for data.
98/// Unlike `AsyncReader`, this struct does not require exclusive ownership, because `read_at` does
99/// not rely on the file offset state in the connection. This is useful if one wants to efficiently
100/// read different parts of the file at the same time.
101#[derive(Debug)]
102pub struct AsyncFile {
103    file: fio::FileProxy,
104    read_at_state: ReadAtState,
105    get_attributes_fut: Option<
106        QueryResponseFut<
107            Result<(fio::MutableNodeAttributes, fio::ImmutableNodeAttributes), i32>,
108            Dialect,
109        >,
110    >,
111}
112
113#[derive(Debug)]
114enum ReadAtState {
115    Empty,
116    Forwarding {
117        fut: QueryResponseFut<Result<Vec<u8>, i32>, Dialect>,
118        file_offset: u64,
119        zero_byte_request: bool,
120    },
121    Bytes {
122        bytes: Vec<u8>,
123        file_offset: u64,
124    },
125}
126
127impl AsyncFile {
128    pub fn from_proxy(file: fio::FileProxy) -> Self {
129        Self { file, read_at_state: ReadAtState::Empty, get_attributes_fut: None }
130    }
131}
132
133impl AsyncReadAt for AsyncFile {
134    fn poll_read_at(
135        mut self: Pin<&mut Self>,
136        cx: &mut Context<'_>,
137        offset: u64,
138        buf: &mut [u8],
139    ) -> Poll<std::io::Result<usize>> {
140        loop {
141            match self.read_at_state {
142                ReadAtState::Empty => {
143                    let len = if let Ok(len) = buf.len().try_into() {
144                        min(len, fio::MAX_BUF)
145                    } else {
146                        fio::MAX_BUF
147                    };
148                    self.read_at_state = ReadAtState::Forwarding {
149                        fut: self.file.read_at(len, offset),
150                        file_offset: offset,
151                        zero_byte_request: len == 0,
152                    };
153                }
154                ReadAtState::Forwarding { ref mut fut, file_offset, zero_byte_request } => {
155                    match futures::ready!(Pin::new(fut).poll(cx)) {
156                        Ok(result) => {
157                            match result {
158                                Err(s) => {
159                                    self.read_at_state = ReadAtState::Empty;
160                                    return Poll::Ready(Err(
161                                        zx_status::Status::err_from_raw(s).into_io_error()
162                                    ));
163                                }
164                                Ok(bytes) => {
165                                    // If the File.ReadAt request was for zero bytes, but the current
166                                    // poll_read_at is not (because the File.ReadAt request was made by an
167                                    // earlier call to poll_read_at with a zero length buffer) then we
168                                    // should not advance to ReadAtState::Bytes because that would return
169                                    // Ready(Ok(0)), which would indicate EOF to the client.
170                                    // This handling is done here instead of short-circuiting at the
171                                    // beginning of the function so that zero-length poll_read_ats still
172                                    // trigger the validation performed by File.ReadAt.
173                                    if zero_byte_request && buf.len() != 0 {
174                                        self.read_at_state = ReadAtState::Empty;
175                                    } else {
176                                        self.read_at_state =
177                                            ReadAtState::Bytes { bytes, file_offset };
178                                    }
179                                }
180                            }
181                        }
182                        Err(e) => {
183                            self.read_at_state = ReadAtState::Empty;
184                            return Poll::Ready(Err(std::io::Error::other(e)));
185                        }
186                    }
187                }
188                ReadAtState::Bytes { ref bytes, file_offset } => {
189                    if offset < file_offset {
190                        self.read_at_state = ReadAtState::Empty;
191                        continue;
192                    }
193                    let bytes_offset = match (offset - file_offset).try_into() {
194                        Ok(offset) => offset,
195                        Err(_) => {
196                            self.read_at_state = ReadAtState::Empty;
197                            continue;
198                        }
199                    };
200                    if bytes_offset != 0 && bytes_offset >= bytes.len() {
201                        self.read_at_state = ReadAtState::Empty;
202                        continue;
203                    }
204                    let n = min(buf.len(), bytes.len() - bytes_offset);
205                    let () = buf[..n].copy_from_slice(&bytes[bytes_offset..bytes_offset + n]);
206                    return Poll::Ready(Ok(n));
207                }
208            }
209        }
210    }
211}
212
213impl AsyncGetSize for AsyncFile {
214    fn poll_get_size(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<std::io::Result<u64>> {
215        if self.get_attributes_fut.is_none() {
216            self.get_attributes_fut =
217                Some(self.file.get_attributes(fio::NodeAttributesQuery::CONTENT_SIZE));
218        }
219        let fut = self.get_attributes_fut.as_mut().unwrap();
220        let get_attributes_fut_result = futures::ready!(fut.poll_unpin(cx));
221        self.get_attributes_fut = None;
222        match get_attributes_fut_result {
223            Ok(get_attributes_response) => match get_attributes_response {
224                Ok((_mutable_attr, immutable_attr)) => {
225                    if let Some(content_size) = immutable_attr.content_size {
226                        return Poll::Ready(Ok(content_size));
227                    }
228                    return Poll::Ready(Err(zx_status::Status::NOT_SUPPORTED.into_io_error()));
229                }
230                Err(status) => {
231                    return Poll::Ready(Err(
232                        zx_status::Status::err_from_raw(status).into_io_error()
233                    ));
234                }
235            },
236            Err(e) => {
237                return Poll::Ready(Err(std::io::Error::other(e)));
238            }
239        }
240    }
241}
242
243/// Adapter to implement AsyncReadAt + AsyncGetSize for AsyncRead + AsyncSeek.
244#[derive(Debug)]
245pub struct Adapter<T> {
246    inner: T,
247}
248
249impl<T> Adapter<T> {
250    pub fn new(inner: T) -> Adapter<T> {
251        Self { inner }
252    }
253}
254
255impl<T: AsyncRead + AsyncSeek + Unpin> AsyncReadAt for Adapter<T> {
256    fn poll_read_at(
257        mut self: Pin<&mut Self>,
258        cx: &mut Context<'_>,
259        offset: u64,
260        buf: &mut [u8],
261    ) -> Poll<std::io::Result<usize>> {
262        futures::ready!(Pin::new(&mut self.inner).poll_seek(cx, SeekFrom::Start(offset)))?;
263        Pin::new(&mut self.inner).poll_read(cx, buf)
264    }
265}
266
267impl<T: AsyncSeek + Unpin> AsyncGetSize for Adapter<T> {
268    fn poll_get_size(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<std::io::Result<u64>> {
269        Pin::new(&mut self.inner).poll_seek(cx, SeekFrom::End(0))
270    }
271}
272
273#[cfg(test)]
274mod tests {
275    use super::*;
276    use crate::file::{self, AsyncReadAtExt};
277    use assert_matches::assert_matches;
278    use fidl::endpoints;
279    use fuchsia_async as fasync;
280    use futures::future::{self, poll_fn};
281    use futures::{StreamExt as _, TryStreamExt as _};
282    use std::convert::TryFrom as _;
283    use std::io::Write;
284    use tempfile::{NamedTempFile, TempDir};
285
286    async fn poll_read_at_with_specific_buf_size(
287        poll_read_size: u64,
288        expected_file_read_size: u64,
289    ) {
290        let (proxy, mut stream) = endpoints::create_proxy_and_stream::<fio::FileMarker>();
291
292        let mut reader = AsyncFile::from_proxy(proxy);
293
294        let () = poll_fn(|cx| {
295            let mut buf = vec![0u8; poll_read_size.try_into().unwrap()];
296            assert_matches!(
297                Pin::new(&mut reader).poll_read_at(cx, 0, buf.as_mut_slice()),
298                Poll::Pending
299            );
300            Poll::Ready(())
301        })
302        .await;
303
304        match stream.next().await.unwrap().unwrap() {
305            fio::FileRequest::ReadAt { count, .. } => {
306                assert_eq!(count, expected_file_read_size);
307            }
308            req => panic!("unhandled request {:?}", req),
309        }
310    }
311
312    #[fasync::run_singlethreaded(test)]
313    async fn poll_read_at_empty_buf() {
314        poll_read_at_with_specific_buf_size(0, 0).await;
315    }
316
317    #[fasync::run_singlethreaded(test)]
318    async fn poll_read_at_caps_buf_size() {
319        poll_read_at_with_specific_buf_size(fio::MAX_BUF * 2, fio::MAX_BUF).await;
320    }
321
322    #[fasync::run_singlethreaded(test)]
323    async fn poll_read_at_pending_saves_future() {
324        let (proxy, mut stream) = endpoints::create_proxy_and_stream::<fio::FileMarker>();
325
326        let mut reader = AsyncFile::from_proxy(proxy);
327
328        // This poll_read_at call will create a File.ReadAt future and poll it. The poll of the
329        // File.ReadAt future will return Pending because nothing is handling the FileRequestStream
330        // yet. The reader should save this File.ReadAt future for handling subsequent poll_read_at
331        // calls.
332        let () = poll_fn(|cx| {
333            assert_matches!(
334                Pin::new(&mut reader).poll_read_at(cx, 2, &mut [0u8; 1]),
335                Poll::Pending
336            );
337            Poll::Ready(())
338        })
339        .await;
340
341        // Call poll_read_at until we get a byte out. This byte should be from the first and only
342        // File.ReadAt request.
343        let poll_read_at = async move {
344            let mut buf = [0u8; 1];
345            assert_eq!(reader.read_at(2, &mut buf).await.unwrap(), buf.len());
346            assert_eq!(&buf, &[1]);
347        };
348
349        let mut file_read_requests = 0u8;
350        let handle_file_stream = async {
351            while let Some(req) = stream.try_next().await.unwrap() {
352                file_read_requests += 1;
353                match req {
354                    fio::FileRequest::ReadAt { count, offset, responder } => {
355                        assert_eq!(count, 1);
356                        assert_eq!(offset, 2);
357                        responder.send(Ok(&[file_read_requests])).unwrap();
358                    }
359                    req => panic!("unhandled request {:?}", req),
360                }
361            }
362        };
363
364        let ((), ()) = future::join(poll_read_at, handle_file_stream).await;
365        assert_eq!(file_read_requests, 1);
366    }
367
368    #[fasync::run_singlethreaded(test)]
369    async fn poll_read_at_with_smaller_buf_after_pending() {
370        let (proxy, mut stream) = endpoints::create_proxy_and_stream::<fio::FileMarker>();
371
372        let mut reader = AsyncFile::from_proxy(proxy);
373
374        // Call poll_read_at with a buf of length 3. This is the first poll_read_at call, so the
375        // reader will create a File.ReadAt future for 3 bytes. poll_read_at will return Pending
376        // because nothing is handling the FileRequestStream yet.
377        let () = poll_fn(|cx| {
378            assert_matches!(
379                Pin::new(&mut reader).poll_read_at(cx, 0, &mut [0u8; 3]),
380                Poll::Pending
381            );
382            Poll::Ready(())
383        })
384        .await;
385
386        // Respond to the three byte File.ReadAt request.
387        let () = async {
388            match stream.next().await.unwrap().unwrap() {
389                fio::FileRequest::ReadAt { count, offset, responder } => {
390                    assert_eq!(count, 3);
391                    assert_eq!(offset, 0);
392                    responder.send(Ok(b"012")).unwrap();
393                }
394                req => panic!("unhandled request {:?}", req),
395            }
396        }
397        .await;
398
399        // Call poll_read_at with a buf of length 1. This should resolve the previously created 3
400        // byte File.ReadAt future and return the first byte from it while saving the remaining two
401        // bytes.
402        let mut buf = [0u8; 1];
403        assert_eq!(reader.read_at(0, &mut buf).await.unwrap(), buf.len());
404        assert_eq!(&buf, b"0");
405
406        // Call poll_read_at with a buf of len 1. This should return the first saved byte, which
407        // should be the second byte from the original File.ReadAt request.
408        let mut buf = [0u8; 1];
409        assert_eq!(reader.read_at(1, &mut buf).await.unwrap(), buf.len());
410        assert_eq!(&buf, b"1");
411
412        // Call poll_read_at with a buf of len 2. There should only be one remaining saved byte
413        // from the original File.ReadAt request, so poll_read_at should only return one byte.
414        let mut buf = [0u8; 2];
415        assert_eq!(reader.read_at(2, &mut buf).await.unwrap(), 1);
416        assert_eq!(&buf[..1], b"2");
417
418        // There should be no saved bytes remaining, so a poll_read_at of four bytes should cause a
419        // new File.ReadAt request.
420        let mut buf = [0u8; 4];
421        let poll_read_at = reader.read_at(3, &mut buf);
422
423        let handle_second_file_request = async {
424            match stream.next().await.unwrap().unwrap() {
425                fio::FileRequest::ReadAt { count, offset, responder } => {
426                    assert_eq!(count, 4);
427                    assert_eq!(offset, 3);
428                    responder.send(Ok(b"3456")).unwrap();
429                }
430                req => panic!("unhandled request {:?}", req),
431            }
432        };
433
434        let (read_res, ()) = future::join(poll_read_at, handle_second_file_request).await;
435        assert_eq!(read_res.unwrap(), 4);
436        assert_eq!(&buf, b"3456");
437    }
438
439    #[fasync::run_singlethreaded(test)]
440    async fn transition_to_empty_on_fidl_error() {
441        let (proxy, _) = endpoints::create_proxy_and_stream::<fio::FileMarker>();
442
443        let mut reader = AsyncFile::from_proxy(proxy);
444
445        // poll_read_at will fail because the channel is closed because the server end was dropped.
446        let () = poll_fn(|cx| {
447            assert_matches!(
448                Pin::new(&mut reader).poll_read_at(cx, 0, &mut [0u8; 1]),
449                Poll::Ready(Err(_))
450            );
451            Poll::Ready(())
452        })
453        .await;
454
455        // This test is accessing internal state because the only fidl error that is easy to inject
456        // is ZX_ERR_PEER_CLOSED (by closing the channel). Once the channel is closed, all new
457        // futures created by the AsyncFile will fail, but, if poll'ed, the old future would also
458        // continue to fail (not panic) because it is Fused.
459        assert_matches!(reader.read_at_state, ReadAtState::Empty);
460    }
461
462    #[fasync::run_singlethreaded(test)]
463    async fn recover_from_file_read_error() {
464        let (proxy, mut stream) = endpoints::create_proxy_and_stream::<fio::FileMarker>();
465
466        let mut reader = AsyncFile::from_proxy(proxy);
467
468        // Call poll_read_at until failure.
469        let mut buf = [0u8; 1];
470        let poll_read_at = reader.read_at(0, &mut buf);
471
472        let failing_file_response = async {
473            match stream.next().await.unwrap().unwrap() {
474                fio::FileRequest::ReadAt { count, offset, responder } => {
475                    assert_eq!(count, 1);
476                    assert_eq!(offset, 0);
477                    responder.send(Err(zx_status::Status::NO_MEMORY.into_raw())).unwrap();
478                }
479                req => panic!("unhandled request {:?}", req),
480            }
481        };
482
483        let (read_res, ()) = future::join(poll_read_at, failing_file_response).await;
484        assert_matches!(read_res, Err(_));
485
486        // Calling poll_read_at again should create a new File.ReadAt request instead of reusing
487        // the old future.
488        let mut buf = [0u8; 1];
489        let poll_read_at = reader.read_at(0, &mut buf);
490
491        let succeeding_file_response = async {
492            match stream.next().await.unwrap().unwrap() {
493                fio::FileRequest::ReadAt { count, offset, responder } => {
494                    assert_eq!(count, 1);
495                    assert_eq!(offset, 0);
496                    responder.send(Ok(b"0")).unwrap();
497                }
498                req => panic!("unhandled request {:?}", req),
499            }
500        };
501
502        let (read_res, ()) = future::join(poll_read_at, succeeding_file_response).await;
503        assert_eq!(read_res.unwrap(), 1);
504        assert_eq!(&buf, b"0");
505    }
506
507    #[fasync::run_singlethreaded(test)]
508    async fn poll_read_at_zero_then_read_nonzero() {
509        let (proxy, mut stream) = endpoints::create_proxy_and_stream::<fio::FileMarker>();
510
511        let mut reader = AsyncFile::from_proxy(proxy);
512
513        // Call poll_read_at with a zero-length buffer.
514        let () = poll_fn(|cx| {
515            assert_matches!(Pin::new(&mut reader).poll_read_at(cx, 0, &mut []), Poll::Pending);
516            Poll::Ready(())
517        })
518        .await;
519
520        // Handle the zero-length File.ReadAt request.
521        match stream.next().await.unwrap().unwrap() {
522            fio::FileRequest::ReadAt { count, offset, responder } => {
523                assert_eq!(count, 0);
524                assert_eq!(offset, 0);
525                responder.send(Ok(&[])).unwrap();
526            }
527            req => panic!("unhandled request {:?}", req),
528        }
529
530        // Call poll_read_at with a length 1 buffer until Ready is returned;
531        let mut buf = vec![0u8; 1];
532        let poll_read_at = reader.read_at(0, &mut buf);
533
534        // The AsyncFile will discard the File.ReadAt response from the first poll_read, and create
535        // another request, this handles that second request. The AsyncFile discards the first
536        // response because the first poll_read_at was for zero bytes, but the current poll_read_at
537        // is not.
538        let handle_file_request = async {
539            match stream.next().await.unwrap().unwrap() {
540                fio::FileRequest::ReadAt { count, offset, responder } => {
541                    assert_eq!(count, 1);
542                    assert_eq!(offset, 0);
543                    responder.send(Ok(&[1])).unwrap();
544                }
545                req => panic!("unhandled request {:?}", req),
546            }
547        };
548
549        let (poll_read, ()) = future::join(poll_read_at, handle_file_request).await;
550
551        // poll_read_at should read 1 byte, even though the first poll_read_at request was for zero
552        // bytes and returned Pending.
553        assert_eq!(poll_read.unwrap(), 1);
554        assert_eq!(&buf[..], &[1]);
555    }
556
557    #[fasync::run_singlethreaded(test)]
558    async fn different_poll_read_at_and_file_sizes() {
559        for first_poll_read_len in 0..5 {
560            for file_size in 0..5 {
561                for second_poll_offset in 0..file_size {
562                    for second_poll_read_len in 0..5 {
563                        let (proxy, mut stream) =
564                            endpoints::create_proxy_and_stream::<fio::FileMarker>();
565
566                        let mut reader = AsyncFile::from_proxy(proxy);
567
568                        // poll_read_at causes the AsyncFile to create a File.ReadAt request.
569                        let () = poll_fn(|cx| {
570                            let mut buf = vec![0u8; first_poll_read_len];
571                            assert_matches!(
572                                Pin::new(&mut reader).poll_read_at(cx, 0, &mut buf),
573                                Poll::Pending
574                            );
575                            Poll::Ready(())
576                        })
577                        .await;
578
579                        // Respond to the File.ReadAt request with at most as many bytes as the
580                        // poll_read_at requested.
581                        match stream.next().await.unwrap().unwrap() {
582                            fio::FileRequest::ReadAt { count, offset, responder } => {
583                                assert_eq!(count, u64::try_from(first_poll_read_len).unwrap());
584                                assert_eq!(offset, 0);
585                                let resp = vec![7u8; min(file_size, first_poll_read_len)];
586                                responder.send(Ok(&resp)).unwrap();
587                            }
588                            req => panic!("unhandled request {:?}", req),
589                        }
590
591                        // Call poll_read_at until it returns Ready. If the first poll_read_at was
592                        // for zero bytes and this poll_read_at is not or this poll_read_at offset
593                        // is outside the buffer, the AsyncFile will make another File.ReadAt
594                        // request.
595                        let mut buf = vec![0u8; second_poll_read_len];
596                        let poll_read_at = reader.read_at(second_poll_offset as u64, &mut buf);
597
598                        let second_request = first_poll_read_len == 0 && second_poll_read_len != 0
599                            || second_poll_offset != 0 && second_poll_offset >= first_poll_read_len;
600                        let handle_conditional_file_request = async {
601                            if second_request {
602                                match stream.next().await.unwrap().unwrap() {
603                                    fio::FileRequest::ReadAt { count, offset, responder } => {
604                                        assert_eq!(
605                                            count,
606                                            u64::try_from(second_poll_read_len).unwrap()
607                                        );
608                                        assert_eq!(
609                                            offset,
610                                            u64::try_from(second_poll_offset).unwrap()
611                                        );
612                                        let resp = vec![
613                                            7u8;
614                                            min(
615                                                file_size - second_poll_offset,
616                                                second_poll_read_len
617                                            )
618                                        ];
619                                        responder.send(Ok(&resp)).unwrap();
620                                    }
621                                    req => panic!("unhandled request {:?}", req),
622                                }
623                            }
624                        };
625
626                        let (read_res, ()) =
627                            future::join(poll_read_at, handle_conditional_file_request).await;
628
629                        let expected_len = if second_request {
630                            min(file_size - second_poll_offset, second_poll_read_len)
631                        } else {
632                            min(
633                                min(file_size, first_poll_read_len) - second_poll_offset,
634                                second_poll_read_len,
635                            )
636                        };
637                        let expected = vec![7u8; expected_len];
638                        assert_eq!(read_res.unwrap(), expected_len);
639                        assert_eq!(&buf[..expected_len], &expected[..]);
640                    }
641                }
642            }
643        }
644    }
645
646    async fn get_size_file_with_contents(contents: &[u8]) {
647        let dir = TempDir::new().unwrap();
648        let path = dir.path().join("get_size_file_with_contents").to_str().unwrap().to_owned();
649        let () = file::write_in_namespace(&path, contents).await.unwrap();
650        let file = file::open_in_namespace(&path, fio::PERM_READABLE).unwrap();
651
652        let mut reader = AsyncFile::from_proxy(file);
653
654        assert_eq!(reader.get_size().await.unwrap(), contents.len() as u64);
655    }
656
657    #[fasync::run_singlethreaded(test)]
658    async fn get_size_empty() {
659        get_size_file_with_contents(&[]).await;
660    }
661
662    #[fasync::run_singlethreaded(test)]
663    async fn get_size_large() {
664        let expected_contents = vec![7u8; (fio::MAX_BUF * 3).try_into().unwrap()];
665        get_size_file_with_contents(&expected_contents[..]).await;
666    }
667
668    #[fasync::run_singlethreaded(test)]
669    async fn get_size_changing_size() {
670        let (mut file, path) = NamedTempFile::new().unwrap().into_parts();
671        let proxy = file::open_in_namespace(path.to_str().unwrap(), fio::PERM_READABLE).unwrap();
672
673        let mut reader = AsyncFile::from_proxy(proxy);
674
675        assert_eq!(reader.get_size().await.unwrap(), 0);
676        file.write_all(&[1; 3][..]).unwrap();
677        assert_eq!(reader.get_size().await.unwrap(), 3);
678        file.write_all(&[2; 5][..]).unwrap();
679        assert_eq!(reader.get_size().await.unwrap(), 8);
680    }
681
682    #[fasync::run_singlethreaded(test)]
683    async fn adapter_for_cursor() {
684        let data = (0..1000).map(|i| (i % 256) as u8).collect::<Vec<_>>();
685        let cursor = futures::io::Cursor::new(data.clone());
686        let mut adapter = Adapter::new(cursor);
687
688        assert_eq!(adapter.get_size().await.unwrap(), 1000);
689
690        let mut buffer = vec![];
691        adapter.read_to_end(&mut buffer).await.unwrap();
692        assert_eq!(buffer, data);
693
694        let mut buffer = vec![0; 100];
695        adapter.read_at_exact(333, &mut buffer).await.unwrap();
696        assert_eq!(buffer, &data[333..433]);
697    }
698}
fuchsia_fs/file/async_read_at.rs

fuchsia_fs/file/
async_read_at.rs
