block_server/callback_interface/
block_service.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::callback_interface::{Interface, Request, SessionManager};
6use crate::{Operation, RequestId};
7use anyhow::{Error, ensure};
8use mapping::reader::{BlockService, MAX_READ_BUFFER_SIZE};
9use std::borrow::Borrow;
10use std::sync::{Arc, Weak};
11use storage_device::buffer::OwnedBuffer;
12use storage_device::buffer_allocator::{BufferAllocator, BufferSource};
13
14const DEFAULT_BUFFER_POOL_CAPACITY: usize = 64 * 1024;
15
16/// Default fallback implementation of [`BlockService`] for
17/// [`callback_interface::Interface`] backends (such as C++ drivers / virtio-block)
18/// that return the default implementation from `into_block_service`.
19pub struct DefaultCallbackBlockService<I: Interface + ?Sized> {
20    orchestrator: Weak<I::Orchestrator>,
21    allocator: Arc<BufferAllocator>,
22    block_size: u32,
23}
24
25impl<I: Interface + ?Sized> DefaultCallbackBlockService<I> {
26    pub fn new(orchestrator: &Arc<I::Orchestrator>) -> Self {
27        Self::new_with_pool_capacity(orchestrator, DEFAULT_BUFFER_POOL_CAPACITY)
28    }
29
30    pub fn new_with_pool_capacity(
31        orchestrator: &Arc<I::Orchestrator>,
32        pool_capacity: usize,
33    ) -> Self {
34        let sm: &SessionManager<I> = orchestrator.as_ref().borrow();
35        let block_size = sm.block_size();
36        let source = BufferSource::new(pool_capacity);
37        let allocator = Arc::new(BufferAllocator::new(
38            std::cmp::max(block_size as usize, zx::system_get_page_size() as usize),
39            source,
40        ));
41        Self { orchestrator: Arc::downgrade(orchestrator), allocator, block_size }
42    }
43
44    pub fn orchestrator(&self) -> Option<Arc<I::Orchestrator>> {
45        self.orchestrator.upgrade()
46    }
47
48    pub fn allocator(&self) -> &Arc<BufferAllocator> {
49        &self.allocator
50    }
51}
52
53impl<I: Interface + ?Sized> BlockService for DefaultCallbackBlockService<I> {
54    fn allocate_buffer(&self, max_len: usize) -> OwnedBuffer {
55        let max_len = std::cmp::min(
56            std::cmp::min(max_len, MAX_READ_BUFFER_SIZE),
57            self.allocator.buffer_source().size(),
58        );
59        self.allocator.allocate_buffer_sync_owned(max_len)
60    }
61
62    fn read_blocks(
63        &self,
64        device_offset: u64,
65        dest_buffer: OwnedBuffer,
66        on_complete: Box<dyn FnOnce(Result<OwnedBuffer, Error>) + Send>,
67    ) -> Result<(), Error> {
68        let orchestrator =
69            self.orchestrator.upgrade().ok_or_else(|| anyhow::anyhow!("Orchestrator dropped"))?;
70        let sm: &SessionManager<I> = orchestrator.as_ref().borrow();
71        let block_size = self.block_size as u64;
72        ensure!(
73            device_offset % block_size == 0
74                && dest_buffer.len() as u64 % block_size == 0
75                && dest_buffer.range().start as u64 % block_size == 0,
76            "Unaligned read request: device_offset={device_offset}, len={}, vmo_offset={}",
77            dest_buffer.len(),
78            dest_buffer.range().start
79        );
80        let device_block_offset = device_offset / block_size;
81        let block_count = (dest_buffer.len() as u64 / block_size) as u32;
82
83        let vmo_offset = dest_buffer.range().start as u64;
84        let vmo = self.allocator.buffer_source().vmo().clone();
85
86        sm.submit_internal_request(
87            |request_id: RequestId| Request {
88                request_id,
89                operation: Operation::Read {
90                    device_block_offset,
91                    block_count,
92                    _unused: 0,
93                    vmo_offset,
94                    options: block_protocol::ReadOptions::default(),
95                },
96                trace_flow_id: None,
97                vmo: Some(vmo),
98            },
99            Box::new(move |status| {
100                if status.is_ok() {
101                    on_complete(Ok(dest_buffer));
102                } else {
103                    on_complete(Err(anyhow::anyhow!("Block read failed: {:?}", status)));
104                }
105            }),
106        );
107        Ok(())
108    }
109}
110
111#[cfg(test)]
112mod tests {
113    use super::*;
114    use crate::callback_interface::{Session, SessionManager};
115    use crate::{BlockInfo, DeviceInfo};
116    use fidl_fuchsia_storage_block as fblock;
117    use fuchsia_sync::Mutex;
118    use mapping::Extents;
119    use mapping::reader::read_aligned_range;
120    use std::borrow::Cow;
121    use std::ops::ControlFlow;
122
123    const BLOCK_SIZE: u32 = 512;
124
125    use crate::testing::MockInterface;
126
127    #[test]
128    fn test_into_block_service_memoization() {
129        let (tx, _rx) = std::sync::mpsc::channel();
130        let interface = Arc::new(MockInterface::new(tx));
131        let session_manager = Arc::new(SessionManager::new(interface.clone(), BLOCK_SIZE));
132
133        let service1 = session_manager.into_block_service(&session_manager);
134        let service2 = session_manager.into_block_service(&session_manager);
135        assert!(Arc::ptr_eq(&service1, &service2));
136    }
137
138    struct CustomBlockService;
139    impl BlockService for CustomBlockService {
140        fn allocate_buffer(&self, _max_len: usize) -> OwnedBuffer {
141            unimplemented!()
142        }
143        fn read_blocks(
144            &self,
145            _device_offset: u64,
146            _dest_buffer: OwnedBuffer,
147            _on_complete: Box<dyn FnOnce(Result<OwnedBuffer, Error>) + Send>,
148        ) -> Result<(), Error> {
149            unimplemented!()
150        }
151    }
152
153    struct CustomInterface {
154        custom_service: Arc<CustomBlockService>,
155    }
156
157    impl Interface for CustomInterface {
158        type Orchestrator = SessionManager<Self>;
159
160        fn get_info(&self) -> Cow<'_, DeviceInfo> {
161            Cow::Owned(DeviceInfo::Block(BlockInfo {
162                block_count: 100,
163                max_transfer_blocks: None,
164                device_flags: fblock::DeviceFlag::empty(),
165            }))
166        }
167
168        fn spawn_session(&self, _session: Arc<Session<Self>>) {}
169
170        fn on_requests(&self, _requests: &[Request]) {}
171
172        fn into_block_service(
173            self: Arc<Self>,
174            _orchestrator: &Arc<Self::Orchestrator>,
175        ) -> Arc<dyn BlockService> {
176            self.custom_service.clone()
177        }
178    }
179
180    #[test]
181    fn test_custom_into_block_service_override() {
182        let custom_service = Arc::new(CustomBlockService);
183        let interface = Arc::new(CustomInterface { custom_service: custom_service.clone() });
184        let session_manager = Arc::new(SessionManager::new(interface, BLOCK_SIZE));
185        let service = session_manager.into_block_service(&session_manager);
186        assert!(Arc::ptr_eq(&service, &(custom_service as Arc<dyn BlockService>)));
187    }
188
189    #[test]
190    fn test_default_callback_block_service_weak_ref_no_cycle() {
191        let (tx, _rx) = std::sync::mpsc::channel();
192        let interface = Arc::new(MockInterface::new(tx));
193        let session_manager = Arc::new(SessionManager::new(interface, BLOCK_SIZE));
194        let service = DefaultCallbackBlockService::<MockInterface>::new(&session_manager);
195        assert!(service.orchestrator().is_some());
196
197        let weak_sm = Arc::downgrade(&session_manager);
198        drop(session_manager);
199        assert!(weak_sm.upgrade().is_none());
200        assert!(service.orchestrator().is_none());
201    }
202
203    #[fuchsia::test]
204    async fn test_default_callback_block_service_read_aligned_range() {
205        let (tx, rx) = std::sync::mpsc::channel();
206        let interface = Arc::new(MockInterface::new(tx));
207        let session_manager = Arc::new(SessionManager::new(interface.clone(), BLOCK_SIZE));
208        let service = session_manager.into_block_service(&session_manager);
209
210        let test_data = vec![0xaa_u8; 4096];
211        let encoded = (8u64 << 32) | 0u64;
212        let extents = Extents::from_encoded([encoded], 0).unwrap();
213
214        let (send, recv) = futures::channel::oneshot::channel();
215        let send = Mutex::new(Some(send));
216        let read_buf = Mutex::new(Vec::new());
217
218        let service_clone = service.clone();
219        read_aligned_range(&extents, 0..4096, &*service_clone, move |buffer_result| {
220            let buffer = buffer_result.unwrap();
221            let mut read_guard = read_buf.lock();
222            buffer.as_ref().append_to(&mut *read_guard);
223            if read_guard.len() == 4096 {
224                if let Some(s) = send.lock().take() {
225                    let _ = s.send(read_guard.clone());
226                }
227            }
228            ControlFlow::Continue(())
229        });
230
231        let req = rx.recv().unwrap();
232        if let Operation::Read { vmo_offset, .. } = &req.operation {
233            req.vmo.as_ref().unwrap().write(&test_data, *vmo_offset).unwrap();
234        } else {
235            panic!("Expected Operation::Read");
236        }
237        session_manager.complete_request(req.request_id, Ok(()));
238
239        let result = recv.await.unwrap();
240        assert_eq!(result, test_data);
241    }
242
243    #[fuchsia::test]
244    async fn test_default_callback_block_service_read_aligned_range_failure() {
245        let (tx, rx) = std::sync::mpsc::channel();
246        let interface = Arc::new(MockInterface::new(tx));
247        let session_manager = Arc::new(SessionManager::new(interface.clone(), BLOCK_SIZE));
248        let service = session_manager.into_block_service(&session_manager);
249
250        let encoded = (8u64 << 32) | 0u64;
251        let extents = Extents::from_encoded([encoded], 0).unwrap();
252
253        let (send, recv) = futures::channel::oneshot::channel();
254        let send = Mutex::new(Some(send));
255
256        let service_clone = service.clone();
257        read_aligned_range(&extents, 0..4096, &*service_clone, move |buffer_result| {
258            assert!(buffer_result.is_err());
259            if let Some(s) = send.lock().take() {
260                let _ = s.send(());
261            }
262            ControlFlow::Break(())
263        });
264
265        let req = rx.recv().unwrap();
266        session_manager.complete_request(req.request_id, Err(zx::Status::IO));
267
268        recv.await.unwrap();
269    }
270}
block_server/callback_interface/block_service.rs

block_server/callback_interface/
block_service.rs
