disk/
asynchronous.rs

1// Copyright 2022 The ChromiumOS Authors
2// Use of this source code is governed by a BSD-style license that can be
3// found in the LICENSE file.
4
5//! Asynchronous disk image helpers.
6
7use std::io;
8use std::sync::Arc;
9use std::time::Duration;
10
11use async_trait::async_trait;
12use base::AsRawDescriptors;
13use base::FileAllocate;
14use base::FileSetLen;
15use base::FileSync;
16use base::PunchHole;
17use base::RawDescriptor;
18use base::WriteZeroesAt;
19use cros_async::BackingMemory;
20use cros_async::BlockingPool;
21use cros_async::Executor;
22
23use crate::AsyncDisk;
24use crate::DiskFile;
25use crate::DiskGetLen;
26use crate::Error;
27use crate::Result;
28
29/// Async wrapper around a non-async `DiskFile` using a `BlockingPool`.
30///
31/// This is meant to be a transitional type, not a long-term solution for async disk support. Disk
32/// formats should be migrated to support async instead (b/219595052).
33pub struct AsyncDiskFileWrapper<T: DiskFile + Send> {
34    blocking_pool: BlockingPool,
35    inner: Arc<T>,
36}
37
38impl<T: DiskFile + Send> AsyncDiskFileWrapper<T> {
39    #[allow(dead_code)] // Only used if qcow or android-sparse features are enabled
40    pub fn new(disk_file: T, _ex: &Executor) -> Self {
41        Self {
42            blocking_pool: BlockingPool::new(1, Duration::from_secs(10)),
43            inner: Arc::new(disk_file),
44        }
45    }
46}
47
48impl<T: DiskFile + Send> DiskGetLen for AsyncDiskFileWrapper<T> {
49    fn get_len(&self) -> io::Result<u64> {
50        self.inner.get_len()
51    }
52}
53
54impl<T: DiskFile + Send + FileSetLen> FileSetLen for AsyncDiskFileWrapper<T> {
55    fn set_len(&self, len: u64) -> io::Result<()> {
56        self.inner.set_len(len)
57    }
58}
59
60impl<T: DiskFile + Send + FileAllocate> FileAllocate for AsyncDiskFileWrapper<T> {
61    fn allocate(&self, offset: u64, len: u64) -> io::Result<()> {
62        self.inner.allocate(offset, len)
63    }
64}
65
66impl<T: DiskFile + Send> AsRawDescriptors for AsyncDiskFileWrapper<T> {
67    fn as_raw_descriptors(&self) -> Vec<RawDescriptor> {
68        self.inner.as_raw_descriptors()
69    }
70}
71
72pub trait DiskFlush {
73    /// Flush intermediary buffers and/or dirty state to file. fsync not required.
74    fn flush(&self) -> io::Result<()>;
75}
76
77#[async_trait(?Send)]
78impl<
79        T: 'static
80            + DiskFile
81            + DiskFlush
82            + Send
83            + Sync
84            + FileAllocate
85            + FileSetLen
86            + FileSync
87            + PunchHole
88            + WriteZeroesAt,
89    > AsyncDisk for AsyncDiskFileWrapper<T>
90{
91    async fn flush(&self) -> Result<()> {
92        let inner_clone = self.inner.clone();
93        self.blocking_pool
94            .spawn(move || inner_clone.flush().map_err(Error::IoFlush))
95            .await
96    }
97
98    async fn fsync(&self) -> Result<()> {
99        let inner_clone = self.inner.clone();
100        self.blocking_pool
101            .spawn(move || inner_clone.fsync().map_err(Error::IoFsync))
102            .await
103    }
104
105    async fn fdatasync(&self) -> Result<()> {
106        let inner_clone = self.inner.clone();
107        self.blocking_pool
108            .spawn(move || inner_clone.fdatasync().map_err(Error::IoFdatasync))
109            .await
110    }
111
112    async fn read_to_mem<'a>(
113        &'a self,
114        mut file_offset: u64,
115        mem: Arc<dyn BackingMemory + Send + Sync>,
116        mem_offsets: cros_async::MemRegionIter<'a>,
117    ) -> Result<usize> {
118        let inner_clone = self.inner.clone();
119        let mem_offsets: Vec<cros_async::MemRegion> = mem_offsets.collect();
120        self.blocking_pool
121            .spawn(move || {
122                let mut size = 0;
123                for region in mem_offsets {
124                    let mem_slice = mem.get_volatile_slice(region).unwrap();
125                    let bytes_read = inner_clone
126                        .read_at_volatile(mem_slice, file_offset)
127                        .map_err(Error::ReadingData)?;
128                    size += bytes_read;
129                    if bytes_read < mem_slice.size() {
130                        break;
131                    }
132                    file_offset += bytes_read as u64;
133                }
134                Ok(size)
135            })
136            .await
137    }
138
139    async fn write_from_mem<'a>(
140        &'a self,
141        mut file_offset: u64,
142        mem: Arc<dyn BackingMemory + Send + Sync>,
143        mem_offsets: cros_async::MemRegionIter<'a>,
144    ) -> Result<usize> {
145        let inner_clone = self.inner.clone();
146        let mem_offsets: Vec<cros_async::MemRegion> = mem_offsets.collect();
147        self.blocking_pool
148            .spawn(move || {
149                let mut size = 0;
150                for region in mem_offsets {
151                    let mem_slice = mem.get_volatile_slice(region).unwrap();
152                    let bytes_written = inner_clone
153                        .write_at_volatile(mem_slice, file_offset)
154                        .map_err(Error::ReadingData)?;
155                    size += bytes_written;
156                    if bytes_written < mem_slice.size() {
157                        break;
158                    }
159                    file_offset += bytes_written as u64;
160                }
161                Ok(size)
162            })
163            .await
164    }
165
166    async fn punch_hole(&self, file_offset: u64, length: u64) -> Result<()> {
167        let inner_clone = self.inner.clone();
168        self.blocking_pool
169            .spawn(move || {
170                inner_clone
171                    .punch_hole(file_offset, length)
172                    .map_err(Error::IoPunchHole)
173            })
174            .await
175    }
176
177    async fn write_zeroes_at(&self, file_offset: u64, length: u64) -> Result<()> {
178        let inner_clone = self.inner.clone();
179        self.blocking_pool
180            .spawn(move || {
181                inner_clone
182                    .write_zeroes_all_at(file_offset, length as usize)
183                    .map_err(Error::WriteZeroes)
184            })
185            .await
186    }
187}
disk/asynchronous.rs

disk/
asynchronous.rs
