block_server/
callback_interface.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::{
6    ActiveRequests, DecodedRequest, DeviceInfo, HandleRequestResult, IntoOrchestrator, OffsetMap,
7    Operation, RequestId, SessionHelper, TraceFlowId, WriteFlags,
8};
9use anyhow::Error;
10use block_protocol::{BlockFifoRequest, BlockFifoResponse};
11use fidl_fuchsia_storage_block as fblock;
12use fuchsia_sync::{Condvar, Mutex};
13use futures::TryStreamExt as _;
14use futures::stream::{AbortHandle, Abortable};
15use std::borrow::{Borrow, Cow};
16use std::collections::{HashMap, VecDeque};
17use std::mem::MaybeUninit;
18use std::sync::{Arc, Weak};
19
20/// An in-flight request.
21#[derive(Clone, Debug)]
22pub struct Request {
23    /// The ID that a request is associated with, for later completion in
24    /// [`SessionManager::complete_request`].  Note that this is not necessarily a unique
25    /// identifier, and multiple Requests may have the same request_id (e.g. due to request
26    /// splitting).  The library internally reference-counts requests which use this ID.
27    pub request_id: RequestId,
28    pub operation: Operation,
29    pub trace_flow_id: TraceFlowId,
30    /// `vmo` is always Some for Operation::Read or Operation::Write, and None otherwise.
31    pub vmo: Option<Arc<zx::Vmo>>,
32}
33
34pub trait Interface: Send + Sync + Unpin + 'static {
35    type Orchestrator: Borrow<SessionManager<Self>> + Send + Sync;
36
37    /// Called to get block/partition information.
38    fn get_info(&self) -> Cow<'_, DeviceInfo>;
39
40    /// Start running a new session.  The interface must ensure that [`Session::run`] is called
41    /// on a different thread.
42    fn spawn_session(&self, session: Arc<Session<Self>>);
43
44    /// Starts a batch of requests.  The implementation may block if there are too many in-flight
45    /// requests, providing pushback.  The interface is responsible for eventually calling
46    /// [`SessionManager::complete_request`] for each request (even during server shutdown).
47    fn on_requests(&self, requests: &[Request]);
48}
49
50struct SessionManagerInner<I: Interface + ?Sized> {
51    open_sessions: HashMap<usize, Weak<Session<I>>>,
52}
53
54// The signals used on the session's FIFO.
55
56/// Signalled on the session's FIFO to wake up the FIFO loop.
57const FIFO_WAKE_SIGNAL: zx::Signals = zx::Signals::USER_0;
58
59/// Signalled on the session's FIFO to terminate the FIFO loop.
60const SHUTDOWN_SIGNAL: zx::Signals = zx::Signals::USER_1;
61
62pub struct SessionManager<I: Interface + ?Sized> {
63    interface: Arc<I>,
64    // These represent active *client* requests, which correspond to one or more in-flight requests.
65    active_requests: ActiveRequests<Arc<Session<I>>>,
66    inflight_requests: Mutex<usize>,
67    no_inflight_requests_condvar: Condvar,
68    inner: Mutex<SessionManagerInner<I>>,
69    no_open_sessions_condvar: Condvar,
70}
71
72impl<I: Interface + ?Sized> super::SessionManager for SessionManager<I> {
73    const SUPPORTS_DECOMPRESSION: bool = false;
74
75    type Orchestrator = I::Orchestrator;
76    type Session = Arc<Session<I>>;
77
78    fn session_eq(a: &Arc<Session<I>>, b: &Arc<Session<I>>) -> bool {
79        Arc::ptr_eq(a, b)
80    }
81
82    async fn on_attach_vmo(
83        _orchestrator: Arc<Self::Orchestrator>,
84        _vmo: &Arc<zx::Vmo>,
85    ) -> Result<(), zx::Status> {
86        Ok(())
87    }
88
89    async fn open_session(
90        orchestrator: Arc<Self::Orchestrator>,
91        mut stream: fblock::SessionRequestStream,
92        offset_map: OffsetMap,
93        block_size: u32,
94    ) -> Result<(), Error> {
95        let (helper, fifo) = SessionHelper::new(orchestrator.clone(), offset_map, block_size)?;
96        let (abort_handle, registration) = AbortHandle::new_pair();
97        let session = Arc::new(Session {
98            helper,
99            fifo,
100            queue: Mutex::default(),
101            abort_handle,
102            close_callback: Mutex::new(None),
103        });
104        let sm = orchestrator.as_ref().borrow();
105        sm.inner
106            .lock()
107            .open_sessions
108            .insert(Arc::as_ptr(&session) as usize, Arc::downgrade(&session));
109
110        sm.interface.spawn_session(session.clone());
111
112        let result = Abortable::new(
113            async {
114                while let Some(request) = stream.try_next().await? {
115                    match session.helper.handle_request(request).await? {
116                        HandleRequestResult::Ok => {}
117                        HandleRequestResult::Closed(callback) => {
118                            *session.close_callback.lock() = Some(callback);
119                            break;
120                        }
121                    }
122                }
123                Ok(())
124            },
125            registration,
126        )
127        .await
128        .unwrap_or_else(|e| Err(e.into()));
129
130        let _ = session.fifo.signal(zx::Signals::empty(), SHUTDOWN_SIGNAL);
131
132        result
133    }
134
135    fn get_info(&self) -> Cow<'_, super::DeviceInfo> {
136        self.interface.get_info()
137    }
138
139    fn active_requests(&self) -> &ActiveRequests<Arc<Session<I>>> {
140        &self.active_requests
141    }
142}
143
144impl<I: Interface + ?Sized> SessionManager<I> {
145    pub fn new(interface: Arc<I>) -> Self {
146        Self {
147            interface,
148            active_requests: ActiveRequests::default(),
149            inflight_requests: Mutex::new(0),
150            no_inflight_requests_condvar: Condvar::new(),
151            inner: Mutex::new(SessionManagerInner { open_sessions: HashMap::new() }),
152            no_open_sessions_condvar: Condvar::new(),
153        }
154    }
155
156    /// Reports the given task as complete with a given status.
157    pub fn complete_request(&self, request_id: RequestId, status: zx::Status) {
158        let notify = {
159            let mut inflight_requests = self.inflight_requests.lock();
160            *inflight_requests -= 1;
161            *inflight_requests == 0
162        };
163        self.complete_unsubmitted_request(request_id, status);
164        if notify {
165            self.no_inflight_requests_condvar.notify_all();
166        }
167    }
168
169    fn submit_requests(&self, requests: &[Request]) {
170        *self.inflight_requests.lock() += requests.len();
171        self.interface.on_requests(requests);
172    }
173
174    /// Waits for there to be no requests in-flight.
175    ///
176    /// NOTE: To void TOCTOUs, this must be called on the same thread which calls
177    /// [`Self::submit_requests`].
178    fn wait_for_no_inflight_requests(&self) {
179        let mut guard = self.inflight_requests.lock();
180        self.no_inflight_requests_condvar.wait_while(&mut guard, |count| *count > 0);
181    }
182
183    /// Called instead of `[Self::complete_request]` when a request is completed before it was
184    /// actually submitted.
185    fn complete_unsubmitted_request(&self, request_id: RequestId, status: zx::Status) {
186        if let Some((session, response)) =
187            self.active_requests.complete_and_take_response(request_id, status)
188        {
189            session.send_response(response);
190        }
191    }
192
193    /// Terminates the session manager.  Blocks until all sessions have terminated.
194    pub fn terminate(&self) {
195        {
196            // We must drop references to sessions whilst we're not holding the lock for
197            // `open_sessions` because `Session::drop` needs to take that same lock.
198            #[allow(clippy::collection_is_never_read)]
199            let mut terminated_sessions = Vec::new();
200            for (_, session) in &self.inner.lock().open_sessions {
201                if let Some(session) = session.upgrade() {
202                    session.terminate_async();
203                    terminated_sessions.push(session);
204                }
205            }
206        }
207        let mut guard = self.inner.lock();
208        self.no_open_sessions_condvar.wait_while(&mut guard, |s| !s.open_sessions.is_empty());
209    }
210}
211
212pub struct Session<I: Interface + ?Sized> {
213    helper: SessionHelper<SessionManager<I>>,
214    fifo: zx::Fifo<BlockFifoRequest, BlockFifoResponse>,
215    queue: Mutex<SessionQueue>,
216    abort_handle: AbortHandle,
217    close_callback: Mutex<Option<Box<dyn FnOnce() + Send + 'static>>>,
218}
219
220#[derive(Default)]
221struct SessionQueue {
222    responses: VecDeque<BlockFifoResponse>,
223}
224
225pub const MAX_REQUESTS: usize = super::FIFO_MAX_REQUESTS;
226
227struct DecodedRequests {
228    requests: [MaybeUninit<Request>; MAX_REQUESTS],
229    count: usize,
230}
231
232impl Default for DecodedRequests {
233    fn default() -> Self {
234        Self { requests: unsafe { MaybeUninit::uninit().assume_init() }, count: 0 }
235    }
236}
237
238impl DecodedRequests {
239    fn push(&mut self, request: Request) {
240        assert!(self.count < MAX_REQUESTS);
241        // To ensure drop runs, we must initialize each element at most once.
242        // As long as we only go via this function, that's satisfied.
243        self.requests[self.count].write(request);
244        self.count += 1;
245    }
246
247    fn is_empty(&self) -> bool {
248        self.count == 0
249    }
250
251    fn is_full(&self) -> bool {
252        self.count == MAX_REQUESTS
253    }
254
255    fn clear(&mut self) {
256        for i in 0..self.count {
257            // SAFETY: We initialized `count` elements via [`Self::push`].
258            unsafe { self.requests[i].assume_init_drop() };
259        }
260        self.count = 0;
261    }
262}
263
264impl Drop for DecodedRequests {
265    fn drop(&mut self) {
266        self.clear();
267    }
268}
269
270impl std::ops::Deref for DecodedRequests {
271    type Target = [Request];
272
273    fn deref(&self) -> &Self::Target {
274        // SAFETY: We wrote the request in [`Self::push`].
275        unsafe { std::slice::from_raw_parts(self.requests[0].as_ptr(), self.count) }
276    }
277}
278
279impl std::ops::DerefMut for DecodedRequests {
280    fn deref_mut(&mut self) -> &mut Self::Target {
281        // SAFETY: We wrote the request in [`Self::push`].
282        unsafe { std::slice::from_raw_parts_mut(self.requests[0].as_mut_ptr(), self.count) }
283    }
284}
285
286impl<I: Interface + ?Sized> Session<I> {
287    /// Begins processing the session's FIFO.  Blocks until completion (either because of
288    /// termination, or due to unrecoverable error).
289    pub fn run(self: &Arc<Self>) {
290        self.fifo_loop();
291        self.abort_handle.abort();
292        // NB: We cannot call [`drop_active_requests`] here, because requests which have already
293        // been submitted are no longer in control by this thread, and will be later completed by
294        // another thread.  If we dropped them here, then they would be completed twice.
295        self.helper.close_active_groups(|s| Arc::ptr_eq(s, self));
296    }
297
298    fn fifo_loop(self: &Arc<Self>) {
299        let mut requests = [MaybeUninit::uninit(); MAX_REQUESTS];
300
301        loop {
302            // Send queued responses.
303            let is_queue_empty = {
304                let mut queue = self.queue.lock();
305                while !queue.responses.is_empty() {
306                    let (front, _) = queue.responses.as_slices();
307                    match self.fifo.write(front) {
308                        Ok(count) => {
309                            let full = count < front.len();
310                            queue.responses.drain(..count);
311                            if full {
312                                break;
313                            }
314                        }
315                        Err(zx::Status::SHOULD_WAIT) => break,
316                        Err(_) => return,
317                    }
318                }
319                queue.responses.is_empty()
320            };
321
322            // Process pending reads.
323            match self.fifo.read_uninit(&mut requests) {
324                Ok(valid_requests) => self.handle_requests(valid_requests.iter_mut()),
325                Err(zx::Status::SHOULD_WAIT) => {
326                    let mut signals =
327                        zx::Signals::OBJECT_READABLE | SHUTDOWN_SIGNAL | FIFO_WAKE_SIGNAL;
328                    if !is_queue_empty {
329                        signals |= zx::Signals::OBJECT_WRITABLE;
330                    }
331                    let Ok(signals) =
332                        self.fifo.wait_one(signals, zx::MonotonicInstant::INFINITE).to_result()
333                    else {
334                        return;
335                    };
336                    if signals.contains(SHUTDOWN_SIGNAL) {
337                        return;
338                    }
339                    // Clear FIFO_WAKE_SIGNAL if it's set.
340                    if signals.contains(FIFO_WAKE_SIGNAL) {
341                        let _ = self.fifo.signal(FIFO_WAKE_SIGNAL, zx::Signals::empty());
342                    }
343                }
344                Err(_) => return,
345            }
346        }
347    }
348
349    /// Synchronously performs a device flush.
350    fn pre_flush(self: &Arc<Self>, request_id: RequestId) -> Result<(), zx::Status> {
351        let trace_flow_id = {
352            let mut request = self.helper.session_manager().active_requests.request(request_id);
353            if let Some(id) = request.trace_flow_id {
354                fuchsia_trace::async_instant!(
355                    fuchsia_trace::Id::from(id.get()),
356                    c"storage",
357                    c"block_server::SimulatedBarrier",
358                    "request_id" => request_id.0
359                );
360            }
361            request.count += 1;
362            request.trace_flow_id
363        };
364        self.helper.session_manager().submit_requests(&[Request {
365            request_id,
366            operation: Operation::Flush,
367            trace_flow_id,
368            vmo: None,
369        }]);
370        self.helper.session_manager().wait_for_no_inflight_requests();
371        let status = self.helper.session_manager().active_requests.request(request_id).status;
372        match status {
373            zx::Status::OK => Ok(()),
374            status => {
375                // Respond for the unsubmitted request too.
376                self.helper.session_manager().complete_unsubmitted_request(request_id, status);
377                Err(status)
378            }
379        }
380    }
381
382    /// Synchronously completes `decoded_requests`, and inserts a post-flush into `decoded_requests`
383    /// to be submitted later.
384    fn post_flush(self: &Arc<Self>, request_id: RequestId, decoded_requests: &mut DecodedRequests) {
385        if !decoded_requests.is_empty() {
386            self.helper.session_manager().submit_requests(decoded_requests);
387            decoded_requests.clear();
388        }
389        self.helper.session_manager().wait_for_no_inflight_requests();
390        let request = self.helper.session_manager().active_requests.request(request_id);
391        match request.status {
392            zx::Status::OK => decoded_requests.push(Request {
393                request_id,
394                operation: Operation::Flush,
395                trace_flow_id: request.trace_flow_id,
396                vmo: None,
397            }),
398            status => {
399                drop(request);
400                self.helper.session_manager().complete_unsubmitted_request(request_id, status)
401            }
402        }
403    }
404
405    fn handle_requests<'a>(
406        self: &Arc<Self>,
407        requests: impl Iterator<Item = &'a mut BlockFifoRequest>,
408    ) {
409        let manager = &self.helper.session_manager();
410        let mut decoded_requests = DecodedRequests::default();
411
412        for request in requests {
413            match self.helper.decode_fifo_request(self.clone(), request) {
414                Ok(DecodedRequest { operation: Operation::CloseVmo, request_id, .. }) => {
415                    manager.complete_unsubmitted_request(request_id, zx::Status::OK);
416                }
417                Ok(mut request) => {
418                    let request_id = request.request_id;
419
420                    // Strip the PRE_BARRIER flag if we don't support it, and simulate the barrier
421                    // with a pre-flush.
422                    if !manager
423                        .interface
424                        .get_info()
425                        .device_flags()
426                        .contains(fblock::DeviceFlag::BARRIER_SUPPORT)
427                        && request.operation.take_write_flag(WriteFlags::PRE_BARRIER)
428                        && self.pre_flush(request_id).is_err()
429                    {
430                        continue;
431                    }
432                    // Strip the FORCE_ACCESS flag if we don't support it, and simulate the FUA with
433                    // a post-flush.
434                    let simulate_fua = !manager
435                        .interface
436                        .get_info()
437                        .device_flags()
438                        .contains(fblock::DeviceFlag::FUA_SUPPORT)
439                        && request.operation.take_write_flag(WriteFlags::FORCE_ACCESS);
440
441                    if simulate_fua {
442                        // Account for the additional request we need at the end.
443                        manager.active_requests.request(request_id).count += 1;
444                    }
445
446                    loop {
447                        let result = self
448                            .helper
449                            .map_request(request, &mut manager.active_requests.request(request_id));
450                        match result {
451                            Ok((
452                                DecodedRequest { request_id, operation, vmo, trace_flow_id },
453                                remainder,
454                            )) => {
455                                decoded_requests.push(Request {
456                                    request_id,
457                                    operation,
458                                    trace_flow_id,
459                                    vmo,
460                                });
461
462                                if decoded_requests.is_full() {
463                                    manager.submit_requests(&*decoded_requests);
464                                    decoded_requests.clear();
465                                }
466
467                                if let Some(r) = remainder {
468                                    request = r;
469                                } else {
470                                    break;
471                                }
472                            }
473                            Err(status) => {
474                                manager.complete_unsubmitted_request(request_id, status);
475                                break;
476                            }
477                        }
478                    }
479
480                    if simulate_fua {
481                        self.post_flush(request_id, &mut decoded_requests);
482                    }
483                }
484                Err(None) => {}
485                Err(Some(response)) => self.send_response(response),
486            }
487        }
488
489        if !decoded_requests.is_empty() {
490            manager.submit_requests(&decoded_requests);
491        }
492    }
493
494    fn send_response(&self, response: BlockFifoResponse) {
495        let mut queue = self.queue.lock();
496        if queue.responses.is_empty() {
497            match self.fifo.write_one(&response) {
498                Ok(()) => {
499                    return;
500                }
501                Err(_) => {
502                    // Wake the FIFO loop up so that we can send the response later.
503                    let _ = self.fifo.signal(zx::Signals::empty(), FIFO_WAKE_SIGNAL);
504                }
505            }
506        }
507        queue.responses.push_back(response);
508    }
509
510    /// Asynchronously request to terminate the Session.  The session's thread will eventually stop
511    /// running.
512    pub fn terminate_async(&self) {
513        let _ = self.fifo.signal(zx::Signals::empty(), SHUTDOWN_SIGNAL);
514        self.abort_handle.abort();
515    }
516}
517
518impl<I: Interface + ?Sized> Drop for Session<I> {
519    fn drop(&mut self) {
520        let callback = std::mem::take(&mut *self.close_callback.lock());
521        if let Some(callback) = callback {
522            callback();
523        }
524        let notify = {
525            let mut inner = self.helper.session_manager().inner.lock();
526            inner.open_sessions.remove(&(self as *const _ as usize));
527            inner.open_sessions.is_empty()
528        };
529        if notify {
530            self.helper.session_manager().no_open_sessions_condvar.notify_all();
531        }
532    }
533}
534
535impl<I: Interface + ?Sized> Drop for SessionManager<I> {
536    fn drop(&mut self) {
537        self.terminate();
538    }
539}
540
541impl<I: Interface<Orchestrator = SessionManager<I>>> IntoOrchestrator for Arc<SessionManager<I>> {
542    type SM = SessionManager<I>;
543
544    fn into_orchestrator(self) -> Arc<I::Orchestrator> {
545        self
546    }
547}
548
549#[cfg(test)]
550mod tests {
551    use super::*;
552    use crate::BlockInfo;
553    use block_protocol::{BlockFifoCommand, BlockFifoRequest, BlockFifoResponse};
554    use fidl::endpoints::create_proxy_and_stream;
555    use fidl_fuchsia_storage_block as fblock;
556    use fuchsia_async as fasync;
557
558    struct MockInterface {
559        request_sender: std::sync::mpsc::Sender<Request>,
560    }
561
562    impl Interface for MockInterface {
563        type Orchestrator = SessionManager<Self>;
564
565        fn get_info(&self) -> Cow<'_, DeviceInfo> {
566            Cow::Owned(DeviceInfo::Block(BlockInfo { block_count: 1024, ..Default::default() }))
567        }
568
569        fn spawn_session(&self, session: Arc<Session<Self>>) {
570            std::thread::spawn(move || {
571                session.run();
572            });
573        }
574
575        fn on_requests(&self, requests: &[Request]) {
576            for request in requests {
577                self.request_sender.send(request.clone()).unwrap();
578            }
579        }
580    }
581
582    #[fasync::run_singlethreaded(test)]
583    async fn test_basic_request() {
584        let (tx, rx) = std::sync::mpsc::channel();
585        let interface = Arc::new(MockInterface { request_sender: tx });
586        let session_manager = Arc::new(SessionManager::new(interface.clone()));
587
588        let sm_clone = session_manager.clone();
589        let (proxy, stream) = create_proxy_and_stream::<fblock::BlockMarker>();
590        let _server_task = fasync::Task::spawn(async move {
591            let server = crate::BlockServer::new(512, sm_clone);
592            server.handle_requests(stream).await.unwrap();
593        })
594        .detach();
595
596        let (session_proxy, session_server_end) =
597            fidl::endpoints::create_proxy::<fblock::SessionMarker>();
598        proxy.open_session(session_server_end).unwrap();
599
600        let fifo_handle = session_proxy.get_fifo().await.unwrap().unwrap();
601        let fifo: zx::Fifo<BlockFifoResponse, BlockFifoRequest> = zx::Fifo::from(fifo_handle);
602
603        let vmo = zx::Vmo::create(8192).unwrap();
604        let vmo_id = session_proxy
605            .attach_vmo(vmo.duplicate_handle(zx::Rights::SAME_RIGHTS).unwrap())
606            .await
607            .unwrap()
608            .unwrap();
609
610        let req = BlockFifoRequest {
611            command: BlockFifoCommand {
612                opcode: fblock::BlockOpcode::Read.into_primitive(),
613                ..Default::default()
614            },
615            reqid: 123,
616            group: 0,
617            vmoid: vmo_id.id,
618            length: 1,
619            vmo_offset: 0,
620            dev_offset: 0,
621            trace_flow_id: 0,
622            ..Default::default()
623        };
624        fifo.write(&[req]).unwrap();
625
626        let r = rx.recv().unwrap();
627        assert_eq!(r.request_id.0, 0);
628        assert!(matches!(r.operation, Operation::Read { .. }));
629
630        session_manager.complete_request(r.request_id, zx::Status::OK);
631
632        let signals =
633            fifo.wait_one(zx::Signals::FIFO_READABLE, zx::MonotonicInstant::INFINITE).unwrap();
634        assert!(signals.contains(zx::Signals::FIFO_READABLE));
635
636        let mut resp = [BlockFifoResponse::default()];
637        fifo.read(&mut resp).unwrap();
638        assert_eq!(resp[0].reqid, 123);
639        assert_eq!(resp[0].status, zx::sys::ZX_OK);
640
641        std::mem::drop(proxy);
642    }
643    #[fasync::run_singlethreaded(test)]
644    async fn test_write_request() {
645        let (tx, rx) = std::sync::mpsc::channel();
646        let interface = Arc::new(MockInterface { request_sender: tx });
647        let session_manager = Arc::new(SessionManager::new(interface.clone()));
648
649        let sm_clone = session_manager.clone();
650        let (proxy, stream) = create_proxy_and_stream::<fblock::BlockMarker>();
651        let _server_task = fasync::Task::spawn(async move {
652            let server = crate::BlockServer::new(512, sm_clone);
653            server.handle_requests(stream).await.unwrap();
654        })
655        .detach();
656
657        let (session_proxy, session_server_end) =
658            fidl::endpoints::create_proxy::<fblock::SessionMarker>();
659        proxy.open_session(session_server_end).unwrap();
660
661        let fifo_handle = session_proxy.get_fifo().await.unwrap().unwrap();
662        let fifo: zx::Fifo<BlockFifoResponse, BlockFifoRequest> = zx::Fifo::from(fifo_handle);
663
664        let vmo = zx::Vmo::create(8192).unwrap();
665        let vmo_id = session_proxy
666            .attach_vmo(vmo.duplicate_handle(zx::Rights::SAME_RIGHTS).unwrap())
667            .await
668            .unwrap()
669            .unwrap();
670
671        let req = BlockFifoRequest {
672            command: BlockFifoCommand {
673                opcode: fblock::BlockOpcode::Write.into_primitive(),
674                ..Default::default()
675            },
676            reqid: 124,
677            group: 0,
678            vmoid: vmo_id.id,
679            length: 1,
680            vmo_offset: 0,
681            dev_offset: 0,
682            trace_flow_id: 0,
683            ..Default::default()
684        };
685        fifo.write(&[req]).unwrap();
686
687        let r = rx.recv().unwrap();
688        assert_eq!(r.request_id.0, 0);
689        assert!(matches!(r.operation, Operation::Write { .. }));
690
691        session_manager.complete_request(r.request_id, zx::Status::OK);
692
693        let signals =
694            fifo.wait_one(zx::Signals::FIFO_READABLE, zx::MonotonicInstant::INFINITE).unwrap();
695        assert!(signals.contains(zx::Signals::FIFO_READABLE));
696
697        let mut resp = [BlockFifoResponse::default()];
698        fifo.read(&mut resp).unwrap();
699        assert_eq!(resp[0].reqid, 124);
700        assert_eq!(resp[0].status, zx::sys::ZX_OK);
701    }
702
703    #[fasync::run_singlethreaded(test)]
704    async fn test_flush_request() {
705        let (tx, rx) = std::sync::mpsc::channel();
706        let interface = Arc::new(MockInterface { request_sender: tx });
707        let session_manager = Arc::new(SessionManager::new(interface.clone()));
708
709        let sm_clone = session_manager.clone();
710        let (proxy, stream) = create_proxy_and_stream::<fblock::BlockMarker>();
711        let _server_task = fasync::Task::spawn(async move {
712            let server = crate::BlockServer::new(512, sm_clone);
713            server.handle_requests(stream).await.unwrap();
714        })
715        .detach();
716
717        let (session_proxy, session_server_end) =
718            fidl::endpoints::create_proxy::<fblock::SessionMarker>();
719        proxy.open_session(session_server_end).unwrap();
720
721        let fifo_handle = session_proxy.get_fifo().await.unwrap().unwrap();
722        let fifo: zx::Fifo<BlockFifoResponse, BlockFifoRequest> = zx::Fifo::from(fifo_handle);
723
724        let req = BlockFifoRequest {
725            command: BlockFifoCommand {
726                opcode: fblock::BlockOpcode::Flush.into_primitive(),
727                ..Default::default()
728            },
729            reqid: 125,
730            group: 0,
731            vmoid: fblock::VMOID_INVALID,
732            length: 0,
733            vmo_offset: 0,
734            dev_offset: 0,
735            trace_flow_id: 0,
736            ..Default::default()
737        };
738        fifo.write(&[req]).unwrap();
739
740        let r = rx.recv().unwrap();
741        assert_eq!(r.request_id.0, 0);
742        assert!(matches!(r.operation, Operation::Flush { .. }));
743
744        session_manager.complete_request(r.request_id, zx::Status::OK);
745
746        let signals =
747            fifo.wait_one(zx::Signals::FIFO_READABLE, zx::MonotonicInstant::INFINITE).unwrap();
748        assert!(signals.contains(zx::Signals::FIFO_READABLE));
749
750        let mut resp = [BlockFifoResponse::default()];
751        fifo.read(&mut resp).unwrap();
752        assert_eq!(resp[0].reqid, 125);
753        assert_eq!(resp[0].status, zx::sys::ZX_OK);
754    }
755
756    #[fasync::run_singlethreaded(test)]
757    async fn test_trim_request() {
758        let (tx, rx) = std::sync::mpsc::channel();
759        let interface = Arc::new(MockInterface { request_sender: tx });
760        let session_manager = Arc::new(SessionManager::new(interface.clone()));
761
762        let sm_clone = session_manager.clone();
763        let (proxy, stream) = create_proxy_and_stream::<fblock::BlockMarker>();
764        let _server_task = fasync::Task::spawn(async move {
765            let server = crate::BlockServer::new(512, sm_clone);
766            server.handle_requests(stream).await.unwrap();
767        })
768        .detach();
769
770        let (session_proxy, session_server_end) =
771            fidl::endpoints::create_proxy::<fblock::SessionMarker>();
772        proxy.open_session(session_server_end).unwrap();
773
774        let fifo_handle = session_proxy.get_fifo().await.unwrap().unwrap();
775        let fifo: zx::Fifo<BlockFifoResponse, BlockFifoRequest> = zx::Fifo::from(fifo_handle);
776
777        let req = BlockFifoRequest {
778            command: BlockFifoCommand {
779                opcode: fblock::BlockOpcode::Trim.into_primitive(),
780                ..Default::default()
781            },
782            reqid: 126,
783            group: 0,
784            vmoid: fblock::VMOID_INVALID,
785            length: 1,
786            vmo_offset: 0,
787            dev_offset: 0,
788            trace_flow_id: 0,
789            ..Default::default()
790        };
791        fifo.write(&[req]).unwrap();
792
793        let r = rx.recv().unwrap();
794        assert_eq!(r.request_id.0, 0);
795        assert!(matches!(r.operation, Operation::Trim { .. }));
796
797        session_manager.complete_request(r.request_id, zx::Status::OK);
798
799        let signals =
800            fifo.wait_one(zx::Signals::FIFO_READABLE, zx::MonotonicInstant::INFINITE).unwrap();
801        assert!(signals.contains(zx::Signals::FIFO_READABLE));
802
803        let mut resp = [BlockFifoResponse::default()];
804        fifo.read(&mut resp).unwrap();
805        assert_eq!(resp[0].reqid, 126);
806        assert_eq!(resp[0].status, zx::sys::ZX_OK);
807    }
808
809    #[fasync::run_singlethreaded(test)]
810    async fn test_close_vmo() {
811        let (tx, rx) = std::sync::mpsc::channel();
812        let interface = Arc::new(MockInterface { request_sender: tx });
813        let session_manager = Arc::new(SessionManager::new(interface.clone()));
814
815        let sm_clone = session_manager.clone();
816        let (proxy, stream) = create_proxy_and_stream::<fblock::BlockMarker>();
817        let _server_task = fasync::Task::spawn(async move {
818            let server = crate::BlockServer::new(512, sm_clone);
819            server.handle_requests(stream).await.unwrap();
820        })
821        .detach();
822
823        let (session_proxy, session_server_end) =
824            fidl::endpoints::create_proxy::<fblock::SessionMarker>();
825        proxy.open_session(session_server_end).unwrap();
826
827        let fifo_handle = session_proxy.get_fifo().await.unwrap().unwrap();
828        let fifo: zx::Fifo<BlockFifoResponse, BlockFifoRequest> = zx::Fifo::from(fifo_handle);
829
830        let vmo = zx::Vmo::create(8192).unwrap();
831        let vmo_id = session_proxy
832            .attach_vmo(vmo.duplicate_handle(zx::Rights::SAME_RIGHTS).unwrap())
833            .await
834            .unwrap()
835            .unwrap();
836
837        let req = BlockFifoRequest {
838            command: BlockFifoCommand {
839                opcode: fblock::BlockOpcode::CloseVmo.into_primitive(),
840                ..Default::default()
841            },
842            reqid: 127,
843            group: 0,
844            vmoid: vmo_id.id,
845            length: 0,
846            vmo_offset: 0,
847            dev_offset: 0,
848            trace_flow_id: 0,
849            ..Default::default()
850        };
851        fifo.write(&[req]).unwrap();
852
853        let signals =
854            fifo.wait_one(zx::Signals::FIFO_READABLE, zx::MonotonicInstant::INFINITE).unwrap();
855        assert!(signals.contains(zx::Signals::FIFO_READABLE));
856
857        let mut resp = [BlockFifoResponse::default()];
858        fifo.read(&mut resp).unwrap();
859        assert_eq!(resp[0].reqid, 127);
860        assert_eq!(resp[0].status, zx::sys::ZX_OK);
861
862        // CloseVmo is handled automatically. Interface should not receive anything.
863        assert!(rx.try_recv().is_err());
864    }
865
866    #[fasync::run_singlethreaded(test)]
867    async fn test_error() {
868        let (tx, rx) = std::sync::mpsc::channel();
869        let interface = Arc::new(MockInterface { request_sender: tx });
870        let session_manager = Arc::new(SessionManager::new(interface.clone()));
871
872        let sm_clone = session_manager.clone();
873        let (proxy, stream) = create_proxy_and_stream::<fblock::BlockMarker>();
874        let _server_task = fasync::Task::spawn(async move {
875            let server = crate::BlockServer::new(512, sm_clone);
876            server.handle_requests(stream).await.unwrap();
877        })
878        .detach();
879
880        let (session_proxy, session_server_end) =
881            fidl::endpoints::create_proxy::<fblock::SessionMarker>();
882        proxy.open_session(session_server_end).unwrap();
883
884        let fifo_handle = session_proxy.get_fifo().await.unwrap().unwrap();
885        let fifo: zx::Fifo<BlockFifoResponse, BlockFifoRequest> = zx::Fifo::from(fifo_handle);
886
887        let req = BlockFifoRequest {
888            command: BlockFifoCommand {
889                opcode: fblock::BlockOpcode::Flush.into_primitive(),
890                ..Default::default()
891            },
892            reqid: 128,
893            group: 0,
894            vmoid: fblock::VMOID_INVALID,
895            length: 0,
896            vmo_offset: 0,
897            dev_offset: 0,
898            trace_flow_id: 0,
899            ..Default::default()
900        };
901        fifo.write(&[req]).unwrap();
902
903        let r = rx.recv().unwrap();
904        session_manager.complete_request(r.request_id, zx::Status::IO);
905
906        let signals =
907            fifo.wait_one(zx::Signals::FIFO_READABLE, zx::MonotonicInstant::INFINITE).unwrap();
908        assert!(signals.contains(zx::Signals::FIFO_READABLE));
909
910        let mut resp = [BlockFifoResponse::default()];
911        fifo.read(&mut resp).unwrap();
912        assert_eq!(resp[0].reqid, 128);
913        assert_eq!(resp[0].status, zx::sys::ZX_ERR_IO);
914    }
915
916    #[fasync::run_singlethreaded(test)]
917    async fn test_teardown_with_active_requests() {
918        let (tx, rx) = std::sync::mpsc::channel();
919        let interface = Arc::new(MockInterface { request_sender: tx });
920        let session_manager = Arc::new(SessionManager::new(interface.clone()));
921
922        let sm_clone = session_manager.clone();
923        let (proxy, stream) = create_proxy_and_stream::<fblock::BlockMarker>();
924        let server_task = fasync::Task::spawn(async move {
925            let server = crate::BlockServer::new(512, sm_clone);
926            server.handle_requests(stream).await.unwrap();
927        });
928
929        let (session_proxy, session_server_end) =
930            fidl::endpoints::create_proxy::<fblock::SessionMarker>();
931        proxy.open_session(session_server_end).unwrap();
932
933        let fifo_handle = session_proxy.get_fifo().await.unwrap().unwrap();
934        let fifo: zx::Fifo<BlockFifoResponse, BlockFifoRequest> = zx::Fifo::from(fifo_handle);
935
936        let vmo = zx::Vmo::create(8192).unwrap();
937        let vmo_id = session_proxy
938            .attach_vmo(vmo.duplicate_handle(zx::Rights::SAME_RIGHTS).unwrap())
939            .await
940            .unwrap()
941            .unwrap();
942
943        let req = BlockFifoRequest {
944            command: BlockFifoCommand {
945                opcode: fblock::BlockOpcode::Read.into_primitive(),
946                ..Default::default()
947            },
948            reqid: 129,
949            group: 0,
950            vmoid: vmo_id.id,
951            length: 1,
952            vmo_offset: 0,
953            dev_offset: 0,
954            trace_flow_id: 0,
955            ..Default::default()
956        };
957        // Start a request and wait until it's been submitted.
958        fifo.write(&[req]).unwrap();
959        let r = rx.recv().unwrap();
960
961        // Close the client, which will eventually cause the FIFO loop to exit.
962        drop(session_proxy);
963        fasync::Timer::new(std::time::Duration::from_millis(50)).await;
964
965        // Complete the request, simulating a completion after the FIFO loop has exited.
966        session_manager.complete_request(r.request_id, zx::Status::OK);
967
968        drop(proxy);
969        fasync::unblock(move || session_manager.terminate()).await;
970        server_task.await;
971    }
972
973    #[fasync::run_singlethreaded(test)]
974    async fn test_teardown_with_active_grouped_requests() {
975        let (tx, rx) = std::sync::mpsc::channel();
976        let interface = Arc::new(MockInterface { request_sender: tx });
977        let session_manager = Arc::new(SessionManager::new(interface.clone()));
978
979        let sm_clone = session_manager.clone();
980        let (proxy, stream) = create_proxy_and_stream::<fblock::BlockMarker>();
981        let server_task = fasync::Task::spawn(async move {
982            let server = crate::BlockServer::new(512, sm_clone);
983            server.handle_requests(stream).await.unwrap();
984        });
985
986        let (session_proxy, session_server_end) =
987            fidl::endpoints::create_proxy::<fblock::SessionMarker>();
988        proxy.open_session(session_server_end).unwrap();
989
990        let fifo_handle = session_proxy.get_fifo().await.unwrap().unwrap();
991        let fifo: zx::Fifo<BlockFifoResponse, BlockFifoRequest> = zx::Fifo::from(fifo_handle);
992
993        let vmo = zx::Vmo::create(8192).unwrap();
994        let vmo_id = session_proxy
995            .attach_vmo(vmo.duplicate_handle(zx::Rights::SAME_RIGHTS).unwrap())
996            .await
997            .unwrap()
998            .unwrap();
999
1000        let mut req = BlockFifoRequest {
1001            command: BlockFifoCommand {
1002                opcode: fblock::BlockOpcode::Read.into_primitive(),
1003                flags: fblock::BlockIoFlag::GROUP_ITEM.bits(),
1004                ..Default::default()
1005            },
1006            reqid: 1,
1007            group: 1,
1008            vmoid: vmo_id.id,
1009            length: 1,
1010            vmo_offset: 0,
1011            dev_offset: 0,
1012            trace_flow_id: 0,
1013            ..Default::default()
1014        };
1015
1016        // Start two groups of requests and wait until they're been submitted.
1017        fifo.write(&[req]).unwrap();
1018        req.reqid = 2;
1019        req.group = 2;
1020        fifo.write(&[req]).unwrap();
1021        let r1 = rx.recv().unwrap();
1022        let r2 = rx.recv().unwrap();
1023        req.reqid = 3;
1024        req.group = 2;
1025        fifo.write(&[req]).unwrap();
1026        let r3 = rx.recv().unwrap();
1027
1028        // Complete requests 1,2.  This leaves two groups in the following states:
1029        // - Group 1: 0 active requests, still waiting for END
1030        // - Group 2: 1 active request, still waiting for END
1031        // Neither group will be able to complete yet.
1032        session_manager.complete_request(r1.request_id, zx::Status::OK);
1033        session_manager.complete_request(r2.request_id, zx::Status::OK);
1034
1035        // Close the client, which will eventually cause the FIFO loop to exit.
1036        // Group 1 should complete now.  Group 2 can't yet.
1037        drop(session_proxy);
1038        fasync::Timer::new(std::time::Duration::from_millis(50)).await;
1039
1040        // At some later time, complete request 3, which should complete group 2.
1041        session_manager.complete_request(r3.request_id, zx::Status::OK);
1042
1043        drop(proxy);
1044
1045        fasync::unblock(move || session_manager.terminate()).await;
1046        server_task.await;
1047    }
1048
1049    #[fasync::run_singlethreaded(test)]
1050    async fn test_session_close_is_synchronous() {
1051        use futures::FutureExt as _;
1052
1053        let (tx, rx) = std::sync::mpsc::channel();
1054        let interface = Arc::new(MockInterface { request_sender: tx });
1055        let session_manager = Arc::new(SessionManager::new(interface.clone()));
1056
1057        let sm_clone = session_manager.clone();
1058        let (proxy, stream) = create_proxy_and_stream::<fblock::BlockMarker>();
1059        let server_task = fasync::Task::spawn(async move {
1060            let server = crate::BlockServer::new(512, sm_clone);
1061            server.handle_requests(stream).await.unwrap();
1062        });
1063
1064        let (session_proxy, session_server_end) =
1065            fidl::endpoints::create_proxy::<fblock::SessionMarker>();
1066        proxy.open_session(session_server_end).unwrap();
1067
1068        let fifo_handle = session_proxy.get_fifo().await.unwrap().unwrap();
1069        let fifo: zx::Fifo<BlockFifoResponse, BlockFifoRequest> = zx::Fifo::from(fifo_handle);
1070
1071        let vmo = zx::Vmo::create(8192).unwrap();
1072        let vmo_id = session_proxy
1073            .attach_vmo(vmo.duplicate_handle(zx::Rights::SAME_RIGHTS).unwrap())
1074            .await
1075            .unwrap()
1076            .unwrap();
1077
1078        let req = BlockFifoRequest {
1079            command: BlockFifoCommand {
1080                opcode: fblock::BlockOpcode::Read.into_primitive(),
1081                ..Default::default()
1082            },
1083            reqid: 123,
1084            group: 0,
1085            vmoid: vmo_id.id,
1086            length: 1,
1087            vmo_offset: 0,
1088            dev_offset: 0,
1089            trace_flow_id: 0,
1090            ..Default::default()
1091        };
1092        fifo.write(&[req]).unwrap();
1093
1094        let r = rx.recv().unwrap();
1095
1096        // The close request shouldn't complete yet because the read is still hanging.
1097        let mut close_fut = std::pin::pin!(session_proxy.close().fuse());
1098        let mut timer_fut =
1099            std::pin::pin!(fasync::Timer::new(std::time::Duration::from_millis(100)).fuse());
1100        futures::select! {
1101            res = close_fut => panic!("close completed too early: {:?}", res),
1102            _ = timer_fut => {}
1103        }
1104
1105        session_manager.complete_request(r.request_id, zx::Status::OK);
1106
1107        // Verify that close() now completes.
1108        close_fut.await.unwrap().unwrap();
1109
1110        std::mem::drop(proxy);
1111        server_task.await;
1112    }
1113}
block_server/callback_interface.rs

block_server/
callback_interface.rs
