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

disk/
asynchronous.rs
