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 == zx::Status::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    struct MockInterface {
126        request_sender: std::sync::mpsc::Sender<Request>,
127    }
128
129    impl Interface for MockInterface {
130        type Orchestrator = SessionManager<Self>;
131
132        fn get_info(&self) -> Cow<'_, DeviceInfo> {
133            Cow::Owned(DeviceInfo::Block(BlockInfo { block_count: 1024, ..Default::default() }))
134        }
135
136        fn spawn_session(&self, _session: Arc<Session<Self>>) {}
137
138        fn on_requests(&self, requests: &[Request]) {
139            for request in requests {
140                self.request_sender.send(request.clone()).unwrap();
141            }
142        }
143    }
144
145    #[test]
146    fn test_into_block_service_memoization() {
147        let (tx, _rx) = std::sync::mpsc::channel();
148        let interface = Arc::new(MockInterface { request_sender: tx });
149        let session_manager = Arc::new(SessionManager::new(interface.clone(), BLOCK_SIZE));
150
151        let service1 = session_manager.into_block_service(&session_manager);
152        let service2 = session_manager.into_block_service(&session_manager);
153        assert!(Arc::ptr_eq(&service1, &service2));
154    }
155
156    struct CustomBlockService;
157    impl BlockService for CustomBlockService {
158        fn allocate_buffer(&self, _max_len: usize) -> OwnedBuffer {
159            unimplemented!()
160        }
161        fn read_blocks(
162            &self,
163            _device_offset: u64,
164            _dest_buffer: OwnedBuffer,
165            _on_complete: Box<dyn FnOnce(Result<OwnedBuffer, Error>) + Send>,
166        ) -> Result<(), Error> {
167            unimplemented!()
168        }
169    }
170
171    struct CustomInterface {
172        custom_service: Arc<CustomBlockService>,
173    }
174
175    impl Interface for CustomInterface {
176        type Orchestrator = SessionManager<Self>;
177
178        fn get_info(&self) -> Cow<'_, DeviceInfo> {
179            Cow::Owned(DeviceInfo::Block(BlockInfo {
180                block_count: 100,
181                max_transfer_blocks: None,
182                device_flags: fblock::DeviceFlag::empty(),
183            }))
184        }
185
186        fn spawn_session(&self, _session: Arc<Session<Self>>) {}
187
188        fn on_requests(&self, _requests: &[Request]) {}
189
190        fn into_block_service(
191            self: Arc<Self>,
192            _orchestrator: &Arc<Self::Orchestrator>,
193        ) -> Arc<dyn BlockService> {
194            self.custom_service.clone()
195        }
196    }
197
198    #[test]
199    fn test_custom_into_block_service_override() {
200        let custom_service = Arc::new(CustomBlockService);
201        let interface = Arc::new(CustomInterface { custom_service: custom_service.clone() });
202        let session_manager = Arc::new(SessionManager::new(interface, BLOCK_SIZE));
203        let service = session_manager.into_block_service(&session_manager);
204        assert!(Arc::ptr_eq(&service, &(custom_service as Arc<dyn BlockService>)));
205    }
206
207    #[test]
208    fn test_default_callback_block_service_weak_ref_no_cycle() {
209        let (tx, _rx) = std::sync::mpsc::channel();
210        let interface = Arc::new(MockInterface { request_sender: tx });
211        let session_manager = Arc::new(SessionManager::new(interface, BLOCK_SIZE));
212        let service = DefaultCallbackBlockService::<MockInterface>::new(&session_manager);
213        assert!(service.orchestrator().is_some());
214
215        let weak_sm = Arc::downgrade(&session_manager);
216        drop(session_manager);
217        assert!(weak_sm.upgrade().is_none());
218        assert!(service.orchestrator().is_none());
219    }
220
221    #[fuchsia::test]
222    async fn test_default_callback_block_service_read_aligned_range() {
223        let (tx, rx) = std::sync::mpsc::channel();
224        let interface = Arc::new(MockInterface { request_sender: tx });
225        let session_manager = Arc::new(SessionManager::new(interface.clone(), BLOCK_SIZE));
226        let service = session_manager.into_block_service(&session_manager);
227
228        let test_data = vec![0xaa_u8; 4096];
229        let encoded = (8u64 << 32) | 0u64;
230        let extents = Extents::from_encoded(&[encoded]).unwrap();
231
232        let (send, recv) = futures::channel::oneshot::channel();
233        let send = Mutex::new(Some(send));
234        let read_buf = Mutex::new(Vec::new());
235
236        let service_clone = service.clone();
237        read_aligned_range(&extents, 0..4096, &*service_clone, move |buffer_result| {
238            let buffer = buffer_result.unwrap();
239            let mut read_guard = read_buf.lock();
240            buffer.as_ref().append_to(&mut *read_guard);
241            if read_guard.len() == 4096 {
242                if let Some(s) = send.lock().take() {
243                    let _ = s.send(read_guard.clone());
244                }
245            }
246            ControlFlow::Continue(())
247        });
248
249        let req = rx.recv().unwrap();
250        if let Operation::Read { vmo_offset, .. } = &req.operation {
251            req.vmo.as_ref().unwrap().write(&test_data, *vmo_offset).unwrap();
252        } else {
253            panic!("Expected Operation::Read");
254        }
255        session_manager.complete_request(req.request_id, zx::Status::OK);
256
257        let result = recv.await.unwrap();
258        assert_eq!(result, test_data);
259    }
260
261    #[fuchsia::test]
262    async fn test_default_callback_block_service_read_aligned_range_failure() {
263        let (tx, rx) = std::sync::mpsc::channel();
264        let interface = Arc::new(MockInterface { request_sender: tx });
265        let session_manager = Arc::new(SessionManager::new(interface.clone(), BLOCK_SIZE));
266        let service = session_manager.into_block_service(&session_manager);
267
268        let encoded = (8u64 << 32) | 0u64;
269        let extents = Extents::from_encoded(&[encoded]).unwrap();
270
271        let (send, recv) = futures::channel::oneshot::channel();
272        let send = Mutex::new(Some(send));
273
274        let service_clone = service.clone();
275        read_aligned_range(&extents, 0..4096, &*service_clone, move |buffer_result| {
276            assert!(buffer_result.is_err());
277            if let Some(s) = send.lock().take() {
278                let _ = s.send(());
279            }
280            ControlFlow::Break(())
281        });
282
283        let req = rx.recv().unwrap();
284        session_manager.complete_request(req.request_id, zx::Status::IO);
285
286        recv.await.unwrap();
287    }
288}
block_server/callback_interface/block_service.rs

block_server/callback_interface/
block_service.rs
