fxfs_platform_testing/fuchsia/
layer_pager.rs

1// Copyright 2026 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 anyhow::{Context, Error};
6use async_trait::async_trait;
7use fidl_fuchsia_storage_block as fblock;
8use fuchsia_async as fasync;
9use futures::lock::Mutex as AsyncMutex;
10use fxfs::filesystem::LayerPager;
11use fxfs::object_handle::{LayerObject, ObjectHandle, ReadObjectHandle};
12use fxfs::object_store::{DataObjectHandle, FileExtent, ObjectStore};
13use fxfs_crypto::UnwrappedKey;
14use mapping::{
15    Extent, Extents, MAPPING_VMO_SIZE, MappingCommand, PENDING_COMMANDS_CAPACITY, RawMappingCommand,
16};
17use std::sync::Arc;
18use std::sync::atomic::{AtomicBool, Ordering};
19use storage_device::buffer::{BufferFuture, MutableBufferRef};
20use storage_units::BlockSize;
21use vmo_fifo::AsyncSender;
22use zx;
23
24struct MappedVmo {
25    vmo: zx::Vmo,
26    vaddr: usize,
27    len: usize,
28}
29
30impl MappedVmo {
31    fn new(vmo: zx::Vmo, len: usize) -> Result<Self, zx::Status> {
32        let page_size = zx::system_get_page_size() as usize;
33        let aligned_len = len.next_multiple_of(page_size);
34        let vaddr = if aligned_len > 0 {
35            fuchsia_runtime::vmar_root_self().map(
36                0,
37                &vmo,
38                0,
39                aligned_len,
40                zx::VmarFlags::PERM_READ,
41            )?
42        } else {
43            0
44        };
45        Ok(Self { vmo, vaddr, len })
46    }
47
48    fn as_slice(&self) -> &[u8] {
49        if self.vaddr == 0 || self.len == 0 {
50            &[]
51        } else {
52            // SAFETY: self.vaddr is mapped for at least self.len bytes with PERM_READ.  Note:
53            // This creates a `&[u8]` slice over memory backed by a pager-supplied VMO.  In
54            // theory, the driver could write to it which is undefined behavior in Rust, but the
55            // driver only supplies pages once (immutable) and this trade-off is acceptable for
56            // now (we could fix this with kernel changes).
57            unsafe { std::slice::from_raw_parts(self.vaddr as *const u8, self.len) }
58        }
59    }
60
61    fn has_io_error(&self) -> bool {
62        self.vmo
63            .wait_one(zx::Signals::USER_0, zx::MonotonicInstant::INFINITE_PAST)
64            .to_result()
65            .is_ok()
66    }
67
68    fn purge_cached_data(&self) {
69        if self.len > 0 {
70            // `DONT_NEED` doesn't actually purge the data immediately; it only serves as a hint
71            // to the kernel, which is the best we can do here.
72            let _ = self.vmo.op_range(zx::VmoOp::DONT_NEED, 0, self.len as u64);
73        }
74    }
75
76    fn prefetch(&self, range: std::ops::Range<u64>) {
77        if range.end > range.start {
78            let _ = self.vmo.op_range(zx::VmoOp::PREFETCH, range.start, range.end - range.start);
79        }
80    }
81}
82
83impl Drop for MappedVmo {
84    fn drop(&mut self) {
85        if self.vaddr != 0 {
86            let page_size = zx::system_get_page_size() as usize;
87            let aligned_len = self.len.next_multiple_of(page_size);
88            // SAFETY: self.vaddr was mapped by vmar_root_self().map with aligned_len.
89            unsafe {
90                let _ = fuchsia_runtime::vmar_root_self().unmap(self.vaddr, aligned_len);
91            }
92        }
93    }
94}
95
96struct PagedLayerObject {
97    handle: DataObjectHandle<ObjectStore>,
98    mapped: MappedVmo,
99    sender: Arc<AsyncMutex<AsyncSender<RawMappingCommand>>>,
100    scope: fasync::ScopeHandle,
101    key: u64,
102    closed: AtomicBool,
103}
104
105impl ObjectHandle for PagedLayerObject {
106    fn object_id(&self) -> u64 {
107        self.handle.object_id()
108    }
109
110    fn block_size(&self) -> BlockSize {
111        self.handle.block_size()
112    }
113
114    fn allocate_buffer(&self, size: usize) -> BufferFuture<'_> {
115        self.handle.allocate_buffer(size)
116    }
117
118    fn set_trace(&self, v: bool) {
119        self.handle.set_trace(v)
120    }
121}
122
123#[async_trait]
124impl ReadObjectHandle for PagedLayerObject {
125    async fn read_aligned(&self, offset: u64, buf: MutableBufferRef<'_>) -> Result<usize, Error> {
126        self.handle.read_aligned(offset, buf).await
127    }
128
129    fn get_size(&self) -> u64 {
130        self.handle.get_size()
131    }
132}
133
134#[async_trait]
135impl LayerObject for PagedLayerObject {
136    fn as_slice(&self) -> Option<&[u8]> {
137        Some(self.mapped.as_slice())
138    }
139
140    fn has_io_error(&self) -> bool {
141        self.mapped.has_io_error()
142    }
143
144    fn purge_cached_data(&self) {
145        self.mapped.purge_cached_data();
146    }
147
148    fn prefetch(&self, range: std::ops::Range<u64>) {
149        self.mapped.prefetch(range);
150    }
151
152    async fn close(&self) {
153        if !self.closed.swap(true, Ordering::Relaxed) {
154            let mut sender = self.sender.lock().await;
155            let _ = sender.push(MappingCommand::CloseBlob { key: self.key }.into()).await;
156        }
157    }
158
159    fn get_vmo(&self) -> Option<&zx::Vmo> {
160        Some(&self.mapped.vmo)
161    }
162}
163
164impl Drop for PagedLayerObject {
165    fn drop(&mut self) {
166        // Layers are only explicitly `close()`d during compaction; when a volume is unmounted or
167        // locked (or if `LayerData::open` fails), the layer is dropped directly.
168        if !*self.closed.get_mut() {
169            let sender = self.sender.clone();
170            let key = self.key;
171            self.scope.spawn(async move {
172                let mut sender = sender.lock().await;
173                let _ = sender.push(MappingCommand::CloseBlob { key }.into()).await;
174            });
175        }
176    }
177}
178
179/// A read-only `LayerObject` backed by a `zx::Vmo` mapped into the current process's address
180/// space. Used by remote volume clients to read layer files synchronously via
181/// `SyncPersistentLayer`.
182pub struct VmoLayerObject {
183    mapped: MappedVmo,
184    object_id: u64,
185    block_size: BlockSize,
186}
187
188impl VmoLayerObject {
189    pub fn new(vmo: zx::Vmo, object_id: u64, block_size: BlockSize) -> Result<Self, Error> {
190        let len = vmo.get_content_size().context("Failed to get VMO content size")? as usize;
191        let mapped = MappedVmo::new(vmo, len).context("Failed to map layer VMO")?;
192        Ok(Self { mapped, object_id, block_size })
193    }
194}
195
196impl ObjectHandle for VmoLayerObject {
197    fn object_id(&self) -> u64 {
198        self.object_id
199    }
200
201    fn block_size(&self) -> BlockSize {
202        self.block_size
203    }
204
205    fn allocate_buffer(&self, _size: usize) -> BufferFuture<'_> {
206        unreachable!("VmoLayerObject is slice-backed and does not allocate buffers")
207    }
208}
209
210#[async_trait]
211impl ReadObjectHandle for VmoLayerObject {
212    async fn read_aligned(&self, _offset: u64, _buf: MutableBufferRef<'_>) -> Result<usize, Error> {
213        unreachable!("VmoLayerObject is slice-backed and does not support read_aligned")
214    }
215
216    fn get_size(&self) -> u64 {
217        self.mapped.len as u64
218    }
219}
220
221#[async_trait]
222impl LayerObject for VmoLayerObject {
223    fn as_slice(&self) -> Option<&[u8]> {
224        Some(self.mapped.as_slice())
225    }
226
227    fn has_io_error(&self) -> bool {
228        self.mapped.has_io_error()
229    }
230
231    fn purge_cached_data(&self) {
232        self.mapped.purge_cached_data();
233    }
234
235    fn prefetch(&self, range: std::ops::Range<u64>) {
236        self.mapped.prefetch(range);
237    }
238
239    async fn close(&self) {}
240
241    fn get_vmo(&self) -> Option<&zx::Vmo> {
242        Some(&self.mapped.vmo)
243    }
244}
245
246/// Services registration of layer files with a pager backed by the block mapping driver.
247pub struct LayerPagerImpl {
248    mapper_session: fblock::MapperSessionProxy,
249    sender: Arc<AsyncMutex<AsyncSender<RawMappingCommand>>>,
250    scope: fasync::Scope,
251}
252
253impl LayerPagerImpl {
254    pub async fn new(mapper_proxy: &fblock::MapperProxy) -> Result<Self, Error> {
255        let mapping_vmo = zx::Vmo::create(MAPPING_VMO_SIZE)?;
256        let sender = AsyncSender::<RawMappingCommand>::new(
257            mapping_vmo.duplicate_handle(zx::Rights::SAME_RIGHTS)?,
258            8,
259            PENDING_COMMANDS_CAPACITY,
260        )?;
261
262        let (mapper_session, mapper_session_server) = fidl::endpoints::create_proxy();
263        mapper_proxy
264            .open_session(mapper_session_server, mapping_vmo, None, None)
265            .await
266            .context("FIDL error on Mapper.OpenSession")?
267            .map_err(zx::Status::err_from_raw)
268            .context("Failed to open mapper session")?;
269
270        Ok(Self {
271            mapper_session,
272            sender: Arc::new(AsyncMutex::new(sender)),
273            scope: fasync::Scope::new_with_name("layer_pager"),
274        })
275    }
276
277    async fn register_layer(
278        &self,
279        size: u64,
280        extents: &[FileExtent],
281        raw_key: Option<&[u8]>,
282    ) -> Result<(u64, zx::Vmo), Error> {
283        let mut mapping_extents = Vec::with_capacity(extents.len());
284        let mut current_offset = 0u64;
285        for ext in extents {
286            if ext.logical_offset() > current_offset {
287                mapping_extents.push(Extent::try_new(current_offset..ext.logical_offset(), None)?);
288            }
289            mapping_extents
290                .push(Extent::try_new(ext.logical_range(), Some(ext.device_range().start))?);
291            current_offset = ext.logical_range().end;
292        }
293        let aligned_size = size.next_multiple_of(mapping::BLOCK_SIZE);
294        if current_offset < aligned_size {
295            mapping_extents.push(Extent::try_new(current_offset..aligned_size, None)?);
296        }
297        let data_extents = Extents::try_new(&mapping_extents, 0)?;
298        let extent_count = data_extents.len() as u32;
299        let encrypted = raw_key.is_some();
300        let key_bytes_len = if encrypted { 32 } else { 0 };
301        let allocation_size = (extent_count as usize * 8) + key_bytes_len;
302
303        let (key, vmo) = self
304            .mapper_session
305            .create_vmo(size, fblock::CreateVmoOptions::SUPPLY_ZEROES_ON_ERROR)
306            .await
307            .context("FIDL error calling MapperSession.CreateVmo")?
308            .map_err(zx::Status::err_from_raw)
309            .context("MapperSession.CreateVmo returned error")?;
310
311        if allocation_size > 0 {
312            let mut sender = self.sender.lock().await;
313            let mut payload = sender.reserve_payload(allocation_size).await?;
314            let offset_in_vmo = payload.offset();
315
316            let extent_bytes = extent_count as usize * 8;
317            for (mut chunk, val) in payload
318                .data()
319                .subslice_mut(0..extent_bytes)
320                .chunks_mut(8)
321                .zip(Extents::encode_extents(&data_extents))
322            {
323                chunk.copy_from_slice(&val.to_le_bytes());
324            }
325
326            if let Some(k) = raw_key {
327                payload.data().subslice_mut(extent_bytes..extent_bytes + 32).copy_from_slice(k);
328            }
329
330            let command = MappingCommand::Mappings {
331                key,
332                offset: offset_in_vmo as u32,
333                stored_size: size,
334                device_offset: 0,
335                metadata_count: 0,
336                extent_count,
337                encrypted,
338            };
339            payload.commit(command.into()).await?;
340        }
341
342        Ok((key, vmo))
343    }
344}
345
346#[async_trait]
347impl LayerPager for LayerPagerImpl {
348    async fn open_layer(
349        &self,
350        handle: DataObjectHandle<ObjectStore>,
351        unwrapped_key: Option<UnwrappedKey>,
352    ) -> Result<Arc<dyn LayerObject>, Error> {
353        let size = handle.get_size();
354        if size == 0 || handle.block_size() > BlockSize::SIZE_4KIB {
355            return Ok(Arc::new(handle) as Arc<dyn LayerObject>);
356        }
357        let extents = handle.device_extents().await?;
358        let (key, vmo) = self
359            .register_layer(size, &extents, unwrapped_key.as_deref().map(|k| k.as_slice()))
360            .await?;
361        drop(unwrapped_key);
362        let mapped = MappedVmo::new(vmo, size as usize).context("Failed to map layer VMO")?;
363        Ok(Arc::new(PagedLayerObject {
364            handle,
365            mapped,
366            sender: self.sender.clone(),
367            scope: self.scope.to_handle(),
368            key,
369            closed: AtomicBool::new(false),
370        }) as Arc<dyn LayerObject>)
371    }
372}
373
374#[cfg(test)]
375mod tests {
376    use super::*;
377    use futures::StreamExt;
378    use vmo_fifo::Receiver;
379
380    #[fuchsia::test]
381    fn test_mapped_vmo_io_error_signal() {
382        let vmo = zx::Vmo::create(4096).unwrap();
383        vmo.write(&[0x42u8; 4096], 0).unwrap();
384        let vmo_dup = vmo.duplicate_handle(zx::Rights::SAME_RIGHTS).unwrap();
385
386        let mapped = MappedVmo::new(vmo, 4096).unwrap();
387        assert_eq!(mapped.as_slice(), &[0x42u8; 4096]);
388        assert!(!mapped.has_io_error());
389
390        vmo_dup.signal(zx::Signals::empty(), zx::Signals::USER_0).unwrap();
391        assert!(mapped.has_io_error());
392    }
393
394    #[fuchsia::test]
395    async fn test_register_layer_large_extent() {
396        let (mapper_proxy, mut mapper_stream) =
397            fidl::endpoints::create_proxy_and_stream::<fblock::MapperMarker>();
398
399        let large_len = (0x3fff_ffffu64 + 5) * mapping::BLOCK_SIZE;
400        let start_dev = 10 * mapping::BLOCK_SIZE;
401
402        let client_task = async move {
403            let layer_pager = LayerPagerImpl::new(&mapper_proxy).await.expect("new failed");
404            let file_extents = [FileExtent::new(0, start_dev..(start_dev + large_len)).unwrap()];
405            let (key, _vmo) = layer_pager
406                .register_layer(large_len, &file_extents, None)
407                .await
408                .expect("register_layer failed");
409            assert_eq!(key, 1);
410        };
411
412        let server_task = async move {
413            let Some(Ok(fblock::MapperRequest::OpenSession {
414                session,
415                mapping_vmo,
416                responder,
417                ..
418            })) = mapper_stream.next().await
419            else {
420                panic!("Expected OpenSession");
421            };
422            responder.send(Ok(())).unwrap();
423
424            let mut session_stream = session.into_stream();
425            let Some(Ok(fblock::MapperSessionRequest::CreateVmo { responder, .. })) =
426                session_stream.next().await
427            else {
428                panic!("Expected CreateVmo");
429            };
430            let vmo = zx::Vmo::create(4096).unwrap();
431            responder.send(Ok((1, vmo))).unwrap();
432
433            fasync::unblock(move || {
434                let mut receiver =
435                    Receiver::<RawMappingCommand>::new(mapping_vmo, PENDING_COMMANDS_CAPACITY)
436                        .unwrap();
437                let cmd_raw = receiver.peek().expect("peek failed");
438                let cmd = MappingCommand::try_from(*cmd_raw).expect("try_from failed");
439
440                let (offset, extent_count) = match cmd {
441                    MappingCommand::Mappings {
442                        key,
443                        offset,
444                        stored_size,
445                        device_offset,
446                        metadata_count,
447                        extent_count,
448                        encrypted,
449                    } => {
450                        assert_eq!(key, 1);
451                        assert_eq!(stored_size, large_len);
452                        assert_eq!(device_offset, 0);
453                        assert_eq!(metadata_count, 0);
454                        assert_eq!(extent_count, 2);
455                        assert!(!encrypted);
456                        (offset, extent_count)
457                    }
458                    _ => panic!("Expected Mappings command"),
459                };
460
461                let buffer = cmd_raw.payload_slice(offset, extent_count * 8).to_vec();
462                let expected_extents =
463                    Extents::try_new([Extent::new(0..large_len, Some(start_dev))], 0).unwrap();
464                let mut expected_payload = Vec::new();
465                for val in Extents::encode_extents(&expected_extents) {
466                    expected_payload.extend_from_slice(&val.to_le_bytes());
467                }
468                assert_eq!(buffer, expected_payload);
469                cmd_raw.pop().expect("pop failed");
470            })
471            .await;
472        };
473
474        futures::join!(client_task, server_task);
475    }
476}
fxfs_platform_testing/fuchsia/layer_pager.rs

fxfs_platform_testing/fuchsia/
layer_pager.rs
