block_client/
lib.rs

1// Copyright 2020 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
5//! A Rust client library for the Fuchsia block protocol.
6//!
7//! This crate provides a low-level client for interacting with block devices over the
8//! `fuchsia.hardware.block` FIDL protocol and the FIFO interface for block I/O.
9//!
10//! See the [`BlockClient`] trait.
11
12use fidl_fuchsia_storage_block as block;
13use fidl_fuchsia_storage_block::{MAX_TRANSFER_UNBOUNDED, VMOID_INVALID};
14use fuchsia_async as fasync;
15use fuchsia_sync::Mutex;
16use futures::channel::oneshot;
17use futures::executor::block_on;
18use std::borrow::Borrow;
19use std::collections::HashMap;
20use std::future::Future;
21use std::hash::{Hash, Hasher};
22use std::mem::MaybeUninit;
23use std::num::NonZero;
24use std::ops::{DerefMut, Range};
25use std::pin::Pin;
26use std::sync::Arc;
27use std::sync::atomic::{AtomicU16, Ordering};
28use std::task::{Context, Poll, Waker};
29use storage_trace as trace;
30
31pub use cache::Cache;
32
33pub use block::DeviceFlag as BlockDeviceFlag;
34
35pub use block_protocol::*;
36
37pub mod cache;
38
39const TEMP_VMO_SIZE: usize = 65536;
40
41/// If a trace flow ID isn't specified for requests, one will be generated.
42pub const NO_TRACE_ID: u64 = 0;
43
44pub use fidl_fuchsia_storage_block::{BlockIoFlag, BlockOpcode};
45
46fn fidl_to_status(error: fidl::Error) -> zx::Status {
47    match error {
48        fidl::Error::ClientChannelClosed { epitaph, .. } => match epitaph.into() {
49            Err(s) => s,
50            Ok(()) => zx::Status::PEER_CLOSED,
51        },
52        _ => zx::Status::INTERNAL,
53    }
54}
55
56fn opcode_str(opcode: u8) -> &'static str {
57    match BlockOpcode::from_primitive(opcode) {
58        Some(BlockOpcode::Read) => "read",
59        Some(BlockOpcode::Write) => "write",
60        Some(BlockOpcode::Flush) => "flush",
61        Some(BlockOpcode::Trim) => "trim",
62        Some(BlockOpcode::CloseVmo) => "close_vmo",
63        None => "unknown",
64    }
65}
66
67pub enum BufferSlice<'a> {
68    VmoId { vmo_id: &'a VmoId, offset: u64, length: u64 },
69    Memory(&'a [u8]),
70}
71
72impl<'a> BufferSlice<'a> {
73    pub fn new_with_vmo_id(vmo_id: &'a VmoId, offset: u64, length: u64) -> Self {
74        BufferSlice::VmoId { vmo_id, offset, length }
75    }
76}
77
78impl<'a> From<&'a [u8]> for BufferSlice<'a> {
79    fn from(buf: &'a [u8]) -> Self {
80        BufferSlice::Memory(buf)
81    }
82}
83
84pub enum MutableBufferSlice<'a> {
85    VmoId { vmo_id: &'a VmoId, offset: u64, length: u64 },
86    Memory(&'a mut [u8]),
87}
88
89impl<'a> MutableBufferSlice<'a> {
90    pub fn new_with_vmo_id(vmo_id: &'a VmoId, offset: u64, length: u64) -> Self {
91        MutableBufferSlice::VmoId { vmo_id, offset, length }
92    }
93}
94
95impl<'a> From<&'a mut [u8]> for MutableBufferSlice<'a> {
96    fn from(buf: &'a mut [u8]) -> Self {
97        MutableBufferSlice::Memory(buf)
98    }
99}
100
101#[derive(Default)]
102struct RequestState {
103    result: Option<Result<(), zx::Status>>,
104    waker: Option<Waker>,
105}
106
107#[derive(Default)]
108struct FifoState {
109    // The fifo.
110    fifo: Option<fasync::Fifo<BlockFifoResponse, BlockFifoRequest>>,
111
112    // The next request ID to be used.
113    next_request_id: u32,
114
115    // A queue of messages to be sent on the fifo.
116    queue: std::collections::VecDeque<BlockFifoRequest>,
117
118    // Map from request ID to RequestState.
119    map: HashMap<u32, RequestState>,
120
121    // The waker for the FifoPoller.
122    poller_waker: Option<Waker>,
123}
124
125impl FifoState {
126    fn terminate(&mut self) {
127        self.fifo.take();
128        for (_, request_state) in self.map.iter_mut() {
129            request_state.result.get_or_insert(Err(zx::Status::CANCELED));
130            if let Some(waker) = request_state.waker.take() {
131                waker.wake();
132            }
133        }
134        if let Some(waker) = self.poller_waker.take() {
135            waker.wake();
136        }
137    }
138
139    // Returns true if polling should be terminated.
140    fn poll_send_requests(&mut self, context: &mut Context<'_>) -> bool {
141        let fifo = if let Some(fifo) = self.fifo.as_ref() {
142            fifo
143        } else {
144            return true;
145        };
146
147        loop {
148            let slice = self.queue.as_slices().0;
149            if slice.is_empty() {
150                return false;
151            }
152            match fifo.try_write(context, slice) {
153                Poll::Ready(Ok(sent)) => {
154                    self.queue.drain(0..sent.get());
155                }
156                Poll::Ready(Err(_)) => {
157                    self.terminate();
158                    return true;
159                }
160                Poll::Pending => {
161                    return false;
162                }
163            }
164        }
165    }
166}
167
168type FifoStateRef = Arc<Mutex<FifoState>>;
169
170// A future used for fifo responses.
171struct ResponseFuture {
172    request_id: u32,
173    fifo_state: FifoStateRef,
174}
175
176impl ResponseFuture {
177    fn new(fifo_state: FifoStateRef, request_id: u32) -> Self {
178        ResponseFuture { request_id, fifo_state }
179    }
180}
181
182impl Future for ResponseFuture {
183    type Output = Result<(), zx::Status>;
184
185    fn poll(self: Pin<&mut Self>, context: &mut Context<'_>) -> Poll<Self::Output> {
186        let mut state = self.fifo_state.lock();
187        let request_state = state.map.get_mut(&self.request_id).unwrap();
188        if let Some(result) = request_state.result {
189            Poll::Ready(result)
190        } else {
191            request_state.waker.replace(context.waker().clone());
192            Poll::Pending
193        }
194    }
195}
196
197impl Drop for ResponseFuture {
198    fn drop(&mut self) {
199        let mut state = self.fifo_state.lock();
200        if let Some(request_state) = state.map.remove(&self.request_id) {
201            if request_state.result.is_none() {
202                // Request was still pending!  This is a cancellation which we do not support.  We
203                // do not know the disposition of any VMO that the far end might still be writing
204                // to.  To avoid potential corruption (e.g. a client reuses a buffer that might
205                // still be being used by the driver), terminate the connection to prevent any
206                // future use.
207                state.terminate();
208            }
209        }
210        update_outstanding_requests_counter(state.map.len());
211    }
212}
213
214/// Wraps a vmo-id. Will panic if you forget to detach.
215#[derive(Debug)]
216#[must_use]
217pub struct VmoId(AtomicU16);
218
219impl VmoId {
220    /// VmoIds will normally be vended by attach_vmo, but this might be used in some tests
221    pub fn new(id: u16) -> Self {
222        Self(AtomicU16::new(id))
223    }
224
225    /// Invalidates self and returns a new VmoId with the same underlying ID.
226    pub fn take(&self) -> Self {
227        Self(AtomicU16::new(self.0.swap(VMOID_INVALID, Ordering::Relaxed)))
228    }
229
230    pub fn is_valid(&self) -> bool {
231        self.id() != VMOID_INVALID
232    }
233
234    /// Takes the ID.  The caller assumes responsibility for detaching.
235    #[must_use]
236    pub fn into_id(self) -> u16 {
237        self.0.swap(VMOID_INVALID, Ordering::Relaxed)
238    }
239
240    pub fn id(&self) -> u16 {
241        self.0.load(Ordering::Relaxed)
242    }
243}
244
245impl PartialEq for VmoId {
246    fn eq(&self, other: &Self) -> bool {
247        self.id() == other.id()
248    }
249}
250
251impl Eq for VmoId {}
252
253impl Drop for VmoId {
254    fn drop(&mut self) {
255        assert_eq!(self.0.load(Ordering::Relaxed), VMOID_INVALID, "Did you forget to detach?");
256    }
257}
258
259impl Hash for VmoId {
260    fn hash<H: Hasher>(&self, state: &mut H) {
261        self.id().hash(state);
262    }
263}
264
265/// Represents a client connection to a block device. This is a simplified version of the block.fidl
266/// interface.
267/// Most users will use the RemoteBlockClient instantiation of this trait.
268pub trait BlockClient: Send + Sync {
269    /// Wraps AttachVmo from fuchsia.hardware.block::Block.
270    ///
271    /// # Safety
272    ///
273    /// The caller must ensure that no references are held during I/O as this would be
274    /// undefined behavior.  The caller may hold pointers, which does not lead to undefined
275    /// behavior; Rust does not make the same assumptions as references for pointers.
276    ///
277    /// Whilst a connection failure or client-side cancellation does not immediately lead to
278    /// Rust undefined behavior, the caller must thereafter assume the VMO is poisoned.  No
279    /// assumptions can be made regarding what might be written to the VMO by the far end, or
280    /// anything the far end might have delegated access to.  The only safe thing to do is
281    /// discard the VMO and not attempt to use or re-attach it.
282    ///
283    /// Attaching the VMO once for its lifetime is safe, if strictly more than necessary.
284    unsafe fn attach_vmo(
285        &self,
286        vmo: &zx::Vmo,
287    ) -> impl Future<Output = Result<VmoId, zx::Status>> + Send;
288
289    /// Detaches the given vmo-id from the device.
290    fn detach_vmo(&self, vmo_id: VmoId) -> impl Future<Output = Result<(), zx::Status>> + Send;
291
292    /// Registers an inline encryption key handle with the session, returning a session-scoped slot.
293    fn register_key(
294        &self,
295        _key_token: zx::EventPair,
296    ) -> impl Future<Output = Result<u8, zx::Status>> + Send {
297        async { Err(zx::Status::NOT_SUPPORTED) }
298    }
299
300    /// Reads from the device at |device_offset| into the given buffer slice.
301    fn read_at(
302        &self,
303        buffer_slice: MutableBufferSlice<'_>,
304        device_offset: u64,
305    ) -> impl Future<Output = Result<(), zx::Status>> + Send {
306        self.read_at_with_opts_traced(buffer_slice, device_offset, ReadOptions::default(), 0)
307    }
308
309    fn read_at_with_opts(
310        &self,
311        buffer_slice: MutableBufferSlice<'_>,
312        device_offset: u64,
313        opts: ReadOptions,
314    ) -> impl Future<Output = Result<(), zx::Status>> + Send {
315        self.read_at_with_opts_traced(buffer_slice, device_offset, opts, 0)
316    }
317
318    fn read_at_with_opts_traced(
319        &self,
320        buffer_slice: MutableBufferSlice<'_>,
321        device_offset: u64,
322        opts: ReadOptions,
323        trace_flow_id: u64,
324    ) -> impl Future<Output = Result<(), zx::Status>> + Send;
325
326    /// Writes the data in |buffer_slice| to the device.
327    fn write_at(
328        &self,
329        buffer_slice: BufferSlice<'_>,
330        device_offset: u64,
331    ) -> impl Future<Output = Result<(), zx::Status>> + Send {
332        self.write_at_with_opts_traced(
333            buffer_slice,
334            device_offset,
335            WriteOptions::default(),
336            NO_TRACE_ID,
337        )
338    }
339
340    fn write_at_with_opts(
341        &self,
342        buffer_slice: BufferSlice<'_>,
343        device_offset: u64,
344        opts: WriteOptions,
345    ) -> impl Future<Output = Result<(), zx::Status>> + Send {
346        self.write_at_with_opts_traced(buffer_slice, device_offset, opts, NO_TRACE_ID)
347    }
348
349    fn write_at_with_opts_traced(
350        &self,
351        buffer_slice: BufferSlice<'_>,
352        device_offset: u64,
353        opts: WriteOptions,
354        trace_flow_id: u64,
355    ) -> impl Future<Output = Result<(), zx::Status>> + Send;
356
357    /// Trims the given range on the block device.
358    fn trim(
359        &self,
360        device_range: Range<u64>,
361    ) -> impl Future<Output = Result<(), zx::Status>> + Send {
362        self.trim_traced(device_range, NO_TRACE_ID)
363    }
364
365    fn trim_traced(
366        &self,
367        device_range: Range<u64>,
368        trace_flow_id: u64,
369    ) -> impl Future<Output = Result<(), zx::Status>> + Send;
370
371    fn flush(&self) -> impl Future<Output = Result<(), zx::Status>> + Send {
372        self.flush_traced(NO_TRACE_ID)
373    }
374
375    /// Sends a flush request to the underlying block device.
376    fn flush_traced(
377        &self,
378        trace_flow_id: u64,
379    ) -> impl Future<Output = Result<(), zx::Status>> + Send;
380
381    /// Closes the fifo.
382    fn close(&self) -> impl Future<Output = Result<(), zx::Status>> + Send;
383
384    /// Returns the block size of the device.
385    fn block_size(&self) -> u32;
386
387    /// Returns the size, in blocks, of the device.
388    fn block_count(&self) -> u64;
389
390    /// Returns the maximum number of blocks which can be transferred in a single request.
391    fn max_transfer_blocks(&self) -> Option<NonZero<u32>>;
392
393    /// Returns the block flags reported by the device.
394    fn block_flags(&self) -> BlockDeviceFlag;
395
396    /// Returns true if the remote fifo is still connected.
397    fn is_connected(&self) -> bool;
398
399    /// Connects to the device's Mapper protocol.
400    fn connect_mapper(
401        &self,
402        server_end: fidl::endpoints::ServerEnd<block::MapperMarker>,
403    ) -> impl Future<Output = Result<(), zx::Status>> + Send {
404        let _ = server_end.close_with_epitaph(zx::Status::NOT_SUPPORTED);
405        async { Err(zx::Status::NOT_SUPPORTED) }
406    }
407}
408
409struct Common {
410    block_size: u32,
411    block_count: u64,
412    max_transfer_blocks: Option<NonZero<u32>>,
413    block_flags: BlockDeviceFlag,
414    fifo_state: FifoStateRef,
415    temp_vmo: futures::lock::Mutex<zx::Vmo>,
416    temp_vmo_id: VmoId,
417}
418
419impl Common {
420    fn new(
421        fifo: fasync::Fifo<BlockFifoResponse, BlockFifoRequest>,
422        info: &block::BlockInfo,
423        temp_vmo: zx::Vmo,
424        temp_vmo_id: VmoId,
425    ) -> Self {
426        let fifo_state = Arc::new(Mutex::new(FifoState { fifo: Some(fifo), ..Default::default() }));
427        fasync::Task::spawn(FifoPoller { fifo_state: fifo_state.clone() }).detach();
428        Self {
429            block_size: info.block_size,
430            block_count: info.block_count,
431            max_transfer_blocks: if info.max_transfer_size != MAX_TRANSFER_UNBOUNDED {
432                NonZero::new(info.max_transfer_size / info.block_size)
433            } else {
434                None
435            },
436            block_flags: info.flags,
437            fifo_state,
438            temp_vmo: futures::lock::Mutex::new(temp_vmo),
439            temp_vmo_id,
440        }
441    }
442
443    fn to_blocks(&self, bytes: u64) -> Result<u64, zx::Status> {
444        if bytes % self.block_size as u64 != 0 {
445            Err(zx::Status::INVALID_ARGS)
446        } else {
447            Ok(bytes / self.block_size as u64)
448        }
449    }
450
451    // Sends the request and waits for the response.
452    async fn send(&self, mut request: BlockFifoRequest) -> Result<(), zx::Status> {
453        let (request_id, trace_flow_id) = {
454            let mut state = self.fifo_state.lock();
455
456            if state.fifo.is_none() {
457                // Fifo has been closed.
458                return Err(zx::Status::CANCELED);
459            }
460            trace::duration!(
461                "storage",
462                "block_client::send::start",
463                "op" => opcode_str(request.command.opcode),
464                "len" => request.length * self.block_size
465            );
466            let request_id = state.next_request_id;
467            state.next_request_id = state.next_request_id.overflowing_add(1).0;
468            assert!(
469                state.map.insert(request_id, RequestState::default()).is_none(),
470                "request id in use!"
471            );
472            update_outstanding_requests_counter(state.map.len());
473            request.reqid = request_id;
474            if request.trace_flow_id == NO_TRACE_ID {
475                request.trace_flow_id = trace::Id::new().into();
476            }
477            let trace_flow_id = request.trace_flow_id;
478            trace::flow_begin!("storage", "block_client::send", trace_flow_id.into());
479            state.queue.push_back(request);
480            if let Some(waker) = state.poller_waker.clone() {
481                state.poll_send_requests(&mut Context::from_waker(&waker));
482            }
483            (request_id, trace_flow_id)
484        };
485        let res = ResponseFuture::new(self.fifo_state.clone(), request_id).await;
486        trace::duration!("storage", "block_client::send::end");
487        trace::flow_end!("storage", "block_client::send", trace_flow_id.into());
488        res
489    }
490
491    fn detach_vmo(&self, vmo_id: VmoId) -> impl Future<Output = Result<(), zx::Status>> {
492        self.send(BlockFifoRequest {
493            command: BlockFifoCommand {
494                opcode: BlockOpcode::CloseVmo.into_primitive(),
495                flags: 0,
496                ..Default::default()
497            },
498            vmoid: vmo_id.into_id(),
499            ..Default::default()
500        })
501    }
502
503    async fn read_at(
504        &self,
505        buffer_slice: MutableBufferSlice<'_>,
506        device_offset: u64,
507        opts: ReadOptions,
508        trace_flow_id: u64,
509    ) -> Result<(), zx::Status> {
510        let mut flags = BlockIoFlag::empty();
511
512        if opts.inline_crypto.is_enabled {
513            flags |= BlockIoFlag::INLINE_ENCRYPTION_ENABLED;
514        }
515
516        match buffer_slice {
517            MutableBufferSlice::VmoId { vmo_id, offset, length } => {
518                self.send(BlockFifoRequest {
519                    command: BlockFifoCommand {
520                        opcode: BlockOpcode::Read.into_primitive(),
521                        flags: flags.bits(),
522                        ..Default::default()
523                    },
524                    vmoid: vmo_id.id(),
525                    length: self
526                        .to_blocks(length)?
527                        .try_into()
528                        .map_err(|_| zx::Status::INVALID_ARGS)?,
529                    vmo_offset: self.to_blocks(offset)?,
530                    dev_offset: self.to_blocks(device_offset)?,
531                    trace_flow_id,
532                    dun: opts.inline_crypto.dun,
533                    slot: opts.inline_crypto.slot,
534                    ..Default::default()
535                })
536                .await?
537            }
538            MutableBufferSlice::Memory(mut slice) => {
539                let temp_vmo = self.temp_vmo.lock().await;
540                let mut device_block = self.to_blocks(device_offset)?;
541                loop {
542                    let to_do = std::cmp::min(TEMP_VMO_SIZE, slice.len());
543                    let block_count = self.to_blocks(to_do as u64)? as u32;
544                    self.send(BlockFifoRequest {
545                        command: BlockFifoCommand {
546                            opcode: BlockOpcode::Read.into_primitive(),
547                            flags: flags.bits(),
548                            ..Default::default()
549                        },
550                        vmoid: self.temp_vmo_id.id(),
551                        length: block_count,
552                        vmo_offset: 0,
553                        dev_offset: device_block,
554                        trace_flow_id,
555                        dun: opts.inline_crypto.dun,
556                        slot: opts.inline_crypto.slot,
557                        ..Default::default()
558                    })
559                    .await?;
560                    temp_vmo.read(&mut slice[..to_do], 0)?;
561                    if to_do == slice.len() {
562                        break;
563                    }
564                    device_block += block_count as u64;
565                    slice = &mut slice[to_do..];
566                }
567            }
568        }
569        Ok(())
570    }
571
572    async fn write_at(
573        &self,
574        buffer_slice: BufferSlice<'_>,
575        device_offset: u64,
576        opts: WriteOptions,
577        trace_flow_id: u64,
578    ) -> Result<(), zx::Status> {
579        let mut flags = BlockIoFlag::empty();
580
581        if opts.flags.contains(WriteFlags::FORCE_ACCESS) {
582            flags |= BlockIoFlag::FORCE_ACCESS;
583        }
584
585        if opts.flags.contains(WriteFlags::PRE_BARRIER) {
586            flags |= BlockIoFlag::PRE_BARRIER;
587        }
588
589        if opts.inline_crypto.is_enabled {
590            flags |= BlockIoFlag::INLINE_ENCRYPTION_ENABLED;
591        }
592
593        match buffer_slice {
594            BufferSlice::VmoId { vmo_id, offset, length } => {
595                self.send(BlockFifoRequest {
596                    command: BlockFifoCommand {
597                        opcode: BlockOpcode::Write.into_primitive(),
598                        flags: flags.bits(),
599                        ..Default::default()
600                    },
601                    vmoid: vmo_id.id(),
602                    length: self
603                        .to_blocks(length)?
604                        .try_into()
605                        .map_err(|_| zx::Status::INVALID_ARGS)?,
606                    vmo_offset: self.to_blocks(offset)?,
607                    dev_offset: self.to_blocks(device_offset)?,
608                    trace_flow_id,
609                    dun: opts.inline_crypto.dun,
610                    slot: opts.inline_crypto.slot,
611                    ..Default::default()
612                })
613                .await?;
614            }
615            BufferSlice::Memory(mut slice) => {
616                let temp_vmo = self.temp_vmo.lock().await;
617                let mut device_block = self.to_blocks(device_offset)?;
618                loop {
619                    let to_do = std::cmp::min(TEMP_VMO_SIZE, slice.len());
620                    let block_count = self.to_blocks(to_do as u64)? as u32;
621                    temp_vmo.write(&slice[..to_do], 0)?;
622                    self.send(BlockFifoRequest {
623                        command: BlockFifoCommand {
624                            opcode: BlockOpcode::Write.into_primitive(),
625                            flags: flags.bits(),
626                            ..Default::default()
627                        },
628                        vmoid: self.temp_vmo_id.id(),
629                        length: block_count,
630                        vmo_offset: 0,
631                        dev_offset: device_block,
632                        trace_flow_id,
633                        dun: opts.inline_crypto.dun,
634                        slot: opts.inline_crypto.slot,
635                        ..Default::default()
636                    })
637                    .await?;
638                    if to_do == slice.len() {
639                        break;
640                    }
641                    device_block += block_count as u64;
642                    slice = &slice[to_do..];
643                }
644            }
645        }
646        Ok(())
647    }
648
649    async fn trim(&self, device_range: Range<u64>, trace_flow_id: u64) -> Result<(), zx::Status> {
650        let length = self.to_blocks(device_range.end - device_range.start)? as u32;
651        let dev_offset = self.to_blocks(device_range.start)?;
652        self.send(BlockFifoRequest {
653            command: BlockFifoCommand {
654                opcode: BlockOpcode::Trim.into_primitive(),
655                flags: 0,
656                ..Default::default()
657            },
658            vmoid: VMOID_INVALID,
659            length,
660            dev_offset,
661            trace_flow_id,
662            ..Default::default()
663        })
664        .await
665    }
666
667    fn flush(&self, trace_flow_id: u64) -> impl Future<Output = Result<(), zx::Status>> {
668        self.send(BlockFifoRequest {
669            command: BlockFifoCommand {
670                opcode: BlockOpcode::Flush.into_primitive(),
671                flags: 0,
672                ..Default::default()
673            },
674            vmoid: VMOID_INVALID,
675            trace_flow_id,
676            ..Default::default()
677        })
678    }
679
680    fn block_size(&self) -> u32 {
681        self.block_size
682    }
683
684    fn block_count(&self) -> u64 {
685        self.block_count
686    }
687
688    fn max_transfer_blocks(&self) -> Option<NonZero<u32>> {
689        self.max_transfer_blocks.clone()
690    }
691
692    fn block_flags(&self) -> BlockDeviceFlag {
693        self.block_flags
694    }
695
696    fn is_connected(&self) -> bool {
697        self.fifo_state.lock().fifo.is_some()
698    }
699}
700
701impl Drop for Common {
702    fn drop(&mut self) {
703        // It's OK to leak the VMO id because the server will dump all VMOs when the fifo is torn
704        // down.
705        let _ = self.temp_vmo_id.take().into_id();
706        self.fifo_state.lock().terminate();
707    }
708}
709
710// RemoteBlockClient is a BlockClient that communicates with a real block device over FIDL.
711pub struct RemoteBlockClient {
712    remote: Mutex<Option<block::BlockProxy>>,
713    session: block::SessionProxy,
714    common: Common,
715}
716
717impl RemoteBlockClient {
718    /// Returns a connection to a remote block device via the given channel.
719    pub async fn new(remote: impl Borrow<block::BlockProxy>) -> Result<Self, zx::Status> {
720        let remote = remote.borrow();
721        let info =
722            remote.get_info().await.map_err(fidl_to_status)?.map_err(zx::Status::err_from_raw)?;
723        let (session, server) = fidl::endpoints::create_proxy();
724        let () = remote.open_session(server).map_err(fidl_to_status)?;
725        let client = Self::from_session(info, session).await?;
726        *client.remote.lock() = Some(remote.clone());
727        Ok(client)
728    }
729
730    pub async fn from_session(
731        info: block::BlockInfo,
732        session: block::SessionProxy,
733    ) -> Result<Self, zx::Status> {
734        const SCRATCH_VMO_NAME: zx::Name = zx::Name::new_lossy("block-client-scratch-vmo");
735        let fifo =
736            session.get_fifo().await.map_err(fidl_to_status)?.map_err(zx::Status::err_from_raw)?;
737        let fifo = fasync::Fifo::from_fifo(fifo);
738        let temp_vmo = zx::Vmo::create(TEMP_VMO_SIZE as u64)?;
739        temp_vmo.set_name(&SCRATCH_VMO_NAME)?;
740        let dup = temp_vmo.duplicate_handle(zx::Rights::SAME_RIGHTS)?;
741        let vmo_id = session
742            .attach_vmo(dup)
743            .await
744            .map_err(fidl_to_status)?
745            .map_err(zx::Status::err_from_raw)?;
746        let vmo_id = VmoId::new(vmo_id.id);
747        Ok(RemoteBlockClient {
748            remote: Mutex::new(None),
749            session,
750            common: Common::new(fifo, &info, temp_vmo, vmo_id),
751        })
752    }
753}
754
755impl BlockClient for RemoteBlockClient {
756    async unsafe fn attach_vmo(&self, vmo: &zx::Vmo) -> Result<VmoId, zx::Status> {
757        let dup = vmo.duplicate_handle(zx::Rights::SAME_RIGHTS)?;
758        let vmo_id = self
759            .session
760            .attach_vmo(dup)
761            .await
762            .map_err(fidl_to_status)?
763            .map_err(zx::Status::err_from_raw)?;
764        Ok(VmoId::new(vmo_id.id))
765    }
766
767    fn detach_vmo(&self, vmo_id: VmoId) -> impl Future<Output = Result<(), zx::Status>> {
768        self.common.detach_vmo(vmo_id)
769    }
770
771    async fn register_key(&self, key_token: zx::EventPair) -> Result<u8, zx::Status> {
772        self.session
773            .register_key(key_token)
774            .await
775            .map_err(fidl_to_status)?
776            .map_err(zx::Status::err_from_raw)
777    }
778
779    fn read_at_with_opts_traced(
780        &self,
781        buffer_slice: MutableBufferSlice<'_>,
782        device_offset: u64,
783        opts: ReadOptions,
784        trace_flow_id: u64,
785    ) -> impl Future<Output = Result<(), zx::Status>> {
786        self.common.read_at(buffer_slice, device_offset, opts, trace_flow_id)
787    }
788
789    fn write_at_with_opts_traced(
790        &self,
791        buffer_slice: BufferSlice<'_>,
792        device_offset: u64,
793        opts: WriteOptions,
794        trace_flow_id: u64,
795    ) -> impl Future<Output = Result<(), zx::Status>> {
796        self.common.write_at(buffer_slice, device_offset, opts, trace_flow_id)
797    }
798
799    fn trim_traced(
800        &self,
801        range: Range<u64>,
802        trace_flow_id: u64,
803    ) -> impl Future<Output = Result<(), zx::Status>> {
804        self.common.trim(range, trace_flow_id)
805    }
806
807    fn flush_traced(&self, trace_flow_id: u64) -> impl Future<Output = Result<(), zx::Status>> {
808        self.common.flush(trace_flow_id)
809    }
810
811    async fn close(&self) -> Result<(), zx::Status> {
812        let _ = self.remote.lock().take();
813        let () = self
814            .session
815            .close()
816            .await
817            .map_err(fidl_to_status)?
818            .map_err(zx::Status::err_from_raw)?;
819        Ok(())
820    }
821
822    fn block_size(&self) -> u32 {
823        self.common.block_size()
824    }
825
826    fn block_count(&self) -> u64 {
827        self.common.block_count()
828    }
829
830    fn max_transfer_blocks(&self) -> Option<NonZero<u32>> {
831        self.common.max_transfer_blocks()
832    }
833
834    fn block_flags(&self) -> BlockDeviceFlag {
835        self.common.block_flags()
836    }
837
838    fn is_connected(&self) -> bool {
839        self.common.is_connected()
840    }
841
842    async fn connect_mapper(
843        &self,
844        server_end: fidl::endpoints::ServerEnd<block::MapperMarker>,
845    ) -> Result<(), zx::Status> {
846        let remote = self.remote.lock().clone();
847        if let Some(remote) = remote {
848            remote
849                .connect_mapper(server_end)
850                .await
851                .map_err(fidl_to_status)?
852                .map_err(zx::Status::err_from_raw)
853        } else {
854            let _ = server_end.close_with_epitaph(zx::Status::NOT_SUPPORTED);
855            Err(zx::Status::NOT_SUPPORTED)
856        }
857    }
858}
859
860pub struct RemoteBlockClientSync {
861    session: block::SessionSynchronousProxy,
862    common: Common,
863}
864
865impl RemoteBlockClientSync {
866    /// Returns a connection to a remote block device via the given channel, but spawns a separate
867    /// thread for polling the fifo which makes it work in cases where no executor is configured for
868    /// the calling thread.
869    pub fn new(
870        client_end: fidl::endpoints::ClientEnd<block::BlockMarker>,
871    ) -> Result<Self, zx::Status> {
872        let remote = block::BlockSynchronousProxy::new(client_end.into_channel());
873        let info = remote
874            .get_info(zx::MonotonicInstant::INFINITE)
875            .map_err(fidl_to_status)?
876            .map_err(zx::Status::err_from_raw)?;
877        let (client, server) = fidl::endpoints::create_endpoints();
878        let () = remote.open_session(server).map_err(fidl_to_status)?;
879        let session = block::SessionSynchronousProxy::new(client.into_channel());
880        let fifo = session
881            .get_fifo(zx::MonotonicInstant::INFINITE)
882            .map_err(fidl_to_status)?
883            .map_err(zx::Status::err_from_raw)?;
884        let temp_vmo = zx::Vmo::create(TEMP_VMO_SIZE as u64)?;
885        let dup = temp_vmo.duplicate_handle(zx::Rights::SAME_RIGHTS)?;
886        let vmo_id = session
887            .attach_vmo(dup, zx::MonotonicInstant::INFINITE)
888            .map_err(fidl_to_status)?
889            .map_err(zx::Status::err_from_raw)?;
890        let vmo_id = VmoId::new(vmo_id.id);
891
892        // The fifo needs to be instantiated from the thread that has the executor as that's where
893        // the fifo registers for notifications to be delivered.
894        let (sender, receiver) = oneshot::channel::<Result<Self, zx::Status>>();
895        std::thread::spawn(move || {
896            let mut executor = fasync::LocalExecutor::default();
897            let fifo = fasync::Fifo::from_fifo(fifo);
898            let common = Common::new(fifo, &info, temp_vmo, vmo_id);
899            let fifo_state = common.fifo_state.clone();
900            let _ = sender.send(Ok(RemoteBlockClientSync { session, common }));
901            executor.run_singlethreaded(FifoPoller { fifo_state });
902        });
903        block_on(receiver).map_err(|_| zx::Status::CANCELED)?
904    }
905
906    /// Wraps AttachVmo from fuchsia.hardware.block::Block.
907    ///
908    /// # Safety
909    ///
910    /// See `BlockClient::attach_vmo`.
911    pub unsafe fn attach_vmo(&self, vmo: &zx::Vmo) -> Result<VmoId, zx::Status> {
912        let dup = vmo.duplicate_handle(zx::Rights::SAME_RIGHTS)?;
913        let vmo_id = self
914            .session
915            .attach_vmo(dup, zx::MonotonicInstant::INFINITE)
916            .map_err(fidl_to_status)?
917            .map_err(zx::Status::err_from_raw)?;
918        Ok(VmoId::new(vmo_id.id))
919    }
920
921    pub fn detach_vmo(&self, vmo_id: VmoId) -> Result<(), zx::Status> {
922        block_on(self.common.detach_vmo(vmo_id))
923    }
924
925    pub fn read_at(
926        &self,
927        buffer_slice: MutableBufferSlice<'_>,
928        device_offset: u64,
929    ) -> Result<(), zx::Status> {
930        block_on(self.common.read_at(
931            buffer_slice,
932            device_offset,
933            ReadOptions::default(),
934            NO_TRACE_ID,
935        ))
936    }
937
938    pub fn write_at(
939        &self,
940        buffer_slice: BufferSlice<'_>,
941        device_offset: u64,
942    ) -> Result<(), zx::Status> {
943        block_on(self.common.write_at(
944            buffer_slice,
945            device_offset,
946            WriteOptions::default(),
947            NO_TRACE_ID,
948        ))
949    }
950
951    pub fn flush(&self) -> Result<(), zx::Status> {
952        block_on(self.common.flush(NO_TRACE_ID))
953    }
954
955    pub fn close(&self) -> Result<(), zx::Status> {
956        let () = self
957            .session
958            .close(zx::MonotonicInstant::INFINITE)
959            .map_err(fidl_to_status)?
960            .map_err(zx::Status::err_from_raw)?;
961        Ok(())
962    }
963
964    pub fn block_size(&self) -> u32 {
965        self.common.block_size()
966    }
967
968    pub fn block_count(&self) -> u64 {
969        self.common.block_count()
970    }
971
972    pub fn is_connected(&self) -> bool {
973        self.common.is_connected()
974    }
975}
976
977impl Drop for RemoteBlockClientSync {
978    fn drop(&mut self) {
979        // Ignore errors here as there is not much we can do about it.
980        let _ = self.close();
981    }
982}
983
984// FifoPoller is a future responsible for sending and receiving from the fifo.
985struct FifoPoller {
986    fifo_state: FifoStateRef,
987}
988
989impl Future for FifoPoller {
990    type Output = ();
991
992    fn poll(self: Pin<&mut Self>, context: &mut Context<'_>) -> Poll<Self::Output> {
993        let mut state_lock = self.fifo_state.lock();
994        let state = state_lock.deref_mut(); // So that we can split the borrow.
995
996        // Send requests.
997        if state.poll_send_requests(context) {
998            return Poll::Ready(());
999        }
1000
1001        // Receive responses.
1002        let fifo = state.fifo.as_ref().unwrap(); // Safe because poll_send_requests checks.
1003        loop {
1004            let mut response = MaybeUninit::uninit();
1005            match fifo.try_read(context, &mut response) {
1006                Poll::Pending => {
1007                    state.poller_waker = Some(context.waker().clone());
1008                    return Poll::Pending;
1009                }
1010                Poll::Ready(Ok(_)) => {
1011                    let response = unsafe { response.assume_init() };
1012                    let request_id = response.reqid;
1013                    // If the request isn't in the map, assume that it's a cancelled read.
1014                    if let Some(request_state) = state.map.get_mut(&request_id) {
1015                        request_state.result.replace(zx::Status::ok(response.status));
1016                        if let Some(waker) = request_state.waker.take() {
1017                            waker.wake();
1018                        }
1019                    }
1020                }
1021                Poll::Ready(Err(_)) => {
1022                    state.terminate();
1023                    return Poll::Ready(());
1024                }
1025            }
1026        }
1027    }
1028}
1029
1030fn update_outstanding_requests_counter(outstanding: usize) {
1031    trace::counter!("storage", "block-requests", 0, "outstanding" => outstanding);
1032}
1033
1034#[cfg(test)]
1035mod tests {
1036    use super::{
1037        BlockClient, BlockFifoRequest, BlockFifoResponse, BufferSlice, MutableBufferSlice,
1038        RemoteBlockClient, RemoteBlockClientSync, WriteOptions,
1039    };
1040    use block_protocol::ReadOptions;
1041    use block_server::{BlockServer, DeviceInfo, PartitionInfo};
1042    use fidl::endpoints::RequestStream as _;
1043    use fidl_fuchsia_storage_block as block;
1044    use fuchsia_async as fasync;
1045    use futures::future::{AbortHandle, Abortable, TryFutureExt as _};
1046    use futures::join;
1047    use futures::stream::StreamExt as _;
1048    use futures::stream::futures_unordered::FuturesUnordered;
1049    use ramdevice_client::RamdiskClient;
1050    use std::borrow::Cow;
1051    use std::num::NonZero;
1052    use std::sync::Arc;
1053    use std::sync::atomic::{AtomicBool, Ordering};
1054
1055    const RAMDISK_BLOCK_SIZE: u64 = 1024;
1056    const RAMDISK_BLOCK_COUNT: u64 = 1024;
1057
1058    pub async fn make_ramdisk() -> (RamdiskClient, block::BlockProxy, RemoteBlockClient) {
1059        let ramdisk = RamdiskClient::create(RAMDISK_BLOCK_SIZE, RAMDISK_BLOCK_COUNT)
1060            .await
1061            .expect("RamdiskClient::create failed");
1062        let client_end = ramdisk.open().expect("ramdisk.open failed");
1063        let proxy = client_end.into_proxy();
1064        let block_client = RemoteBlockClient::new(proxy).await.expect("new failed");
1065        assert_eq!(block_client.block_size(), 1024);
1066        let client_end = ramdisk.open().expect("ramdisk.open failed");
1067        let proxy = client_end.into_proxy();
1068        (ramdisk, proxy, block_client)
1069    }
1070
1071    #[fuchsia::test]
1072    async fn test_against_ram_disk() {
1073        let (_ramdisk, _block_proxy, block_client) = make_ramdisk().await;
1074
1075        let vmo = zx::Vmo::create(131072).expect("Vmo::create failed");
1076        vmo.write(b"hello", 5).expect("vmo.write failed");
1077        // SAFETY: Test code, only attach once, no other mappings.
1078        let vmo_id = unsafe { block_client.attach_vmo(&vmo) }.await.expect("attach_vmo failed");
1079        block_client
1080            .write_at(BufferSlice::new_with_vmo_id(&vmo_id, 0, 1024), 0)
1081            .await
1082            .expect("write_at failed");
1083        block_client
1084            .read_at(MutableBufferSlice::new_with_vmo_id(&vmo_id, 1024, 2048), 0)
1085            .await
1086            .expect("read_at failed");
1087        let mut buf: [u8; 5] = Default::default();
1088        vmo.read(&mut buf, 1029).expect("vmo.read failed");
1089        assert_eq!(&buf, b"hello");
1090        block_client.detach_vmo(vmo_id).await.expect("detach_vmo failed");
1091    }
1092
1093    #[fuchsia::test]
1094    async fn test_alignment() {
1095        let (_ramdisk, _block_proxy, block_client) = make_ramdisk().await;
1096        let vmo = zx::Vmo::create(131072).expect("Vmo::create failed");
1097        // SAFETY: Test code, only attach once, no other mappings.
1098        let vmo_id = unsafe { block_client.attach_vmo(&vmo) }.await.expect("attach_vmo failed");
1099        block_client
1100            .write_at(BufferSlice::new_with_vmo_id(&vmo_id, 0, 1024), 1)
1101            .await
1102            .expect_err("expected failure due to bad alignment");
1103        block_client.detach_vmo(vmo_id).await.expect("detach_vmo failed");
1104    }
1105
1106    #[fuchsia::test]
1107    async fn test_parallel_io() {
1108        let (_ramdisk, _block_proxy, block_client) = make_ramdisk().await;
1109        let vmo = zx::Vmo::create(131072).expect("Vmo::create failed");
1110        // SAFETY: Test code, only attach once, no other mappings.
1111        let vmo_id = unsafe { block_client.attach_vmo(&vmo) }.await.expect("attach_vmo failed");
1112        let mut reads = Vec::new();
1113        for _ in 0..1024 {
1114            reads.push(
1115                block_client
1116                    .read_at(MutableBufferSlice::new_with_vmo_id(&vmo_id, 0, 1024), 0)
1117                    .inspect_err(|e| panic!("read should have succeeded: {}", e)),
1118            );
1119        }
1120        futures::future::join_all(reads).await;
1121        block_client.detach_vmo(vmo_id).await.expect("detach_vmo failed");
1122    }
1123
1124    #[fuchsia::test]
1125    async fn test_closed_device() {
1126        let (ramdisk, _block_proxy, block_client) = make_ramdisk().await;
1127        let vmo = zx::Vmo::create(131072).expect("Vmo::create failed");
1128        // SAFETY: Test code, only attach once, no other mappings.
1129        let vmo_id = unsafe { block_client.attach_vmo(&vmo) }.await.expect("attach_vmo failed");
1130        let mut reads = Vec::new();
1131        for _ in 0..1024 {
1132            reads.push(
1133                block_client.read_at(MutableBufferSlice::new_with_vmo_id(&vmo_id, 0, 1024), 0),
1134            );
1135        }
1136        assert!(block_client.is_connected());
1137        let _ = futures::join!(futures::future::join_all(reads), async {
1138            std::mem::drop(ramdisk);
1139        });
1140        // Destroying the ramdisk is asynchronous. Keep issuing reads until they start failing.
1141        while block_client
1142            .read_at(MutableBufferSlice::new_with_vmo_id(&vmo_id, 0, 1024), 0)
1143            .await
1144            .is_ok()
1145        {}
1146
1147        // Sometimes the FIFO will start rejecting requests before FIFO is actually closed, so we
1148        // get false-positives from is_connected.
1149        while block_client.is_connected() {
1150            // Sleep for a bit to minimise lock contention.
1151            fasync::Timer::new(fasync::MonotonicInstant::after(
1152                zx::MonotonicDuration::from_millis(500),
1153            ))
1154            .await;
1155        }
1156
1157        // But once is_connected goes negative, it should stay negative.
1158        assert_eq!(block_client.is_connected(), false);
1159        let _ = block_client.detach_vmo(vmo_id).await;
1160    }
1161
1162    #[fuchsia::test]
1163    async fn test_cancelled_reads() {
1164        let (_ramdisk, _block_proxy, block_client) = make_ramdisk().await;
1165        let vmo = zx::Vmo::create(131072).expect("Vmo::create failed");
1166        // SAFETY: Test code, only attach once, no other mappings.
1167        let vmo_id = unsafe { block_client.attach_vmo(&vmo) }.await.expect("attach_vmo failed");
1168        {
1169            let mut reads = FuturesUnordered::new();
1170            for _ in 0..1024 {
1171                reads.push(
1172                    block_client.read_at(MutableBufferSlice::new_with_vmo_id(&vmo_id, 0, 1024), 0),
1173                );
1174            }
1175            // Read the first 500 results and then dump the rest.
1176            for _ in 0..500 {
1177                reads.next().await;
1178            }
1179        }
1180
1181        // Since we dropped pending futures (cancellation), the client must forcibly
1182        // terminate the connection to prevent any future unsafe I/O. Thus, further
1183        // calls like `detach_vmo` must fail with CANCELED.
1184        assert_eq!(
1185            block_client.read_at(MutableBufferSlice::new_with_vmo_id(&vmo_id, 0, 1024), 0).await,
1186            Err(zx::Status::CANCELED)
1187        );
1188        assert_eq!(block_client.detach_vmo(vmo_id).await, Err(zx::Status::CANCELED));
1189    }
1190
1191    #[fuchsia::test]
1192    async fn test_parallel_large_read_and_write_with_memory_succeds() {
1193        let (_ramdisk, _block_proxy, block_client) = make_ramdisk().await;
1194        let block_client_ref = &block_client;
1195        let test_one = |offset, len, fill| async move {
1196            let buf = vec![fill; len];
1197            block_client_ref.write_at(buf[..].into(), offset).await.expect("write_at failed");
1198            // Read back an extra block either side.
1199            let mut read_buf = vec![0u8; len + 2 * RAMDISK_BLOCK_SIZE as usize];
1200            block_client_ref
1201                .read_at(read_buf.as_mut_slice().into(), offset - RAMDISK_BLOCK_SIZE)
1202                .await
1203                .expect("read_at failed");
1204            assert_eq!(
1205                &read_buf[0..RAMDISK_BLOCK_SIZE as usize],
1206                &[0; RAMDISK_BLOCK_SIZE as usize][..]
1207            );
1208            assert_eq!(
1209                &read_buf[RAMDISK_BLOCK_SIZE as usize..RAMDISK_BLOCK_SIZE as usize + len],
1210                &buf[..]
1211            );
1212            assert_eq!(
1213                &read_buf[RAMDISK_BLOCK_SIZE as usize + len..],
1214                &[0; RAMDISK_BLOCK_SIZE as usize][..]
1215            );
1216        };
1217        const WRITE_LEN: usize = super::TEMP_VMO_SIZE * 3 + RAMDISK_BLOCK_SIZE as usize;
1218        join!(
1219            test_one(RAMDISK_BLOCK_SIZE, WRITE_LEN, 0xa3u8),
1220            test_one(2 * RAMDISK_BLOCK_SIZE + WRITE_LEN as u64, WRITE_LEN, 0x7fu8)
1221        );
1222    }
1223
1224    // Implements dummy server which can be used by test cases to verify whether
1225    // channel messages and fifo operations are being received - by using set_channel_handler or
1226    // set_fifo_hander respectively
1227    struct FakeBlockServer<'a> {
1228        server_channel: Option<fidl::endpoints::ServerEnd<block::BlockMarker>>,
1229        channel_handler: Box<dyn Fn(&block::SessionRequest) -> bool + 'a>,
1230        fifo_handler: Box<dyn Fn(BlockFifoRequest) -> BlockFifoResponse + 'a>,
1231    }
1232
1233    impl<'a> FakeBlockServer<'a> {
1234        // Creates a new FakeBlockServer given a channel to listen on.
1235        //
1236        // 'channel_handler' and 'fifo_handler' closures allow for customizing the way how the server
1237        // handles requests received from channel or the fifo respectfully.
1238        //
1239        // 'channel_handler' receives a message before it is handled by the default implementation
1240        // and can return 'true' to indicate all processing is done and no further processing of
1241        // that message is required
1242        //
1243        // 'fifo_handler' takes as input a BlockFifoRequest and produces a response which the
1244        // FakeBlockServer will send over the fifo.
1245        fn new(
1246            server_channel: fidl::endpoints::ServerEnd<block::BlockMarker>,
1247            channel_handler: impl Fn(&block::SessionRequest) -> bool + 'a,
1248            fifo_handler: impl Fn(BlockFifoRequest) -> BlockFifoResponse + 'a,
1249        ) -> FakeBlockServer<'a> {
1250            FakeBlockServer {
1251                server_channel: Some(server_channel),
1252                channel_handler: Box::new(channel_handler),
1253                fifo_handler: Box::new(fifo_handler),
1254            }
1255        }
1256
1257        // Runs the server
1258        async fn run(&mut self) {
1259            let server = self.server_channel.take().unwrap();
1260
1261            // Set up a mock server.
1262            let (server_fifo, client_fifo) =
1263                zx::Fifo::<BlockFifoRequest, BlockFifoResponse>::create(16)
1264                    .expect("Fifo::create failed");
1265            let maybe_server_fifo = fuchsia_sync::Mutex::new(Some(client_fifo));
1266
1267            let (fifo_future_abort, fifo_future_abort_registration) = AbortHandle::new_pair();
1268            let fifo_future = Abortable::new(
1269                async {
1270                    let mut fifo = fasync::Fifo::from_fifo(server_fifo);
1271                    let (mut reader, mut writer) = fifo.async_io();
1272                    let mut request = BlockFifoRequest::default();
1273                    loop {
1274                        match reader.read_entries(&mut request).await {
1275                            Ok(n) if n.get() == 1 => {}
1276                            Err(zx::Status::PEER_CLOSED) => break,
1277                            Err(e) => panic!("read_entry failed {:?}", e),
1278                            _ => unreachable!(),
1279                        };
1280
1281                        let response = self.fifo_handler.as_ref()(request);
1282                        writer
1283                            .write_entries(std::slice::from_ref(&response))
1284                            .await
1285                            .expect("write_entries failed");
1286                    }
1287                },
1288                fifo_future_abort_registration,
1289            );
1290
1291            let channel_future = async {
1292                server
1293                    .into_stream()
1294                    .for_each_concurrent(None, |request| async {
1295                        let request = request.expect("unexpected fidl error");
1296
1297                        match request {
1298                            block::BlockRequest::GetInfo { responder } => {
1299                                responder
1300                                    .send(Ok(&block::BlockInfo {
1301                                        block_count: 1024,
1302                                        block_size: 512,
1303                                        max_transfer_size: 1024 * 1024,
1304                                        flags: block::DeviceFlag::empty(),
1305                                    }))
1306                                    .expect("send failed");
1307                            }
1308                            block::BlockRequest::OpenSession { session, control_handle: _ } => {
1309                                let stream = session.into_stream();
1310                                stream
1311                                    .for_each(|request| async {
1312                                        let request = request.expect("unexpected fidl error");
1313                                        // Give a chance for the test to register and potentially
1314                                        // handle the event
1315                                        if self.channel_handler.as_ref()(&request) {
1316                                            return;
1317                                        }
1318                                        match request {
1319                                            block::SessionRequest::GetFifo { responder } => {
1320                                                match maybe_server_fifo.lock().take() {
1321                                                    Some(fifo) => {
1322                                                        responder.send(Ok(fifo.downcast()))
1323                                                    }
1324                                                    None => responder.send(Err(
1325                                                        zx::Status::NO_RESOURCES.into_raw(),
1326                                                    )),
1327                                                }
1328                                                .expect("send failed")
1329                                            }
1330                                            block::SessionRequest::AttachVmo {
1331                                                vmo: _,
1332                                                responder,
1333                                            } => responder
1334                                                .send(Ok(&block::VmoId { id: 1 }))
1335                                                .expect("send failed"),
1336                                            block::SessionRequest::RegisterKey {
1337                                                responder,
1338                                                ..
1339                                            } => responder
1340                                                .send(Err(zx::Status::NOT_SUPPORTED.into_raw()))
1341                                                .expect("send failed"),
1342                                            block::SessionRequest::Close { responder } => {
1343                                                fifo_future_abort.abort();
1344                                                responder.send(Ok(())).expect("send failed")
1345                                            }
1346                                        }
1347                                    })
1348                                    .await
1349                            }
1350                            _ => panic!("Unexpected message"),
1351                        }
1352                    })
1353                    .await;
1354            };
1355
1356            let _result = join!(fifo_future, channel_future);
1357            //_result can be Err(Aborted) since FifoClose calls .abort but that's expected
1358        }
1359    }
1360
1361    #[fuchsia::test]
1362    async fn test_block_close_is_called() {
1363        let close_called = fuchsia_sync::Mutex::new(false);
1364        let (client_end, server) = fidl::endpoints::create_endpoints::<block::BlockMarker>();
1365
1366        std::thread::spawn(move || {
1367            let _block_client =
1368                RemoteBlockClientSync::new(client_end).expect("RemoteBlockClientSync::new failed");
1369            // The drop here should cause Close to be sent.
1370        });
1371
1372        let channel_handler = |request: &block::SessionRequest| -> bool {
1373            if let block::SessionRequest::Close { .. } = request {
1374                *close_called.lock() = true;
1375            }
1376            false
1377        };
1378        FakeBlockServer::new(server, channel_handler, |_| unreachable!()).run().await;
1379
1380        // After the server has finished running, we can check to see that close was called.
1381        assert!(*close_called.lock());
1382    }
1383
1384    #[fuchsia::test]
1385    async fn test_block_flush_is_called() {
1386        let (proxy, stream) = fidl::endpoints::create_proxy_and_stream::<block::BlockMarker>();
1387
1388        struct Interface {
1389            flush_called: Arc<AtomicBool>,
1390        }
1391        impl block_server::async_interface::Interface for Interface {
1392            fn get_info(&self) -> Cow<'_, DeviceInfo> {
1393                Cow::Owned(DeviceInfo::Partition(PartitionInfo {
1394                    device_flags: fidl_fuchsia_storage_block::DeviceFlag::empty(),
1395                    max_transfer_blocks: None,
1396                    start_block_offset: None,
1397                    block_count: 1000,
1398                    type_guid: [0; 16],
1399                    instance_guid: [0; 16],
1400                    name: "foo".to_string(),
1401                    ..Default::default()
1402                }))
1403            }
1404
1405            async fn read(
1406                &self,
1407                _device_block_offset: u64,
1408                _block_count: u32,
1409                _vmo: &Arc<zx::Vmo>,
1410                _vmo_offset: u64,
1411                _opts: ReadOptions,
1412                _trace_flow_id: Option<NonZero<u64>>,
1413            ) -> Result<(), zx::Status> {
1414                unreachable!();
1415            }
1416
1417            async fn write(
1418                &self,
1419                _device_block_offset: u64,
1420                _block_count: u32,
1421                _vmo: &Arc<zx::Vmo>,
1422                _vmo_offset: u64,
1423                _opts: WriteOptions,
1424                _trace_flow_id: Option<NonZero<u64>>,
1425            ) -> Result<(), zx::Status> {
1426                unreachable!();
1427            }
1428
1429            async fn flush(&self, _trace_flow_id: Option<NonZero<u64>>) -> Result<(), zx::Status> {
1430                self.flush_called.store(true, Ordering::Relaxed);
1431                Ok(())
1432            }
1433
1434            async fn trim(
1435                &self,
1436                _device_block_offset: u64,
1437                _block_count: u32,
1438                _trace_flow_id: Option<NonZero<u64>>,
1439            ) -> Result<(), zx::Status> {
1440                unreachable!();
1441            }
1442        }
1443
1444        let flush_called = Arc::new(AtomicBool::new(false));
1445
1446        futures::join!(
1447            async {
1448                let block_client = RemoteBlockClient::new(proxy).await.expect("new failed");
1449
1450                block_client.flush().await.expect("flush failed");
1451            },
1452            async {
1453                let block_server = BlockServer::new(
1454                    512,
1455                    Arc::new(Interface { flush_called: flush_called.clone() }),
1456                );
1457                block_server.handle_requests(stream.cast_stream()).await.unwrap();
1458            }
1459        );
1460
1461        assert!(flush_called.load(Ordering::Relaxed));
1462    }
1463
1464    #[fuchsia::test]
1465    async fn test_trace_flow_ids_set() {
1466        let (proxy, server) = fidl::endpoints::create_proxy();
1467
1468        futures::join!(
1469            async {
1470                let block_client = RemoteBlockClient::new(proxy).await.expect("new failed");
1471                block_client.flush().await.expect("flush failed");
1472            },
1473            async {
1474                let flow_id: fuchsia_sync::Mutex<Option<u64>> = fuchsia_sync::Mutex::new(None);
1475                let fifo_handler = |request: BlockFifoRequest| -> BlockFifoResponse {
1476                    if request.trace_flow_id > 0 {
1477                        *flow_id.lock() = Some(request.trace_flow_id);
1478                    }
1479                    BlockFifoResponse {
1480                        status: zx::sys::ZX_OK,
1481                        reqid: request.reqid,
1482                        ..Default::default()
1483                    }
1484                };
1485                FakeBlockServer::new(server, |_| false, fifo_handler).run().await;
1486                // After the server has finished running, verify the trace flow ID was set to some value.
1487                assert!(flow_id.lock().is_some());
1488            }
1489        );
1490    }
1491}
block_client/lib.rs

block_client/
lib.rs
