fxfs_platform/fuchsia/fxblob/
mapping_server.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 crate::fuchsia::fxblob::directory::BlobDirectory;
6use crate::fuchsia::pager::PagerBacked;
7use anyhow::{Error, anyhow};
8use fuchsia_merkle::Hash;
9use futures::lock::Mutex;
10use mapping::{Extents, MappingCommand, RawMappingCommand};
11use std::sync::Arc;
12use std::sync::atomic::{AtomicU64, Ordering};
13use vmo_fifo::AsyncSender;
14
15// The `vmo-fifo` divides the VMO into two regions: a fixed-size command slots region, and a
16// dynamically allocated payload region where the actual extents are written.
17//
18// The following is the layout for a 512KB VMO with 256 capacity:
19// [ Headers (64B) | Command Slots: 256 * 24B = 6,144B | .. Padding to 8KB .. | Payload (504KB) ]
20// Note: Each command slot takes 24 bytes because `RawMappingCommand` has six 4-byte fields.
21//
22// 504KB / 8-bytes per extent = 64,512 maximum extents bounded by the payload block.
23const MAPPING_VMO_SIZE: u64 = 512 * 1024;
24
25// With a maximum capacity of 256 pending mapping commands, this allows for an average of ~252
26// extents per blob. In the worst case of maximum fragmentation (every 4KB block maps to one
27// extent), 64,512 extents can map up to ~252MB of blob data (or ~504MB if block size is 8KB).
28const PENDING_COMMANDS_CAPACITY: u32 = 256;
29
30/// This maintains state for mappings between the driver paging system and Fxfs.
31///
32/// Note: It is the responsibility of the client to track concurrent attempts to open files and
33/// broker them properly.
34pub struct BlobMappingServer {
35    blob_directory: Arc<BlobDirectory>,
36    sender: Mutex<AsyncSender<RawMappingCommand>>,
37    next_key: AtomicU64,
38}
39
40#[derive(Copy, Clone, Debug, PartialEq)]
41pub struct SessionState {
42    /// The unique identifier for the session.
43    pub key: u64,
44    /// The uncompressed byte size of the blob.
45    pub size: u64,
46}
47
48impl BlobMappingServer {
49    pub fn new(blob_directory: Arc<BlobDirectory>) -> Result<Self, Error> {
50        let vmo = zx::Vmo::create(MAPPING_VMO_SIZE)
51            .map_err(|s| anyhow!("Failed to create VMO: {}", s))?;
52
53        // We allow up to 256 pending commands. This is a guess - we may have to adjust this.
54        let sender = AsyncSender::<RawMappingCommand>::new(
55            vmo,
56            8,                         // alignment
57            PENDING_COMMANDS_CAPACITY, // capacity of commands
58        )
59        .map_err(|s| anyhow!("Failed to create Sender: {}", s))?;
60
61        Ok(Self { blob_directory, sender: Mutex::new(sender), next_key: AtomicU64::new(1) })
62    }
63
64    /// Returns a duplicated handle to the mapping VMO.
65    pub async fn clone_mapping(&self) -> Result<zx::Vmo, zx::Status> {
66        let sender = self.sender.lock().await;
67        sender.vmo().duplicate_handle(zx::Rights::SAME_RIGHTS)
68    }
69
70    /// Retrieves the extent mappings for the blob and registers a new mapping session. Returns a
71    // `SessionState` containing the node size and the uniquely generated session key.
72    pub async fn create_session(&self, hash: Hash) -> Result<SessionState, Error> {
73        let node = self
74            .blob_directory
75            .open_blob(&hash.into())
76            .await?
77            .ok_or_else(|| anyhow!("Blob not found"))?;
78
79        let extents = node.get_mapping_extents().await?;
80        let size = node.as_ref().byte_size();
81        let blob_count = extents.data.len() as u32;
82        let metadata_count = extents.merkle.len() as u32;
83
84        let key = self.next_key.fetch_add(1, Ordering::Relaxed);
85        let allocation_size = (blob_count + metadata_count) as usize * std::mem::size_of::<u64>();
86
87        if allocation_size > 0 {
88            let mut sender = self.sender.lock().await;
89
90            let mut payload = sender.reserve_payload(allocation_size).await?;
91            let offset_in_vmo = payload.offset();
92
93            for (mut chunk, val_res) in payload.data().chunks_mut(std::mem::size_of::<u64>()).zip(
94                Extents::encode_extents_iter(&extents.data)
95                    .chain(Extents::encode_extents_iter(&extents.merkle)),
96            ) {
97                chunk.write(val_res.to_le());
98            }
99
100            let command = MappingCommand::Mappings {
101                key,
102                offset: offset_in_vmo as u32,
103                metadata_count,
104                blob_count,
105            };
106
107            payload.commit(command.into()).await?;
108        }
109
110        Ok(SessionState { key, size })
111    }
112
113    /// Unregisters the blob mapping and signals the block driver to terminate tracking.
114    pub async fn close_session(&self, session_key: u64) -> Result<(), Error> {
115        let mut sender = self.sender.lock().await;
116        sender.push(MappingCommand::CloseBlob { key: session_key }.into()).await?;
117        Ok(())
118    }
119}
120
121#[cfg(test)]
122mod tests {
123    use super::*;
124    use crate::fuchsia::fxblob::testing::{BlobFixture, new_blob_fixture};
125    use delivery_blob::CompressionMode;
126    use fuchsia_async as fasync;
127    use vmo_fifo::Receiver;
128
129    #[fuchsia::test]
130    async fn test_blob_mapping_server() {
131        let fixture = new_blob_fixture().await;
132        // Test with a large amount of non-compressible data to generate many extents
133        let data = vec![42; 300_000];
134        let hash = fixture.write_blob(&data, CompressionMode::Never).await;
135
136        let blob_dir = fixture
137            .volume()
138            .root()
139            .clone()
140            .as_node()
141            .into_any()
142            .downcast::<BlobDirectory>()
143            .expect("Failed to downcast root directory to BlobDirectory");
144
145        let node = blob_dir
146            .open_blob(&hash.into())
147            .await
148            .expect("Failed to open blob in Fxfs")
149            .expect("open_blob returned None instead of node");
150        let extents = node.get_mapping_extents().await.expect("Failed to retrieve extents");
151        let data_extents = extents.data;
152        let merkle_extents = extents.merkle;
153
154        // Un-dropped nodes pin the Blob as actively opened. The unmount routine in fixture.close()
155        // will wait forever for this blob to be fully closed, causing a test timeout.
156        drop(node);
157
158        let server = BlobMappingServer::new(blob_dir).expect("Failed to create BlobMappingServer");
159        let client_mapping = server.clone_mapping().await.expect("Failed to clone VMO mapping");
160
161        let receiver_task = fasync::unblock(move || {
162            let mut receiver = Receiver::<RawMappingCommand>::new(client_mapping, 256)
163                .expect("Failed to create the Receiver wrapper");
164
165            let pop_cmd = |receiver: &mut Receiver<RawMappingCommand>| loop {
166                match receiver.pop_reserve() {
167                    Ok(cmd) => break cmd,
168                    Err(zx::Status::SHOULD_WAIT) => {
169                        std::thread::sleep(std::time::Duration::from_millis(10))
170                    }
171                    Err(e) => panic!("pop_reserve failed: {:?}", e),
172                }
173            };
174
175            // First Open Command
176            let cmd1_raw = pop_cmd(&mut receiver);
177            let cmd1 =
178                MappingCommand::try_from(cmd1_raw).expect("Failed to convert raw mapping command");
179
180            let (cmd1_offset, cmd1_blob_count, cmd1_metadata_count) = match cmd1 {
181                MappingCommand::Mappings { key, offset, metadata_count, blob_count } => {
182                    assert_eq!(key, 1);
183                    assert_eq!(blob_count, data_extents.len() as u32);
184                    assert_eq!(metadata_count, merkle_extents.len() as u32);
185                    (offset, blob_count, metadata_count)
186                }
187                _ => panic!("Expected Mappings command"),
188            };
189
190            // Verify payload
191            let total_extents = cmd1_blob_count + cmd1_metadata_count;
192            let buffer = receiver.payload_slice(cmd1_offset, total_extents * 8).to_vec();
193
194            let mut expected_payload = Vec::new();
195            for val in Extents::encode_extents_iter(&data_extents)
196                .chain(Extents::encode_extents_iter(&merkle_extents))
197            {
198                expected_payload.extend_from_slice(&val.to_le_bytes());
199            }
200            assert_eq!(buffer, expected_payload);
201            receiver.pop_commit().expect("Failed pop_commit");
202
203            let cmd2_raw = pop_cmd(&mut receiver);
204            let cmd2 =
205                MappingCommand::try_from(cmd2_raw).expect("Failed to convert raw mapping command");
206            match cmd2 {
207                MappingCommand::CloseBlob { key } => assert_eq!(key, 1),
208                _ => panic!("Expected CloseBlob command"),
209            };
210            receiver.pop_commit().expect("Failed pop_commit");
211        });
212
213        let server_task = async move {
214            let SessionState { key, .. } =
215                server.create_session(hash).await.expect("create_session failed");
216            assert_eq!(key, 1);
217
218            server.close_session(key).await.expect("close_session failed on existing key");
219
220            std::mem::drop(server);
221        };
222
223        futures::join!(receiver_task, server_task);
224
225        fixture.close().await;
226    }
227
228    #[fuchsia::test]
229    async fn test_missing_blob() {
230        let fixture = new_blob_fixture().await;
231        let blob_dir = fixture
232            .volume()
233            .root()
234            .clone()
235            .as_node()
236            .into_any()
237            .downcast::<BlobDirectory>()
238            .expect("Failed to downcast");
239
240        let server = BlobMappingServer::new(blob_dir).expect("Failed to create server");
241        let hash = Hash::from([1u8; 32]);
242        server
243            .create_session(hash)
244            .await
245            .expect_err("create_session should fail with blob that doesn't exist");
246
247        std::mem::drop(server);
248        fixture.close().await;
249    }
250
251    #[fuchsia::test]
252    async fn test_invalid_key() {
253        let fixture = new_blob_fixture().await;
254        let blob_dir = fixture
255            .volume()
256            .root()
257            .clone()
258            .as_node()
259            .into_any()
260            .downcast::<BlobDirectory>()
261            .expect("Failed to downcast");
262
263        let server = BlobMappingServer::new(blob_dir).expect("Failed to create server");
264        server.close_session(42).await.expect("close_session should return Ok with invalid key");
265        std::mem::drop(server);
266
267        fixture.close().await;
268    }
269}
fxfs_platform/fuchsia/fxblob/mapping_server.rs

fxfs_platform/fuchsia/fxblob/
mapping_server.rs
