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 = MAX_READ_BUFFER_SIZE * 16;
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    max_transfer_bytes: usize,
24}
25
26impl<I: Interface + ?Sized> DefaultCallbackBlockService<I> {
27    pub fn new(orchestrator: &Arc<I::Orchestrator>) -> Self {
28        Self::new_with_pool_capacity(orchestrator, DEFAULT_BUFFER_POOL_CAPACITY)
29    }
30
31    pub fn new_with_pool_capacity(
32        orchestrator: &Arc<I::Orchestrator>,
33        pool_capacity: usize,
34    ) -> Self {
35        let sm: &SessionManager<I> = orchestrator.as_ref().borrow();
36        let block_size = sm.block_size();
37        let max_transfer_bytes = sm
38            .interface
39            .get_info()
40            .max_transfer_blocks()
41            .map(|blocks| {
42                let bytes = (blocks.get() as usize).saturating_mul(block_size as usize);
43                std::cmp::max(
44                    bytes & !(mapping::BLOCK_SIZE as usize - 1),
45                    mapping::BLOCK_SIZE as usize,
46                )
47            })
48            .unwrap_or(MAX_READ_BUFFER_SIZE);
49        let source = BufferSource::new(pool_capacity);
50        let allocator = Arc::new(BufferAllocator::new(
51            std::cmp::max(block_size as usize, zx::system_get_page_size() as usize),
52            source,
53        ));
54        Self {
55            orchestrator: Arc::downgrade(orchestrator),
56            allocator,
57            block_size,
58            max_transfer_bytes,
59        }
60    }
61
62    pub fn orchestrator(&self) -> Option<Arc<I::Orchestrator>> {
63        self.orchestrator.upgrade()
64    }
65
66    pub fn allocator(&self) -> &Arc<BufferAllocator> {
67        &self.allocator
68    }
69}
70
71impl<I: Interface + ?Sized> BlockService for DefaultCallbackBlockService<I> {
72    fn allocate_buffer(&self, max_len: usize) -> OwnedBuffer {
73        let max_len = max_len
74            .min(MAX_READ_BUFFER_SIZE)
75            .min(self.max_transfer_bytes)
76            .min(self.allocator.buffer_source().size());
77        self.allocator.allocate_buffer_sync_owned(max_len)
78    }
79
80    fn read_blocks(
81        &self,
82        device_offset: u64,
83        dest_buffer: OwnedBuffer,
84        on_complete: Box<dyn FnOnce(Result<OwnedBuffer, Error>) + Send>,
85    ) -> Result<(), Error> {
86        let orchestrator =
87            self.orchestrator.upgrade().ok_or_else(|| anyhow::anyhow!("Orchestrator dropped"))?;
88        let sm: &SessionManager<I> = orchestrator.as_ref().borrow();
89        let block_size = self.block_size as u64;
90        ensure!(
91            device_offset % block_size == 0
92                && dest_buffer.len() as u64 % block_size == 0
93                && dest_buffer.range().start as u64 % block_size == 0,
94            "Unaligned read request: device_offset={device_offset}, len={}, vmo_offset={}",
95            dest_buffer.len(),
96            dest_buffer.range().start
97        );
98        let device_block_offset = device_offset / block_size;
99        let block_count = (dest_buffer.len() as u64 / block_size) as u32;
100
101        let vmo_offset = dest_buffer.range().start as u64;
102        let vmo = self.allocator.buffer_source().vmo().clone();
103
104        sm.submit_internal_request(
105            |request_id: RequestId| Request {
106                request_id,
107                operation: Operation::Read {
108                    device_block_offset,
109                    block_count,
110                    _unused: 0,
111                    vmo_offset,
112                    options: block_protocol::ReadOptions::default(),
113                },
114                trace_flow_id: None,
115                vmo: Some(vmo),
116            },
117            Box::new(move |status| {
118                if status.is_ok() {
119                    on_complete(Ok(dest_buffer));
120                } else {
121                    on_complete(Err(anyhow::anyhow!("Block read failed: {:?}", status)));
122                }
123            }),
124        );
125        Ok(())
126    }
127}
128
129#[cfg(test)]
130mod tests {
131    use super::*;
132    use crate::callback_interface::{Session, SessionManager};
133    use crate::{BlockInfo, DeviceInfo};
134    use fidl_fuchsia_storage_block as fblock;
135    use fuchsia_sync::Mutex;
136    use mapping::Extents;
137    use mapping::reader::read_aligned_range;
138    use std::borrow::Cow;
139    use std::ops::ControlFlow;
140
141    const BLOCK_SIZE: u32 = 512;
142
143    use crate::testing::MockInterface;
144
145    #[test]
146    fn test_into_block_service_memoization() {
147        let (tx, _rx) = std::sync::mpsc::channel();
148        let interface = Arc::new(MockInterface::new(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::new(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::new(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], 0).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, 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::new(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], 0).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, Err(zx::Status::IO));
285
286        recv.await.unwrap();
287    }
288
289    #[test]
290    fn test_default_callback_block_service_respects_max_transfer_blocks() {
291        struct MaxTransferInterface {
292            max_transfer_blocks: Option<std::num::NonZero<u32>>,
293        }
294        impl Interface for MaxTransferInterface {
295            type Orchestrator = SessionManager<Self>;
296            fn get_info(&self) -> Cow<'_, DeviceInfo> {
297                Cow::Owned(DeviceInfo::Block(BlockInfo {
298                    block_count: 1024,
299                    max_transfer_blocks: self.max_transfer_blocks,
300                    device_flags: fblock::DeviceFlag::empty(),
301                }))
302            }
303            fn spawn_session(&self, _session: Arc<Session<Self>>) {}
304            fn on_requests(&self, _requests: &[Request]) {}
305        }
306
307        // 16 blocks * 512 bytes = 8192 bytes max transfer size.
308        let interface =
309            Arc::new(MaxTransferInterface { max_transfer_blocks: std::num::NonZero::new(16) });
310        let session_manager = Arc::new(SessionManager::new(interface, BLOCK_SIZE));
311        let service = session_manager.into_block_service(&session_manager);
312
313        let buf = service.allocate_buffer(64 * 1024);
314        assert_eq!(buf.len(), 8192);
315    }
316}
block_server/callback_interface/block_service.rs

block_server/callback_interface/
block_service.rs
