fuchsia_async/handle/zircon/
fifo.rs

1// Copyright 2018 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 super::rwhandle::{RWHandle, ReadableHandle as _, WritableHandle as _};
6use futures::ready;
7use std::fmt;
8use std::future::poll_fn;
9use std::mem::MaybeUninit;
10use std::num::NonZeroUsize;
11use std::task::{Context, Poll};
12use zerocopy::{FromBytes, Immutable, IntoBytes};
13use zx::{self as zx, AsHandleRef};
14
15/// Marker trait for types that can be read/written with a `Fifo`.
16///
17/// An implementation is provided for all types that implement
18/// [`IntoBytes`], [`FromBytes`], and [`Immutable`].
19pub trait FifoEntry: IntoBytes + FromBytes + Immutable {}
20
21impl<O: IntoBytes + FromBytes + Immutable> FifoEntry for O {}
22
23/// A buffer used to write `T` into [`Fifo`] objects.
24pub trait FifoWriteBuffer<T> {
25    fn as_slice(&self) -> &[T];
26}
27
28/// A buffer used to read `T` from [`Fifo`] objects.
29///
30/// # Safety
31///
32/// This trait is unsafe because the compiler cannot verify a correct
33/// implementation of `as_bytes_ptr_mut`. See
34/// [`FifoReadBuffer::as_bytes_ptr_mut`] for safety notes.
35pub unsafe trait FifoReadBuffer<T> {
36    /// Returns the number of slots available in the buffer to be rceived.
37    fn count(&self) -> usize;
38    /// Returns a mutable pointer to the buffer contents where FIFO entries must
39    /// be written into.
40    ///
41    /// # Safety
42    ///
43    /// The returned memory *must* be at least `count() * sizeof<T>()` bytes
44    /// long.
45    fn as_mut_ptr(&mut self) -> *mut T;
46}
47
48impl<T: FifoEntry, const N: usize> FifoWriteBuffer<T> for [T; N] {
49    fn as_slice(&self) -> &[T] {
50        self
51    }
52}
53
54impl<T: FifoEntry> FifoWriteBuffer<T> for [T] {
55    fn as_slice(&self) -> &[T] {
56        self
57    }
58}
59
60unsafe impl<T: FifoEntry, const N: usize> FifoReadBuffer<T> for [T; N] {
61    fn count(&self) -> usize {
62        N
63    }
64
65    fn as_mut_ptr(&mut self) -> *mut T {
66        self.as_mut_slice().as_mut_ptr()
67    }
68}
69
70unsafe impl<T: FifoEntry> FifoReadBuffer<T> for [T] {
71    fn count(&self) -> usize {
72        self.len()
73    }
74
75    fn as_mut_ptr(&mut self) -> *mut T {
76        self.as_mut_ptr()
77    }
78}
79
80impl<T: FifoEntry> FifoWriteBuffer<T> for T {
81    fn as_slice(&self) -> &[T] {
82        std::slice::from_ref(self)
83    }
84}
85
86unsafe impl<T: FifoEntry> FifoReadBuffer<T> for T {
87    fn count(&self) -> usize {
88        1
89    }
90
91    fn as_mut_ptr(&mut self) -> *mut T {
92        self as *mut T
93    }
94}
95
96unsafe impl<T: FifoEntry> FifoReadBuffer<T> for MaybeUninit<T> {
97    fn count(&self) -> usize {
98        1
99    }
100
101    fn as_mut_ptr(&mut self) -> *mut T {
102        self.as_mut_ptr()
103    }
104}
105
106unsafe impl<T: FifoEntry> FifoReadBuffer<T> for [MaybeUninit<T>] {
107    fn count(&self) -> usize {
108        self.len()
109    }
110
111    fn as_mut_ptr(&mut self) -> *mut T {
112        // TODO(https://github.com/rust-lang/rust/issues/63569): Use
113        // `MaybeUninit::slice_as_mut_ptr` once stable.
114        self.as_mut_ptr() as *mut T
115    }
116}
117
118/// An I/O object representing a `Fifo`.
119pub struct Fifo<R, W = R> {
120    handle: RWHandle<zx::Fifo<R, W>>,
121}
122
123impl<R: FromBytes + IntoBytes, W: FromBytes + IntoBytes> AsRef<zx::Fifo<R, W>> for Fifo<R, W> {
124    fn as_ref(&self) -> &zx::Fifo<R, W> {
125        self.handle.get_ref()
126    }
127}
128
129impl<R: FromBytes + IntoBytes, W: FromBytes + IntoBytes> AsHandleRef for Fifo<R, W> {
130    fn as_handle_ref(&self) -> zx::HandleRef<'_> {
131        self.handle.get_ref().as_handle_ref()
132    }
133}
134
135impl<R: FromBytes + IntoBytes, W: FromBytes + IntoBytes> From<Fifo<R, W>> for zx::Fifo<R, W> {
136    fn from(fifo: Fifo<R, W>) -> zx::Fifo<R, W> {
137        fifo.handle.into_inner()
138    }
139}
140
141impl<R: FromBytes + IntoBytes, W: FromBytes + IntoBytes> Fifo<R, W> {
142    /// Creates a new `Fifo` from a previously-created `zx::Fifo`.
143    ///
144    /// # Panics
145    ///
146    /// If called on a thread that does not have a current async executor.
147    pub fn from_fifo(fifo: impl Into<zx::Fifo<R, W>>) -> Self {
148        Fifo { handle: RWHandle::new(fifo.into()) }
149    }
150
151    /// Writes entries to the fifo and registers this `Fifo` as needing a write on receiving a
152    /// `zx::Status::SHOULD_WAIT`.
153    ///
154    /// Returns the number of elements processed.
155    ///
156    /// NOTE: Only one writer is supported; this will overwrite any waker registered with a previous
157    /// invocation to `try_write`.
158    pub fn try_write<B: ?Sized + FifoWriteBuffer<W>>(
159        &self,
160        cx: &mut Context<'_>,
161        entries: &B,
162    ) -> Poll<Result<NonZeroUsize, zx::Status>> {
163        ready!(self.handle.poll_writable(cx)?);
164
165        let entries = entries.as_slice();
166        let fifo = self.as_ref();
167        // SAFETY: Safety relies on us keeping the slice alive over the call to `write_raw`, which
168        // we do.
169        loop {
170            let result = unsafe { fifo.write_raw(entries.as_ptr(), entries.len()) };
171            match result {
172                Err(zx::Status::SHOULD_WAIT) => ready!(self.handle.need_writable(cx)?),
173                Err(e) => return Poll::Ready(Err(e)),
174                Ok(count) => return Poll::Ready(Ok(count)),
175            }
176        }
177    }
178
179    /// Reads entries from the fifo into `entries` and registers this `Fifo` as needing a read on
180    /// receiving a `zx::Status::SHOULD_WAIT`.
181    ///
182    /// NOTE: Only one reader is supported; this will overwrite any waker registered with a previous
183    /// invocation to `try_read`.
184    pub fn try_read<B: ?Sized + FifoReadBuffer<R>>(
185        &self,
186        cx: &mut Context<'_>,
187        entries: &mut B,
188    ) -> Poll<Result<NonZeroUsize, zx::Status>> {
189        ready!(self.handle.poll_readable(cx)?);
190
191        let buf = entries.as_mut_ptr();
192        let count = entries.count();
193        let fifo = self.as_ref();
194
195        loop {
196            // SAFETY: Safety relies on the pointer returned by `B` being valid,
197            // which itself depends on a correct implementation of `FifoEntry` for
198            // `R`.
199            let result = unsafe { fifo.read_raw(buf, count) };
200
201            match result {
202                Err(zx::Status::SHOULD_WAIT) => ready!(self.handle.need_readable(cx)?),
203                Err(e) => return Poll::Ready(Err(e)),
204                Ok(count) => return Poll::Ready(Ok(count)),
205            }
206        }
207    }
208
209    /// Returns a reader and writer which have async functions that can be used to read and write
210    /// requests.
211    pub fn async_io(&mut self) -> (FifoReader<'_, R, W>, FifoWriter<'_, R, W>) {
212        (FifoReader(self), FifoWriter(self))
213    }
214}
215
216pub struct FifoWriter<'a, R, W>(&'a Fifo<R, W>);
217
218impl<R: FifoEntry, W: FifoEntry> FifoWriter<'_, R, W> {
219    /// NOTE: If this future is dropped or there is an error, there is no indication how many
220    /// entries were successfully written.
221    pub async fn write_entries(
222        &mut self,
223        entries: &(impl ?Sized + FifoWriteBuffer<W>),
224    ) -> Result<(), zx::Status> {
225        let mut entries = entries.as_slice();
226        poll_fn(|cx| {
227            while !entries.is_empty() {
228                match ready!(self.0.try_write(cx, entries)) {
229                    Ok(count) => entries = &entries[count.get()..],
230                    Err(status) => return Poll::Ready(Err(status)),
231                }
232            }
233            Poll::Ready(Ok(()))
234        })
235        .await
236    }
237
238    /// Same as Fifo::try_write.
239    pub fn try_write<B: ?Sized + FifoWriteBuffer<W>>(
240        &mut self,
241        cx: &mut Context<'_>,
242        entries: &B,
243    ) -> Poll<Result<NonZeroUsize, zx::Status>> {
244        self.0.try_write(cx, entries)
245    }
246}
247
248pub struct FifoReader<'a, R, W>(&'a Fifo<R, W>);
249
250impl<R: FifoEntry, W: FifoEntry> FifoReader<'_, R, W> {
251    pub async fn read_entries(
252        &mut self,
253        entries: &mut (impl ?Sized + FifoReadBuffer<R>),
254    ) -> Result<NonZeroUsize, zx::Status> {
255        poll_fn(|cx| self.0.try_read(cx, entries)).await
256    }
257
258    /// Same as Fifo::try_read.
259    pub fn try_read<B: ?Sized + FifoReadBuffer<R>>(
260        &mut self,
261        cx: &mut Context<'_>,
262        entries: &mut B,
263    ) -> Poll<Result<NonZeroUsize, zx::Status>> {
264        self.0.try_read(cx, entries)
265    }
266}
267
268impl<R: FromBytes + IntoBytes, W: FromBytes + IntoBytes> fmt::Debug for Fifo<R, W> {
269    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
270        self.handle.get_ref().fmt(f)
271    }
272}
273
274#[cfg(test)]
275mod tests {
276    use super::*;
277    use crate::{DurationExt, TestExecutor, Timer};
278    use futures::future::try_join;
279    use futures::prelude::*;
280    use zerocopy::{Immutable, KnownLayout};
281
282    #[derive(
283        Copy, Clone, Debug, PartialEq, Eq, Default, IntoBytes, KnownLayout, FromBytes, Immutable,
284    )]
285    #[repr(C)]
286    struct Entry {
287        a: u32,
288        b: u32,
289    }
290
291    #[derive(
292        Clone, Debug, PartialEq, Eq, Default, IntoBytes, KnownLayout, FromBytes, Immutable,
293    )]
294    #[repr(C)]
295    struct WrongEntry {
296        a: u16,
297    }
298
299    #[test]
300    fn can_read_write() {
301        let mut exec = TestExecutor::new();
302        let element = Entry { a: 10, b: 20 };
303
304        let (tx, rx) = zx::Fifo::<Entry>::create(2).expect("failed to create zx fifo");
305        let (mut tx, mut rx) = (Fifo::from_fifo(tx), Fifo::from_fifo(rx));
306        let (_, mut tx) = tx.async_io();
307        let (mut rx, _) = rx.async_io();
308
309        let mut buffer = Entry::default();
310        let receiver = rx.read_entries(&mut buffer).map_ok(|count| {
311            assert_eq!(count.get(), 1);
312        });
313
314        // Sends an entry after the timeout has passed
315        let sender = Timer::new(zx::MonotonicDuration::from_millis(10).after_now())
316            .then(|()| tx.write_entries(&element));
317
318        let done = try_join(receiver, sender);
319        exec.run_singlethreaded(done).expect("failed to run receive future on executor");
320        assert_eq!(buffer, element);
321    }
322
323    #[test]
324    fn read_wrong_size() {
325        let mut exec = TestExecutor::new();
326        let elements = &[Entry { a: 10, b: 20 }][..];
327
328        let (tx, rx) = zx::Fifo::<Entry>::create(2).expect("failed to create zx fifo");
329        let wrong_rx = zx::Fifo::<WrongEntry>::from(rx.into_handle());
330        let (mut tx, mut rx) = (Fifo::from_fifo(tx), Fifo::from_fifo(wrong_rx));
331        let (_, mut tx) = tx.async_io();
332        let (mut rx, _) = rx.async_io();
333
334        let mut buffer = WrongEntry::default();
335        let receiver = rx
336            .read_entries(&mut buffer)
337            .map_ok(|count| panic!("read should have failed, got {count}"));
338
339        // Sends an entry after the timeout has passed
340        let sender = Timer::new(zx::MonotonicDuration::from_millis(10).after_now())
341            .then(|()| tx.write_entries(elements));
342
343        let done = try_join(receiver, sender);
344        let res = exec.run_singlethreaded(done);
345        match res {
346            Err(zx::Status::OUT_OF_RANGE) => (),
347            _ => panic!("did not get out-of-range error"),
348        }
349    }
350
351    #[test]
352    fn write_wrong_size() {
353        let mut exec = TestExecutor::new();
354        let elements = &[WrongEntry { a: 10 }][..];
355
356        let (tx, rx) = zx::Fifo::<Entry>::create(2).expect("failed to create zx fifo");
357        let wrong_tx = zx::Fifo::<WrongEntry>::from(tx.into_handle());
358        let wrong_rx = zx::Fifo::<WrongEntry>::from(rx.into_handle());
359        let (mut tx, _rx) = (Fifo::from_fifo(wrong_tx), Fifo::from_fifo(wrong_rx));
360        let (_, mut tx) = tx.async_io();
361
362        let sender = Timer::new(zx::MonotonicDuration::from_millis(10).after_now())
363            .then(|()| tx.write_entries(elements));
364
365        let res = exec.run_singlethreaded(sender);
366        match res {
367            Err(zx::Status::OUT_OF_RANGE) => (),
368            _ => panic!("did not get out-of-range error"),
369        }
370    }
371
372    #[test]
373    fn write_into_full() {
374        use std::sync::atomic::{AtomicUsize, Ordering};
375
376        let mut exec = TestExecutor::new();
377        let elements =
378            &[Entry { a: 10, b: 20 }, Entry { a: 30, b: 40 }, Entry { a: 50, b: 60 }][..];
379
380        let (tx, rx) = zx::Fifo::<Entry>::create(2).expect("failed to create zx fifo");
381        let (mut tx, mut rx) = (Fifo::from_fifo(tx), Fifo::from_fifo(rx));
382
383        // Use `writes_completed` to verify that not all writes
384        // are transmitted at once, and the last write is actually blocked.
385        let writes_completed = AtomicUsize::new(0);
386        let sender = async {
387            let (_, mut writer) = tx.async_io();
388            writer.write_entries(&elements[..2]).await?;
389            writes_completed.fetch_add(1, Ordering::SeqCst);
390            writer.write_entries(&elements[2..]).await?;
391            writes_completed.fetch_add(1, Ordering::SeqCst);
392            Ok::<(), zx::Status>(())
393        };
394
395        // Wait 10 ms, then read the messages from the fifo.
396        let receiver = async {
397            Timer::new(zx::MonotonicDuration::from_millis(10).after_now()).await;
398            let mut buffer = Entry::default();
399            let (mut reader, _) = rx.async_io();
400            let count = reader.read_entries(&mut buffer).await?;
401            assert_eq!(writes_completed.load(Ordering::SeqCst), 1);
402            assert_eq!(count.get(), 1);
403            assert_eq!(buffer, elements[0]);
404            let count = reader.read_entries(&mut buffer).await?;
405            // At this point, the last write may or may not have
406            // been written.
407            assert_eq!(count.get(), 1);
408            assert_eq!(buffer, elements[1]);
409            let count = reader.read_entries(&mut buffer).await?;
410            assert_eq!(writes_completed.load(Ordering::SeqCst), 2);
411            assert_eq!(count.get(), 1);
412            assert_eq!(buffer, elements[2]);
413            Ok::<(), zx::Status>(())
414        };
415
416        let done = try_join(receiver, sender);
417
418        exec.run_singlethreaded(done).expect("failed to run receive future on executor");
419    }
420
421    #[test]
422    fn write_more_than_full() {
423        let mut exec = TestExecutor::new();
424        let elements =
425            &[Entry { a: 10, b: 20 }, Entry { a: 30, b: 40 }, Entry { a: 50, b: 60 }][..];
426
427        let (tx, rx) = zx::Fifo::<Entry>::create(2).expect("failed to create zx fifo");
428        let (mut tx, mut rx) = (Fifo::from_fifo(tx), Fifo::from_fifo(rx));
429        let (_, mut tx) = tx.async_io();
430        let (mut rx, _) = rx.async_io();
431
432        let sender = tx.write_entries(elements);
433
434        // Wait 10 ms, then read the messages from the fifo.
435        let receiver = async {
436            Timer::new(zx::MonotonicDuration::from_millis(10).after_now()).await;
437            for e in elements {
438                let mut buffer = [Entry::default(); 1];
439                let count = rx.read_entries(&mut buffer[..]).await?;
440                assert_eq!(count.get(), 1);
441                assert_eq!(&buffer[0], e);
442            }
443            Ok::<(), zx::Status>(())
444        };
445
446        let done = try_join(receiver, sender);
447
448        exec.run_singlethreaded(done).expect("failed to run receive future on executor");
449    }
450
451    #[test]
452    fn read_multiple() {
453        let mut exec = TestExecutor::new();
454        let elements =
455            &[Entry { a: 10, b: 20 }, Entry { a: 30, b: 40 }, Entry { a: 50, b: 60 }][..];
456        let (tx, rx) = zx::Fifo::<Entry>::create(elements.len()).expect("failed to create zx fifo");
457        let (mut tx, mut rx) = (Fifo::from_fifo(tx), Fifo::from_fifo(rx));
458
459        let write_fut = async {
460            tx.async_io().1.write_entries(elements).await.expect("failed write entries");
461        };
462        let read_fut = async {
463            // Use a larger buffer to show partial reads.
464            let mut buffer = [Entry::default(); 5];
465            let count = rx
466                .async_io()
467                .0
468                .read_entries(&mut buffer[..])
469                .await
470                .expect("failed to read entries");
471            assert_eq!(count.get(), elements.len());
472            assert_eq!(&buffer[..count.get()], elements);
473        };
474        let ((), ()) = exec.run_singlethreaded(futures::future::join(write_fut, read_fut));
475    }
476
477    #[test]
478    fn read_one() {
479        let mut exec = TestExecutor::new();
480        let elements =
481            &[Entry { a: 10, b: 20 }, Entry { a: 30, b: 40 }, Entry { a: 50, b: 60 }][..];
482        let (tx, rx) = zx::Fifo::<Entry>::create(elements.len()).expect("failed to create zx fifo");
483        let (mut tx, mut rx) = (Fifo::from_fifo(tx), Fifo::from_fifo(rx));
484
485        let write_fut = async {
486            tx.async_io().1.write_entries(elements).await.expect("failed write entries");
487        };
488        let read_fut = async {
489            let (mut reader, _) = rx.async_io();
490            for e in elements {
491                let mut entry = Entry::default();
492                assert_eq!(
493                    reader.read_entries(&mut entry).await.expect("failed to read entry").get(),
494                    1
495                );
496                assert_eq!(&entry, e);
497            }
498        };
499        let ((), ()) = exec.run_singlethreaded(futures::future::join(write_fut, read_fut));
500    }
501
502    #[test]
503    fn maybe_uninit_single() {
504        let mut exec = TestExecutor::new();
505        let element = Entry { a: 10, b: 20 };
506        let (tx, rx) = zx::Fifo::<Entry>::create(1).expect("failed to create zx fifo");
507        let (mut tx, mut rx) = (Fifo::from_fifo(tx), Fifo::from_fifo(rx));
508
509        let write_fut = async {
510            tx.async_io().1.write_entries(&element).await.expect("failed write entries");
511        };
512        let read_fut = async {
513            let mut buffer = MaybeUninit::<Entry>::uninit();
514            let count =
515                rx.async_io().0.read_entries(&mut buffer).await.expect("failed to read entries");
516            assert_eq!(count.get(), 1);
517            // SAFETY: We just read a new entry into the buffer.
518            let read = unsafe { buffer.assume_init() };
519            assert_eq!(read, element);
520        };
521        let ((), ()) = exec.run_singlethreaded(futures::future::join(write_fut, read_fut));
522    }
523
524    #[test]
525    fn maybe_uninit_slice() {
526        let mut exec = TestExecutor::new();
527        let elements =
528            &[Entry { a: 10, b: 20 }, Entry { a: 30, b: 40 }, Entry { a: 50, b: 60 }][..];
529        let (tx, rx) = zx::Fifo::<Entry>::create(elements.len()).expect("failed to create zx fifo");
530        let (mut tx, mut rx) = (Fifo::from_fifo(tx), Fifo::from_fifo(rx));
531
532        let write_fut = async {
533            tx.async_io().1.write_entries(elements).await.expect("failed write entries");
534        };
535        let read_fut = async {
536            // Use a larger buffer to show partial reads.
537            let mut buffer = [MaybeUninit::<Entry>::uninit(); 15];
538            let count = rx
539                .async_io()
540                .0
541                .read_entries(&mut buffer[..])
542                .await
543                .expect("failed to read entries");
544            assert_eq!(count.get(), elements.len());
545            let read = &mut buffer[..count.get()];
546            for (i, v) in read.iter_mut().enumerate() {
547                // SAFETY: This is the read region of the buffer, initialized by
548                // reading from the FIFO.
549                let read = unsafe { v.assume_init_ref() };
550                assert_eq!(read, &elements[i]);
551                // SAFETY: The buffer was partially initialized by reading from
552                // the FIFO, the correct thing to do here is to manually drop
553                // the elements that were initialized.
554                unsafe {
555                    v.assume_init_drop();
556                }
557            }
558        };
559        let ((), ()) = exec.run_singlethreaded(futures::future::join(write_fut, read_fut));
560    }
561}
fuchsia_async/handle/zircon/fifo.rs

fuchsia_async/handle/zircon/
fifo.rs
