block_client/
cache.rs

1// Copyright 2020 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 super::{BufferSlice, MutableBufferSlice, RemoteBlockClientSync, VmoId};
6use anyhow::{Error, ensure};
7
8use linked_hash_map::LinkedHashMap;
9use log::error;
10use std::io::{SeekFrom, Write};
11
12const VMO_SIZE: u64 = 262_144;
13const BLOCK_SIZE: u64 = 8192;
14const BLOCK_COUNT: usize = (VMO_SIZE / BLOCK_SIZE) as usize;
15
16struct CacheEntry {
17    vmo_offset: u64,
18    dirty: bool,
19}
20
21#[derive(Debug, Default, Eq, PartialEq)]
22pub struct Stats {
23    read_count: u64,
24    write_count: u64,
25    cache_hits: u64,
26}
27
28/// Wraps RemoteBlockDeviceSync providing a simple LRU cache and trait implementations for
29/// std::io::{Read, Seek, Write}. This is unlikely to be performant; the implementation is single
30/// threaded. The cache works by dividing up a VMO into BLOCK_COUNT blocks of BLOCK_SIZE bytes, and
31/// maintaining mappings from device offsets to offsets in the VMO.
32pub struct Cache {
33    device: RemoteBlockClientSync,
34    vmo: zx::Vmo,
35    vmo_id: VmoId,
36    map: LinkedHashMap<u64, CacheEntry>,
37    offset: u64, // For std::io::{Read, Seek, Write}
38    stats: Stats,
39}
40
41impl Cache {
42    /// Returns a new Cache wrapping the given RemoteBlockClientSync.
43    pub fn new(device: RemoteBlockClientSync) -> Result<Self, Error> {
44        ensure!(
45            BLOCK_SIZE % device.block_size() as u64 == 0,
46            "underlying block size not supported"
47        );
48        let vmo = zx::Vmo::create(VMO_SIZE)?;
49        // SAFETY: The VMO is newly created and only attached once here. We do not map the VMO in
50        // this process (we only use vmo.read/write system calls), so no Rust references to the VMO
51        // memory are ever held.
52        let vmo_id = unsafe { device.attach_vmo(&vmo) }?;
53        Ok(Self {
54            device,
55            vmo,
56            vmo_id,
57            map: Default::default(),
58            offset: 0,
59            stats: Stats::default(),
60        })
61    }
62
63    fn device_size(&self) -> u64 {
64        self.device.block_count() * self.device.block_size() as u64
65    }
66
67    // Finds a block that can be used for the given offset, marking dirty if requested. Returns a
68    // tuple with the VMO offset and whether it was a cache hit. If not a cache hit, the caller is
69    // responsible for initializing the data and inserting a cache entry.
70    fn get_block(&mut self, offset: u64, mark_dirty: bool) -> Result<(u64, bool), Error> {
71        if let Some(ref mut entry) = self.map.get_refresh(&offset) {
72            self.stats.cache_hits += 1;
73            if mark_dirty {
74                entry.dirty = true;
75            }
76            Ok((entry.vmo_offset, true))
77        } else {
78            let vmo_offset = if self.map.len() < BLOCK_COUNT {
79                self.map.len() as u64 * BLOCK_SIZE
80            } else {
81                let entry = self.map.pop_front().unwrap();
82                if entry.1.dirty {
83                    self.stats.write_count += 1;
84                    self.device.write_at(
85                        BufferSlice::new_with_vmo_id(
86                            &self.vmo_id,
87                            entry.1.vmo_offset,
88                            std::cmp::min(BLOCK_SIZE, self.device_size() - entry.0),
89                        ),
90                        entry.0,
91                    )?;
92                }
93                entry.1.vmo_offset
94            };
95            Ok((vmo_offset, false))
96        }
97    }
98
99    // Reads the block at the given offset and marks it dirty if requested. Returns the offset in
100    // the VMO.
101    fn read_block(&mut self, offset: u64, mark_dirty: bool) -> Result<u64, Error> {
102        let (vmo_offset, hit) = self.get_block(offset, mark_dirty)?;
103        if !hit {
104            self.stats.read_count += 1;
105            self.device.read_at(
106                MutableBufferSlice::new_with_vmo_id(
107                    &self.vmo_id,
108                    vmo_offset,
109                    std::cmp::min(BLOCK_SIZE, self.device_size() - offset),
110                ),
111                offset,
112            )?;
113            self.map.insert(offset, CacheEntry { vmo_offset, dirty: mark_dirty });
114        }
115        Ok(vmo_offset)
116    }
117
118    /// Reads at |offset| into |buf|.
119    pub fn read_at(&mut self, mut buf: &mut [u8], offset: u64) -> Result<(), Error> {
120        ensure!(
121            offset <= self.device_size() && buf.len() as u64 <= self.device_size() - offset,
122            "read exceeds device size"
123        );
124
125        // Start by reading the head.
126        let mut aligned_offset = offset - offset % BLOCK_SIZE;
127        let end = offset + buf.len() as u64;
128        if aligned_offset < offset {
129            let vmo_offset = self.read_block(aligned_offset, false)?;
130            let to_copy = std::cmp::min(aligned_offset + BLOCK_SIZE, end) - offset;
131            self.vmo.read(&mut buf[..to_copy as usize], vmo_offset + offset - aligned_offset)?;
132            aligned_offset += BLOCK_SIZE;
133            buf = &mut buf[to_copy as usize..];
134        }
135
136        // Now do whole blocks.
137        while aligned_offset + BLOCK_SIZE <= end {
138            let vmo_offset = self.read_block(aligned_offset, false)?;
139            self.vmo.read(&mut buf[..BLOCK_SIZE as usize], vmo_offset)?;
140            aligned_offset += BLOCK_SIZE;
141            buf = &mut buf[BLOCK_SIZE as usize..];
142        }
143
144        // And finally the tail.
145        if end > aligned_offset {
146            let vmo_offset = self.read_block(aligned_offset, false)?;
147            self.vmo.read(buf, vmo_offset)?;
148        }
149        Ok(())
150    }
151
152    /// Writes from |buf| to |offset|.
153    pub fn write_at(&mut self, mut buf: &[u8], offset: u64) -> Result<(), Error> {
154        ensure!(
155            offset <= self.device_size() && buf.len() as u64 <= self.device_size() - offset,
156            "write exceeds device size"
157        );
158
159        // Start by writing the head.
160        let mut aligned_offset = offset - offset % BLOCK_SIZE;
161        let end = offset + buf.len() as u64;
162        if aligned_offset < offset {
163            let vmo_offset = self.read_block(aligned_offset, true)?;
164            let to_copy = std::cmp::min(aligned_offset + BLOCK_SIZE, end) - offset;
165            self.vmo.write(&buf[..to_copy as usize], vmo_offset + offset - aligned_offset)?;
166            aligned_offset += BLOCK_SIZE;
167            buf = &buf[to_copy as usize..];
168        }
169
170        // Now do whole blocks.
171        while aligned_offset + BLOCK_SIZE <= end {
172            let (vmo_offset, hit) = self.get_block(aligned_offset, true)?;
173            self.vmo.write(&buf[..BLOCK_SIZE as usize], vmo_offset)?;
174            if !hit {
175                self.map.insert(aligned_offset, CacheEntry { vmo_offset, dirty: true });
176            }
177            aligned_offset += BLOCK_SIZE;
178            buf = &buf[BLOCK_SIZE as usize..];
179        }
180
181        // And finally the tail.
182        if end > aligned_offset {
183            let vmo_offset = self.read_block(aligned_offset, true)?;
184            self.vmo.write(buf, vmo_offset)?;
185        }
186        Ok(())
187    }
188
189    /// Returns statistics.
190    pub fn stats(&self) -> &Stats {
191        &self.stats
192    }
193
194    /// Returns a reference to the underlying device
195    /// Can be used for additional control, like instructing the device to flush any written data
196    pub fn device(&self) -> &RemoteBlockClientSync {
197        &self.device
198    }
199
200    pub fn flush_device(&self) -> Result<(), Error> {
201        Ok(self.device.flush()?)
202    }
203}
204
205impl Drop for Cache {
206    fn drop(&mut self) {
207        if let Err(e) = self.flush() {
208            error!("Flush failed: {}", e);
209        }
210        let _ = self.vmo_id.take().into_id(); // Ok to leak because fifo will be closed.
211    }
212}
213
214fn into_io_error<E: Into<Box<dyn std::error::Error + Send + Sync>>>(error: E) -> std::io::Error {
215    std::io::Error::other(error)
216}
217
218impl std::io::Read for Cache {
219    fn read(&mut self, mut buf: &mut [u8]) -> std::io::Result<usize> {
220        if self.offset > self.device_size() {
221            return Ok(0);
222        }
223        let max_len = self.device_size() - self.offset;
224        if buf.len() as u64 > max_len {
225            buf = &mut buf[0..max_len as usize];
226        }
227        self.read_at(buf, self.offset).map_err(into_io_error)?;
228        self.offset += buf.len() as u64;
229        Ok(buf.len())
230    }
231}
232
233impl Write for Cache {
234    fn write(&mut self, mut buf: &[u8]) -> std::io::Result<usize> {
235        if self.offset > self.device_size() {
236            return Ok(0);
237        }
238        let max_len = self.device_size() - self.offset;
239        if buf.len() as u64 > max_len {
240            buf = &buf[0..max_len as usize];
241        }
242        self.write_at(&buf, self.offset).map_err(into_io_error)?;
243        self.offset += buf.len() as u64;
244        Ok(buf.len())
245    }
246
247    /// This does *not* issue a flush to the underlying block device; this will only send the
248    /// writes.
249    fn flush(&mut self) -> std::io::Result<()> {
250        let max = self.device_size();
251        for mut entry in self.map.entries() {
252            if entry.get().dirty {
253                self.stats.write_count += 1;
254                self.device
255                    .write_at(
256                        BufferSlice::new_with_vmo_id(
257                            &self.vmo_id,
258                            entry.get().vmo_offset,
259                            std::cmp::min(BLOCK_SIZE, max - *entry.key()),
260                        ),
261                        *entry.key(),
262                    )
263                    .map_err(into_io_error)?;
264                entry.get_mut().dirty = false;
265            }
266        }
267        Ok(())
268    }
269}
270
271impl std::io::Seek for Cache {
272    fn seek(&mut self, pos: SeekFrom) -> std::io::Result<u64> {
273        self.offset = match pos {
274            SeekFrom::Start(offset) => Some(offset),
275            SeekFrom::End(delta) => {
276                if delta >= 0 {
277                    self.device_size().checked_add(delta as u64)
278                } else {
279                    self.device_size().checked_sub(-delta as u64)
280                }
281            }
282            SeekFrom::Current(delta) => {
283                if delta >= 0 {
284                    self.offset.checked_add(delta as u64)
285                } else {
286                    self.offset.checked_sub(-delta as u64)
287                }
288            }
289        }
290        .ok_or_else(|| std::io::Error::new(std::io::ErrorKind::InvalidInput, "bad delta"))?;
291        Ok(self.offset)
292    }
293}
294
295#[cfg(test)]
296mod tests {
297    use super::{Cache, Stats};
298    use crate::RemoteBlockClientSync;
299    use ramdevice_client::RamdiskClient;
300    use std::io::{Read as _, Seek as _, SeekFrom, Write as _};
301
302    const RAMDISK_BLOCK_SIZE: u64 = 1024;
303    const RAMDISK_BLOCK_COUNT: u64 = 1023; // Deliberate for testing max offset.
304    const RAMDISK_SIZE: u64 = RAMDISK_BLOCK_SIZE * RAMDISK_BLOCK_COUNT;
305
306    pub async fn make_ramdisk() -> (RamdiskClient, RemoteBlockClientSync) {
307        let ramdisk = RamdiskClient::create(RAMDISK_BLOCK_SIZE, RAMDISK_BLOCK_COUNT)
308            .await
309            .expect("RamdiskClient::create failed");
310        let client_end = ramdisk.open().expect("ramdisk.open failed");
311        let block_client =
312            RemoteBlockClientSync::new(client_end).expect("RemoteBlockClientSync::new failed");
313        (ramdisk, block_client)
314    }
315
316    #[fuchsia::test]
317    async fn test_cache_read_at_and_write_at_with_no_hits() {
318        let (_ramdisk, block_client) = make_ramdisk().await;
319        let mut cache = Cache::new(block_client).expect("Cache::new failed");
320        let mut offset = 5;
321        const TEST_COUNT: usize = super::BLOCK_COUNT * 2; // Chosen so there are no cache hits.
322        const DATA: &[u8] = b"hello";
323        for _ in 0..TEST_COUNT {
324            cache.write_at(DATA, offset).expect("cache.write failed");
325            // The delta here is deliberately chosen to catch mistakes such as returning data from
326            // the wrong block.
327            offset += super::BLOCK_SIZE + 1;
328        }
329        assert_eq!(
330            cache.stats(),
331            &Stats {
332                read_count: TEST_COUNT as u64,
333                write_count: super::BLOCK_COUNT as u64,
334                cache_hits: 0
335            }
336        );
337        offset = 5;
338        for _ in 0..TEST_COUNT {
339            let mut buf = [0; 5];
340            cache.read_at(&mut buf, offset).expect("cache.read_at failed");
341            assert_eq!(&buf, DATA);
342            offset += super::BLOCK_SIZE + 1;
343        }
344        assert_eq!(
345            cache.stats(),
346            &Stats {
347                read_count: 2 * TEST_COUNT as u64,
348                write_count: TEST_COUNT as u64,
349                cache_hits: 0
350            }
351        );
352    }
353
354    #[fuchsia::test]
355    async fn test_cache_read_at_and_write_at_with_hit() {
356        let (_ramdisk, block_client) = make_ramdisk().await;
357        let mut cache = Cache::new(block_client).expect("Cache::new failed");
358        const OFFSET: u64 = 11;
359        const DATA: &[u8] = b"hello";
360        cache.write_at(DATA, OFFSET).expect("cache.write failed");
361        let mut buf = [0; 5];
362        cache.read_at(&mut buf, OFFSET).expect("cache.read_at failed");
363        assert_eq!(&buf, DATA);
364        assert_eq!(cache.stats(), &Stats { read_count: 1, write_count: 0, cache_hits: 1 });
365    }
366
367    #[fuchsia::test]
368    async fn test_cache_aligned_read_at_and_write_at() {
369        let (_ramdisk, block_client) = make_ramdisk().await;
370        let mut cache = Cache::new(block_client).expect("Cache::new failed");
371        const OFFSET: u64 = 11;
372        const BLOCKS: usize = 3;
373        const DATA_LEN: usize = super::BLOCK_SIZE as usize * BLOCKS + 11;
374        let data = [0xe2u8; DATA_LEN];
375        // This should require alignment at the start, and at the end with some whole blocks.
376        cache.write_at(&data, OFFSET).expect("cache.write failed");
377        let mut buf = [0; DATA_LEN + 2]; // Read an extra byte at the start and at the end.
378        cache.read_at(&mut buf, OFFSET - 1).expect("cache.read_at failed");
379        assert_eq!(buf[0], 0);
380        assert_eq!(buf[DATA_LEN + 1], 0);
381        assert_eq!(&buf[1..DATA_LEN + 1], &data[0..DATA_LEN]);
382        // We should have only read the first and last blocks. The writes to the whole blocks should
383        // not have triggered reads.
384        assert_eq!(
385            cache.stats(),
386            &Stats { read_count: 2, write_count: 0, cache_hits: BLOCKS as u64 + 1 }
387        );
388    }
389
390    #[fuchsia::test]
391    async fn test_cache_aligned_read_at_and_write_at_cold() {
392        // The same as the previous test, but tear down the cache after the writes.
393        let (ramdisk, block_client) = make_ramdisk().await;
394        let mut cache = Cache::new(block_client).expect("Cache::new failed");
395        const OFFSET: u64 = 11;
396        const BLOCKS: usize = 3;
397        const DATA_LEN: usize = super::BLOCK_SIZE as usize * BLOCKS + 11;
398        let data = [0xe2u8; DATA_LEN];
399        // This should require alignment at the start, and at the end with some whole blocks.
400        cache.write_at(&data, OFFSET).expect("cache.write failed");
401        assert_eq!(cache.stats(), &Stats { read_count: 2, write_count: 0, cache_hits: 0 });
402
403        drop(cache);
404        let client_end = ramdisk.open().expect("ramdisk.open failed");
405        let block_client =
406            RemoteBlockClientSync::new(client_end).expect("RemoteBlockClientSync::new failed");
407        let mut cache = Cache::new(block_client).expect("Cache::new failed");
408
409        let mut buf = [0; DATA_LEN + 2]; // Read an extra byte at the start and at the end.
410        cache.read_at(&mut buf, OFFSET - 1).expect("cache.read_at failed");
411        assert_eq!(buf[0], 0);
412        assert_eq!(buf[DATA_LEN + 1], 0);
413        assert_eq!(&buf[1..DATA_LEN + 1], &data[0..DATA_LEN]);
414        // We should have only read the first and last blocks. The writes to the whole blocks should
415        // not have triggered reads.
416        assert_eq!(
417            cache.stats(),
418            &Stats { read_count: BLOCKS as u64 + 1, write_count: 0, cache_hits: 0 }
419        );
420    }
421
422    #[fuchsia::test]
423    async fn test_io_read_write_and_seek() {
424        let (_ramdisk, block_client) = make_ramdisk().await;
425        let mut cache = Cache::new(block_client).expect("Cache::new failed");
426        const OFFSET: u64 = 11;
427        const DATA: &[u8] = b"hello";
428        assert_eq!(cache.seek(SeekFrom::Start(OFFSET)).expect("seek failed"), OFFSET);
429        cache.write_all(DATA).expect("cache.write failed");
430        assert_eq!(
431            cache.seek(SeekFrom::Current(-(DATA.len() as i64))).expect("seek failed"),
432            OFFSET
433        );
434        let mut buf = [0u8; 5];
435        assert_eq!(cache.read(&mut buf).expect("cache.read failed"), DATA.len());
436        assert_eq!(&buf, DATA);
437    }
438
439    #[fuchsia::test]
440    async fn test_io_read_write_and_seek_at_max_offset() {
441        let (_ramdisk, block_client) = make_ramdisk().await;
442        let mut cache = Cache::new(block_client).expect("Cache::new failed");
443        const DATA: &[u8] = b"hello";
444        assert_eq!(cache.seek(SeekFrom::End(-1)).expect("seek failed"), RAMDISK_SIZE - 1);
445        assert_eq!(cache.write(DATA).expect("cache.write failed"), 1);
446        assert_eq!(cache.seek(SeekFrom::End(-4)).expect("seek failed"), RAMDISK_SIZE - 4);
447        let mut buf = [0x56u8; 5];
448        assert_eq!(cache.read(&mut buf).expect("cache.read failed"), 4);
449        assert_eq!(&buf, &[0, 0, 0, b'h', 0x56]);
450    }
451
452    #[fuchsia::test]
453    async fn test_read_beyond_max_offset_returns_error() {
454        let (_ramdisk, block_client) = make_ramdisk().await;
455        let mut cache = Cache::new(block_client).expect("Cache::new failed");
456        let mut buf = [0u8; 2];
457        cache.read_at(&mut buf, RAMDISK_SIZE).expect_err("read_at succeeded");
458        cache.read_at(&mut buf, RAMDISK_SIZE - 1).expect_err("read_at succeeded");
459    }
460
461    #[fuchsia::test]
462    async fn test_write_beyond_max_offset_returns_error() {
463        let (_ramdisk, block_client) = make_ramdisk().await;
464        let mut cache = Cache::new(block_client).expect("Cache::new failed");
465        let buf = [0u8; 2];
466        cache.write_at(&buf, RAMDISK_SIZE).expect_err("write_at succeeded");
467        cache.write_at(&buf, RAMDISK_SIZE - 1).expect_err("write_at succeeded");
468    }
469
470    #[fuchsia::test]
471    async fn test_read_with_overflow_returns_error() {
472        let (_ramdisk, block_client) = make_ramdisk().await;
473        let mut cache = Cache::new(block_client).expect("Cache::new failed");
474        let mut buf = [0u8; 2];
475        cache.read_at(&mut buf, u64::MAX - 1).expect_err("read_at succeeded");
476    }
477
478    #[fuchsia::test]
479    async fn test_write_with_overflow_returns_error() {
480        let (_ramdisk, block_client) = make_ramdisk().await;
481        let mut cache = Cache::new(block_client).expect("Cache::new failed");
482        let buf = [0u8; 2];
483        cache.write_at(&buf, u64::MAX - 1).expect_err("write_at succeeded");
484    }
485
486    #[fuchsia::test]
487    async fn test_read_and_write_at_max_offset_suceeds() {
488        let (_ramdisk, block_client) = make_ramdisk().await;
489        let mut cache = Cache::new(block_client).expect("Cache::new failed");
490        let buf = [0xd4u8; 2];
491        cache.write_at(&buf, RAMDISK_SIZE - buf.len() as u64).expect("write_at failed");
492        let mut read_buf = [0xf3u8; 2];
493        cache.read_at(&mut read_buf, RAMDISK_SIZE - buf.len() as u64).expect("read_at failed");
494        assert_eq!(&buf, &read_buf);
495    }
496
497    #[fuchsia::test]
498    async fn test_seek_with_bad_delta_returns_error() {
499        let (_ramdisk, block_client) = make_ramdisk().await;
500        let mut cache = Cache::new(block_client).expect("Cache::new failed");
501        cache.seek(SeekFrom::End(-(RAMDISK_SIZE as i64) - 1)).expect_err("seek suceeded");
502        cache.seek(SeekFrom::Current(-1)).expect_err("seek succeeded");
503    }
504
505    #[fuchsia::test]
506    async fn test_ramdisk_with_large_block_size_returns_error() {
507        let ramdisk = RamdiskClient::create(super::BLOCK_SIZE * 2, 10)
508            .await
509            .expect("RamdiskClient::create failed");
510        let client_end = ramdisk.open().expect("ramdisk.open failed");
511        let block_client =
512            RemoteBlockClientSync::new(client_end).expect("RemoteBlockClientSync::new failed");
513        Cache::new(block_client).err().expect("Cache::new succeeded");
514    }
515}
block_client/cache.rs

block_client/
cache.rs
